// auto-recovery.ts — AUTOMATIC RECOVERY of FAILED items (the failed-status
// lane), bounded + transparent.
//
// Observed failure mode: children died at the provider layer (bare 400 at
// spawn, empty api_error mid-run, hang-then-die); every one needed a manual
// re-dispatch, and re-dispatches during the degraded window churned. This
// engine turns that manual loop into a DETERMINISTIC policy driven from the
// failed item's recorded `failCause` (BUDGET-GOVERNANCE), with a short backoff,
// per-item bounds, and a provider-health hold:
//
//   budget-capped → re-dispatch with a BIGGER budget (timeoutMs ×1.5, max 3h)
//   verdict       → ONE recovery re-dispatch carrying the P5 recovery context
//   zombie        → ONE recovery re-dispatch carrying the P5 recovery context
//   spawn/infra   → retry up to 2× with backoff, then ESCALATE
//
// BOUNDS: every attempt (including a spawn the provider REJECTED) increments
// the item's `recoveries`; once the per-cause cap (min per-cause, global
// default 2) is spent the item STAYS failed and gets a one-time escalation
// tick — never an unbounded retry loop. DEGRADED-WINDOW HOLD: consecutive
// provider failures (spawn throws / hang-then-die zombies) pause retries and
// emit a user tick instead of churning (recovery-state.ts owns the counter).
//
// RECOVERED ITEMS CARRY THEIR BRANCH (AUTO-SHIP-ON-DONE integration): the
// re-dispatch preserves the new run's parallel-branch tip exactly like every
// other dispatch lane, and the item still flows active → ai-review →
// human-review. Recovery NEVER short-circuits the approval gate and NEVER
// touches main — shipping stays the post-`done` merge-finisher lane.
//
// FINISHER-CLASS ITEMS ARE NEVER AUTO-RE-DISPATCHED (KEY: AUTOPILOT-46). The
// landed-evidence check (AUTOPILOT-34, phase 0) can only skip landings it can
// SEE — an ancestor merge/fast-forward, or a patch-equivalent copy. A
// conflict-resolved cherry-pick and a squash of a multi-commit source produce
// NO evidence by design (patch-id equality cannot settle a rewritten patch),
// and for those a recovery re-dispatch would RE-RUN A MERGE THAT ALREADY
// LANDED — automatically, before the operator can close the item out. Since a
// finisher re-run is the single action that can duplicate a landing whether or
// not there is evidence, the whole class is HELD and handed to the human
// instead (same posture as phase 0), with a visible one-time escalation. The
// cost is stated: a genuinely failed finisher gets NO automatic retry — it is
// announced and waits for a deliberate re-dispatch.
//
// THE HOLD IS THE GENERAL CASE (KEY: AUTOPILOT-49). The finisher class is one
// reason an item must not be re-dispatched automatically; the other is a
// failure retrying cannot fix — an interactive prompt no headless run can
// answer, a missing credential, a decision only the user can make. That state
// was previously INEXPRESSIBLE: `blocked` was unreachable from `failed`, so an
// item parked in prose stayed `failed`, and this engine re-dispatched it 30s
// later (observed live: a device-code prompt re-entered, twice, against a
// shared file outside git). `failed → blocked` is now a legal PARK, and the
// hold below covers BOTH reasons through one seam (`recoveryHold`), with the
// same posture: never re-dispatched, surfaced ONCE, and released deliberately
// (blocked → approved, which re-arms recovery — see queue-store's updateItem).

import { git } from "./git.ts";
import { loadStore, mutateStore, updateItem, isFailCause, unmetDependencies, type BlockerReason, type DispatchClass, type QueueItem, type FailCause, type UpdatePatch } from "../queue-store.ts";
import type { SubagentBackend } from "../backends/types.ts";
import { workerTask } from "./auto-dispatch.ts";
import { loadAutopilotConfig, appendTelemetry } from "../config.ts";
import { resolveSpawnModel } from "../spawn-model.ts";
import { appendOverride, finisherLandedEvidence, isFinisherItem, landedOverride } from "../finisher-evidence.ts";
import { preserveRunWorktree, reviewPointersFor } from "./worktree-preservation.ts";
import { resolveDispatchBudget } from "./run-budget.ts";
import { formatDurationMs } from "../duration.ts";
import {
  loadRecoveryState,
  saveRecoveryState,
  SPAWN_FAILURE_THRESHOLD,
  DEGRADED_COOLDOWN_MS,
} from "./recovery-state.ts";

