// auto-dispatch.ts — the framework ACTS on the queue instead of only nudging.
// Two automations (both gated on the item being FULLY specified):
//
//  A. autoDispatchEligible — when a slot frees, fill it with approved items
//     items that have a complete scope + cwd + low/medium risk. The task is
//     `KEY: <key>` + the scope (the orchestrator writes scopes at
//     proposal/approval time — the dispatch-time scoping moves earlier).
//     HIGH-risk or incomplete items are left for the orchestrator's
//     checkpoint (the dispatch tick still nudges those).
//
//  B. autoRedispatch — on a review FAIL, re-dispatch with the original scope
//     PLUS the reviewer's findings (up to the cap). The cap still surfaces.
//
//  B2. dispatchPendingHumanRedispatch (KEY: AUTOPILOT-59) — the harness
//     consumer for a HUMAN's re-dispatch (decision panel/TUI "redispatch"
//     action). Spawns through B (autoRedispatch) itself — same findings,
//     same task builder — but never touches the review-FAIL `attempts` cap: a
//     human's correction is not a review failure. Panels.ts does the
//     human-review → active move + records the findings; this sweep-time
//     pass turns that into a spawn (and fails loudly, never silently, if the
//     item's repo no longer exists).
//
//  C. autoReview — when a worker completes (active → ai-review), dispatch the
//     reviewer automatically with the SAME fields the dispatch used: KEY +
//     scope + cwd (to locate the work). The approval gate guarantees scope +
//     cwd on every approved item, so both automations just take them off the
//     item. The verdict routing (PASS → done, FAIL → B, cap → failed,
//     NO-CHANGE → no-change) is the engine's; the human handover (flag_for_review)
//     stays the orchestrator's.
//
// The orchestrator keeps all JUDGMENT (intake scanning, approval, high-risk
// checkpoints); these automations remove the mechanical round-trips.

import { existsSync } from "node:fs";
import { loadStore, mutateStore, updateItem, unmetDependencies, dependsOnHeader, type QueueItem, type QueueStore } from "../queue-store.ts";
import { captureFinisherBaseline, isFinisherItem } from "../finisher-evidence.ts";
import { resolveDispatchBudget } from "./run-budget.ts";
import { AUTO_DISPATCH_RISK, type Risk } from "./item-contract.ts";
import type { SubagentBackend } from "../backends/types.ts";
import { parallelBranches, preserveRunWorktree, reviewPointersFor, tipReachableFromMain } from "./worktree-preservation.ts";
import { git } from "./git.ts";
import { isProviderDegraded, recordProviderFailure, recordProviderSuccess } from "./recovery-state.ts";
import { reviewerRunAlive, runStateEvidence } from "./run-liveness.ts";
import { appendTelemetry, loadAutopilotConfig } from "../config.ts";
import { SCOPE_SKELETON } from "./spec-completeness.ts";
import { SPEC_PASS_SCOPE_START, SPEC_PASS_SCOPE_END } from "../verdict.ts";
import { REVIEW_CONTRACT_TEXT } from "../review-contract.ts";
import { acceptanceEvidenceStatus, WORKER_EVIDENCE_INSTRUCTION } from "../acceptance-evidence.ts";
import { resolveSpawnModel } from "../spawn-model.ts";
import { resolveSpawnAgent } from "../agents/install.ts";
import { validateDispatchPairing } from "./dispatch-pairing.ts";

/** B26 rule as a testable PREDICATE: a subagent tool call is a worker spawn
 *  WITHOUT worktree isolation — the class that breaks parallel workers (B20's
 *  strays blocked B22/B23 at launch, dirtying the main checkout). The pi host
 *  wires its tool_call interception to this; any host with a tool-call hook
 *  can adopt the same rule. Management actions (status/steer/stop) are not
 *  spawns — never blocked. */
export function isUnisolatedWorkerSpawn(
  args: { agent?: string; worktree?: boolean; action?: string },
  workerAgents: string[],
): boolean {
  if (args.action !== undefined) return false; // management — not spawns
  if (!args.agent || !workerAgents.includes(args.agent)) return false;
  if (args.worktree === true) return false; // explicitly isolated
  return true;
}

/** A low/medium-risk item with a complete scope + cwd is auto-dispatchable.
 *
 *  RISK MUST BE AN EXPLICIT low/medium (KEY: AUTOPILOT-58 part B) — this used
 *  to read `item.risk !== "high"`, an EXCLUDE-list that let anything but the
 *  literal string "high" through, including an EMPTY risk. That silently
 *  contradicted this function's own doc comment ("low/medium-risk item") and
 *  is the actual mechanism behind the incident this ticket exists for: an
 *  approved item with real scope+cwd but no risk classification has no
 *  completeness check ANYWHERE, so it either dispatches on an unclassified
 *  risk (the old bug) or, once the item-contract table
 *  (framework/item-contract.ts) closes off *new* risk-less approvals at the
 *  approval transition, is simply legacy data this function
 *  must still refuse to treat as safe. An ALLOW-list (low/medium only) closes
 *  both paths: new admission is gated upstream; old/garbage risk values never
 *  silently pass here either. */
export function isAutoDispatchable(item: QueueItem, store?: QueueStore): boolean {
  return (
    item.status === "approved" &&
    // ALLOW-list from the closed vocabulary (item-contract.ts): an empty or
    // non-vocabulary risk ("med", prose) never silently passes.
    AUTO_DISPATCH_RISK.includes(item.risk as Risk) &&
    typeof item.scope === "string" &&
    item.scope.trim().length > 0 &&
    typeof item.cwd === "string" &&
    item.cwd.trim().length > 0 &&
    // DEPENDENCY GATE (KEY: AUTOPILOT-81): an approved item with unmet `after`
    // deps is WAITING, not ready — skipped here exactly like the risk/scope/
    // cwd gates above. `store` is OPTIONAL so every existing single-arg call
    // site (tests especially) keeps its prior meaning unchanged; only a
    // caller that HAS a store to check dependencies against is held to it.
    (!store || unmetDependencies(item, store).length === 0)
  );
}

/** The prior attempt's branch/tip pointers the framework can actually resolve
 *  (KEY: AUTOPILOT-129). TWO sources, unioned, because relying on the journal
 *  alone named nothing in the live profile (the journal is never written there
 *  — see the note above):
 *    1. the worktree journal (`reviewPointersFor`) — the durable record when a
 *       sweep captured it;
 *    2. the item's OWN record of the prior run — `runWorktree.branch` (when it
 *       belongs to the run this item last had, the AUTOPILOT-85 ownership rule)
 *       and the branches resolvable for the prior run id (`parallelBranches`,
 *       path-matched by the FULL run id embedded in the worktree directory).
 *  A freshly rev-parsed tip wins over a journaled one (the journal's tip can
 *  lag a branch that advanced). Best-effort: any failure simply contributes
 *  nothing — a resolver must never block a dispatch. */
