/**
 * Durable HITL approval handling for the agent runtime: the
 * pre-execution `needsApproval` probe (with its rejecting secrets
 * accessor), gated-args validation at the pre-screen, resume-directive
 * processing (grant / deny decisions against persisted pending
 * approvals), and the exactly-once dispatch of resumed approved calls
 * with write-ahead intent checkpoints. Extracted verbatim from
 * `factory.ts` (issue #23).
 *
 * @packageDocumentation
 */

import type {
  AgentEvent,
  AgentResult,
  CompletedToolCall,
  Message,
  RunContext,
  RunState,
  Tool,
  ToolApproval,
  ToolCall,
  ToolExecutionContext,
  UsageAccumulator,
} from '@graphorin/core';
import { NOOP_LOGGER, zeroUsage } from '@graphorin/core';
import { emitToolAudit } from '@graphorin/tools/audit';
import type { ToolExecutor } from '@graphorin/tools/executor';
import type { ToolRegistry } from '@graphorin/tools/registry';
import { SubAgentResumeTargetNotFoundError } from '../errors/index.js';
import { serializeRunState } from '../run-state/index.js';
import type { AgentCallOptions, AgentConfig, ResumeDirective } from '../types.js';
import { getSubAgentToolRefs } from './agent-to-tool.js';
import type { DispatchBatchFn } from './dispatch.js';
import { composeSubRunPath, type HandoffEntry, splitSubRunPath } from './handoff.js';
import { foldChildRunUsage, renderToolErrorMessage } from './messages.js';
import type { MutableRunState } from './run-input.js';

/**
 * Pre-execution approval screen (Adapter G / durable HITL). Evaluates a
 * (registry-resolved) tool's `needsApproval` against the realized args.
 * Returns `true` when the run must suspend before the tool executes.
 *
 * Actual execution flows through the `@graphorin/tools` executor, whose
 * `ApprovalGate` auto-grants because only no-approval / pre-approved
 * calls ever reach it; this probe is what keeps the suspend in the
 * agent so the durable-HITL contract (persist `RunState`, resume via
 * directive) is preserved.
 */
export async function invokeNeedsApproval(
  tool: Pick<Tool, 'needsApproval'> | undefined,
  args: unknown,
  baseCtx: RunContext,
  signal: AbortSignal,
): Promise<boolean> {
  const predicate = tool?.needsApproval;
  if (predicate === undefined || predicate === false) return false;
  if (predicate === true) return true;
  const probeCtx: ToolExecutionContext = {
    toolCallId: 'probe',
    runContext: baseCtx,
    signal,
    tracer: baseCtx.tracer,
    logger: NOOP_LOGGER,
    secrets: probeSecretsAccessor(),
    reportProgress: () => {},
    streamContent: () => {},
  };
  return Boolean(await predicate(args as never, probeCtx));
}

/**
 * Rejecting secrets accessor used only by the {@link invokeNeedsApproval}
 * probe. Real tool execution resolves secrets through the executor's
 * ACL-scoped accessor; an approval predicate has no legitimate need to
 * read secret material, so every `require(...)` rejects.
 */
function probeSecretsAccessor(): ToolExecutionContext['secrets'] {
  const rejector = (_key: string, _options?: { readonly optional?: boolean }): Promise<never> =>
    Promise.reject(new Error('secrets.require is unavailable inside a needsApproval predicate'));
  return { require: rejector } as unknown as ToolExecutionContext['secrets'];
}

/**
 * tools-02: validate an approval-gated call's args at the pre-screen so
 * the gate decision - and what a human is asked to approve - is the input
 * that will actually execute. Structural + defensive: `undefined` when
 * the tool exposes no callable `safeParse` (nothing to validate here; the
 * executor still validates at dispatch); a throwing schema counts as a
 * validation failure rather than crashing the loop.
 */