/** Global per-item auto-recovery cap (attempts, including rejected spawns). */
export const MAX_RECOVERIES = 2;
/** Short backoff between a failure and its recovery attempt (recovery timer). */
export const RECOVERY_BACKOFF_MS = 30_000;
/** budget-capped recovery: grow the wall-clock budget by this factor … */
export const BUDGET_MULTIPLIER = 1.5;
/** … capped at 3h (the outer bound the operator asked for). */
export const BUDGET_MAX_MS = 3 * 60 * 60_000;
/** Recovery attempt cap per cause (clamped to MAX_RECOVERIES). */
const CAUSE_MAX_ATTEMPTS: Record<FailCause, number> = {
  "budget-capped": MAX_RECOVERIES,
  verdict: 1, // one recovery re-dispatch, then escalate
  zombie: 1, // one recovery re-dispatch, then escalate
  spawn: MAX_RECOVERIES, // provider retries (2×) then escalate
};

export interface RecoveryPlan {
  cause: FailCause;
  /** Attempts allowed for this cause (already clamped to the global cap). */
  maxAttempts: number;
  /** The wall-clock budget the re-dispatch runs under (null = runtime default).
   *  Honoured as requested (AUTOPILOT-53) — only the outer `BUDGET_MAX_MS`
   *  bound can hold it below the full ×1.5 escalation. */
  budgetMs: number | null;
  /** budget-capped only: did the budget ACTUALLY grow over the prior run's?
   *  False only when the ×1.5 escalation was absorbed by the outer
   *  `BUDGET_MAX_MS` bound (the prior run was already at/above it). */
  budgetGrew: boolean;
  /** The task block appended after KEY + scope (the P5 recovery context). */
  context: string;
}

/** The deterministic plan for one failed item: cap, budget, task context.
 *
 *  AUTOPILOT-53: the ×1.5 escalation is NOT clamped to a runtime ceiling — an
 *  explicit `timeoutMs` is honoured end to end, so a bigger recorded budget
 *  really does buy the worker more time. The only bound left is the outer
 *  cap the operator asked for (`BUDGET_MAX_MS`, 3h). `budgetGrew` is false
 *  only in the rare case that outer cap was already reached. */
export function recoveryPlan(item: QueueItem, cause: FailCause, globalMax = MAX_RECOVERIES): RecoveryPlan {
  const maxAttempts = Math.max(1, Math.min(CAUSE_MAX_ATTEMPTS[cause] ?? globalMax, globalMax));
  if (cause === "budget-capped") {
    const base = item.timeoutMs && item.timeoutMs > 0 ? item.timeoutMs : resolveDispatchBudget(item.timeoutMs).timeoutMs;
    const budgetMs = Math.min(Math.round(base * BUDGET_MULTIPLIER), BUDGET_MAX_MS);
    // Compared against `base`, which IS the budget the prior run ran under —
    // including the explicit default a dispatch applies when nothing was
    // recorded. Comparing against a null-as-0 would call the very first
    // escalation "bigger" when it lands on the same wall the unbudgeted run
    // already hit.
    const budgetGrew = budgetMs > base;
    return { cause, maxAttempts, budgetMs, budgetGrew, context: recoveryContext(cause, item.dispatchClass, budgetGrew) };
  }
  return { cause, maxAttempts, budgetMs: item.timeoutMs ?? null, budgetGrew: false, context: recoveryContext(cause, item.dispatchClass) };
}

/** The P5 recovery context appended to a re-dispatch task. P5 is the
 *  fleet-health rule (prompts/orchestrate.md): CHECK RECOVERABILITY BEFORE A
 *  FULL REDO — the prior attempt's commits may live on its parallel branch.
 *
 *  The main-write rule is DISPATCH-CLASS AWARE (AUTOPILOT-46): "NEVER commit or
 *  merge to main" is the rule for a worker and a contradiction for a finisher,
 *  whose entire scope is to land an approved branch in the target checkout —
 *  telling it not to merge would instruct it to fail. The finisher wording is
 *  the class's real rule instead: land ONLY the declared source, and check
 *  first whether the previous attempt already landed it. (autoRecoverFails
 *  HOLDS finisher-class items rather than re-dispatching them, so today this
 *  branch is reached only through this exported function; it exists so the
 *  text stays correct for the class if that hold is ever narrowed.) */