function priorAttemptPointers(stateDir: string, item: QueueItem): Array<{ branch: string; tipSha: string }> {
  const repo = item.cwd;
  if (!repo) return [];
  const byBranch = new Map<string, string>();
  try {
    for (const p of reviewPointersFor(stateDir, item.key)) byBranch.set(p.branch, p.tipSha);
  } catch {
    // journal unreadable — the item's own record below still applies
  }
  const recorded = item.runWorktree ?? null;
  const currentRunId = item.runId ?? item.lastRunId ?? null;
  const priorRunId = recorded?.runId ?? currentRunId;
  const branches = new Set<string>();
  if (recorded?.branch && (!currentRunId || recorded.runId === currentRunId)) branches.add(recorded.branch);
  if (priorRunId) {
    try {
      for (const b of parallelBranches(repo, priorRunId)) branches.add(b);
    } catch {
      // best-effort
    }
  }
  for (const branch of branches) {
    const tip = git(repo, ["rev-parse", "--verify", `${branch}^{commit}`]);
    if (tip) byBranch.set(branch, tip);
  }
  return [...byBranch.entries()].map(([branch, tipSha]) => ({ branch, tipSha }));
}

/** P5 PRIOR-ATTEMPT POINTERS (KEY: AUTOPILOT-60 part A). A re-dispatch's
 *  worktree is a FRESH `pi-parallel-<runid>-*` checkout the runtime builds at
 *  spawn time — it structurally carries none of an earlier attempt's commits,
 *  which lived only on THAT attempt's branch. The runtime is DOCUMENTED to
 *  delete that branch + worktree at run end, and demonstrably sometimes does
 *  (three fsck-rescue incidents motivated worktree-preservation.ts), but an
 *  AUTOPILOT-74 audit found that deletion is NOT reliable — old worktrees can
 *  and do survive for days, sometimes long after merging into main (see
 *  `pruneStaleWorktrees`, worktree-preservation.ts) — so this comment no
 *  longer asserts deletion as a given; either way, the redo's fresh worktree
 *  does not inherit the prior attempt's commits. This framework has no seam
 *  to choose a redo's base (SubagentBackend.spawn takes no baseRef/branch —
 *  see backends/types.ts), so the redo WILL start clean; what it can do is
 *  TELL the worker so, naming the branch(es) and SHA(s) the framework can
 *  still resolve — the journal when it has them, PLUS the item's OWN record of
 *  the prior run, because the live profile's journal is EMPTY in practice
 *  (verified: no `handoffs.jsonl` under
 *  ~/.local/state/orchestrator/personal, and no keep-refs existed at the time
 *  of the check) and a journal-only resolver named
 *  NOTHING for AUTOPILOT-126. The durable keep-ref
 *  guarantees the commits stay reachable even if/when the runtime does
 *  delete the branch — `git log <sha>` / `git cherry-pick <sha>` — from the
 *  SAME repo's object database the new worktree shares). Observed without
 *  this: DOTFILES-3's
 *  redo landed on main with none of attempt 1's commits and the worker found
 *  that out by chance, then recovered on its own initiative — nothing in the
 *  framework told it to look. A re-dispatch with NO journaled pointer now says
 *  that too ("no preserved work is recorded — stop looking") instead of staying
 *  silent, and a first dispatch says nothing at all (AUTOPILOT-129). */
export function priorAttemptContext(stateDir: string, item: QueueItem): string {
  if (!item.cwd) return "";
  const pointers = priorAttemptPointers(stateDir, item);
  if (pointers.length === 0) {
    // A FIRST dispatch has no prior attempt to speak of — say nothing (noise).
    // A RE-DISPATCH does: state the record positively so the worker stops
    // hunting for a pointer that does not exist, WITHOUT claiming the work
    // never happened (the journal can be empty for reasons other than "no
    // work" — AUTOPILOT-77).
    if (!isRedispatch(item)) return "";
    return [
      "## Prior attempt(s) on this item — NO PRESERVED WORK IS RECORDED (do not go looking for one)",
      "This is a RE-DISPATCH: the item has already run at least once. The framework's worktree journal holds NO branch/tip for it, so there is no preserved reference for you to import — start clean and do not spend time searching for one.",
      "That is the RECORD, not proof: if you happen across prior work anyway (`git log --all --grep`, `git branch -a`), IMPORT it rather than rebuilding — do not ignore work just because this brief did not name it.",
    ].join("\n");
  }
  const lines = pointers.map((p) => {
    // STALE-VS-MAIN GUARD: a tip already reachable from main was probably
    // folded in by a prior recovery/merge — surfaced, not silently treated as
    // still-live work to reapply (the risk named in the task: blindly basing a
    // redo on a branch that may already be superseded, or whose approach was
    // wrong, carries the flaw forward).
    let reachable = false;
    try {
      reachable = tipReachableFromMain(item.cwd!, p.tipSha);
    } catch {
      reachable = false;
    }
    return `- ${p.branch} @ ${p.tipSha.slice(0, 8)}${reachable ? " (ALREADY reachable from main — likely already merged/superseded; verify before reusing, do not blindly reapply)" : " (NOT in main — exists ONLY on this branch/ref)"}`;
  });
  return [
    "## Prior attempt(s) on this item — YOUR FRESH WORKTREE DOES NOT INCLUDE THESE COMMITS",
    "This redo runs in a FRESH isolated worktree/branch, so it does NOT inherit the commits below regardless of what happened to the prior attempt's worktree (the runtime is documented to delete it at run end, and sometimes does, but AUTOPILOT-74 found that is not reliable — do not assume the old one is gone either). Either way they are preserved under a durable ref, not lost, and share this repo's object database with your new worktree:",
    ...lines,
    "",
    "IMPORT, DO NOT REBUILD: if the work on a branch above already satisfies (or nearly satisfies) this item's deliverable, cherry-pick/reconcile it onto your branch and go straight to what is actually missing. Re-implementing committed work that is one `git log` away is the exact cost this section exists to remove.",
    "CHECK RECOVERABILITY BEFORE REDOING ANYTHING (P5): `git log <sha>` / `git show <sha>` to see what is there — the commits stay reachable even if the branch itself was deleted (each tip is preserved under a durable `refs/orchestrator/keep/<runId>` ref). If the work is still good, cherry-pick or reconcile it onto your branch DELIBERATELY. If this is a review-FAIL redo and the prior approach was WRONG, start clean ON PURPOSE and say so in your report — do not blindly inherit a flawed approach, and do not silently re-implement work that already exists.",
  ].join("\n");
}

/** Has this item already had at least one run? The discriminator between a
 *  FIRST dispatch (nothing prior to name, so the brief stays silent) and a
 *  RE-DISPATCH (the brief must say where the prior work is, or that none is
 *  recorded). Every marker below is a fact stamped by a lane the framework
 *  owns: `lastRunId` (terminal-transition clearing, updateItem), `runId`
 *  (ai-review keeps it), `attempts`/`recoveries` (the retry counters), or a
 *  recorded failure. An `approved` item that has never run carries none of
 *  them (KEY: AUTOPILOT-129). */
export function isRedispatch(item: QueueItem): boolean {
  return (
    Boolean(item.lastRunId) ||
    Boolean(item.runId) ||
    (item.attempts ?? 0) > 0 ||
    (item.recoveries ?? 0) > 0 ||
    item.status === "failed" ||
    item.failCause != null
  );
}

/** KEY: AUTOPILOT-70 — observed live: a worker completed correct, fully
 *  tested work, left it STAGED (never committed) on its branch, and reported
 *  success as "uncommitted, not merged" — conflating the one rule it was
 *  given (never merge/push to main) with a rule it was never given (never
 *  commit at all). A worktree the runtime deletes at run end can only be
 *  rescued from a real commit; staged changes are not a git object and are
 *  gone with no error, no failed status, nothing. This line is WORKER-CLASS
 *  ONLY (dispatch-class aware like recoveryContext's own version of this
 *  rule): a finisher-class item's entire scope is landing an approved branch
 *  in the target checkout, so "never merge to main" would instruct it to
 *  fail its own task. */
