// The confined stage-execution seam (`runStage`) — the headless stage-drive primitive.
//
// Drives ONE read-write stage (`implement`/`address`) end-to-end on an already-prepared worktree,
// running the SAME `@mgiles/perk` extension package, with a locked resource set, Pi's compaction
// and auto-retry following the merged settings (as in a warm session), and a budget/timeout
// watchdog. It seeds the stage's initial prompt, lets the model work (calling perk's real tools),
// detects the stage's terminal signal, and returns a structured `RunOutcome`. This implements the
// contract locked in `docs/design/headless-worker.md` §B — the event-stream substrate and the e2e
// harness consume.
//
// Scope here is the in-process drive primitive only. Positioning (worktree create, handoff/plan-ref
// /plan-body materialization, `run_id` mint) is the cold-door/runner's job and is a PREPARED-
// WORKTREE input (audit Gap 7): the worker inherits `PERK_RUN_ID` from the env and never re-mints.
//
// Budget semantics: `budget.tokens` counts FRESH WORK from Pi's own usage records, once each —
// per `turn_end`, the assistant's `input + output` plus every tool result's reported
// `input + output` (Pi folds a tool's nested calls — a codemode script's `models.*` aggregate, a
// nested tool roll-up — onto the parent result, so nested ends are never summed), and per
// `compaction_end`, the summarization usage Pi also persists on the `compaction` entry. Usage
// reported by failed or aborted calls counts. Excluded: cache reads/writes, the provider
// `reasoning` breakdown (a subset of `output` in pi-ai's normalization), branch-summary usage (the
// worker never navigates the tree) and perk's report-wave children (separate sessions). Cache
// warming is off, so the fold observes every usage record the worker can produce at one of two
// boundaries — a turn or a compaction — and the watchdog checks the cap at both. Pi's
// `getSessionStats().tokens.input + output` is the reference census the e2e tier asserts equality
// against; see the adapter's `freshTokensOf`.
//
// CONFINEMENT: this seam's caller surface carries no SDK shapes. Every `@earendil-works` import
// AND the session-drive mechanics (construction, raw events, prompt/abort) live in the private
// `./sdkAdapter.ts` — the seam drives the session only through the adapter's drive-session
// handle, whose listener receives the perk-owned `StageEvent` union (the adapter translates raw
// SDK events at the boundary; ALL policy folding — budget counters, terminal capture, outcome —
// happens here on perk shapes). The one opaque model input (`StageRunOptions.model`) is the
// nominal `WorkerModelRequest` (see the adapter header for the exact — narrow — opacity
// guarantee: an import-edge ban plus nominal minting, nothing stronger) — an UNRESOLVED request
// (the raw `--model` pattern) that the adapter resolves only after the worktree's extensions
// registered their providers. The thin entry `workerMain.ts` carries zero SDK imports: it reaches
// the SDK-free envelope (`runEnvelope.ts`) statically and imports this seam DYNAMICALLY, only
// after admitting the loaded SDK against the host floor (guard Rule F; contracts §8.11).
//
// The initialization boundary: model-runtime creation → services (extension load + provider
// registration) → selection → admission → session construction → bind all run inside the
// outcome boundary. A selection refusal (`no_model` / `model_not_found` / `model_auth`) is a
// typed zero-turn `failed`/`model_error` outcome; any other failure before bind is
// `runtime_init`.

import { env } from "node:process";
import { type PlanRef, readPlanRef } from "../substrate/cache.ts";
import { capForModel } from "../substrate/modelVisible.ts";
import { planReadInstruction, render } from "../substrate/prompts.ts";
import { captureSessionPointer } from "../substrate/sessionPointers.ts";
import { rebuildWorkflowState } from "../substrate/workflowState.ts";
import {
  assembleOutcome,
  createEventEmitter,
  type DriveBudget,
  type DriveStage,
  defaultEventSink,
  EVENT_SUMMARY_CAP,
  type RunEventSink,
  type RunOutcome,
  type TerminalVerdict,
} from "./runEnvelope.ts";
import {
  createDriveSession,
  type DriveRuntimeLike,
  type DriveSessionHandle,
  defaultCreateRuntime,
  type StageEvent,
  WorkerModelRequest,
} from "./sdkAdapter.ts";