export function recoveryContext(cause: FailCause, dispatchClass: DispatchClass = "worker", budgetGrew = false): string {
  const lines = [
    "## Recovery re-dispatch (P5)",
    `This item previously FAILED (cause: ${cause}); this is an automated recovery re-dispatch.`,
    ...(cause === "budget-capped"
      ? [
          // AUTOPILOT-47 A: only claim a bigger budget when there IS one. The
          // runtime ceiling normally absorbs the escalation, and telling a
          // worker it has more time than it does is the exact failure this item
          // exists to end — it plans work that cannot finish.
          budgetGrew
            ? "The prior run was CUT OFF by its wall-clock budget mid-task — this re-dispatch runs with a BIGGER budget (see the launch). Land a committable increment EARLY (commit on your branch before long test/fix cycles), then continue."
            : `The prior run was CUT OFF by its wall-clock budget mid-task, and this re-dispatch gets THE SAME WALL — it is already at the ${formatDurationMs(BUDGET_MAX_MS)} recovery ceiling this framework allows, so the ×1.5 escalation has nothing left to add. Do NOT plan a run longer than that. COMMIT EARLY AND OFTEN on your branch: a committed partial survives the cap and the next attempt resumes from it, an uncommitted worktree does not.`,
        ]
      : []),
    ...(cause === "spawn"
      ? [
          "The provider/infra layer rejected the previous run AT SPAWN (bare 400 / empty api_error) — no work was produced. If this attempt fails the same way, the harness escalates instead of retrying forever.",
        ]
      : []),
    "P5 — CHECK RECOVERABILITY BEFORE REDOING ANYTHING: the prior attempt ran in an isolated worktree and pushed to its own parallel branch (pi-parallel-<runid>-0). Its commits may exist there, unmerged. Check `git log --all`, `git branch -a`, and the run's session log; if the work is present, RESTORE/reconcile it ON THE BRANCH instead of re-implementing it.",
    dispatchClass === "finisher"
      ? "This item is FINISHER-CLASS: landing the declared source in the target checkout IS its scope, so the usual 'never merge to main' rule does not apply to that landing. CHECK FIRST whether the previous attempt already landed it (`git log`, `git cherry <target> <source>`) — including as a conflict-resolved or squashed commit, which no automatic check can recognise; if it is already in, land NOTHING and report that. Land ONLY the declared source; any other work still goes on your branch."
      : "NEVER commit or merge to main — main is touched ONLY by the human-approved merge-finisher AFTER review. Commit early on your branch; this item still goes through the normal AI review + human approval.",
  ];
  return lines.join("\n");
}

// PRIOR-RUN WORKTREE STATE (KEY: AUTOPILOT-97) ------------------------------
//
// A recovery re-dispatch must not send a fresh worker to re-implement a
// deliverable that already exists. The old brief named a prior branch/tip only
// when the journal happened to carry one, and said nothing otherwise — so a
// capped run that COMMITTED first was indistinguishable from a run that
// produced nothing, and the redo redid the work. The runtime's own recovery
// summary reports only the UNCOMMITTED field ("changed tracked files: none"),
// which is exactly the conflation this item exists to end; the fix here is
// OURS to make — state the COMMITTED state as well, and make "no uncommitted
// changes" and "no work produced" two different statements.

/** What the prior run left behind, as far as the framework can HONESTLY say.
 *  `uncommittedReadable` separates "the tree is clean" (a fact) from "the tree
 *  could not be read" (not a fact) — an unreadable tree must never be reported
 *  as clean, and never as empty. */
export interface PriorRunState {
  /** The run's parallel branch (null when none could be resolved). */
  branch: string | null;
  /** The branch tip captured for this run (null when unresolved). */
  tipSha: string | null;
  /** The run's base — the merge-base of the branch with main (null when
   *  unresolved). `tipSha !== baseSha` means the run COMMITTED. */
  baseSha: string | null;
  /** Files with uncommitted changes in the run's worktree. */
  uncommittedFiles: string[];
  /** true when the worktree status WAS readable, so an empty file list is a
   *  real "clean", not a failed read. */
  uncommittedReadable: boolean;
}

/** The four honest readings of a prior run. UNKNOWN is "the framework cannot
 *  tell", and is NEVER silently folded into `empty`: a false "produced nothing"
 *  causes exactly the redo this item exists to prevent, so the fail direction
 *  is toward telling the worker to verify. */
export type PriorRunOutcome = "committed" | "uncommitted" | "empty" | "unknown";

/** Classify a resolved prior-run state. `empty` is a POSITIVE claim — no
 *  committed tip exists for this run AND the worktree is clean AND readable.
 *  A tip with NO commits beyond its base is deliberately NOT empty: it is
 *  either a fresh branch at a main commit (nothing committed) or a branch whose
 *  commits are already folded into main, and nothing on the item tells them
 *  apart — so it is `unknown`, never a false "produced nothing". */
export function classifyPriorRun(state: PriorRunState): PriorRunOutcome {
  if (state.uncommittedReadable && state.uncommittedFiles.length > 0) return "uncommitted";
  if (state.tipSha && state.baseSha && state.tipSha !== state.baseSha) return "committed";
  if (state.tipSha) return "unknown";
  if (state.uncommittedReadable) return "empty";
  return "unknown";
}

/** The human-readable prior-run brief appended to a recovery re-dispatch. The
 *  four outcomes must read DIFFERENTLY — in particular (committed work exists)
 *  must never read like (nothing produced), which was the bug. */