const COMMIT_NOT_MERGE =
  '## Commit discipline\nCOMMIT your work on your own branch as you go — this is REQUIRED and SAFE, never optional: a completion report is not done while `git status` still shows staged or unstaged changes. MERGE or PUSH to main is a DIFFERENT action and is FORBIDDEN unless this task explicitly says otherwise — main is touched only by the human-approved merge-finisher after review. "Do not merge to main" means commit everything on your branch and stop there, not withhold commits.';

/** Build the worker task: the KEY line + the scope + (when known) the prior
 *  attempt's branch/SHAs (`stateDir`, optional — callers with no queue state
 *  to consult, e.g. tests, simply omit it) + optional review findings. */
export function workerTask(item: QueueItem, findings?: string, stateDir?: string): string {
  const parts: string[] = [];
  // DEPENDENCY HEADER (KEY: AUTOPILOT-81): when this item names `after`
  // dependencies, PREFIX the task with whether each one's work is actually on
  // main yet — `done` only means the WORK is done; shipping can be a separate,
  // later, manual step (see dependsOnHeader, queue-store.ts). Best-effort: a
  // missing/unreadable store must never block a dispatch that is otherwise
  // ready to go out.
  if (stateDir && item.after && item.after.length) {
    try {
      const store = loadStore(stateDir);
      const header = store ? dependsOnHeader(item, store) : "";
      if (header) parts.push(header);
    } catch {
      // best-effort — never blocks dispatch
    }
  }
  parts.push(`KEY: ${item.key}\n${item.scope.trim()}`);
  if (!isFinisherItem(item)) parts.push(COMMIT_NOT_MERGE);
  // WORKER ACCEPTANCE ARTIFACT (KEY: AUTOPILOT-108) — the pi-subagents worker
  // never reads worker/canonical.md (see that file's HOST SCOPE note), so the
  // evidence requirement reaches it through the TASK TEXT, the same channel
  // COMMIT_NOT_MERGE uses.
  parts.push(WORKER_EVIDENCE_INSTRUCTION);
  if (stateDir) {
    const prior = priorAttemptContext(stateDir, item);
    if (prior) parts.push(prior);
  }
  if (findings && findings.trim()) {
    // Staleness guard: findings describe the repo AS THE REVIEWER SAW IT —
    // between the failed attempt and this re-dispatch, the orchestrator (or a
    // recovery pass) may have already fixed some/all of them (observed live:
    // ATL-DOC-REVIEW-APPLY's FAIL auto-re-dispatch landed AFTER the work was
    // recovered onto main). Blindly applying stale findings duplicates or
    // corrupts; verify-first makes the redo safe by construction.
    parts.push(`## Review findings (address them — VERIFY FIRST)\n${findings.trim()}\n\nSTALENESS WARNING: these findings may be OUT OF DATE — the repo may have changed since the failed attempt (recovery commits, merges, manual fixes). BEFORE applying anything: check current state (git log/status, read the files). Apply ONLY what is genuinely still missing; skip anything already addressed and say so in your report.`);
  }
  return parts.join("\n\n");
}

/** Zombie-grace default (minutes) used when a caller supplies none — mirrors
 *  AutopilotConfig.zombieGraceMinutes's own default. */
const DEFAULT_ZOMBIE_GRACE_MINUTES = 30;

export interface OccupiedSlotsOpts {
  /** The backend's own active-run count. The maximum is the honest floor — an
   *  undercount (a status RPC lagging a just-started run, or a fleet view this
   *  backend cannot provide at all) must never invent free capacity. Unknown
   *  (null/undefined) contributes nothing rather than a fake 0. */
  fleetTotalActive?: number | null;
  /** Backend used for PER-RUN liveness (the run's status.json). Absent → the
   *  no-backend grace fallback below. */
  backend?: Pick<SubagentBackend, "asyncDirFor"> | null;
  /** Grace window (minutes) for the no-backend fallback. */
  zombieGraceMinutes?: number;
  now?: number;
}

/** Does ONE recorded run still hold a worker slot? (KEY: AUTOPILOT-86)
 *
 *  The capacity lane's fail direction is the OPPOSITE of the reviewer lane's:
 *  a wrong "free" over-subscribes and can strand a dispatch behind a phantom
 *  count, so when liveness CANNOT be determined the run counts as occupied.
 *  With a backend, the run's own status.json is authoritative:
 *    - a parsed TERMINAL phase (stopped/killed/completed/failed/…) releases
 *      the slot with NO completion event needed — this is the deterministic
 *      STOP-path release (the stop action writes the run record, the
 *      derivation reads it);
 *    - `running`/`queued`/`paused`/`blocked` is live;
 *    - unreadable/absent/unknown (and no run id at all — a failed spawn) is
 *      fail-closed OCCUPIED, never a guessed free slot.
 *  With NO backend the run dir is unreadable, so the only signal is how long
 *  the item has sat in flight: within `zombieGraceMinutes` it holds its slot;
 *  past it the zombie/dead-run sweeps reap it, so it releases here too. This
 *  is the "no runs table in this repo" fallback (the actuator has one). */
function runOccupiesSlot(
  runId: string | null,
  updatedAt: string,
  backend: Pick<SubagentBackend, "asyncDirFor"> | null | undefined,
  graceMs: number,
  now: number,
): boolean {
  if (backend) return runStateEvidence(backend, runId ?? "").kind !== "terminal";
  const startedAt = Date.parse(updatedAt);
  if (!Number.isFinite(startedAt)) return true; // unparseable stamp → fail-closed
  return now - startedAt < graceMs;
}

/** THE ONE occupancy derivation (KEY: AUTOPILOT-86) — count the store items
 *  genuinely in flight: `active` workers (or `ai-review` items whose reviewer
 *  is live) whose run is still live. NEVER a separately-incremented counter:
 *  a stopped run, a failed spawn and a lost completion all release their slot
 *  from the run record, so this number cannot ratchet up. Every reader
 *  (queue_dispatch's refusal, auto-dispatch's free-slot math, the tick's
 *  FLEET line, /autopilot status, queue_list's fleet block) calls THIS.
 *
 *  The returned number is the STORE half; callers union it with the backend's
 *  fleetTotalActive (see effectiveOccupiedSlots) so runs the queue never
 *  tracked still bind the cap. */
export function liveTrackedSlots(store: QueueStore, opts: OccupiedSlotsOpts = {}): number {
  const now = opts.now ?? Date.now();
  const graceMs = (opts.zombieGraceMinutes ?? DEFAULT_ZOMBIE_GRACE_MINUTES) * 60_000;
  let live = 0;
  for (const item of Object.values(store.items)) {
    let runId: string | null = null;
    if (item.status === "active") runId = item.runId ?? null;
    else if (item.status === "ai-review" && item.reviewerRunId) runId = item.reviewerRunId;
    else continue; // not an in-flight shape (ai-review with NO reviewer holds no slot)
    if (runOccupiesSlot(runId, item.updatedAt, opts.backend, graceMs, now)) live += 1;
  }
  return live;
}