// The envelope types live in the SDK-free `runEnvelope.ts` (the thin entry reaches them before any
// SDK links); re-exported so every importer of the seam keeps its import path.
export type {
  DriveBudget,
  DriveStage,
  RunEvent,
  RunEventSink,
  RunOutcome,
  RunStatus,
  TerminalSignal,
} from "./runEnvelope.ts";
// Re-exports so `workerMain.ts` never imports the adapter (guard Rule F), plus the transitive type
// closure of `runStage`'s signature: the `StageRunDeps.createRuntime` fake-construction types and
// the `StageRunOptions.model` request class (a VALUE re-export so `workerMain.ts` mints the
// request from the raw `--model` text with zero SDK imports).
export type { DriveEvent, DriveRuntimeLike, DriveSessionLike } from "./sdkAdapter.ts";
export { WorkerModelRequest } from "./sdkAdapter.ts";

export interface StageRunOptions {
  /** Absolute path to the already-positioned worktree (Gap 7). */
  worktree: string;
  stage: DriveStage;
  /** The seeded first prompt (see `initialPromptFor`). */
  initialPrompt: string;
  /**
   * The one opaque model input: the raw `--model` request, resolved inside the runtime factory
   * only AFTER the worktree's extensions registered their providers/virtual models. Absent (or
   * no pattern) ⇒ the SDK's own default chain picks the model at session creation (settings
   * `defaultProvider`/`defaultModel` with configured auth → pi's per-provider defaults → first
   * available). Never pre-pinned here: `getAvailable()` sorts alphabetically, so `[0]` is the
   * *oldest* model of the first provider (a since-removed `claude-3-5-haiku` date-pin 404'd a
   * whole remote drive). The request's runtime field is the test-injection seam.
   */
  model?: WorkerModelRequest;
  budget: DriveBudget;
  /** External cancellation; OR'd with the budget watchdog. */
  signal?: AbortSignal;
}

/**
 * The offline seam. `createRuntime` overrides the production runtime factory so tests drive
 * synthetic sessions; `now` injects the clock for deterministic `elapsed_ms`.
 */
export interface StageRunDeps {
  createRuntime?: (opts: StageRunOptions) => Promise<DriveRuntimeLike>;
  now?: () => number;
  /** The structured run-event sink. Absent ⇒ the default run-scoped NDJSON file sink. */
  eventSink?: RunEventSink;
}

// --- pure helpers (module-local policy fold) -----------------------------------------------------

/** Mutable counters/captures the drive's policy fold accumulates over the run. */
interface DriveCounters {
  turns: number;
  tokens: number;
  /** Latest submit-bearing evidence (standalone submit or the nested finalizer submit). */
  submitDetails: Record<string, unknown> | null;
  finalizeDetails: Record<string, unknown> | null;
  modelError: { message: string } | null;
}

function freshCounters(): DriveCounters {
  return { turns: 0, tokens: 0, submitDetails: null, finalizeDetails: null, modelError: null };
}

/**
 * Fold one perk-owned drive event into the running counters (pure) — the seam's policy fold over
 * the adapter's translated `StageEvent` union. Counts turns and fresh-work tokens at both usage
 * boundaries (a turn's sum is adapter-computed — assistant plus tool-result usage; a compaction
 * adds its summarization usage without a turn), captures the `submit`/`finalize_address` terminal
 * tool details, and REPLACES the recorded model error on every turn: an error Pi recovers (a
 * retried attempt, an overflow compacted and continued) is cleared by the recovery turn, so only
 * an error on the run's last turn survives to classification. A scheduled retry changes no
 * counter. Exhaustive over the union: a new event kind fails type-checking here.
 */