export function priorRunBrief(state: PriorRunState): string {
  const where =
    state.branch && state.tipSha
      ? `branch \`${state.branch}\` at tip \`${state.tipSha.slice(0, 8)}\`` +
        (state.baseSha ? ` (base \`${state.baseSha.slice(0, 8)}\`)` : "")
      : "its branch/tip could not be resolved";
  switch (classifyPriorRun(state)) {
    case "committed":
      return [
        "## Prior run state — COMMITTED WORK EXISTS (confirm it; do NOT re-implement)",
        `The prior run COMMITTED work on ${where}; the tip is AHEAD of the base, so the deliverable may already exist on that branch.`,
        `CONFIRM before writing anything: read the diff (\`git diff ${state.baseSha?.slice(0, 8) ?? "<base>"}..${state.tipSha?.slice(0, 8) ?? "<tip>"}\` / \`git show ${state.tipSha?.slice(0, 8) ?? "<tip>"}\`), re-run the gates, and report the result. Do NOT re-implement work that already exists.`,
      ].join("\n");
    case "uncommitted":
      return [
        "## Prior run state — UNCOMMITTED CHANGES (judge them; do not blindly redo)",
        `The prior run left UNCOMMITTED changes${state.branch && state.tipSha ? ` on ${where}` : ""}: ${state.uncommittedFiles.join(", ")}.`,
        "Judge whether they are PART OF THE FIX or scratch: if they belong to the fix, keep and COMMIT them on your branch; if they are scratch, discard them deliberately. Do NOT re-implement the deliverable from scratch while these are unexamined.",
      ].join("\n");
    case "empty":
      return [
        "## Prior run state — NOTHING PRODUCED (fresh start expected)",
        "The prior run produced NOTHING: no commits ahead of its base and no uncommitted changes in its worktree. A fresh start is expected — there is no prior work to recover.",
      ].join("\n");
    default:
      return [
        "## Prior run state — UNKNOWN (do not assume it produced nothing)",
        "The framework could NOT determine what the prior run left behind (no readable branch/tip/base and/or an unreadable worktree). This is NOT the same as producing nothing — check `git log --all`, `git branch -a`, and any preserved worktree before re-implementing.",
      ].join("\n");
  }
}

/** Merge-base of `tip` with main (local first, then origin/main); null when no
 *  main ref resolves — the caller then reads the state as unknown. */
function baseAgainstMain(repo: string, tip: string): string | null {
  for (const ref of ["main", "origin/main"]) {
    const mb = git(repo, ["merge-base", ref, tip]);
    if (mb) return mb;
  }
  return null;
}

/** Best-effort resolve of the prior run's state from the item + journal + git.
 *  NEVER throws: any failure leaves the fields null/unreadable so the brief
 *  says UNKNOWN rather than a false "produced nothing". */
export function resolvePriorRunState(stateDir: string, item: QueueItem): PriorRunState {
  const state: PriorRunState = { branch: null, tipSha: null, baseSha: null, uncommittedFiles: [], uncommittedReadable: false };
  const repo = item.cwd;
  if (!repo) return state;

  // Branch + tip: prefer the run's own recorded worktree branch, else the
  // newest journaled pointer for this item (worktree-preservation.ts captures
  // the tip BEFORE the runtime deletes the branch).
  let branch = item.runWorktree?.branch?.trim() || null;
  try {
    const pointers = reviewPointersFor(stateDir, item.key);
    const chosen = branch ? pointers.find((p) => p.branch === branch) : pointers[pointers.length - 1];
    if (chosen) {
      branch = branch ?? chosen.branch;
      state.tipSha = chosen.tipSha;
    }
  } catch {
    // journal unreadable — fall through; the branch ref may still resolve
  }
  state.branch = branch;
  if (!state.tipSha && branch) state.tipSha = git(repo, ["rev-parse", "--verify", `${branch}^{commit}`]);

  // Uncommitted files: only from the CURRENT run's worktree (a stale record
  // from a superseded run would describe someone else's tree).
  const rw = item.runWorktree;
  if (rw && rw.runId === (item.runId ?? item.lastRunId) && rw.path) {
    const status = git(rw.path, ["status", "--porcelain"]);
    if (status !== null) {
      state.uncommittedReadable = true;
      state.uncommittedFiles = status
        .split("\n")
        .map((l) => l.slice(3).trim())
        .filter(Boolean);
    }
  }

  if (state.tipSha) state.baseSha = baseAgainstMain(repo, state.tipSha);
  return state;
}

/** Build the recovery task: the original worker task (KEY + scope + prior-
 *  attempt branch/SHAs when `stateDir` is given) + the P5 context, then the
 *  prior-run state brief (committed / uncommitted / empty / unknown) so a
 *  re-dispatch CONFIRMS existing work instead of re-implementing it. The
 *  context comes from the PLAN so the budget the task describes and the
 *  budget the spawn actually gets are one decision, not two that can drift. */
export function recoveryTask(item: QueueItem, cause: FailCause, stateDir?: string): string {
  const base = `${workerTask(item, undefined, stateDir)}\n\n${recoveryPlan(item, cause).context}`;
  if (!stateDir) return base;
  try {
    return `${base}\n\n${priorRunBrief(resolvePriorRunState(stateDir, item))}`;
  } catch {
    return base; // the prior-run brief must never break a dispatch
  }
}