/** Occupied slots as the MANUAL lane must count them (KEY: AUTOPILOT-48/86):
 *  the higher of the store's live-item derivation and the backend's fleet
 *  count. The maximum is the honest floor — a fleet undercount must never
 *  invent free capacity, and an UNKNOWN fleet contributes nothing rather than
 *  a fake 0.
 *
 *  KEY: AUTOPILOT-61 — this used to ALSO take the engine's run ledger
 *  (`Autopilot.status().running`), a separately-incremented Map that can only
 *  recover to zero if every started run's completion event later arrives with
 *  the EXACT SAME id. A live session watched it climb monotonically ("12 of
 *  6", "13 of 6", "15 of 6", "17 of 6") while every async-run status.json
 *  reported ZERO live processes — one leaked entry permanently inflates every
 *  future reading. AUTOPILOT-86 removes the counter class entirely: occupancy
 *  is now derived from the run RECORDS (liveTrackedSlots), so a stopped run
 *  releases its slot from its own status.json. */
export function effectiveOccupiedSlots(stateDir: string, opts: OccupiedSlotsOpts = {}): number {
  const store = loadStore(stateDir);
  const tracked = store ? liveTrackedSlots(store, opts) : 0;
  return Math.max(tracked, opts.fleetTotalActive ?? 0);
}

/** One countable `dispatch.at-capacity {occupied, max, ready}` telemetry line
 *  when the harness gate stalls with READY work (KEY: AUTOPILOT-96). The gate
 *  used to return [] silently at capacity, so a full lane and an idle queue
 *  looked identical in the log; this is the event that tells them apart.
 *
 *  One line per (stateDir, occupied/max/ready) episode: a persistent stall is
 *  not re-logged on every sweep (the same "one notice per episode" discipline
 *  the fleet-RPC-degraded and spec-pass notices use), and the entry is cleared
 *  the moment the gate is no longer at capacity (or has no ready work), so a
 *  LATER stall of the same shape is recorded again. Map is module-scoped and
 *  keyed by stateDir — one runner/host process owns one state dir. */
const atCapacityEpisode = new Map<string, string>();
function recordAtCapacity(stateDir: string, d: { occupied: number; max: number; ready: number; foreignRuns?: number; unattributedRuns?: number }): void {
  const episode = `${d.occupied}/${d.max}/${d.ready}/${d.foreignRuns ?? ""}/${d.unattributedRuns ?? ""}`;
  if (atCapacityEpisode.get(stateDir) === episode) return;
  atCapacityEpisode.set(stateDir, episode);
  appendTelemetry(stateDir, JSON.stringify({
    t: new Date().toISOString(),
    type: "dispatch.at-capacity",
    occupied: d.occupied,
    max: d.max,
    ready: d.ready,
    // WHERE KNOWABLE (AUTOPILOT-98): how many in-flight runs in the shared
    // async-run root belong to a DIFFERENT mode and were excluded from
    // `occupied`, plus how many counted runs carried no mode marker (counted
    // conservatively as ours). Absent when the backend cannot attribute — a
    // backend without a shared inventory has no foreign runs to name.
    ...(d.foreignRuns !== undefined ? { foreignRuns: d.foreignRuns } : {}),
    ...(d.unattributedRuns !== undefined ? { unattributedRuns: d.unattributedRuns } : {}),
  }));
}

/** One countable telemetry line when a harness lane REFUSES to spawn because
 *  the item (or the config default) named an unknown/invalid model level
 *  (KEY: AUTOPILOT-110) — the loud failure that replaces a silent fall-back to
 *  inherit. Never throws; the item is simply skipped (not spawned). */
function recordModelLevelRefusal(stateDir: string, lane: string, key: string, reason: string): void {
  appendTelemetry(stateDir, JSON.stringify({ t: new Date().toISOString(), type: "spawn-model-level-unknown", lane, key, reason }));
}

/**
 * A — fill free slots with auto-dispatchable approved items. Returns the
 * number dispatched. Never throws — a spawn failure stops the batch.
 *
 * `fleetTotalActive` is the caller's authoritative occupancy (the runner passes
 * `effectiveOccupiedSlots`' number for the pass); it is unioned with the
 * store's own live-item count here, so a caller with only a raw fleet number
 * still gets the fail-closed floor. When the lane is full with ready work, a
 * `dispatch.at-capacity {occupied, max, ready}` telemetry event is recorded
 * (KEY: AUTOPILOT-96) — the gate refuses loudly instead of returning [] in
 * silence, which read the same as an empty queue. `fleetForeignRuns` /
 * `fleetUnattributedRuns` (optional, KEY: AUTOPILOT-98) ride onto that event
 * where knowable: how many in-flight runs belong to ANOTHER mode (excluded
 * from occupancy) and how many counted runs had no mode marker. */