function applyStageEvent(counters: DriveCounters, event: StageEvent): void {
  if (event.kind === "turn_ended") {
    counters.turns += 1;
    counters.tokens += event.freshTokens;
    counters.modelError = event.modelError;
    return;
  }
  if (event.kind === "compaction_ended") {
    counters.tokens += event.freshTokens;
    return;
  }
  if (event.kind === "model_retrying") return;
  if (event.kind === "tool_ended") {
    if (event.tool === "submit") counters.submitDetails = event.details;
    else if (event.tool === "finalize_address") {
      counters.finalizeDetails = event.details;
      // A finalizer carries the submit that immediately preceded resolution. Recording it into the
      // same latest-evidence slot means a later standalone submit naturally supersedes it after a
      // conflict-resolver re-drive.
      const nestedSubmit = event.details?.submit;
      if (nestedSubmit && typeof nestedSubmit === "object" && !Array.isArray(nestedSubmit)) {
        // The finalizer only exposes this nested block after submit succeeded; restore the
        // success marker stripped from its nested public shape so a later failed standalone
        // submit cannot accidentally satisfy the address completion predicate.
        counters.submitDetails = { ok: true, ...(nestedSubmit as Record<string, unknown>) };
      }
    }
    return;
  }
  const unhandled: never = event;
  throw new Error(`unhandled stage event ${JSON.stringify(unhandled)}`);
}

/**
 * The one abort verdict — shared by the entry sample, the pre-prompt sample and the mid-drive
 * `classify()` arm so all three abort routes yield a byte-identical outcome.
 */
const EXTERNAL_ABORT_VERDICT: TerminalVerdict = {
  status: "aborted",
  terminal_signal: "external_abort",
  pr: null,
  errorType: "external_abort",
  errorMessage: "drive aborted by external signal.",
};

/** True when the budget watchdog should trip from the current counters. */
function budgetTripped(counters: DriveCounters, budget: DriveBudget): boolean {
  return counters.turns >= budget.maxTurns || counters.tokens >= budget.maxTokens;
}

/**
 * Classify a natural-idle terminal from the captured state (pure). `modelError` wins (a
 * post-acceptance provider error on the run's last turn — one Pi did not recover, §B #4); else
 * the stage success predicate:
 *  - implement: a successful `submit` carrying a `pr` → completed/submit_tool;
 *  - address: `finalize_address` ok, `last_review_batch` appended, and the latest submit-bearing
 *    evidence is successful and not definitively unmergeable → completed/address_resolved;
 *  - otherwise the agent went idle without completing the stage → failed/agent_idle_incomplete.
 */