export interface RecoveryOptions {
  now?: number;
  /** Global per-item attempt cap (default MAX_RECOVERIES = 2). */
  maxRecoveries?: number;
  /** Backoff between attempts (default RECOVERY_BACKOFF_MS = 30s). */
  backoffMs?: number;
  spawnFailureThreshold?: number;
  degradedCooldownMs?: number;
  /** Free-slot accounting: recovery never over-spawns past capacity. */
  maxSlots?: number;
  /** Authoritative active count (the runner's fleet union) — the floor. */
  totalActive?: number;
}

export interface RecoveryOutcome {
  recovered: Array<{ key: string; attempt: number; cause: FailCause; budgetMs: number | null; runId: string }>;
  escalated: Array<{ key: string; cause: FailCause; attempts: number }>;
  /** true when the degraded window is currently pausing attempts. */
  held: boolean;
  /** true on the false→true transition — emit the one-time "paused" tick. */
  heldNotice: boolean;
  /** Consecutive provider failures at the end of this pass (for the message). */
  spawnFailures: number;
  /** Soonest future due time (epoch ms) — the runner arms its recovery timer.
   */
  nextAt: number | null;
  /** FINISHER-EVIDENCE (AUTOPILOT-34): failed items whose work is EVIDENCED AS
   *  LANDED (the source they were sent to land is in the declared cwd's
   *  history, and was not at dispatch). Recovery refuses to re-dispatch
   *  them — a re-run would duplicate a merge that already landed — and
   *  surfaces them once for the human's close-out call. */
  landedSkipped: Array<{ key: string; repo: string; sha: string; surfaced: boolean }>;
  /** THE HOLD (AUTOPILOT-46 finishers, generalised in AUTOPILOT-49): items
   *  recovery refuses to re-dispatch even without landed evidence, because a
   *  retry cannot help and may harm — a finisher re-run can duplicate a
   *  landing, and a PARKED item failed on something only a human can supply.
   *  They keep their status (`failed` / `blocked`) and are surfaced ONCE. */
  heldForHuman: Array<{ key: string; hold: RecoveryHold; cause: FailCause | null; blocker: BlockerReason; attempts: number; surfaced: boolean }>;
}

/** WHY recovery will not touch an item (null = ordinary, recoverable failure).
 *  `finisher`: AUTOPILOT-46 — an automatic re-run is the one action that can
 *  duplicate a landing. `parked`: AUTOPILOT-49 — a human deliberately moved the
 *  failed item to `blocked` because retrying cannot fix what went wrong. */
export type RecoveryHold = "finisher" | "parked";

export function recoveryHold(item: QueueItem): RecoveryHold | null {
  if (item.status === "blocked") return "parked";
  if (isFinisherItem(item)) return "finisher";
  return null;
}

function appendNote(existing: string, note: string): string {
  return existing ? `${existing}\n\n${note}` : note;
}

/**
 * One recovery pass. Deterministic and idempotent per (item, now): items whose
 * backoff has not elapsed are deferred (reported via nextAt), exhausted items
 * escalate once, and provider failures drive the degraded-window hold. Never
 * throws. The caller (runner) owns tick delivery.
 */