export async function autoDispatchEligible(
  stateDir: string,
  backend: SubagentBackend,
  maxSlots: number,
  fleetTotalActive?: number,
  fleetForeignRuns?: number,
  fleetUnattributedRuns?: number,
): Promise<Array<{ key: string; runId: string }>> {
  const store = loadStore(stateDir);
  if (!store) return [];
  // DEGRADED-WINDOW HOLD (AUTO-RECOVER-FAILS): while the provider is failing
  // spawns back-to-back, do NOT keep churning auto-dispatches into a dead
  // provider — the recovery engine emits the user tick that pauses retries.
  if (isProviderDegraded(stateDir)) return [];
  const eligible = Object.values(store.items)
    .filter((item) => isAutoDispatchable(item, store))
    .sort((a, b) => (a.updatedAt < b.updatedAt ? -1 : 1)); // oldest first
  if (eligible.length === 0) {
    atCapacityEpisode.delete(stateDir); // no ready work → an "idle queue", not a stall
    return [];
  }
  const cfg = loadAutopilotConfig(stateDir);
  // THE GATE'S OCCUPANCY (KEY: AUTOPILOT-96). effectiveOccupiedSlots is the ONE
  // derivation (max of the store's live items and the authoritative fleet
  // count, in RUNS not entries — AUTOPILOT-95). The runner already computed it
  // for this pass and passes it as the 4th argument, so recomputing here from
  // the same store + inputs yields the SAME number the tick is a projection of
  // — never a second view. The config's zombieGraceMinutes (not the default)
  // keeps this identical to the runner's/engine's derivation.
  const occupied = effectiveOccupiedSlots(stateDir, {
    backend,
    zombieGraceMinutes: cfg.zombieGraceMinutes,
    fleetTotalActive: fleetTotalActive ?? null,
  });
  const slots = Math.max(0, maxSlots - occupied);
  if (slots <= 0) {
    // VISIBLE STALL: ready work + a full lane is recorded as its own event
    // instead of the silent [] that was indistinguishable from an idle queue.
    recordAtCapacity(stateDir, { occupied, max: maxSlots, ready: eligible.length, foreignRuns: fleetForeignRuns, unattributedRuns: fleetUnattributedRuns });
    return [];
  }
  atCapacityEpisode.delete(stateDir); // capacity freed — the next stall logs afresh
  // AUTOPILOT-92: the worker lane SELECTS its agent through the registry —
  // before this, `workerAgents` was probe-checked but never used to pick the
  // spawn agent, so a read-only name could never be caught here. The resolved
  // name is passed explicitly; for the default it equals the backend's own
  // "worker" fallback, so the spawn is unchanged.
  const worker = resolveSpawnAgent("worker", cfg.workerAgents[0]);
  const dispatched: Array<{ key: string; runId: string }> = [];
  for (const item of eligible.slice(0, slots)) {
    try {
      // FINISHER-EVIDENCE (AUTOPILOT-34): a finisher-class item dispatched by
      // the harness gets a FRESH baseline for THIS run, captured before the
      // spawn. Without it the run has none (entering `active` clears the
      // previous one) and its verdict stands as the runtime reported it.
      const baseline = isFinisherItem(item) ? captureFinisherBaseline(item.cwd!, null, item.finisherSource ?? null) : null;
      // The item's recorded budget rides along — harness dispatches must not
      // silently fall back to the runtime default (the timeout-plumbing fix).
      // AUTOPILOT-53: an item with NO recorded budget gets an explicit,
      // visible default instead of a silent `undefined` passthrough.
      const dispatchBudget = resolveDispatchBudget(item.timeoutMs);
      // AUTOPILOT-60: name any prior attempt's journaled branch/SHA(s).
      // AUTOPILOT-79/80/110: param(none here) > item.model > param tier(none)
      // > item.tier > config.model > config default tier > inherit.
      const task = workerTask(item, undefined, stateDir);
      // AUTOPILOT-92: refuse a read-only agent given worker work AT THE
      // DECISION (countable telemetry, item named) instead of letting the
      // runtime refuse it at spawn with neither.
      const pairing = validateDispatchPairing({ stateDir, lane: "auto-dispatch", key: item.key, agent: worker, task, role: "worker" });
      if (!pairing.ok) continue; // refused — one mismatch line, NO spawn
      // AUTOPILOT-110: an unknown/invalid tier REFUSES the spawn (never a
      // silent inherit) with one telemetry line naming the item.
      const modelResolution = resolveSpawnModel({ item, config: cfg, role: "worker", repo: item.cwd });
      if (!modelResolution.ok) {
        recordModelLevelRefusal(stateDir, "auto-dispatch", item.key, modelResolution.error);
        continue;
      }
      const runId = await backend.spawn(task, {
        agent: worker.name,
        cwd: item.cwd!,
        timeoutMs: dispatchBudget.timeoutMs,
        model: modelResolution.model,
      });
      if (!runId) {
        // A falsy run id is a provider-layer rejection too — same hold signal.
        recordProviderFailure(stateDir);
        break;
      }
      recordProviderSuccess(stateDir);
      // Recorded per dispatch, through the safe path: the batch used to hold one
      // in-memory snapshot across every spawn and write it whole at the end,
      // erasing anything a tool or another lane wrote during those awaits.
      mutateStore(stateDir, (s) => {
        if (s.items[item.key]) {
          updateItem(s, item.key, {
            status: "active",
            runId,
            ...(dispatchBudget.appliedDefault ? { timeoutMs: dispatchBudget.timeoutMs } : {}),
            ...(baseline ? { finisherBaseline: { ...baseline, runId } } : {}),
          });
        }
      });
      dispatched.push({ key: item.key, runId });
      try {
        preserveRunWorktree({ stateDir, repo: item.cwd!, runId, key: item.key });
      } catch {
        // preservation never breaks dispatch
      }
    } catch {
      // bare-400 / infra at spawn: count the provider failure (degraded-window
      // hold) and stop the batch — the orchestrator/recovery handles it.
      recordProviderFailure(stateDir);
      break; // a spawn failure stops the batch — the orchestrator handles it
    }
  }
  return dispatched;
}

/** KEY: AUTOPILOT-64 — the review-FAIL retry path (core.ts) NULLS the item's
 *  runId and relies on THIS spawn to repopulate it (autoRedispatch is invoked
 *  fire-and-forget from runner.ts). A spawn failure here used to be swallowed
 *  completely: `return false` with no note, no telemetry — nothing let an
 *  operator or a later reconcile pass tell "the re-dispatch spawn itself
 *  failed" apart from "waiting for a live worker". The item was left
 *  status:active, runId:null — invisible to deadRunReconcile (needs a runId
 *  to check liveness against) and to zombieReconcile until the WHOLE fleet
 *  went idle (rare mid-session; this exact shape was observed live on
 *  AUTOPILOT-42: status=failed, runId=null, after zombieReconcile eventually
 *  caught it hours later while the actual worker had long since completed on
 *  a real branch).
 *
 *  This records the failure (note on the item + a telemetry line, the same
 *  two channels every other framework-level failure in this file uses — see
 *  recordProviderFailure/appendTelemetry above) WITHOUT touching status: the
 *  existing callers' contracts (this function still returns false; the
 *  fire-and-forget caller still does nothing further; dispatchPendingHumanRedispatch's
 *  own marker-restore-for-retry stays intact) are unchanged. The actual
 *  recovery is NOT a new bounded-retry mechanism — it reuses the one this
 *  repo already built for exactly this cause: zombieReconcile now checks a
 *  null-runId `active` item on EVERY sweep instead of gating it behind the
 *  whole fleet going idle (see core.ts), flips it to `failed`/`zombie` once
 *  its grace window elapses, and autoRecoverFails already knows how to retry
 *  (bounded) and escalate a `zombie`-cause failure. */
function recordRedispatchSpawnFailure(stateDir: string, key: string, reason: string): void {
  const note =
    `[auto-redispatch] ${new Date().toISOString()} — the re-dispatch spawn itself FAILED after a review FAIL (${reason}); ` +
    `this item stays active with NO run. It is not stuck forever: zombie reconciliation checks a null-run active item every ` +
    `sweep (not gated on the whole fleet going idle) and will flip it to failed once it has sat idle past the grace window, ` +
    `which auto-recovery then retries (bounded) and escalates if it keeps failing. If this repeats, check provider/backend health.`;
  try {
    mutateStore(stateDir, (s) => {
      const fresh = s.items[key];
      if (!fresh) return;
      updateItem(s, key, { notes: fresh.notes ? `${fresh.notes}\n\n${note}` : note });
    });
  } catch {
    // best-effort — recording the failure must never itself throw
  }
  appendTelemetry(stateDir, JSON.stringify({ t: new Date().toISOString(), type: "auto-redispatch-spawn-failed", key, reason }));
}

/**
 * B — re-dispatch a review-FAILED item with the reviewer's findings (up to
 * the cap). The caller decides the cap (the engine already flipped to active
 * when attempts < cap). Returns true when re-dispatched.
 */