function evaluateTerminal(args: {
  stage: DriveStage;
  submitDetails: Record<string, unknown> | null;
  finalizeDetails: Record<string, unknown> | null;
  lastReviewBatchPresent: boolean;
  modelError: { message: string } | null;
}): TerminalVerdict {
  if (args.modelError !== null) {
    return {
      status: "failed",
      terminal_signal: "model_error",
      pr: null,
      errorType: "model_error",
      errorMessage: args.modelError.message,
    };
  }

  if (args.stage === "implement") {
    const pr = extractPr(args.submitDetails);
    if (args.submitDetails?.ok === true && pr !== null) {
      // Completion additionally requires the submit to be mergeable: a definitively-
      // unmergeable PR (merge conflicts unresolved) is NOT done. `mergeable === true`/`null`/
      // absent all allow completion (fail-open); only a definitive `false` blocks it. On the
      // happy path the resolver follow-up turns re-submit, overwriting submitDetails with a
      // mergeable result, so the natural-idle classification then passes.
      if (args.submitDetails.mergeable === false) {
        return {
          status: "failed",
          terminal_signal: "agent_idle_incomplete",
          pr: null,
          errorType: "incomplete",
          errorMessage:
            "implement drive went idle with an unmergeable PR (merge conflicts unresolved).",
        };
      }
      return {
        status: "completed",
        terminal_signal: "submit_tool",
        pr,
        errorType: null,
        errorMessage: null,
      };
    }
    return {
      status: "failed",
      terminal_signal: "agent_idle_incomplete",
      pr: null,
      errorType: "incomplete",
      errorMessage: "implement drive went idle without an opened PR (no successful submit).",
    };
  }

  // address. `applyEvent` keeps submitDetails as the latest submit-bearing evidence: the nested
  // finalizer submit first, then a later standalone submit from the conflict-resolution re-drive.
  const nestedSubmit = args.finalizeDetails?.submit;
  const fallbackSubmit: Record<string, unknown> | null =
    nestedSubmit && typeof nestedSubmit === "object" && !Array.isArray(nestedSubmit)
      ? { ok: true, ...(nestedSubmit as Record<string, unknown>) }
      : null;
  const effectiveSubmit = args.submitDetails ?? fallbackSubmit;
  if (
    args.finalizeDetails?.ok === true &&
    args.lastReviewBatchPresent &&
    effectiveSubmit?.ok === true &&
    effectiveSubmit.mergeable !== false
  ) {
    return {
      status: "completed",
      terminal_signal: "address_resolved",
      pr: null,
      errorType: null,
      errorMessage: null,
    };
  }
  return {
    status: "failed",
    terminal_signal: "agent_idle_incomplete",
    pr: null,
    errorType: "incomplete",
    errorMessage:
      "address drive went idle without fully finalizing feedback " +
      "(publication, thread resolution, and last_review_batch are required).",
  };
}

/**
 * The post-bind preflight rule (pure): the stage's terminating perk tool must be registered —
 * `implement` → `submit`, `address` → `finalize_address`. Returns the required tool name
 * when absent, else `null`. Deliberately does NOT require the `subagent` tool for `address` — the
 * subagent-under-worker live smoke stays the §8.11 carried risk.
 */
function missingTerminatingTool(stage: DriveStage, toolNames: string[]): string | null {
  const required = stage === "implement" ? "submit" : "finalize_address";
  return toolNames.includes(required) ? null : required;
}

/** Pull a `{ number, url }` PR from a captured `submit` details block; null when malformed. */
function extractPr(
  details: Record<string, unknown> | null,
): { number: number; url: string } | null {
  if (!details || typeof details.pr !== "object" || details.pr === null) return null;
  const pr = details.pr as { number?: unknown; url?: unknown };
  if (typeof pr.number === "number" && typeof pr.url === "string") {
    return { number: pr.number, url: pr.url };
  }
  return null;
}

// --- run-event helpers (offline-testable) ---------------------------------------------

/**
 * Compute a `tool_outcome` `{ tool, ok, summary }` from a translated `tool_ended` event (pure).
 * `summary` is `null` on success and, on failure, a capped (route-don't-relay) rendering of the
 * adapter's pre-cap error text — never the raw tool result. The `errorText` nullish fallback
 * below is defensive only: the adapter always supplies `errorText` for failures (pinned in
 * `sdkAdapter.test.ts`), so it is unreachable through the production path.
 */
function toolOutcomeOf(event: Extract<StageEvent, { kind: "tool_ended" }>): {
  tool: string;
  ok: boolean;
  summary: string | null;
} {
  const summary = event.ok
    ? null
    : capForModel(event.errorText ?? `tool ${event.tool} failed`, EVENT_SUMMARY_CAP).shown;
  return { tool: event.tool, ok: event.ok, summary };
}