export function safeParseGatedArgs(
  tool: { readonly inputSchema?: unknown },
  args: unknown,
):
  | { readonly success: true; readonly data: unknown }
  | { readonly success: false; readonly message: string }
  | undefined {
  const schema = tool.inputSchema as { safeParse?: (value: unknown) => unknown } | null | undefined;
  const safeParse = schema?.safeParse;
  if (typeof safeParse !== 'function') return undefined;
  try {
    const parsed = safeParse.call(schema, args) as {
      readonly success?: boolean;
      readonly data?: unknown;
      readonly error?: { readonly message?: string };
    };
    if (parsed.success === true) return { success: true, data: parsed.data };
    return {
      success: false,
      message: parsed.error?.message ?? 'schema validation failed',
    };
  } catch (cause) {
    return {
      success: false,
      message: cause instanceof Error ? cause.message : String(cause),
    };
  }
}

/** The run-scoped context the resume-directive pass operates on. */
export interface ResumeRunEnv {
  readonly state: MutableRunState & RunState;
  readonly messages: Message[];
}

/**
 * Process resume directive - apply approval decisions to any
 * pending approvals captured in the previous suspend.
 *
 * Fills `resumedApprovedCalls` / `grantedApprovals` in place (they are
 * declared by the run loop, which dispatches the granted subset next).
 */
/** One decision routed into a parked sub-run (W-001). */
export interface RoutedSubRunDecision {
  readonly toolCallId: string;
  readonly granted: boolean;
  readonly reason?: string;
  /** Remaining routing path for nested parks (one segment per level). */
  readonly subRunToolCallId?: string;
}

/**
 * Composite decision key: child-local toolCallIds of two
 * different parked children may collide, so a decision matches a
 * pending approval only when BOTH `toolCallId` and `subRunToolCallId`
 * agree (`undefined` on both sides = the parent's own approvals).
 */
function decisionKey(toolCallId: string, subRunToolCallId: string | undefined): string {
  return `${toolCallId} ${subRunToolCallId ?? ''}`;
}