export async function autoRedispatch(
  stateDir: string,
  backend: SubagentBackend,
  key: string,
  findings: string,
): Promise<boolean> {
  const store = loadStore(stateDir);
  const item = store?.items[key];
  if (!item || item.status !== "active") return false; // only the FAIL→active flip is re-dispatchable
  if (typeof item.cwd !== "string" || !item.cwd.trim()) return false; // no repo → orchestrator
  // DEPENDENCY GATE (KEY: AUTOPILOT-81): a re-dispatch (review-FAIL retry OR
  // human re-dispatch, both funnel through here) must not run while `after`
  // names an unfinished dependency. This is the ONE spawn choke point both
  // producers share, so both get the skip for free. The caller-side fallback
  // already restores the pending marker on a `false` return
  // (dispatchPendingHumanRedispatch) — the next sweep simply retries once the
  // dependency clears.
  if (store) {
    const unmet = unmetDependencies(item, store);
    if (unmet.length > 0) {
      recordRedispatchSpawnFailure(stateDir, key, `waiting on unmet dependencies: ${unmet.join(", ")}`);
      return false;
    }
  }
  try {
    // FINISHER-EVIDENCE (AUTOPILOT-34): same rule as every other lane — this
    // run is judged against a baseline captured for THIS run, never against the
    // previous run's (which would let a re-run that wrote nothing inherit the
    // first run's landing as its own evidence).
    const baseline = isFinisherItem(item) ? captureFinisherBaseline(item.cwd, null, item.finisherSource ?? null) : null;
    // AUTOPILOT-53: same no-null-budget guarantee as every other dispatch lane.
    const dispatchBudget = resolveDispatchBudget(item.timeoutMs);
    // AUTOPILOT-60: name any prior attempt's journaled branch/SHA(s).
    // AUTOPILOT-79/80/110: param(none here) > item.model > param tier(none)
    // > item.tier > config.model > config default tier > inherit.
    const cfg = loadAutopilotConfig(stateDir);
    const worker = resolveSpawnAgent("worker", cfg.workerAgents[0]);
    const task = workerTask(item, findings, stateDir);
    // AUTOPILOT-92: same decision-time pairing guard as every worker lane.
    const pairing = validateDispatchPairing({ stateDir, lane: "auto-redispatch", key, agent: worker, task, role: "worker" });
    if (!pairing.ok) {
      recordRedispatchSpawnFailure(stateDir, key, pairing.reason);
      return false;
    }
    // AUTOPILOT-110: an unknown/invalid tier refuses the re-dispatch (never a
    // silent inherit), recorded through the same visible channel.
    const modelResolution = resolveSpawnModel({ item, config: cfg, role: "worker", repo: item.cwd });
    if (!modelResolution.ok) {
      recordRedispatchSpawnFailure(stateDir, key, modelResolution.error);
      return false;
    }
    const runId = await backend.spawn(task, {
      agent: worker.name,
      cwd: item.cwd,
      timeoutMs: dispatchBudget.timeoutMs,
      model: modelResolution.model,
    });
    if (!runId) {
      // A falsy run id is a provider-layer rejection — same shape as the catch
      // below, just without a thrown error to describe.
      recordRedispatchSpawnFailure(stateDir, key, "provider returned no run id");
      return false;
    }
    mutateStore(stateDir, (s) => {
      if (s.items[key])
        updateItem(s, key, {
          runId,
          // AUTOPILOT-85: a stale item.runWorktree (from the FAILed attempt
          // this re-dispatch supersedes) must never ride onto the new runId —
          // shipping.ts's resolveWorkTip source (c) trusts it only when
          // runWorktree.runId still matches the item's run identity, but the
          // safer, visible fix is to never let it go stale in the first place.
          // Cleared in the SAME patch that assigns the new runId; a later sweep
          // repopulates it once it sees a LIVE worktree for THIS run
          // (recordActiveWorktrees).
          runWorktree: null,
          ...(dispatchBudget.appliedDefault ? { timeoutMs: dispatchBudget.timeoutMs } : {}),
          ...(baseline ? { finisherBaseline: { ...baseline, runId } } : {}),
        });
    });
    try {
      preserveRunWorktree({ stateDir, repo: item.cwd, runId, key });
    } catch {
      // preservation never breaks re-dispatch
    }
    return true;
  } catch (e) {
    recordRedispatchSpawnFailure(stateDir, key, e instanceof Error ? e.message : String(e));
    return false; // the orchestrator re-dispatches manually
  }
}

/**
 * B2 — HUMAN RE-DISPATCH (KEY: AUTOPILOT-59): the harness consumer TWO
 * producers hand off to — the decision panel's "redispatch" action (panels.ts,
 * a human-operated host UI) and queue_update's `findings` param
 * (queue-ops.ts, KEY: AUTOPILOT-69 — the tool surface the orchestrator itself
 * calls, which has no panel to click). Both move the item human-review →
 * active and record `pendingRedispatchFindings` in ONE mutateStore write (the
 * single point of truth against double-dispatch: once that write lands the
 * item is active, so a late/duplicate re-dispatch call for the same key is
 * refused at ITS producer — this pass never sees it twice). Each
 * pending item is CLAIMED (the marker cleared) before the spawn, so two
 * overlapping sweeps racing this same pass cannot both dispatch it — the
 * loser's claim fails (the marker it read is already gone) and it skips.
 *
 * Spawns through path B (autoRedispatch) — the SAME findings-carrying
 * dispatcher a review FAIL uses (workerTask(item, findings), staleness
 * warning included). No cap consumed: `attempts` is never touched here — a
 * human's correction is not a review failure, and the cap exists to stop
 * hopeless automatic retries, not a human's deliberate redo.
 *
 * The redo re-enters the NORMAL lane (active → ai-review → human-review) by
 * construction: this only re-spawns a worker exactly like any other dispatch;
 * the existing completion routing (onCompletion → autoReview → PASS/FAIL/NO-CHANGE)
 * decides what happens next, same as for any other active item.
 *
 * FAILS LOUDLY when the item's declared repo no longer exists (deleted,
 * moved, or otherwise gone since the item entered human-review): the item is
 * flipped to `failed` with a clear note instead of silently spawning against
 * a missing path, and instead of silently stalling forever in `active` with
 * no runId and no operator-visible record of why.
 */
export async function dispatchPendingHumanRedispatch(
  stateDir: string,
  backend: SubagentBackend,
): Promise<Array<{ key: string; runId: string }>> {
  if (isProviderDegraded(stateDir)) return [];
  const store = loadStore(stateDir);
  if (!store) return [];
  const pending = Object.values(store.items).filter(
    (i) => i.status === "active" && !i.runId && typeof i.pendingRedispatchFindings === "string" && i.pendingRedispatchFindings.trim(),
  );
  const dispatched: Array<{ key: string; runId: string }> = [];
  for (const item of pending) {
    const findings = item.pendingRedispatchFindings!.trim();
    const cwd = item.cwd;
    // CLAIM the marker before spawning: a concurrent sweep that reads the
    // store between this claim and the spawn below sees the marker already
    // gone (still-null, unchanged) and skips it on its own pass.
    const claimed = mutateStore(stateDir, (s) => {
      const fresh = s.items[item.key];
      if (!fresh || fresh.status !== "active" || fresh.runId || fresh.pendingRedispatchFindings !== item.pendingRedispatchFindings) return false;
      updateItem(s, item.key, { pendingRedispatchFindings: null });
      return true;
    });
    if (!claimed) continue; // another sweep already claimed (or resolved) it
    if (!cwd || !existsSync(cwd)) {
      mutateStore(stateDir, (s) => {
        const fresh = s.items[item.key];
        if (!fresh) return;
        const note = `[human re-dispatch] cannot re-dispatch — repo '${cwd ?? "(none)"}' no longer exists (deleted/moved since the item entered human-review). Re-approve with a valid repo, or drop the item.`;
        updateItem(s, item.key, {
          status: "failed",
          failCause: "spawn",
          notes: fresh.notes ? `${fresh.notes}\n\n${note}` : note,
        });
      });
      continue;
    }
    const ok = await autoRedispatch(stateDir, backend, item.key, findings);
    if (ok) {
      const runId = loadStore(stateDir)?.items[item.key]?.runId;
      if (runId) dispatched.push({ key: item.key, runId });
    } else {
      // The spawn failed (provider rejection / infra) — restore the marker so
      // the next sweep retries; the claim above must never permanently strand
      // an item without ever spawning it.
      mutateStore(stateDir, (s) => {
        const fresh = s.items[item.key];
        if (fresh && fresh.status === "active" && !fresh.runId) updateItem(s, item.key, { pendingRedispatchFindings: findings });
      });
    }
  }
  return dispatched;
}