/**
 * Re-derive the stage's initial prompt from the plan-ref — the TS twin of
 * `perk/run/launch.py._implement_prompt`/`_address_prompt`. INVARIANT: textual parity with the Python
 * plane (asserted reciprocally in `stageExecution.test.ts` + `tests/test_worker_prompt_parity.py`). No
 * skill-binding suffix is appended here: in the driven session the bindings arrive via Mechanism A
 * (bindingDelivery.ts injects the handoff stage's render because this prompt carries no
 * `BINDING_HEADER`) — content byte-identical to the cold door's suffix (contracts.md §8.38).
 * Returns `null` when there is no plan-ref (nothing to prime).
 *
 * The implement primer's wording lives in the canonical template `prompts/stages/implement.md`,
 * rendered by the shared seam (contracts.md §8.31); branching stays in code — only the `read_cmd`
 * var differs. This implement output is byte-identical to the warm `implementHandoffPrompt`.
 *
 * The `address` wording lives in the shared canonical template `prompts/stages/address/action.md`
 * rendered via the cross-plane render seam (contracts.md §8.31); the worker has no preview path
 * (preview is a warm/cold flag only), so it always renders the action body. The classify step is
 * the `classify_review_feedback` tool, which reads the configured classifier model at execute
 * time — nothing model-shaped rides the prompt.
 */
function initialPromptFor(stage: DriveStage, planRef: PlanRef | null): string | null {
  if (planRef === null) return null;
  const provider = String(planRef.provider ?? "");
  const prId = String(planRef.pr_id ?? "");
  const url = String(planRef.url ?? "");
  if (stage === "implement") {
    const readCmd = planReadInstruction(provider, prId, url);
    return render("stages/implement.md", { provider, pr_id: prId, url, read_cmd: readCmd });
  }
  // address
  return render("stages/address/action.md", { provider, pr_id: prId, url });
}

// --- the drive primitive ------------------------------------------------------------------------

/**
 * Drive one stage to terminal and return a structured `RunOutcome` — never throws (fail-soft like
 * the delivery submit operation). Seeds `initialPrompt`, races the driving `prompt()` against the budget watchdog and
 * the external `signal`, classifies the terminal at idle, and disposes the adapter handle in
 * `finally` (guarded — a cleanup error can never replace the computed outcome). All session
 * mechanics go through the adapter's drive-session handle; policy stays here.
 *
 * The `run_started` + `run_finished` pair is emitted at entry / at every exit, so each path —
 * pre-aborted, `runtime_init`, a selection refusal (`no_model` / `model_not_found` /
 * `model_auth`), the preflight, a pre-prompt abort, the drive — is a well-formed
 * zero-or-more-turn stream. Model-runtime creation, extension registration, model selection +
 * admission, runtime construction and the bind all run inside the outcome boundary: a selection
 * refusal comes back from the adapter as a typed value, and the catch arm keys its `error.type`
 * on the bind boundary (`runtime_init` before the session is bound, `drive_error` after). The external `signal` is
 * SAMPLED at entry and again immediately before `prompt()` (an idle session has nothing to abort,
 * so an aborted sample returns `aborted` directly) and SUBSCRIBED only for the drive — the
 * listener is registered synchronously after the pre-prompt sample because `AbortSignal` never
 * replays an earlier abort to a late listener.
 */