export async function* processResumeDirective<TOutput>(
  env: ResumeRunEnv,
  approvals: NonNullable<ResumeDirective['approvals']>,
  resumedApprovedCalls: ToolCall[],
  grantedApprovals: ToolApproval[],
  subRunDecisionsOut?: Map<string, Array<RoutedSubRunDecision>>,
): AsyncGenerator<AgentEvent<TOutput>, void, void> {
  const { state, messages } = env;
  // TOOL-AUDI-01: the durable-HITL grant/deny decisions resolve in the agent
  // (the executor's approval phase is skipped on a pre-approved replay), so
  // emit the audited outcome here. runId + resume step number attribute it.
  const resumeStep = state.steps.reduce((m, s) => Math.max(m, s.stepNumber), 0) + 1;
  const emitApprovalAudit = (
    approval: ToolApproval,
    granted: boolean,
    reason: string | undefined,
  ): void => {
    emitToolAudit({
      action: granted ? 'tool:approval:granted' : 'tool:approval:denied',
      actor: { kind: 'tool', id: approval.toolName },
      target: approval.toolName,
      decision: granted ? 'success' : 'denied',
      ts: Date.now(),
      context: { runId: state.id, stepNumber: resumeStep, toolCallId: approval.toolCallId },
      ...(reason !== undefined ? { metadata: { reason } } : {}),
    });
  };
  // Step-journal: tool calls already completed on a prior resume are recorded
  // in the journal (`state.steps`); a re-resume must not run their side effects
  // again. Collect their ids so an approved call already journaled is replayed,
  // not re-executed (exactly-once; AG-1).
  const journaledCallIds = new Set<string>();
  for (const step of state.steps) {
    for (const completed of step.toolCalls) journaledCallIds.add(completed.call.toolCallId);
  }

  const decisions = new Map(
    approvals.map((d) => [decisionKey(d.toolCallId, d.subRunToolCallId), d]),
  );
  const stillPending: ToolApproval[] = [];
  for (const approval of state.pendingApprovals) {
    const decision = decisions.get(decisionKey(approval.toolCallId, approval.subRunToolCallId));
    if (decision === undefined) {
      stillPending.push(approval);
      continue;
    }
    // W-001: a decision addressed to a PARKED sub-run resolves here (the
    // approval leaves the parent queue and the grant/deny event streams)
    // but EXECUTES inside the child: one path segment is stripped and the
    // decision routes to the parked run's own resume (deeper segments
    // route recursively at each level). No parent tool message is written
    // for a denied child id - the parent transcript has no matching
    // tool_use.
    if (approval.subRunToolCallId !== undefined) {
      emitApprovalAudit(approval, decision.granted, decision.reason);
      yield decision.granted
        ? { type: 'tool.approval.granted', toolCallId: approval.toolCallId }
        : {
            type: 'tool.approval.denied',
            toolCallId: approval.toolCallId,
            ...(decision.reason !== undefined ? { reason: decision.reason } : {}),
          };
      if (subRunDecisionsOut !== undefined) {
        const { head, rest } = splitSubRunPath(approval.subRunToolCallId);
        const bucket = subRunDecisionsOut.get(head) ?? [];
        bucket.push({
          toolCallId: approval.toolCallId,
          granted: decision.granted,
          ...(decision.reason !== undefined ? { reason: decision.reason } : {}),
          ...(rest !== undefined ? { subRunToolCallId: rest } : {}),
        });
        subRunDecisionsOut.set(head, bucket);
      }
      continue;
    }
    if (decision.granted) {
      emitApprovalAudit(approval, true, decision.reason);
      yield {
        type: 'tool.approval.granted',
        toolCallId: approval.toolCallId,
      };
      // Step-journal: if this approved call already ran on a prior resume -
      // journaled in `state.steps` with its result still in the message
      // buffer - replay it instead of running the side effect again
      // (exactly-once across re-resumes). If the journal entry exists but its
      // result message was lost, fall through to a single re-execution (the
      // documented "at most one re-execution" bound).
      if (
        journaledCallIds.has(approval.toolCallId) &&
        messages.some((m) => m.role === 'tool' && m.toolCallId === approval.toolCallId)
      ) {
        continue;
      }
      // AG-1: queue the approved call for REAL execution (dispatched
      // below). It runs through the same ToolExecutor as any other tool
      // call - taint / audit / result recording - instead of pushing a
      // "[not actually executed]" placeholder that left the gated side
      // effect unreachable.
      resumedApprovedCalls.push({
        toolCallId: approval.toolCallId,
        toolName: approval.toolName,
        args: approval.args,
      });
      grantedApprovals.push(approval);
    } else {
      emitApprovalAudit(approval, false, decision.reason);
      yield {
        type: 'tool.approval.denied',
        toolCallId: approval.toolCallId,
        ...(decision.reason !== undefined ? { reason: decision.reason } : {}),
      };
      messages.push({
        role: 'tool',
        toolCallId: approval.toolCallId,
        content: `Error: tool approval denied${decision.reason ? `: ${decision.reason}` : ''}`,
      });
      state.messages.push({
        role: 'tool',
        toolCallId: approval.toolCallId,
        content: `Error: tool approval denied${decision.reason ? `: ${decision.reason}` : ''}`,
      });
    }
  }
  // Clear the queue + restore the running status so the loop
  // resumes from where it paused.
  state.pendingApprovals.splice(0, state.pendingApprovals.length, ...stillPending);
  if (stillPending.length === 0) {
    state.status = 'running';
  }
}

/** What the resumed-approval dispatch needs from the run loop's scope. */
export interface ResumedDispatchEnv<TDeps, TOutput> {
  readonly config: Pick<AgentConfig<TDeps, TOutput>, 'checkpointStore'>;
  readonly state: MutableRunState & RunState;
  readonly messages: Message[];
  readonly runContextBase: RunContext<TDeps>;
  readonly toolExecutor: ToolExecutor;
  readonly dispatchBatch: DispatchBatchFn<TDeps, TOutput>;
}

/**
 * AG-1: execute the approved gated calls for REAL before the provider
 * loop - the model sees their genuine results on the first step. They
 * run through the shared ToolExecutor (taint / audit) and record
 * CompletedToolCalls in a resume step. Dispatching here (outside the
 * loop's approval pre-screen) also means the gated call never
 * re-suspends, so there is no livelock.
 *
 * agent-07: this dispatch runs even when OTHER approvals remain
 * pending. A granted call has already been removed from
 * `pendingApprovals`, so skipping it (as the old order did - the
 * suspended-guard returned before the dispatch) stranded it
 * unrunnable forever; the run re-suspends with the remainder below.
 */