/**
 * SPEC PASS (KEY: AUTOPILOT-36) — a bounded, READ-ONLY reconnaissance task
 * against ONE under-specified proposal. The worker reads the code and writes
 * a COMPLETE scope back; it does NOT implement anything, does NOT edit any
 * files, and has NO authority to approve — the item stays a proposal no
 * matter what it reports (approval remains human-only). Output is parsed
 * strictly (see parseSpecPassScope, verdict.ts): only the text between the
 * SPEC_PASS_SCOPE_START/END markers is ever written back.
 */
export function specPassTask(item: QueueItem): string {
  return [
    `KEY: ${item.key} — SPEC PASS (reconnaissance only, this is NOT an implementation task)`,
    `This proposal is under-specified. Your job: READ the code at ${item.cwd} and turn the rough note below into a complete, dispatchable scope. You do NOT implement anything, you do NOT edit any files, and you have NO authority to approve this item — it stays a proposal regardless of what you find; a human decides whether to approve it.`,
    "",
    "## Current (thin) scope",
    item.scope.trim() || "(empty — a bare title/idea only)",
    "",
    "## What to produce",
    "Read the relevant code and produce a scope that follows this exact six-heading shape:",
    SCOPE_SKELETON,
    "EVIDENCE must cite real file:line references you actually read — never invent a location. JUDGMENT CALLS must name decisions a worker implementing this must NOT silently invert (e.g. fail-open vs fail-closed, which field wins a conflict) — if you found none, say so explicitly rather than omitting the heading.",
    "",
    "## Output contract (REQUIRED — the harness parses this mechanically)",
    "Write your completed scope between these two EXACT marker lines, and nothing else between them:",
    SPEC_PASS_SCOPE_START,
    "<the completed scope, following the six-heading shape above>",
    SPEC_PASS_SCOPE_END,
    "Anything you write outside the markers (your reasoning, questions, caveats) is fine and is read by the human — but the markers must wrap ONLY the scope text itself, verbatim as you want it stored.",
  ].join("\n");
}

/** Reconnaissance is bounded — this is a read against the repo, not an
 *  implementation run, so it gets a SHORT budget of its own rather than
 *  inheriting the item's real dispatch budget (`item.timeoutMs` describes
 *  the WORK, not a spec pass against it). Not persisted onto the item —
 *  overwriting `timeoutMs` here would silently change the budget the item's
 *  real dispatch runs under later. */
const SPEC_PASS_TIMEOUT_MS = 15 * 60 * 1000;

/**
 * SPEC PASS DISPATCH (KEY: AUTOPILOT-36) — the harness consumer for the
 * decision panel's `spec-pass` action (panels.ts sets `pendingSpecPass` on a
 * PROPOSAL, never touching status — approval stays human-only). Mirrors
 * dispatchPendingHumanRedispatch's claim-before-spawn shape: the marker is
 * cleared before the spawn so two overlapping sweeps cannot both dispatch the
 * same pass, and a spawn failure restores the marker so the next sweep
 * retries instead of silently dropping the request.
 *
 * READ-ONLY (`worktree: false`, exactly like autoReview's reviewer dispatch)
 * — a spec pass never touches the repo, so it needs no isolated worktree.
 * Its completion is attributed by `specPassRunId` and consumed by
 * core.ts's handleAsyncComplete (parseSpecPassScope), which writes the
 * completed scope back through the SAME validated mutateStore/updateItem
 * path every other queue write uses — never a bare saveStore.
 */
export async function dispatchPendingSpecPass(
  stateDir: string,
  backend: SubagentBackend,
): Promise<Array<{ key: string; runId: string }>> {
  if (isProviderDegraded(stateDir)) return [];
  const store = loadStore(stateDir);
  if (!store) return [];
  const pending = Object.values(store.items).filter((i) => i.status === "proposal" && i.pendingSpecPass === true && !i.specPassRunId);
  const cfg = loadAutopilotConfig(stateDir);
  // AUTOPILOT-92: the recon lane resolves its agent through the registry too
  // (`workerAgents`), so a misconfigured read-only name is caught here rather
  // than silently spawning on the backend's built-in default. A recon task
  // declares itself read-only, so a mutable agent passes.
  const worker = resolveSpawnAgent("worker", cfg.workerAgents[0]);
  const dispatched: Array<{ key: string; runId: string }> = [];
  for (const item of pending) {
    // DEPENDENCY GATE (KEY: AUTOPILOT-81): skip a spec pass on a proposal that
    // still names an unfinished `after` dependency — the pending marker stays
    // set, so a later sweep (once the dependency clears) simply picks it up.
    if (unmetDependencies(item, store).length > 0) continue;
    if (!item.cwd || !existsSync(item.cwd)) {
      mutateStore(stateDir, (s) => {
        const fresh = s.items[item.key];
        if (!fresh || fresh.status !== "proposal") return;
        const note = `[spec pass] cannot run — repo '${item.cwd ?? "(none)"}' is missing/no longer exists. Set a valid repo (refine) and try again.`;
        updateItem(s, item.key, { pendingSpecPass: false, notes: fresh.notes ? `${fresh.notes}\n\n${note}` : note });
      });
      continue;
    }
    // AUTOPILOT-92: validate the pairing BEFORE the claim — a refusal must
    // leave `pendingSpecPass` set so the next sweep can retry, never silently
    // drop the request behind a cleared marker.
    const task = specPassTask(item);
    const pairing = validateDispatchPairing({ stateDir, lane: "spec-pass", key: item.key, agent: worker, task, role: "worker" });
    if (!pairing.ok) continue; // refused — one mismatch line, NO spawn
    // AUTOPILOT-110: resolve the level BEFORE the claim — an unknown tier must
    // not clear the marker (the next sweep retries) and must never silently
    // spawn the inherited model.
    const modelResolution = resolveSpawnModel({ item, config: cfg, role: "worker", repo: item.cwd });
    if (!modelResolution.ok) {
      recordModelLevelRefusal(stateDir, "spec-pass", item.key, modelResolution.error);
      continue;
    }
    // CLAIM before spawning — same double-dispatch guard as AUTOPILOT-59.
    const claimed = mutateStore(stateDir, (s) => {
      const fresh = s.items[item.key];
      if (!fresh || fresh.status !== "proposal" || fresh.pendingSpecPass !== true || fresh.specPassRunId) return false;
      updateItem(s, item.key, { pendingSpecPass: false });
      return true;
    });
    if (!claimed) continue; // another sweep already claimed (or resolved) it
    try {
      const runId = await backend.spawn(task, {
        agent: worker.name,
        worktree: false, // read-only recon — no isolation needed, mirrors autoReview
        cwd: item.cwd,
        timeoutMs: SPEC_PASS_TIMEOUT_MS,
        // AUTOPILOT-79/80/110: param(none here) > item.model > item.tier >
        // config.model > config default tier > inherit.
        model: modelResolution.model,
      });
      if (!runId) {
        recordProviderFailure(stateDir);
        mutateStore(stateDir, (s) => {
          const fresh = s.items[item.key];
          if (fresh && fresh.status === "proposal" && !fresh.specPassRunId) updateItem(s, item.key, { pendingSpecPass: true });
        });
        continue;
      }
      recordProviderSuccess(stateDir);
      mutateStore(stateDir, (s) => {
        if (s.items[item.key]?.status === "proposal") updateItem(s, item.key, { specPassRunId: runId });
      });
      dispatched.push({ key: item.key, runId });
    } catch {
      // bare-400 / infra at spawn — count the provider failure (degraded-window
      // hold) and restore the marker so the next sweep retries.
      recordProviderFailure(stateDir);
      mutateStore(stateDir, (s) => {
        const fresh = s.items[item.key];
        if (fresh && fresh.status === "proposal" && !fresh.specPassRunId) updateItem(s, item.key, { pendingSpecPass: true });
      });
    }
  }
  return dispatched;
}