export async function autoRecoverFails(
  stateDir: string,
  backend: SubagentBackend,
  opts: RecoveryOptions = {},
): Promise<RecoveryOutcome> {
  const now = opts.now ?? Date.now();
  const backoffMs = opts.backoffMs ?? RECOVERY_BACKOFF_MS;
  const globalMax = Math.max(1, opts.maxRecoveries ?? MAX_RECOVERIES);
  const threshold = opts.spawnFailureThreshold ?? SPAWN_FAILURE_THRESHOLD;
  const cooldownMs = opts.degradedCooldownMs ?? DEGRADED_COOLDOWN_MS;
  const maxSlots = opts.maxSlots ?? 3;
  const outcome: RecoveryOutcome = { recovered: [], escalated: [], held: false, heldNotice: false, spawnFailures: 0, nextAt: null, landedSkipped: [], heldForHuman: [] };
  const store = loadStore(stateDir);
  if (!store) return outcome;
  const state = loadRecoveryState(stateDir);
  const degraded = state.providerFailures >= threshold && state.lastProviderFailureAt !== null && now - state.lastProviderFailureAt < cooldownMs;
  outcome.held = degraded;

  // Oldest failed first — the items waiting longest recover first (matches the
  // auto-dispatch ordering). Only items with a recorded, recoverable cause.
  const allFailed = Object.values(store.items)
    .filter((i) => i.status === "failed" && !!i.failCause && isFailCause(i.failCause))
    .sort((a, b) => (a.updatedAt < b.updatedAt ? -1 : 1));

  // PARKED ITEMS (AUTOPILOT-49): `blocked` + a recorded failCause is exactly
  // "parked out of failed" — the store preserves the cause across that one
  // transition and clears it everywhere else, so a proposal blocked before it
  // was ever dispatched is not in this set and never gets a recovery tick.
  // Being `blocked` is already enough to keep every dispatch lane off the item;
  // it is scanned here so the hold is ANNOUNCED like the finisher hold instead
  // of leaving the operator to notice a silently parked item.
  const parked = Object.values(store.items)
    .filter((i) => i.status === "blocked" && !!i.failCause && isFailCause(i.failCause))
    .sort((a, b) => (a.updatedAt < b.updatedAt ? -1 : 1));

  const later = (t: number): void => {
    outcome.nextAt = outcome.nextAt === null ? t : Math.min(outcome.nextAt, t);
  };

  // The pass spans backend.spawn awaits, so it cannot hold the store's mutation
  // window open: it decides against its own snapshot and STAGES each patch, then
  // MERGES them onto the current store in one mutateStore at the end. A patch is
  // a function of the item it lands on (notes append from the FRESH notes), so a
  // change another writer made meanwhile survives instead of being overwritten.
  const staged: Array<{ key: string; patch: (item: QueueItem) => UpdatePatch }> = [];
  const stage = (key: string, patch: (item: QueueItem) => UpdatePatch): void => {
    const local = store.items[key];
    if (local) updateItem(store, key, patch(local)); // keep the local view honest for the rest of the pass
    staged.push({ key, patch });
  };

  let changed = false;

  // Phase 0 — FINISHER-EVIDENCE (AUTOPILOT-34). A `failed` item whose work is
  // EVIDENCED AS LANDED (a finisher-class dispatch whose declared source is now
  // in the declared cwd's history) is NOT a recovery candidate: re-dispatching it
  // would re-run a merge that already landed (duplicate cherry-pick, or a
  // worker pointed at a moved main). The evidence + the override are RECORDED
  // on the item and it is surfaced ONCE for the human's close-out call.
  const failed: QueueItem[] = [];
  for (const item of allFailed) {
    const evidence = finisherLandedEvidence(item, now);
    if (!evidence) {
      failed.push(item);
      continue;
    }
    const alreadySurfaced = !!item.landedEvidence && item.recoveryEscalated === true;
    outcome.landedSkipped.push({ key: item.key, repo: evidence.repo, sha: evidence.sha, surfaced: !alreadySurfaced });
    if (alreadySurfaced) continue; // recorded AND already ticked — never re-announce
    stage(item.key, (fresh) => ({
      landedEvidence: evidence,
      ...(fresh.landedEvidence ? {} : { overrides: appendOverride(fresh, landedOverride(evidence, fresh.runId ?? item.runId, now)) }),
      // `recoveryEscalated` here means BOTH "already announced" (the skip is
      // surfaced once, never re-ticked) and "the later phases stay off this item".
      // Entering `active` does NOT clear it, so a deliberate human re-open
      // keeps auto-recovery hands-off even if the new run fails — INTENDED:
      // this item's declared source is already in the target's history, so an
      // automatic retry is the one action that can duplicate the landing. Who
      // re-opened it is driving it; the escalation notes say what to check.
      recoveryEscalated: true,
      notes: appendNote(
        fresh.notes,
        `[recover-skip: landed] ${new Date(now).toISOString()} — auto-recovery will NOT re-dispatch this item: its work is EVIDENCED AS LANDED (${evidence.repo} ${evidence.ref ?? "HEAD"} ${(evidence.fromSha ?? "?").slice(0, 8)} → ${evidence.sha.slice(0, 8)}). A re-dispatch would duplicate a merge that already landed. Needs your call: verify the landed commit and close the item (failed → done), or re-open it deliberately.`,
      ),
    }));
    changed = true;
  }

  // Phase 1 — THE HOLD (AUTOPILOT-46 finishers; AUTOPILOT-49 parked items).
  //
  // FINISHER HOLD (AUTOPILOT-46). Phase 0 skips the landings it can
  // SEE; the shapes it cannot see (a conflict-resolved cherry-pick, a squash of
  // a multi-commit source) produce no evidence BY DESIGN, and without this hold
  // they fall through as ordinary verdict failures and get RE-DISPATCHED once
  // the backoff elapses — re-running a merge that already landed, before the
  // operator can close the item out. A finisher re-run is the single action
  // that can duplicate a landing, evidence or not, so the whole class is held
  // and handed to the human (phase 0's posture) instead. Finishers are
  // PARTITIONED OUT here, so the phases below cannot reach one at all.
  // The cost, stated: a finisher that genuinely failed gets NO automatic retry.
  // It is never dropped silently — the hold is recorded on the item and ticked
  // once, and a deliberate re-dispatch (queue_dispatch) is always available.
  //
  // PARKED HOLD (AUTOPILOT-49). The same posture for the other non-retryable
  // class: an item a human moved from `failed` to `blocked` because retrying
  // cannot fix what went wrong (an interactive prompt, a missing credential, a
  // pending decision). `blocked` already keeps every dispatch lane off it — the
  // work here is to ANNOUNCE the hold once, name the release, and never nag
  // again. Release is blocked → approved (that transition re-arms recovery by
  // clearing this flag), then queue_dispatch.
  const recoverable: QueueItem[] = [];
  for (const item of [...parked, ...failed]) {
    const hold = recoveryHold(item);
    if (!hold) {
      // DEPENDENCY GATE (KEY: AUTOPILOT-81): a failed item still WAITING on an
      // unmet `after` dependency is not a recovery decision — skip it
      // entirely (no escalation, no attempt spent); once the dependency
      // clears, a later pass simply finds it recoverable again.
      if (unmetDependencies(item, store).length > 0) continue;
      recoverable.push(item);
      continue;
    }
    const cause = (item.failCause ?? null) as FailCause | null;
    const attempts = item.recoveries ?? 0;
    outcome.heldForHuman.push({ key: item.key, hold, cause, blocker: item.blocker ?? null, attempts, surfaced: !item.recoveryEscalated });
    if (item.recoveryEscalated) continue; // already announced — never re-nag
    stage(item.key, (fresh) => ({
      // Same meaning as phase 0's flag: announced once, and the phases below
      // stay off this item.
      recoveryEscalated: true,
      notes: appendNote(
        fresh.notes,
        hold === "finisher"
          ? `[recover-hold: finisher] ${new Date(now).toISOString()} — auto-recovery will NOT re-dispatch this item: it is FINISHER-CLASS (cause: ${cause}, ${attempts} attempt(s) spent) and an automatic re-run is the one action that can DUPLICATE a landing — including landings this framework cannot detect (a conflict-resolved cherry-pick, a squash of a multi-commit source), which leave no evidence by design. The item STAYS failed. Needs your call: check the target checkout (\`git log\`, \`git cherry\`) — if the work is IN, close it out (failed → done with an overrideReason); if it is not, re-dispatch deliberately.`
          : `[recover-hold: parked] ${new Date(now).toISOString()} — this item is PARKED (blocked: ${item.blocker ?? "decision"}) after failing (cause: ${cause ?? "unrecorded"}, ${attempts} attempt(s) spent): auto-recovery will NOT re-dispatch it while it is blocked, because retrying cannot supply what it needs. It is HELD, not lost. RELEASE IT when the blocker is resolved: queue_update('${item.key}', {status: 'approved'}) — which re-arms auto-recovery — then queue_dispatch. Or drop it: queue_update('${item.key}', {status: 'rejected'}).`,
      ),
    }));
    changed = true;
  }

  // Phase 2 — escalate exhausted items. This does NOT spawn, so it runs even
  // while the degraded window pauses attempts: an item that already spent its
  // budget must still be surfaced once.
  for (const item of recoverable) {
    if (item.recoveryEscalated) continue;
    const cause = item.failCause as FailCause;
    const attempts = item.recoveries ?? 0;
    const plan = recoveryPlan(item, cause, globalMax);
    if (attempts >= plan.maxAttempts) {
      stage(item.key, (fresh) => ({
        recoveryEscalated: true,
        notes: appendNote(
          fresh.notes,
          `[recover-exhausted] ${new Date(now).toISOString()} — auto-recovery spent ${attempts} attempt(s) (cause: ${cause}); the item STAYS failed. Needs your call: verify the pi-parallel-* branch, re-dispatch manually with a bigger budget, or drop.`,
        ),
      }));
      outcome.escalated.push({ key: item.key, cause, attempts });
      changed = true;
    }
  }

  // Phase 3 — attempt recovery (paused while degraded). Recovery spawns are
  // slot-bounded so the failed lane can never over-spawn past capacity.
  // FINISHER-CLASS ITEMS NEVER GET HERE (phase 1 partitioned them out), which
  // is why this lane needs no finisher baseline: the only class that lands
  // outside its worktree is not re-dispatched by recovery at all.
  if (!degraded) {
    const storeActive = Object.values(store.items).filter((i) => i.status === "active").length;
    const occupied = Math.max(opts.totalActive ?? 0, storeActive);
    let free = Math.max(0, maxSlots - occupied);
    for (const item of recoverable) {
      if (item.recoveryEscalated) continue;
      const cause = item.failCause as FailCause;
      const attempts = item.recoveries ?? 0;
      const plan = recoveryPlan(item, cause, globalMax);
      if (attempts >= plan.maxAttempts) continue; // phase 1 escalated it
      const notBefore = item.recoveryNotBefore ?? null;
      if (notBefore === null) {
        // A fresh failure — arm the short backoff first. A hang-then-die
        // (zombie) is ALSO provider-layer evidence; count it toward the hold.
        if (cause === "zombie") {
          state.providerFailures += 1;
          state.lastProviderFailureAt = now;
        }
        stage(item.key, () => ({ recoveryNotBefore: now + backoffMs }));
        later(now + backoffMs);
        changed = true;
        continue;
      }
      if (notBefore > now) {
        later(notBefore);
        continue;
      }
      if (free <= 0) {
        // No capacity — try again after the backoff instead of spawning into a
        // full fleet. The engine's own auto-dispatch owns the free-slot fill.
        later(now + backoffMs);
        continue;
      }
      const attempt = attempts + 1;
      // AUTOPILOT-53: never spawn with a null budget — recoveryPlan can still
      // hand back null for a non-budget-capped cause on a legacy item with no
      // recorded timeoutMs, so this is the recovery lane's own
      // resolveDispatchBudget backstop.
      const recoveryBudget = resolveDispatchBudget(plan.budgetMs);
      // AUTOPILOT-79/80/110: param(none here) > item.model > item.tier >
      // config.model > config default tier > inherit. An unknown/invalid tier
      // REFUSES the spawn (never a silent inherit): the item is skipped this
      // pass with one telemetry line, NOT marked failed (a config typo is not
      // a provider failure).
      const modelResolution = resolveSpawnModel({ item, config: loadAutopilotConfig(stateDir), role: "worker", repo: item.cwd });
      if (!modelResolution.ok) {
        appendTelemetry(stateDir, JSON.stringify({ t: new Date(now).toISOString(), type: "spawn-model-level-unknown", lane: "auto-recovery", key: item.key, reason: modelResolution.error }));
        continue;
      }
      let runId: string | null = null;
      try {
        // AUTOPILOT-60: name any prior attempt's journaled branch/SHA(s).
        runId = await backend.spawn(recoveryTask(item, cause, stateDir), {
          cwd: item.cwd ?? undefined,
          timeoutMs: recoveryBudget.timeoutMs,
          model: modelResolution.model,
        });
      } catch {
        runId = null;
      }
      if (runId) {
        stage(item.key, (fresh) => ({
          status: "active",
          runId,
          // AUTOPILOT-85: a stale item.runWorktree (from the run this recovery
          // is superseding) must never survive onto the new runId — it is only
          // refreshed once a LATER sweep sees a LIVE worktree for THIS run (see
          // shipping.ts's resolveWorkTip source (c) guard). Cleared in the SAME
          // patch that assigns the new runId, never a separate write.
          runWorktree: null,
          recoveries: attempt,
          recoveryNotBefore: null,
          timeoutMs: recoveryBudget.timeoutMs,
          notes: appendNote(
            fresh.notes,
            `[recover] ${new Date(now).toISOString()} — auto-recovery re-dispatched (attempt ${attempt}, cause: ${cause}, budget ${recoveryBudget.timeoutMs}ms${recoveryBudget.appliedDefault ? " (no budget recorded — applied the default)" : ""}); prior run ${item.runId ?? "?"} — verify its pi-parallel-* branch.`,
          ),
        }));
        try {
          if (item.cwd) preserveRunWorktree({ stateDir, repo: item.cwd, runId, key: item.key });
        } catch {
          // preservation never breaks recovery
        }
        state.providerFailures = 0;
        state.heldNotified = false;
        outcome.recovered.push({ key: item.key, attempt, cause, budgetMs: recoveryBudget.timeoutMs, runId });
        free -= 1;
      } else {
        // The provider rejected the spawn (bare 400 / empty api_error): record
        // the attempt, switch the cause to `spawn` (the infra class — 2× then
        // escalate), and back off. The global provider-health counter drives
        // the degraded-window hold.
        state.providerFailures += 1;
        state.lastProviderFailureAt = now;
        stage(item.key, (fresh) => ({
          failCause: "spawn",
          recoveries: attempt,
          recoveryNotBefore: now + backoffMs,
          notes: appendNote(
            fresh.notes,
            `[recover] ${new Date(now).toISOString()} — auto-recovery attempt ${attempt} could not spawn (provider/infra failure); retrying after backoff.`,
          ),
        }));
        later(now + backoffMs);
      }
      changed = true;
    }
  } else if (state.lastProviderFailureAt !== null) {
    // Paused until the cooldown lets a probe through.
    later(state.lastProviderFailureAt + cooldownMs);
  }

  if (state.providerFailures >= threshold && !state.heldNotified) {
    state.heldNotified = true;
    outcome.heldNotice = true;
  }
  outcome.spawnFailures = state.providerFailures;
  state.updatedAt = new Date(now).toISOString();
  saveRecoveryState(stateDir, state);
  if (changed && staged.length) {
    mutateStore(stateDir, (s) => {
      for (const { key, patch } of staged) {
        const fresh = s.items[key];
        if (!fresh) continue; // the item is gone (rejected/renamed) — nothing to recover
        try {
          updateItem(s, key, patch(fresh));
        } catch {
          // The item MOVED under us (a human/tool transition this pass could not
          // see), so the staged transition is no longer legal. That is a real
          // conflict resolved in favour of the newer state, not a lost update:
          // the next pass re-decides from the item's actual status.
        }
      }
    });
  }
  return outcome;
}