export async function* dispatchResumedApprovals<TDeps, TOutput>(
  env: ResumedDispatchEnv<TDeps, TOutput>,
  resumedApprovedCalls: ToolCall[],
  grantedApprovals: ToolApproval[],
): AsyncGenerator<AgentEvent<TOutput>, void, void> {
  const { config, state, messages, runContextBase, toolExecutor, dispatchBatch } = env;
  // W-035: the resume step continues the journal's numbering instead of
  // aliasing step 0 - `RunStep.stepNumber` stays unique across any
  // number of suspend/resume cycles (replay itself keys on toolCallId
  // and never depends on this number).
  const resumeStepNumber = state.steps.reduce((max, s) => Math.max(max, s.stepNumber), 0) + 1;
  // agent-02 write-ahead intent: persist a checkpoint equivalent to
  // the pre-dispatch suspended state (granted approvals re-attached
  // to `pendingApprovals`) BEFORE any side effect runs. A crash-retry
  // against this checkpoint re-resumes with the same directive and
  // re-dispatches - the documented at-most-one-re-execution bound -
  // and its nodeName records that a grant arrived and dispatch was in
  // flight.
  if (config.checkpointStore !== undefined) {
    const prevStatus = state.status;
    state.status = 'awaiting_approval';
    state.pendingApprovals.unshift(...grantedApprovals);
    const intentState = serializeRunState(state, { stripTracingApiKey: true });
    state.pendingApprovals.splice(0, grantedApprovals.length);
    state.status = prevStatus;
    await config.checkpointStore.put(
      state.id,
      'agent',
      {
        id: state.id,
        threadId: state.id,
        namespace: 'agent',
        state: intentState,
        channelVersions: {},
        stepNumber: resumeStepNumber,
        createdAt: new Date().toISOString(),
      },
      {
        source: 'sync',
        status: 'suspended',
        nodeName: 'agent.resume.intent',
        sessionId: state.sessionId,
      },
    );
  }
  state.steps.push({
    stepNumber: resumeStepNumber,
    startedAt: new Date().toISOString(),
    endedAt: new Date().toISOString(),
    usage: zeroUsage(),
    toolCalls: [],
    agentId: state.currentAgentId,
  });
  // tools-02: a human granted exactly `approval.args` - the repair
  // hook must not rewrite a pre-approved payload behind the grant, so
  // a (should-be-impossible) validation failure surfaces as
  // `invalid_input` instead of executing args nobody saw. E1:
  // `preApproved` tells the executor's permission/policy phases that
  // the grant already resolved any ask/defer verdict (a hook rewrite
  // is refused the same way the repair hook is).
  yield* dispatchBatch(
    resumedApprovedCalls,
    toolExecutor,
    { ...runContextBase, stepNumber: resumeStepNumber, messages },
    resumeStepNumber,
    { disableRepair: true, preApproved: true },
  );
  // agent-02: persist the journaled post-dispatch state. From THIS
  // checkpoint a re-delivered resume is exactly-once: the granted ids
  // are no longer pending and their journal entries + tool messages
  // are present, so nothing re-dispatches. (For the manual JSON flow,
  // the same state is returned as `result.state` - persist it after
  // every resume to get the same guarantee.)
  if (config.checkpointStore !== undefined) {
    await config.checkpointStore.put(
      state.id,
      'agent',
      {
        id: state.id,
        threadId: state.id,
        namespace: 'agent',
        state: serializeRunState(state, { stripTracingApiKey: true }),
        channelVersions: {},
        stepNumber: resumeStepNumber,
        createdAt: new Date().toISOString(),
      },
      {
        source: 'sync',
        status: state.status === 'awaiting_approval' ? 'suspended' : 'running',
        nodeName: 'agent.resume.dispatched',
        sessionId: state.sessionId,
      },
    );
  }
}

/** What the W-001 sub-run resume router needs from the run loop's scope. */
export interface SubRunResumeEnv<TDeps, TOutput> {
  readonly config: Pick<AgentConfig<TDeps, TOutput>, 'deps' | 'checkpointStore'>;
  readonly options: AgentCallOptions<TDeps>;
  readonly state: MutableRunState & RunState;
  readonly messages: Message[];
  readonly handoffMap: ReadonlyMap<string, HandoffEntry<TDeps>>;
  readonly toolRegistry: ToolRegistry;
  readonly usageAcc: UsageAccumulator;
  readonly signal: AbortSignal;
  readonly sessionId: string;
}