/**
 * C — the review task for an item: the SAME fields the dispatch used (KEY +
 * scope), plus where the work landed (cwd), the repo-scope restriction (KEY:
 * AUTOPILOT-72 — unconditional here, not left to item.scope to repeat), and
 * the locate-it-yourself rule. The reviewer must read the actual work
 * product, not the worker's summary.
 */
export function reviewTask(item: QueueItem): string {
  return [
    `KEY: ${item.key}`,
    // KEY: AUTOPILOT-92 — declare the task KIND first, before the
    // implementation-shaped scope below. The runtime's task-intent classifier
    // reads the whole task; without this line the scope's write imperatives
    // (e.g. ACTUATOR-51's literal "must apply") made the review read as
    // implementation work, so the read-only reviewer was refused AT SPAWN.
    // "review only" is the phrase that classifier keys on; it is also the
    // marker classifyTaskKind (dispatch-pairing.ts) reads at the decision.
    "Review only — this is a review task, NOT an implementation task. Do NOT edit, write, or modify any files; inspect the work and report findings.",
    item.scope.trim(),
    "",
    "## Review the completed work",
    `Verify the work product in the repo at: ${item.cwd}`,
    // KEY: AUTOPILOT-72 — auto-dispatched reviewers carry NO repo-scope
    // restriction unless the orchestrator's own scope text happens to repeat
    // it by hand. Live-observed failure: AUTOPILOT-65's auto-dispatched
    // reviewer grepped `/Users/marc.anders` (the whole home directory, not
    // the repo) and hung until the 300000ms tool timeout killed the run,
    // leaving the item stuck on a dead reviewerRunId until dead-run
    // reconciliation caught it. This line is unconditional on every
    // reviewTask() build — it does not depend on the orchestrator having
    // written it into item.scope.
    `Search scope: THIS REPO ONLY (${item.cwd} above). Do NOT grep, read, or search outside it — a reviewer that greps ~ or a large ancestor directory can hang or time out (observed: AUTOPILOT-65's reviewer died at a 300s grep timeout searching /Users/marc.anders).`,
    "Locate the work yourself: the worker ran in an isolated worktree and pushed to ITS OWN parallel branch (pi-parallel-<runid>-0) — the commits may be there, unmerged. Check `git log --all -- <path>` + `git branch -a -r` + `git show` on those branches BEFORE concluding anything is missing; a FAIL must be based on the work being absent everywhere, never just on main. Read the actual diff/files — do NOT trust the worker's summary.",
    "Judge the CURRENT tree as it exists NOW: between worker completion and this review, the orchestrator may have recovered/merged/patched work (observed failure class: a reviewer read pre-recovery refs and FAILed on work that was already on main). Re-check refs/logs at review time; never conclude from a snapshot you formed earlier in your run.",
    "",
    "The FIRST line of your response MUST be a verdict line: `Verdict: PASS`, `Verdict: FAIL`, or `Verdict: NO-CHANGE` (the LAST only when the correct outcome was to make NO change — a refusal/withdrawal, no deliverable; it is terminal and never re-dispatched). Ordinary formatting around it is tolerated (surrounding markdown emphasis/backticks, a leading heading/list marker, indentation, case variation), but a verdict quoted inside prose is NOT read — keep the word and value adjacent. If FAIL, list each finding as an actionable item.",
    "",
    REVIEW_CONTRACT_TEXT,
    "",
    acceptanceEvidenceStatus(item.acceptanceEvidence),
  ].join("\n");
}

/**
 * C — auto-dispatch the reviewer for a completed item (active → ai-review).
 * Mirrors queueReview but builds the task from the item (no orchestrator
 * round-trip). Returns true when a reviewer was spawned.
 */
export async function autoReview(
  stateDir: string,
  backend: SubagentBackend,
  reviewerAgent: string,
  key: string,
): Promise<string | null> {
  try {
    const store = loadStore(stateDir);
    const item = store?.items[key];
    if (!item || item.status !== "ai-review") return null;
    // STALE-REF GUARD: a live reviewer is never duplicated, but a DEAD one must
    // not wedge the lane. This skip is SILENT (no error surface), so a stale id
    // used to stop auto-review for this item forever. reviewerRunAlive FAILS
    // OPEN (see run-liveness.ts): undeterminable → not alive → re-dispatch. A
    // duplicate read-only reviewer costs tokens; a permanent silent skip costs
    // a manual bypass, which breaks verdict attribution.
    if (item.reviewerRunId && reviewerRunAlive(backend, item.reviewerRunId)) return null;
    if (!(item.scope ?? "").trim() || !item.cwd) return null; // safety net — the approval gate guarantees these
    // AUTOPILOT-53: same no-null-budget guarantee as every other dispatch
    // lane — an item with no recorded budget gets an explicit default,
    // persisted onto the item in the same mutateStore call that records the
    // reviewer run (docs/queue-model.md's every-spawn-site guarantee).
    const dispatchBudget = resolveDispatchBudget(item.timeoutMs);
    const cfg = loadAutopilotConfig(stateDir);
    // AUTOPILOT-92: resolve the reviewer through the registry and validate the
    // pairing BEFORE the spawn. A review task that is not recognisable as a
    // review (rule 2), or any implementation task paired with this read-only
    // agent (rule 1), is refused here with the item key + one mismatch line
    // instead of the runtime's nameless spawn-time refusal.
    const reviewer = resolveSpawnAgent("review", reviewerAgent);
    const task = reviewTask(item);
    const pairing = validateDispatchPairing({ stateDir, lane: "auto-review", key, agent: reviewer, task, role: "review" });
    if (!pairing.ok) return null; // refused at the DECISION — telemetry already emitted
    // AUTOPILOT-79/80/110: param(none here) > item.model > param tier(none) >
    // item.tier > config.model > config default tier (models.reviewDefault for
    // a REVIEWER) > inherit; an unknown tier refuses the spawn loudly.
    const modelResolution = resolveSpawnModel({ item, config: cfg, role: "review", repo: item.cwd });
    if (!modelResolution.ok) {
      recordModelLevelRefusal(stateDir, "auto-review", key, modelResolution.error);
      return null;
    }
    const runId = await backend.spawn(task, {
      agent: reviewer.name,
      worktree: false, // reviewers are read-only — no worktree
      cwd: item.cwd,
      timeoutMs: dispatchBudget.timeoutMs, // the SAME fields the dispatch used — budget included
      model: modelResolution.model,
    });
    if (!runId) return null;
    mutateStore(stateDir, (s) => {
      if (s.items[key])
        updateItem(s, key, {
          reviewerRunId: runId,
          ...(dispatchBudget.appliedDefault ? { timeoutMs: dispatchBudget.timeoutMs } : {}),
        });
    });
    return runId;
  } catch {
    return null; // the review tick nudges the orchestrator
  }
}