export async function runStage(
  opts: StageRunOptions,
  deps: StageRunDeps = {},
): Promise<RunOutcome> {
  const now = deps.now ?? Date.now;
  const startMs = now();
  const counters = freshCounters();
  const elapsed = (): number => Math.max(0, now() - startMs);

  // Structured run-event stream: resolve the sink + run_id once, build the emitter, emit
  // `run_started` once at entry, and route every terminal exit through `finish` so exactly one
  // `run_started` + `run_finished` pair is emitted per drive.
  const runId = env.PERK_RUN_ID ?? "";
  const sink = deps.eventSink ?? defaultEventSink(opts.worktree, runId);
  const emitter = createEventEmitter(sink, now, startMs);
  const finish = (verdict: TerminalVerdict): RunOutcome => {
    const outcome = assembleOutcome({
      stage: opts.stage,
      verdict,
      budget: { turns: counters.turns, tokens: counters.tokens, elapsed_ms: elapsed() },
    });
    emitter.emit({ kind: "run_finished", outcome });
    return outcome;
  };
  emitter.emit({ kind: "run_started", run_id: runId, stage: opts.stage });

  // The entry sample: nothing is resolved, constructed or bound for an already-aborted signal.
  if (opts.signal?.aborted) return finish(EXTERNAL_ABORT_VERDICT);

  let terminationReason: "natural" | "budget" | "abort" = "natural";
  let settled = false;
  let bound = false;
  let handle: DriveSessionHandle | null = null;
  // The cap is checked at BOTH usage boundaries. A pre-request compaction runs before its request
  // starts, so a trip on `compaction_ended` aborts the agent signal before the request is
  // dispatched: zero provider requests follow a compaction-boundary trip.
  const listener = (event: StageEvent): void => {
    applyStageEvent(counters, event);
    if (event.kind === "turn_ended") {
      if (budgetTripped(counters, opts.budget)) trip("budget");
    } else if (event.kind === "compaction_ended") {
      console.error(`perk worker: compaction (${event.reason}) — ${event.freshTokens} tokens`);
      if (budgetTripped(counters, opts.budget)) trip("budget");
    } else if (event.kind === "model_retrying") {
      console.error(
        `perk worker: auto-retry ${event.attempt}/${event.maxAttempts} in ${event.delayMs} ms — ${event.message}`,
      );
    } else if (event.kind === "tool_ended") {
      const o = toolOutcomeOf(event);
      emitter.emit({ kind: "tool_outcome", tool: o.tool, ok: o.ok, summary: o.summary });
    }
  };

  function trip(reason: "budget" | "abort"): void {
    if (settled) return;
    if (terminationReason === "natural") terminationReason = reason;
    handle?.abort();
  }

  // The turn-boundary gate: a turn that reaches the cap ends the run at its own boundary, so no
  // next turn starts (Pi's loop does not re-check the abort signal between turns). The following
  // `turn_ended` still folds the counters and trips the watchdog, which records the verdict and
  // aborts the session's post-run continuation. A run already tripped ends at the next boundary.
  // The running `counters.tokens` already include every earlier compaction's usage.
  const endRunAfterTurn = (turn: { freshTokens: number }): boolean => {
    if (settled) return false;
    if (terminationReason !== "natural") return true;
    const after = {
      ...counters,
      turns: counters.turns + 1,
      tokens: counters.tokens + turn.freshTokens,
    };
    return budgetTripped(after, opts.budget);
  };

  const onSignal = (): void => trip("abort");

  try {
    let runtime: DriveRuntimeLike;
    if (deps.createRuntime) runtime = await deps.createRuntime(opts);
    else {
      // Model selection is a production-path concern only: an injected runtime factory (tests)
      // never touches the default `ModelRuntime.create` (no host file reads). The adapter runs
      // the whole ladder (services → selection → admission → construction) and returns a
      // refusal as a typed value — never a session that cannot drive.
      const built = await defaultCreateRuntime(
        opts.worktree,
        opts.model ?? new WorkerModelRequest({}),
      );
      if (!built.ok) {
        return finish({
          status: "failed",
          terminal_signal: "model_error",
          pr: null,
          errorType: built.refusal.type,
          errorMessage: built.refusal.message,
        });
      }
      runtime = built.runtime;
    }
    handle = createDriveSession(runtime, listener, endRunAfterTurn);
    await handle.bind();
    bound = true;

    // Terminating-tool preflight (presence-gated on the session's extension runner): disk
    // discovery has a silent-zero arm — a missing/unparseable `.pi/settings.json` or an
    // unresolvable local-path package yields ZERO extension tools without throwing — so fail
    // fast (zero turns) instead of burning the whole budget on a drive that can never call its
    // terminating tool. Reuses the `model_error` terminal signal with a distinct `error.type`
    // (the `no_model` precedent).
    const toolNames = handle.registeredToolNames();
    if (toolNames !== null) {
      const missing = missingTerminatingTool(opts.stage, toolNames);
      if (missing !== null) {
        return finish({
          status: "failed",
          terminal_signal: "model_error",
          pr: null,
          errorType: "no_extension_tools",
          errorMessage:
            `perk extension tools did not register — the ${opts.stage} stage's terminating ` +
            `tool \`${missing}\` is missing. Check the worktree's .pi/settings.json packages ` +
            "list (perk init converges it); construction diagnostics are on stderr.",
        });
      }
    }

    // Implementation/worker session pointer (contracts.md §8.35): the headless drive records the
    // inner driven session's file under THIS run id into the shared main checkout (the worktree's
    // `mainCheckoutRoot`), labelled `.worker` by capture site. The inner session's own
    // `session_start` records the matching `.main`. Best-effort + non-fatal (carrier warns).
    if (opts.stage === "implement") {
      captureSessionPointer({
        cwd: opts.worktree,
        runId,
        klass: "implementation",
        site: "worker",
        sessionFile: handle.sessionFile(),
      });
    }

    // The pre-prompt sample, then the drive-only subscription with no `await` in between (no
    // check-then-subscribe window). Budget/abort wiring (Gap 2): wall-clock timer + external
    // signal both trip → handle.abort().
    if (opts.signal?.aborted) return finish(EXTERNAL_ABORT_VERDICT);
    const timer = setTimeout(() => trip("budget"), opts.budget.wallClockMs);
    opts.signal?.addEventListener("abort", onSignal, { once: true });

    try {
      await handle.prompt(opts.initialPrompt);
    } finally {
      clearTimeout(timer);
      opts.signal?.removeEventListener("abort", onSignal);
      settled = true;
    }

    // Defensive rebind (Gap 1): the happy path never replaces the session; a replacement is loud.
    if (await handle.rebindIfReplaced()) {
      console.error("perk worker: unexpected mid-drive session replacement — rebinding listener.");
    }

    const verdict = classify(opts, counters, terminationReason, () => handle?.workflowBranch());
    return finish(verdict);
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    return finish({
      status: "failed",
      terminal_signal: "model_error",
      pr: null,
      errorType: bound ? "drive_error" : "runtime_init",
      errorMessage: bound
        ? `headless drive failed: ${message}`
        : `worker runtime initialization failed: ${message}`,
    });
  } finally {
    if (handle) await handle.dispose();
  }
}