/**
 * W-001: resume the parked sub-agent runs a directive addressed. Each
 * routed decision batch replays into its child via `child.run(parked
 * state, { directive })` on the SAME parent instance's target (the
 * handoff map or the tool registry's `toTool` refs). Outcomes settle
 * exactly like the inline seam: a completed child's shaped output
 * becomes the parent's tool message for the parked toolCallId, a
 * re-suspended child refreshes its park and re-suspends the parent
 * with the remainder, a failed child surfaces the typed tool error.
 * Nesting composes recursively: a grandchild's park rides inside the
 * child's own parked state.
 */
export async function* processSubRunResumes<TDeps, TOutput>(
  env: SubRunResumeEnv<TDeps, TOutput>,
  subRunDecisions: ReadonlyMap<string, ReadonlyArray<RoutedSubRunDecision>>,
): AsyncGenerator<AgentEvent<TOutput>, void, void> {
  if (subRunDecisions.size === 0) return;
  const { config, options, state, messages, handoffMap, toolRegistry } = env;
  const { usageAcc, signal, sessionId } = env;
  const resumeStepNumber = state.steps.reduce((max, s) => Math.max(max, s.stepNumber), 0) + 1;
  const stepEntry = {
    stepNumber: resumeStepNumber,
    startedAt: new Date().toISOString(),
    endedAt: new Date().toISOString(),
    usage: zeroUsage(),
    toolCalls: [] as CompletedToolCall[],
    agentId: state.currentAgentId,
  };
  state.steps.push(stepEntry);

  for (const [parentToolCallId, decisions] of subRunDecisions) {
    const subs = state.pendingSubRuns ?? [];
    const parkedIdx = subs.findIndex((s) => s.toolCallId === parentToolCallId);
    const parked = subs[parkedIdx];
    if (parked === undefined) {
      throw new SubAgentResumeTargetNotFoundError(
        '(unparked)',
        `no parked sub-run matches subRunToolCallId '${parentToolCallId}'`,
      );
    }
    // Resolve the child seam on THIS instance: handoff targets first
    // (their tool names live only in the handoff map), then `toTool`
    // sub-agent tools via their SUBAGENT_TOOL refs.
    const handoffEntry = handoffMap.get(parked.toolName);
    const subRefs =
      handoffEntry === undefined
        ? getSubAgentToolRefs(toolRegistry.get(parked.toolName))
        : undefined;
    if (handoffEntry === undefined && subRefs === undefined) {
      throw new SubAgentResumeTargetNotFoundError(
        parked.toolName,
        'it matches neither a configured handoff target nor a toTool sub-agent tool on this instance',
      );
    }
    const childRun =
      handoffEntry !== undefined
        ? (input: unknown, opts?: Record<string, unknown>) =>
            handoffEntry.agent.run(input as RunState, opts as AgentCallOptions<TDeps>)
        : (subRefs as NonNullable<typeof subRefs>).run;
    const agentName = parked.targetAgentName;
    yield { type: 'tool.execute.start', toolCallId: parked.toolCallId };
    const childStart = Date.now();
    const childResult = (await childRun(parked.state, {
      signal,
      ...(options.deps !== undefined || config.deps !== undefined
        ? { deps: options.deps ?? config.deps }
        : {}),
      sessionId,
      directive: {
        approvals: decisions as unknown as Array<{ toolCallId: string; granted: boolean }>,
      },
    })) as AgentResult<unknown>;
    const childDurationMs = Date.now() - childStart;

    if (childResult.status === 'awaiting_approval') {
      // Nested (or partial) approval round: refresh the park and
      // re-suspend the parent with the remainder. Stale mirrors of this
      // park (unresolved approvals a partial grant left in the parent
      // queue) are dropped first - the child's CURRENT remainder is the
      // single source of truth.
      subs[parkedIdx] = { ...parked, state: childResult.state };
      const keep = state.pendingApprovals.filter(
        (a) =>
          a.subRunToolCallId === undefined ||
          splitSubRunPath(a.subRunToolCallId).head !== parked.toolCallId,
      );
      state.pendingApprovals.splice(0, state.pendingApprovals.length, ...keep);
      for (const approval of childResult.state.pendingApprovals) {
        state.pendingApprovals.push({
          ...approval,
          subRunToolCallId: composeSubRunPath(parked.toolCallId, approval.subRunToolCallId),
        });
        yield {
          type: 'tool.approval.requested',
          toolCallId: approval.toolCallId,
          ...(approval.reason !== undefined ? { reason: approval.reason } : {}),
        };
      }
      state.status = 'awaiting_approval';
      continue;
    }

    // Terminal outcome: fold the child's cumulative usage exactly once.
    foldChildRunUsage(state, usageAcc, childResult.state, agentName);
    subs.splice(parkedIdx, 1);

    if (childResult.status !== 'completed') {
      const toolError = {
        toolCallId: parked.toolCallId,
        toolName: parked.toolName,
        kind:
          childResult.status === 'aborted' ? ('aborted' as const) : ('execution_failed' as const),
        message: `sub-run '${agentName}' ${childResult.status}${
          childResult.error !== undefined ? `: ${childResult.error.message}` : ''
        }`,
      };
      stepEntry.toolCalls.push({
        call: callOf(parked),
        outcome: toolError,
        stepNumber: resumeStepNumber,
      });
      yield { type: 'tool.execute.error', toolCallId: parked.toolCallId, error: toolError };
      const text = renderToolErrorMessage(toolError);
      messages.push({ role: 'tool', toolCallId: parked.toolCallId, content: text });
      state.messages.push({ role: 'tool', toolCallId: parked.toolCallId, content: text });
      continue;
    }

    const shaped =
      subRefs !== undefined
        ? subRefs.shapeCompleted(childResult, collectTurns(childResult))
        : { output: collectTurns(childResult).join('') };
    const rendered =
      typeof shaped.output === 'string' ? shaped.output : (JSON.stringify(shaped.output) ?? '');
    stepEntry.toolCalls.push({
      call: callOf(parked),
      outcome: {
        toolCallId: parked.toolCallId,
        toolName: parked.toolName,
        output: rendered,
        durationMs: childDurationMs,
      },
      stepNumber: resumeStepNumber,
    });
    yield {
      type: 'tool.execute.end',
      toolCallId: parked.toolCallId,
      result: rendered,
      durationMs: childDurationMs,
    };
    messages.push({ role: 'tool', toolCallId: parked.toolCallId, content: rendered });
    state.messages.push({ role: 'tool', toolCallId: parked.toolCallId, content: rendered });
  }

  if (state.pendingSubRuns !== undefined && state.pendingSubRuns.length === 0) {
    delete (state as { pendingSubRuns?: unknown }).pendingSubRuns;
  }

  // Persist the post-phase state (agent-02 shape): from this checkpoint a
  // re-delivered resume is a no-op - the routed approvals are gone and the
  // settled parks removed.
  if (config.checkpointStore !== undefined) {
    await config.checkpointStore.put(
      state.id,
      'agent',
      {
        id: state.id,
        threadId: state.id,
        namespace: 'agent',
        state: serializeRunState(state, { stripTracingApiKey: true }),
        channelVersions: {},
        stepNumber: resumeStepNumber,
        createdAt: new Date().toISOString(),
      },
      {
        source: 'sync',
        status: state.status === 'awaiting_approval' ? 'suspended' : 'running',
        nodeName: 'agent.subrun.resumed',
        sessionId: state.sessionId,
      },
    );
  }
}

/** Rebuild the parent-side ToolCall of a parked sub-run for the journal. */
function callOf(parked: { readonly toolCallId: string; readonly toolName: string }): ToolCall {
  return { toolCallId: parked.toolCallId, toolName: parked.toolName, args: {} };
}

/**
 * A resumed child's `text.complete` turns are not observable through
 * `run()` - derive the final text from the completed state's last
 * assistant message (the same content `output` carries for plain runs).
 */
function collectTurns(result: AgentResult<unknown>): string[] {
  return typeof result.output === 'string' && result.output.length > 0 ? [result.output] : [];
}