/**
 * Pick the terminal verdict: watchdog/abort override the natural-idle classification. `branch` is
 * a thunk so the workflow branch is read only on the natural path (exactly as before the seam).
 */
function classify(
  opts: StageRunOptions,
  counters: DriveCounters,
  terminationReason: "natural" | "budget" | "abort",
  branch: () => unknown[] | undefined,
): TerminalVerdict {
  if (terminationReason === "budget") {
    return {
      status: "budget_exhausted",
      terminal_signal: "budget",
      pr: null,
      errorType: "budget",
      errorMessage: "budget exhausted (turns/tokens/wall-clock) — drive aborted.",
    };
  }
  if (terminationReason === "abort") return EXTERNAL_ABORT_VERDICT;
  const lastReviewBatchPresent =
    rebuildWorkflowState((branch() ?? []) as never).last_review_batch != null;
  return evaluateTerminal({
    stage: opts.stage,
    submitDetails: counters.submitDetails,
    finalizeDetails: counters.finalizeDetails,
    lastReviewBatchPresent,
    modelError: counters.modelError,
  });
}

/** Convenience: re-derive the initial prompt for a prepared worktree (reads its `cache.plan-ref`). */
export function initialPromptForWorktree(worktree: string, stage: DriveStage): string | null {
  return initialPromptFor(stage, readPlanRef(worktree));
}