/**
 * The review lane: what happens to a run after a reviewer says no.
 *
 * Three passes with one subject — revisions, the fresh-correction lane (#1047)
 * and ceiling adjudication — kept together because they share the decision this
 * module exists to make: *how* to relaunch. `resolveReviewLaunch` reads the
 * prior session's health, the model it actually ran on and the round already
 * recorded, and answers resume-or-refresh once, durably, for every caller. Split
 * across three files that answer would be re-derived three times, which is the
 * shape #1035 was.
 *
 * It launches through `dispatch.ts` rather than beside it: a correction is
 * another launch of the same run against the same branch and PR, so the claim,
 * label-swap and settle rules are borrowed, never restated. The dependency runs
 * one way only — nothing in the dispatch path knows this module exists.
 */
import { existsSync, mkdirSync } from "node:fs";
import { dirname, join } from "node:path";
import { hasContinuationBudget, runSpendAllowanceUsd } from "../admission.ts";
import { resolveReleaseGrants, resolveReview, stateDir } from "../config.ts";
import { classifyRun, providerCreditRefusal, providerTransientFault, reviewRoundNeverWorked } from "../failure-class.ts";
import { pushRunBranch, type RunRepoRef } from "../gitops.ts";
import { errText, log, safeEscalate } from "../log.ts";
import { DEFAULT_MODEL_FALLBACK_THRESHOLD, providerFailureFacts, resolveDispatchModel } from "../model-fallback.ts";
import { materializeOmpSettings } from "../omp-settings.ts";
import { harnessLogDir, readSessionError, readTerminalEvidence, recordOperatorStop, recordWorkerDiscoveries, settleWorktree, swapToQueue } from "../settlement.ts";
import { GhPrMissingError } from "../tracker/github.ts";
import { REVIEW_ROUND_INFRA_MAX_RETRIES, type PrDiff, type PrState, type PrVerification, type RepoTarget, type ReviewAdjudicationRecord, type ReviewCorrectionProvenance, type ReviewLaunchMode, type ReviewRevisionOutcome, type ReviewRevisionRecord, type RunRecord, type RunState } from "../types.ts";
import { isHeadMismatch, listenVerbChannel, type VerbListener } from "../verbs/server.ts";
import { ensureVerbSocketDir, verbSocketPath } from "../verbs/socket.ts";
import { renderAdjudicationBrief, renderReviewRevisionPrompt, runAdjudicator, runCorrectionWorker, runWorker, type AdjudicationRound, type CorrectionEvidence, type WorkerPauseControl, type WorkerResult } from "../worker.ts";
import { addRunRepo, mirrorPathFor, removeWorktree, type RunPublisher } from "../worktree.ts";
import type { WorkerPool } from "./admission-pass.ts";
import { workerGateGroups } from "./brief.ts";
import { completionLastError, exhaustedSessionReason, githubRepo, knowledgeRepoKey, recordDaemonSessionSpend, repoSlug, verbDeps, type Deps, type TurnLimitController, type WorkerControlSlot } from "./deps.ts";
import { endedBy, shouldContinueAfterTurnsCap, swapLabel, verifyPushedGreenClaim, workerReleaseBlockRecorder } from "./dispatch.ts";
import { wakeOrchestratorForBlockedRun } from "./supervision.ts";

/**
 * The daemon half of the review-revision transport (#677): wake every pending
 * revision the orchestrator's verb recorded, on the tick, exactly like
 * admission — the verb has no handle on the worker machinery, and the CLI path
 * must behave like the embedded one. Bounded by the same worker slots, so a
 * queued revision can never push the fleet past `maxConcurrentWorkers`; the
 * remaining rows stay pending for the next tick.
 *
 * The claim (`pushed-green` → `running`) is synchronous and atomic per row;
 * only the resumed worker runs in the pool. A row that moved since the verb
 * recorded it — merged after all, settled, re-claimed — fails the claim and is
 * settled `skipped` rather than woken on stale identity.
 */
export async function dispatchReviewRevisions(d: Deps, pool?: WorkerPool): Promise<void> {
  const pending = d.store.pendingReviewRevisions(d.project.name);
  if (pending.length === 0) return;
  const slots = Math.max(0, d.caps.maxConcurrentWorkers - d.store.liveRuns(d.project.name).length);
  const batch = pending.slice(0, slots);
  const launches: Promise<void>[] = [];
  for (const revision of batch) {
    if (d.drain?.draining === true) {
      log(
        `review revisions held: the daemon is draining (${batch.length - launches.length} of ${batch.length} not launched)`,
      );
      break;
    }
    // A capped/failed/stopped run has no pushed-green settle sweep keeping its
    // row honest (#795 review round 1, #1101): nothing transitions a `failed`
    // / `killed` / `stopped` row when its PR merges or closes, so a PR that
    // changed after the verb recorded the round would otherwise be claimed and
    // a worker resumed against a dead PR. The settled-green origin keeps its
    // own interlock (the settle sweep flips the row and the claim below
    // refuses it), so every other revisable origin re-reads the reviewed PR
    // fact here: the round dispatches only while the PR is still open at the
    // exact reviewed head.
    //
    // The skip decision is decisive-fact only (review round 2): a definitively
    // missing PR (`GhPrMissingError` — a corroborated 404, #779), a PR that
    // is definitively not open, and a head that has definitively MOVED are
    // settled skipped, exactly like any other row that moved. Everything
    // transient or unreadable — a bare 404 that could not be corroborated, an
    // undefined answer, a pending check run, a same-head red check that a
    // rerun may clear, a throwing tracker — stays PENDING for the next tick:
    // settling it would permanently discard the orchestrator's findings and
    // consume the round.
    // The row as it stands BEFORE the claim. Read whole rather than by state:
    // the review claim clears the terminal classification (#795), so the facts
    // #1047's launch decision rests on are only readable here.
    const originRun = d.store.getRun(revision.runId);
    const origin = originRun?.state;
    if (origin === "failed" || origin === "killed" || origin === "stopped") {
      let prState: PrState | undefined;
      try {
        prState = await d.tracker.prState(revision.prUrl);
      } catch (err) {
        if (err instanceof GhPrMissingError) {
          d.store.settleReviewRevision(revision.id, "skipped", Date.now());
          log(
            `#${revision.issue} review round ${revision.round} skipped: ${revision.prUrl} does not exist — the reviewed PR is gone`,
          );
          continue;
        }
        log(
          `#${revision.issue} review round ${revision.round} held: PR state re-check failed (${errText(err)}) — retrying next tick`,
        );
        continue;
      }
      if (prState === undefined) {
        log(
          `#${revision.issue} review round ${revision.round} held: ${revision.prUrl} state could not be re-read — retrying next tick`,
        );
        continue;
      }
      if (prState !== "open") {
        d.store.settleReviewRevision(revision.id, "skipped", Date.now());
        log(
          `#${revision.issue} review round ${revision.round} skipped: ${revision.prUrl} is ${prState}, not open — a review round resumes an open PR only`,
        );
        continue;
      }
      let verification: PrVerification | undefined;
      try {
        verification = await d.tracker.verifyPr(revision.prUrl, revision.headSha);
      } catch (err) {
        log(
          `#${revision.issue} review round ${revision.round} held: the reviewed-head check failed (${errText(err)}) — retrying next tick`,
        );
        continue;
      }
      if (verification === undefined || verification.status === "pending") {
        log(
          `#${revision.issue} review round ${revision.round} held: ${revision.prUrl} at ${revision.headSha} is ` +
            `${verification?.status ?? "unverifiable"} — retrying next tick`,
        );
        continue;
      }
      if (verification.status === "failed" && isHeadMismatch(verification.reason)) {
        d.store.settleReviewRevision(revision.id, "skipped", Date.now());
        log(
          `#${revision.issue} review round ${revision.round} skipped: ${revision.prUrl} no longer stands at ` +
            `${revision.headSha} (${verification.reason}) — the reviewed head moved`,
        );
        continue;
      }
      // Anything else — a same-head red check a rerun may clear, an unknown
      // verdict — is not definitive: the round stays pending for the next
      // tick rather than discarding the findings.
      if (verification.status !== "green") {
        log(
          `#${revision.issue} review round ${revision.round} held: ${revision.prUrl} at ${revision.headSha} is ` +
            `${verification.status} — retrying next tick`,
        );
        continue;
      }
    }
    // #1047: decide the transport from the pre-claim row, and write it once.
    // Before the claim on purpose — the claim clears exactly the columns the
    // diagnosis reads — and durable on purpose, so a restart, an infra retry
    // and a second racing pass all honour this one decision instead of
    // re-deciding against a row the claim has already rewritten.
    const launch =
      originRun === undefined ? undefined : resolveReviewLaunch(d, revision, originRun);
    if (!d.store.claimRunForReview(revision.runId)) {
      d.store.settleReviewRevision(revision.id, "skipped", Date.now());
      log(
        `#${revision.issue} review round ${revision.round} skipped: run ${revision.runId} is ${origin ?? "gone"} — no longer reclaimable for review`,
      );
      continue;
    }
    d.store.markReviewRevisionDispatched(revision.id, Date.now());
    log(
      `#${revision.issue} review round ${revision.round} → ${launch?.launchMode ?? "resume-original"} on run ` +
        `${revision.runId} (model requested ${launch?.requestedModel ?? "harness default"}): ` +
        `${launch?.reason ?? "the run row vanished before the decision"}`,
    );
    launches.push(handleReviewRevision(d, revision, launch));
  }
  if (pool !== undefined) {
    for (const launch of launches) pool.launch(launch);
    return;
  }
  // `--once`: like admissions, the tick waits for the workers it launched.
  await Promise.allSettled(launches);
}

/**
 * The turn ceiling for one adjudication (#932).
 *
 * Small on purpose, and not the worker cap: a verdict is a read and an answer,
 * not an implementation. A session still going after this is not deciding — and
 * the ceiling exists so a stuck adjudicator costs a bounded amount rather than a
 * worker's whole budget, on work that has already spent its review rounds.
 */
export const ADJUDICATION_MAX_TURNS = 40;

/**
 * Launch the review-ceiling adjudications this project owes (#932).
 *
 * Deliberately its own pass, next to {@link dispatchReviewRevisions} but never
 * inside it: a revision resumes an implementation worker to change code, an
 * adjudication opens a fresh read-only session to decide about code. Folding
 * them together is how "escalate the automation" quietly becomes "resume the
 * same worker with a different label", which is #873's named silent fake.
 *
 * Only `pending` rows are launched. A `running` row belongs to a live
 * adjudicator — or to one a restart killed, which the startup sweep settles —
 * so a restart re-reads it and never launches a second adjudicator for one head.
 */
export async function dispatchReviewAdjudications(d: Deps, pool?: WorkerPool): Promise<void> {
  const open = d.store.openReviewAdjudications(d.project.name).filter((a) => a.state === "pending");
  if (open.length === 0) return;
  const launches: Promise<void>[] = [];
  for (const adjudication of open) {
    if (d.drain?.draining === true) {
      log(`adjudications held: the daemon is draining (${open.length - launches.length} not launched)`);
      break;
    }
    // The head is re-read before anything launches. A verdict describes the diff
    // it read, so adjudicating a head the PR has since moved past would produce
    // a decision about code nobody is merging — settled `stale-head` with both
    // heads named, never launched.
    let live: string | undefined;
    try {
      live = await d.tracker.prHead(adjudication.prUrl);
    } catch (err) {
      log(`#${adjudication.issue} adjudication held: ${adjudication.prUrl} head unreadable (${errText(err)})`);
      continue;
    }
    if (live === undefined) {
      log(`#${adjudication.issue} adjudication held: ${adjudication.prUrl} head could not be read — retrying next tick`);
      continue;
    }
    if (live.toLowerCase() !== adjudication.headSha.toLowerCase()) {
      d.store.settleReviewAdjudication(
        adjudication.id,
        "stale-head",
        `recorded ${adjudication.headSha}, live ${live} — the pull request moved before this adjudication launched`,
        Date.now(),
      );
      log(`#${adjudication.issue} adjudication stale-head: recorded ${adjudication.headSha}, live ${live}`);
      continue;
    }
    // The claim IS the single-flight guard: the transition is the row's own
    // WHERE clause, so two dispatch passes racing one adjudication cannot both
    // launch — the loser changes nothing and skips.
    const review = resolveReview(d.project);
    const provenance = { role: adjudication.role, model: adjudication.role, resolvedAt: Date.now() };
    if (!d.store.markReviewAdjudicationRunning(adjudication.id, provenance, Date.now())) {
      log(`#${adjudication.issue} adjudication skipped: another pass already claimed it`);
      continue;
    }
    log(`#${adjudication.issue} adjudication → launching role ${adjudication.role} at ${adjudication.headSha}`);
    launches.push(handleReviewAdjudication(d, adjudication, review.maxRounds));
  }
  if (pool !== undefined) {
    for (const launch of launches) pool.launch(launch);
    return;
  }
  await Promise.allSettled(launches);
}

/**
 * Assemble one adjudication's evidence and run it to a durable verdict (#932).
 *
 * Every fact comes from a durable row or a live tracker read — never from the
 * worker's own report, which is the claim under adjudication. The issue body is
 * carried verbatim because the acceptance criteria ARE the standard being
 * judged, and every prior round travels WITH its settled outcome: a finding
 * without its outcome reads as an outstanding complaint even when the worker
 * fixed it, which is how an adjudicator rejects work that was already corrected.
 */
export async function handleReviewAdjudication(
  d: Deps,
  adjudication: ReviewAdjudicationRecord,
  maxRounds: number,
): Promise<void> {
  const issue = adjudication.issue;
  const [title, body, diff, verification] = await Promise.all([
    d.tracker.getIssue(issue).then(
      (row) => row?.title,
      () => undefined,
    ),
    d.tracker.issueBody(issue).then(
      (text) => text,
      () => undefined,
    ),
    d.tracker.prDiff(adjudication.prUrl).then(
      (parsed) => parsed,
      () => undefined,
    ),
    d.tracker.verifyPr(adjudication.prUrl, adjudication.headSha).then(
      (result) => result,
      () => undefined,
    ),
  ]);
  // Prior rounds, oldest first, keyed by PR so a continuation that inherited the
  // PR cannot hide its predecessor's findings. The ceiling call's own findings
  // are appended as the last entry — they are the reason this adjudication
  // exists, and the durable row is where they were kept precisely so this brief
  // could carry them.
  const rounds: AdjudicationRound[] = d.store
    .reviewRevisionsForPr(d.project.name, adjudication.prUrl)
    .map((revision) => ({
      round: revision.round,
      findings: revision.findings,
      ...(revision.outcome === undefined ? {} : { outcome: revision.outcome }),
    }));
  if (adjudication.findings !== undefined) {
    rounds.push({
      round: rounds.length + 1,
      findings: adjudication.findings,
      outcome: "escalated to this adjudication — no worker round was available",
    });
  }
  const brief = renderAdjudicationBrief({
    issue,
    ...(title === undefined ? {} : { issueTitle: title }),
    issueBody: body ?? "(the issue body could not be read — judge the diff against the findings below)",
    prUrl: adjudication.prUrl,
    headSha: adjudication.headSha,
    checks:
      verification === undefined
        ? "(the check state could not be read)"
        : `${verification.status} — ${verification.reason}`,
    diff:
      diff === undefined
        ? "(the diff could not be read — say so in your reasons rather than guessing)"
        : renderDiffForAdjudication(diff),
    rounds,
    maxRounds,
  });

  const sessionDir = join(stateDir(), "sessions", `adjudication-${adjudication.id}`);
  mkdirSync(sessionDir, { recursive: true });
  const ompSettingsFile = materializeOmpSettings(d.project, sessionDir);
  const result = await (d.runAdjudicatorImpl ?? runAdjudicator)({
    brief,
    // No worktree: an adjudicator has no branch, and everything it needs is in
    // the brief. The session directory doubles as its confined cwd.
    cwd: sessionDir,
    sessionDir,
    model: adjudication.role,
    ...(ompSettingsFile === undefined ? {} : { ompSettingsFile }),
    socketPath: join(sessionDir, "ipc.sock"),
    maxTurns: ADJUDICATION_MAX_TURNS,
  });

  // Above every verdict branch below — a stale-head refusal and an unresolvable
  // role cost as much as a clear, and all three used to be free as far as
  // attribution was concerned.
  recordDaemonSessionSpend(d, {
    role: "adjudicator",
    issue,
    ...(adjudication.role === undefined ? {} : { model: adjudication.role }),
    ...(result.model === undefined ? {} : { resolvedModel: result.model }),
    turns: result.turns,
    spendUsd: result.spendUsd,
    at: Date.now(),
  });

  // What actually ran it, named in the evidence when it differs from the role
  // that was asked for: the provenance recorded at claim time is conductor's own
  // resolution (the role it passed to OMP), and the model that wrote the
  // messages is a later observation only this result carries.
  const ran =
    result.model === undefined
      ? ""
      : `adjudicated by ${result.model}${result.provider === undefined ? "" : ` (${result.provider})`}: `;
  if (result.verdict === undefined) {
    // A session that never took a turn did not run: the role could not be
    // resolved or the harness could not start, and the remedy is configuration
    // rather than a retry. One that ran and produced nothing usable is a
    // failure. Neither is ever a clear.
    const state = result.turns === 0 ? "unavailable-model" : "failed";
    d.store.settleReviewAdjudication(
      adjudication.id,
      state,
      `${ran}${result.report === "" ? "no verdict and no output" : result.report}`,
      Date.now(),
    );
    log(`#${issue} adjudication ${state} at ${adjudication.headSha}`);
    return;
  }
  // A verdict that names a different head judged something else. Fail closed:
  // the one-shot is spent on the head it was opened for, and nothing merges on
  // the strength of a decision about another diff.
  if (
    result.verdict.headSha !== undefined &&
    result.verdict.headSha.toLowerCase() !== adjudication.headSha.toLowerCase()
  ) {
    d.store.settleReviewAdjudication(
      adjudication.id,
      "stale-head",
      `${ran}the verdict names head ${result.verdict.headSha}, not the adjudicated ${adjudication.headSha}: ${result.verdict.reasons}`,
      Date.now(),
    );
    log(`#${issue} adjudication stale-head: verdict named ${result.verdict.headSha}`);
    return;
  }
  d.store.settleReviewAdjudication(
    adjudication.id,
    result.verdict.verdict,
    `${ran}${result.verdict.reasons}`,
    Date.now(),
  );
  log(`#${issue} adjudication ${result.verdict.verdict} at ${adjudication.headSha}`);
}

/**
 * Apply the terminal dispositions an adjudication's verdict implies (#876).
 *
 * The verdict is a decision; this is what is DONE about it, and the two are
 * separate rows for exactly that reason. Recording `cleared` while leaving the
 * merge gate refusing, or `rejected` while leaving the PR open to block the
 * issue forever, is the issue's named silent fake — so what this pass changes is
 * tracker and gate state, and the disposition text is only its receipt.
 *
 * Idempotent by that receipt: a row whose `disposition` is already recorded is
 * skipped, so a restart, a retried tick or a partial failure cannot close one PR
 * twice. Nothing here is destructive before its explanation is durable.
 *
 * Neither branch opens a review round, launches another adjudicator, or asks an
 * operator for anything. Neither touches an attempt or continuation budget
 * either: a rejected PR closes, and the ORDINARY settle sweep charges that
 * closure exactly as it charges any PR closed without merging — which is what
 * keeps this composable with #815's separate corrective-worker path instead of
 * resetting its counters.
 */
export async function applyAdjudicationDispositions(d: Deps): Promise<void> {
  const { project, store } = d;
  // Terminal rows only, and only those nothing has disposed of yet. Read from
  // the PRs still in flight: a settled adjudication whose PR is long gone needs
  // no disposition, and scanning history would re-litigate closed work.
  const pending = store
    .activeRuns(project.name)
    .flatMap((run) => (run.prUrl === undefined ? [] : store.reviewAdjudicationsForPr(project.name, run.prUrl)))
    .filter((row) => row.settledAt !== undefined && row.disposition === undefined);
  for (const adjudication of pending) {
    if (adjudication.state === "cleared") {
      // The clear re-enters the ORDINARY merge path, and it does so through the
      // mechanism that path already reads: the #888 gate refuses a head carrying
      // unresolved review evidence, and #913's clearance is the recorded
      // disposition that settles it. Reusing it means no second gate, no
      // adjudication-shaped exception inside `conductor_pr_merge`, and one
      // audit trail for "this head was cleared, by whom, and why".
      //
      // The head is re-read first: a clearance is head-scoped and clears
      // backwards only, so recording one against a head the PR has moved past
      // would bless a diff nobody adjudicated.
      let live: string | undefined;
      try {
        live = await d.tracker.prHead(adjudication.prUrl);
      } catch (err) {
        log(`#${adjudication.issue} adjudication clearance held: head unreadable (${errText(err)})`);
        continue;
      }
      if (live === undefined) {
        log(`#${adjudication.issue} adjudication clearance held: head could not be read — retrying next tick`);
        continue;
      }
      if (live.toLowerCase() !== adjudication.headSha.toLowerCase()) {
        // Visible and non-destructive: the clearance is not recorded, the row
        // says why, and the PR keeps whatever gate state it has.
        store.recordReviewAdjudicationDisposition(
          adjudication.id,
          `no clearance recorded: the head moved from ${adjudication.headSha} to ${live} after the verdict`,
        );
        log(`#${adjudication.issue} adjudication clearance skipped: head moved to ${live}`);
        continue;
      }
      store.recordReviewClearance({
        project: project.name,
        prUrl: adjudication.prUrl,
        headSha: adjudication.headSha,
        by: `adjudication:${adjudication.role}`,
        reason: `cleared by the review-ceiling adjudication: ${adjudication.evidence ?? "no evidence recorded"}`,
      });
      store.recordReviewAdjudicationDisposition(
        adjudication.id,
        `recorded a review clearance at ${adjudication.headSha}; the PR is on the ordinary merge path`,
      );
      log(`#${adjudication.issue} adjudication cleared → merge path open at ${adjudication.headSha}`);
      continue;
    }
    // Every non-clear terminal state disposes the same way, and that is
    // deliberate: `rejected`, `failed`, `unavailable-model` and `stale-head` all
    // mean "this head is not merging and no further round is coming", and the
    // blocking artefact is identical in each case — an open PR that occupies the
    // issue. Only the recorded reason differs.
    const findings = store
      .reviewRevisionsForPr(project.name, adjudication.prUrl)
      .map((revision) => `### Review round ${revision.round} (${revision.outcome ?? "never settled"})

${revision.findings}`);
    const comment = [
      `## Adjudication: ${adjudication.state}`,
      "",
      `The review-round ceiling was reached on ${adjudication.headSha}, and the \`${adjudication.role}\` adjudicator ` +
        `did not clear it. Closing this pull request so the issue is not blocked by work that cannot merge — ` +
        "nothing here asks anyone to repair it by hand.",
      "",
      "### Adjudicator's reasons",
      "",
      adjudication.evidence ?? "(none recorded)",
      ...(findings.length === 0 ? [] : ["", "## Preserved review findings", "", ...findings]),
      "",
      "The branch itself is untouched and remains in the repository.",
    ].join("\n");
    const closed = await d.verbActions?.closePr(adjudication.prUrl, comment);
    if (closed === undefined) {
      log(`#${adjudication.issue} adjudication disposition held: no action surface is wired`);
      continue;
    }
    if (!closed.ok) {
      // Retryable, and no receipt written: the next tick tries again rather than
      // leaving a rejected PR open with a row that claims it was handled.
      log(`#${adjudication.issue} adjudication disposition held: could not close ${adjudication.prUrl} — ${closed.stderr}`);
      continue;
    }
    store.recordReviewAdjudicationDisposition(
      adjudication.id,
      `closed ${adjudication.prUrl} with the adjudicator's reasons and every preserved finding; ` +
        "the settle sweep releases the issue",
    );
    log(`#${adjudication.issue} adjudication ${adjudication.state} → closed ${adjudication.prUrl}`);
  }
}

/** The diff as one text, from the structured read the settlement audit uses.
 *  `renderAdjudicationBrief` bounds the total; this only flattens it, and says
 *  when the adapter itself had to cut the read short — an adjudicator must know
 *  its silence covers only what it saw. */
export function renderDiffForAdjudication(diff: PrDiff): string {
  const files = diff.files.map((file) =>
    [
      `--- ${file.status} ${file.previousPath === undefined ? "" : `${file.previousPath} → `}${file.path}`,
      file.hunks ?? "(no textual hunks: binary, mode-only or pure rename)",
    ].join("\n"),
  );
  return [...(diff.truncated ? ["[the tracker truncated this diff read]"] : []), ...files].join("\n\n");
}

// ============================== review-correction launch selection (#1047)
//
// Why a review round has a *decision* at all, rather than just resuming:
// #1035. A capped implementation session (Ox, `model-empty-stop` at its turn
// ceiling) kept receiving correction rounds by `resume: true` after the fleet
// had already moved its worker model to DeepSeek. Every round replayed an
// exhausted transcript under the model nobody dispatches any more, the PR
// never moved, and fleet status said only `review-revision N`.
//
// So dispatch chooses, once per round, between two transports on ONE durable
// identity:
//
//   - `resume-original` — the cheap path, unchanged: `continueRecent` on the
//     run's own transcript, on the model the green run used. A healthy bounded
//     round must keep this; it is what makes small corrections nearly free.
//   - `fresh-correction` — #1046's transport: a blank session on the SAME
//     branch, worktree, PR and reviewed head, briefed from durable evidence,
//     launched on the model the fleet resolves TODAY.
//
// Two rules make the choice trustworthy rather than clever:
//
//  1. It is made from the run row BEFORE the review claim, because the claim
//     deliberately clears `failureClass`/`recoveryAction`/`recoveredAt` so the
//     resumed execution classifies fresh (#795). The diagnosis this decision
//     rests on lives in exactly those columns, so reading them after the claim
//     would read a blank row and always resume — the bug wearing a decision.
//  2. It is written ONCE per round (#1045) and re-read on every later claim,
//     retry and restart. A round that was decided `fresh-correction` and then
//     infra-killed relaunches fresh; it never quietly reverts because the row
//     it was diagnosed from has since been rewritten.

/**
 * The worker model a dispatch would resolve for this issue right now — the
 * same ladder {@link handleIssue} walks: this chain's one model escalation
 * (#807) outranks the project default, and the provider failover chain (#286)
 * layers on top of whichever is primary.
 *
 * Deliberately NOT re-parsing the issue's `Model:` declaration: that value is
 * an admission-time read of the issue thread, and a correction round reads no
 * comments. The durable escalation marker — the only per-issue override that
 * outranks config — is read, so an escalated chain still corrects on its
 * stronger tier.
 */
export function currentWorkerModel(d: Pick<Deps, "project" | "store">, issue: number): string | undefined {
  const { project, store } = d;
  const facts = providerFailureFacts(store.runsForIssue(project.name, issue));
  const escalation = store.modelEscalation(project.name, issue);
  return resolveDispatchModel({
    workerModel: escalation?.model ?? project.workerModel,
    modelFallbacks: project.modelFallbacks,
    threshold: project.modelFallbackThreshold ?? DEFAULT_MODEL_FALLBACK_THRESHOLD,
    streak: facts.streak,
  }).model;
}

/** One round's launch decision, as dispatch made it or as the store returns
 *  it. `reason` is the clause a human reads and the fresh correction's brief
 *  carries; it is derived, never persisted, because the durable columns are
 *  the facts and this is the sentence about them. */
export interface ReviewLaunchDecision {
  launchMode: ReviewLaunchMode;
  requestedModel?: string;
  originSessionRef?: string;
  reason: string;
}

/**
 * Choose the transport for one PENDING round from the pre-claim run row
 * (#1047).
 *
 * Fresh on any of three diagnosed facts, and resume otherwise:
 *
 *  1. the session is exhausted ({@link exhaustedSessionReason});
 *  2. the transcript the round names is gone, so there is nothing to resume —
 *     which used to escalate the round away by hand;
 *  3. the model moved: the run dispatched on a selector the fleet no longer
 *     resolves. Compared selector-to-selector only. `RunRecord.resolvedModel`
 *     is the harness's own resolution and lives in a different namespace from
 *     a configured selector (`@fast` resolves to a provider/model string), so
 *     comparing across the two would invent a model change on every role-alias
 *     project. A row with no recorded selector therefore cannot prove a
 *     configured change and keeps the cheap path — never a guess.
 */
export function selectReviewLaunch(
  d: Pick<Deps, "project" | "store">,
  revision: ReviewRevisionRecord,
  run: RunRecord,
): ReviewLaunchDecision {
  const originSessionRef = revision.sessionFile ?? run.sessionFile;
  const resume = (reason: string): ReviewLaunchDecision => ({
    launchMode: "resume-original",
    ...(run.model === undefined ? {} : { requestedModel: run.model }),
    ...(originSessionRef === undefined ? {} : { originSessionRef }),
    reason,
  });
  const fresh = (reason: string): ReviewLaunchDecision => {
    const requestedModel = currentWorkerModel(d, revision.issue);
    return {
      launchMode: "fresh-correction",
      ...(requestedModel === undefined ? {} : { requestedModel }),
      ...(originSessionRef === undefined ? {} : { originSessionRef }),
      reason,
    };
  };

  const exhausted = exhaustedSessionReason(run);
  if (exhausted !== undefined) {
    return fresh(`the implementation session cannot be resumed: ${exhausted}`);
  }
  if (originSessionRef === undefined || !existsSync(originSessionRef)) {
    return fresh(
      `the implementation transcript ${originSessionRef ?? "(none recorded)"} is gone, so there is no session to continue`,
    );
  }
  const current = currentWorkerModel(d, revision.issue);
  if (run.model !== undefined && current !== undefined && run.model !== current) {
    return fresh(
      `the fleet's worker model changed since the reviewed run: it ran on ${run.model}, dispatch now resolves ${current}`,
    );
  }
  return resume("the reviewed session is bounded and healthy");
}

/**
 * The decision for one round: the recorded one when there is one, otherwise
 * this pass's selection, written once (#1045, #1047).
 *
 * The write is the arbiter, not this function: `recordReviewRevisionLaunch`
 * refuses a contradictory second decision, and a refusal means another pass
 * already decided — so the refused caller re-reads and follows the recorded
 * decision rather than launching its own. That is what makes a restart, a
 * dispatch retry and two racing passes all produce ONE correction.
 */
export function resolveReviewLaunch(
  d: Pick<Deps, "project" | "store">,
  revision: ReviewRevisionRecord,
  run: RunRecord,
): ReviewLaunchDecision {
  const recorded = d.store.reviewRevisionLaunch(d.project.name, revision.id);
  if (recorded?.launchMode !== undefined) return fromRecordedLaunch(recorded);
  const chosen = selectReviewLaunch(d, revision, run);
  const wrote = d.store.recordReviewRevisionLaunch(d.project.name, revision.id, {
    launchMode: chosen.launchMode,
    ...(chosen.requestedModel === undefined ? {} : { requestedModel: chosen.requestedModel }),
    ...(chosen.originSessionRef === undefined ? {} : { originSessionRef: chosen.originSessionRef }),
    at: Date.now(),
  });
  if (wrote) return chosen;
  const winner = d.store.reviewRevisionLaunch(d.project.name, revision.id);
  if (winner?.launchMode === undefined) {
    // No row, or a row the store refuses to decide: the round is not
    // dispatchable as a correction, and the honest fallback is the transport
    // that changes nothing. The dispatch pass's own claim already refuses a
    // round whose run moved, so this is reachable only for a row that vanished
    // between the two reads.
    return {
      ...chosen,
      launchMode: "resume-original",
      reason: `${chosen.reason} — but no launch decision could be recorded for this round`,
    };
  }
  return fromRecordedLaunch(winner);
}

/** The durable provenance read back as a decision. The reason names the store
 *  as its source, so a log line for a re-read decision cannot be mistaken for
 *  a fresh diagnosis of the row as it stands now. */
export function fromRecordedLaunch(recorded: ReviewCorrectionProvenance): ReviewLaunchDecision {
  const launchMode = recorded.launchMode ?? "resume-original";
  return {
    launchMode,
    ...(recorded.requestedModel === undefined ? {} : { requestedModel: recorded.requestedModel }),
    ...(recorded.originSessionRef === undefined ? {} : { originSessionRef: recorded.originSessionRef }),
    reason: `the decision already recorded for this round (${launchMode})`,
  };
}

/**
 * Everything a fresh correction is told, assembled from durable rows and live
 * reads only (#1046, #1047).
 *
 * The shape of {@link handleReviewAdjudication}'s evidence gathering, and for
 * the same reason: a session with no transcript has to be handed the facts, and
 * the one source that must NOT be trusted is the worker's own prior report —
 * that report is the claim the findings contradict. So the issue body is
 * verbatim (the acceptance criteria are the standard), the diff and the check
 * state come from the tracker at the exact reviewed head, and every prior round
 * travels with its settled outcome so a fixed finding does not read as an
 * outstanding one.
 *
 * Every read is individually best-effort: a tracker that refuses one fact
 * degrades that line to a named absence rather than losing the round. Losing
 * the round is worse — the findings are already durable, the PR is already
 * green, and the correction is the only thing that can move either.
 */
export async function correctionEvidenceFor(
  d: Deps,
  revision: ReviewRevisionRecord,
  run: RunRecord,
  repo: RepoTarget,
  args: { branch: string; worktree: string; reason: string },
): Promise<CorrectionEvidence> {
  const issue = revision.issue;
  const [title, body, diff, verification] = await Promise.all([
    d.tracker.getIssue(issue).then(
      (row) => row?.title,
      () => undefined,
    ),
    d.tracker.issueBody(issue).then(
      (text) => text,
      () => undefined,
    ),
    d.tracker.prDiff(revision.prUrl).then(
      (parsed) => parsed,
      () => undefined,
    ),
    d.tracker.verifyPr(revision.prUrl, revision.headSha).then(
      (result) => result,
      () => undefined,
    ),
  ]);
  // Earlier rounds on this PR, oldest first, with what became of each. This
  // round's own findings are carried separately and verbatim, so they are
  // excluded here rather than repeated.
  const priorRounds: AdjudicationRound[] = d.store
    .reviewRevisionsForPr(d.project.name, revision.prUrl)
    .filter((prior) => prior.round < revision.round)
    .sort((a, b) => a.round - b.round)
    .map((prior) => ({
      round: prior.round,
      findings: prior.findings,
      ...(prior.outcome === undefined ? {} : { outcome: prior.outcome }),
    }));
  const review = resolveReview(d.project);
  return {
    repo: knowledgeRepoKey(repo),
    issue,
    ...(title === undefined ? {} : { issueTitle: title }),
    issueBody: body ?? "(the issue body could not be read — the findings below are your contract)",
    prUrl: revision.prUrl,
    headSha: revision.headSha,
    branch: args.branch,
    worktree: args.worktree,
    checks:
      verification === undefined
        ? "(the check state at this head could not be read)"
        : `${verification.status} — ${verification.reason}`,
    diff:
      diff === undefined
        ? "(the diff could not be read — read it from the worktree before you change anything)"
        : renderDiffForAdjudication(diff),
    round: revision.round,
    maxRounds: review.maxRounds,
    findings: revision.findings,
    priorRounds,
    reason: args.reason,
    // The gates this worker may actually run (#1042): the CI-owned ones are
    // not proof it can produce, and commanding one buys a refusal.
    proofCommands: workerGateGroups(repo).local.map((gate) => `${gate.cmd} (from ${gate.cwd})`),
    // `knowledge` is deliberately omitted: the transport reads the overlay for
    // `evidence.repo` itself, so one reader owns the cap and the heading.
  };
}

/**
 * Run one review-correction round on an already-claimed run row
 * (`pushed-green` → `running` by the dispatch pass), in whichever of the two
 * transports #1047's decision names — and settle the row the way
 * {@link handleIssue} settles any worker.
 *
 * `resume-original` resumes the SAME OMP session: same session directory,
 * `resume: true` at the harness, the `sessionFile` lineage compared afterwards,
 * on the same branch and PR. `fresh-correction` opens a blank session on that
 * same branch, worktree, PR and reviewed head, briefed from durable evidence
 * instead of a transcript (#1046), on the model the fleet resolves today.
 *
 * Both share one identity, and that is the point of the whole lane: the run row
 * is reused, never cloned — same attempt number, same charges, same prUrl, same
 * head, same lease — so a correction can never read as a second attempt or open
 * a second PR. A revision round stays off both budgets because it is a state
 * transition of one already-green row, green → running → green, and no new row
 * exists for either counter to count. Only a correction worker that genuinely
 * settles terminal falls through to the ordinary failure handling, budgets and
 * all, exactly as today.
 *
 * `launch` is passed by the dispatch pass, which decided it from the pre-claim
 * row. Absent (a direct caller, a test), the decision is resolved here from the
 * store — the recorded one when one exists, so a re-entry after a restart still
 * cannot re-decide.
 */
export async function handleReviewRevision(
  d: Deps,
  revision: ReviewRevisionRecord,
  launch?: ReviewLaunchDecision,
): Promise<void> {
  const { project, caps, tracker, store } = d;
  const run = store.getRun(revision.runId);
  if (run === undefined || run.state !== "running") {
    log(
      `#${revision.issue} review round ${revision.round} skipped: run ${revision.runId} is ${run?.state ?? "gone"} — ` +
        "the row moved between the dispatch claim and the resume.",
    );
    store.settleReviewRevision(revision.id, "skipped", Date.now());
    return;
  }
  const issue = run.issue;
  const branch = run.branch;
  const runId = run.id;
  const inProgress = project.stateLabels.inProgress;
  const repo = Object.values(project.routing.repos).find((candidate) => candidate.name === run.repo);
  if (repo === undefined) {
    log(`#${issue} review round ${revision.round} cannot dispatch: repo ${run.repo} is no longer routed`);
    store.updateRun(runId, { state: "pushed-green", endedAt: Date.now() });
    store.settleReviewRevision(revision.id, "skipped", Date.now());
    return;
  }
  const mirrorPath = mirrorPathFor(repo, project.mirrorRoot);

  // The transport for this round (#1047). The dispatch pass decided it from the
  // pre-claim row; a direct caller gets the recorded decision, or — for a round
  // nobody has decided — a selection made from the row as it stands, which
  // after the claim can only ever be more conservative.
  const decision = launch ?? resolveReviewLaunch(d, revision, run);
  const fresh = decision.launchMode === "fresh-correction";

  // The session a resumed round continues is the one the pushed-green row
  // recorded. A resume without that transcript cannot continue the same session
  // — the exact fake this feature exists to prevent — so fail loudly instead of
  // silently starting a fresh one. A `fresh-correction` round is not that case:
  // it never claimed to continue anything, its evidence comes from durable rows
  // and the live PR, and a missing transcript is one of the facts that selected
  // it (#1047). Escalating there would page a human about the absence of a
  // thing this round does not use.
  const priorSessionFile = run.sessionFile;
  if (!fresh && (priorSessionFile === undefined || !existsSync(priorSessionFile))) {
    log(
      `#${issue} review round ${revision.round} cannot resume: the recorded transcript ` +
        `${priorSessionFile ?? "(none)"} is gone`,
    );
    store.updateRun(runId, {
      state: "pushed-green",
      endedAt: Date.now(),
      lastError: `review round ${revision.round} could not resume: the run's transcript is gone`,
    });
    store.settleReviewRevision(revision.id, "skipped", Date.now());
    await safeEscalate(d, {
      tier: 1,
      project: project.name,
      issue,
      runId,
      summary: `#${issue} review round ${revision.round} could not resume — the transcript is missing`,
      detail: [
        revision.prUrl,
        "",
        "The PR is still open and green; the run row was returned to pushed-green.",
        "Resume the manual path (close + unblock + continuation) or merge it as it stands.",
      ].join("\n"),
    });
    return;
  }
  // The run's own session directory either way: a resumed round continues the
  // newest transcript in it, and a fresh correction writes its own beside that
  // one — the lineage of one run stays in one place, and the next round's
  // `continueRecent` finds the correction rather than the transcript it
  // replaced. Only one session per row ever runs at a time, so the shared
  // control socket and staged settings are still one session's.
  const sessionDir =
    priorSessionFile === undefined
      ? join(stateDir(), "sessions", `run-${String(runId)}`)
      : dirname(priorSessionFile);
  mkdirSync(sessionDir, { recursive: true });

  let worktreePath: string | undefined;
  let runRepo: RunRepoRef | undefined;
  let turnLimit: TurnLimitController | undefined;
  let workerControl: WorkerControlSlot | undefined;
  let workerSessionInstalled = false;
  let verbListener: VerbListener | undefined;
  const repoSlug = githubRepo(repo.cloneUrl);

  /**
   * Publishes the run's branch on the privileged side: run repo → mirror →
   * GitHub, fast-forward only — the same route `handleIssue` uses, so the
   * revised head the worker pushed through its own `conductor_push` is
   * re-verified against the remote before the tree dies.
   */
  const publish: RunPublisher = async () => {
    if (runRepo === undefined) return { ok: false, stderr: "the run repository was never provisioned" };
    return pushRunBranch(project, runRepo);
  };

  const settleStopBeforeSession = async (): Promise<boolean> => {
    const reason = workerControl?.requestedStop();
    if (reason === undefined || workerSessionInstalled) return false;
    turnLimit?.close();
    turnLimit = undefined;
    recordOperatorStop(store, {
      project: project.name,
      issue,
      runId,
      inProgress,
      reason,
      patch: {
        endedAt: Date.now(),
        report: ["Operator stopped the review revision before its session started.", `Reason: ${reason}`].join("\n"),
      },
    });
    store.settleReviewRevision(revision.id, "skipped", Date.now());
    log(`#${issue} review round ${revision.round} stopped by operator before its session started: ${reason}`);
    return true;
  };

  const settleDrainBeforeSession = async (): Promise<boolean> => {
    if (d.drain?.draining !== true || workerSessionInstalled) return false;
    turnLimit?.close();
    turnLimit = undefined;
    // A shutdown that interrupts the wake is the same situation as the
    // provisioning-failure branch below, not an operator stop — nobody
    // stopped anything, the PR is still green and still open, and only the
    // wake failed; `settleStopBeforeSession` immediately above covers the
    // case where an operator genuinely did. So mirror that branch: restore
    // `pushed-green` naming the shutdown, and leave the in-progress label
    // alone — the ordinary settle sweep releases it in the same breath it
    // terminalises the row when the PR resolves. Stopping the row here would
    // strand it: `settlePushedGreen` sweeps only pushed-* rows and
    // classification excludes `stopped`, so the issue would carry neither
    // label and nothing would ever revisit the still-green PR.
    //
    // The round itself is RE-QUEUED rather than settled `skipped` (#903). A
    // daemon shutdown is the plainest infrastructure kill there is — no
    // session started, so no round was worked — and a settled row consumed
    // one of the PR's rounds through `latestReviewRound`, which is how an
    // upgrade or a restart used to cost a review round it never spent. The
    // retry is counted and bounded exactly like any other infra kill, so a
    // host that shuts down mid-wake forever still escalates.
    store.updateRun(runId, {
      state: "pushed-green",
      endedAt: Date.now(),
      lastError: "daemon shutdown began after the review revision claim; the round was not launched",
    });
    const retry = store.retryReviewRevisionAfterInfra(revision.id, REVIEW_ROUND_INFRA_MAX_RETRIES);
    if (retry.kind === "exhausted") {
      store.settleReviewRevision(revision.id, "skipped", Date.now());
      log(
        `#${issue} review round ${revision.round} not launched: daemon shutdown began after the claim, and ` +
          `its infra retries are exhausted (${retry.retries}/${REVIEW_ROUND_INFRA_MAX_RETRIES}) — settled skipped`,
      );
      return true;
    }
    log(
      `#${issue} review round ${revision.round} not launched: daemon shutdown began after the claim — ` +
        `re-queued, retry ${retry.retries}/${REVIEW_ROUND_INFRA_MAX_RETRIES}`,
    );
    return true;
  };

  // Live-run controls, opened before any provisioning so the operator stop
  // and shutdown fences cover the whole wake window, exactly as they do for a
  // fresh claim in `handleIssue` (#374). The original run's entries were
  // closed at its settle, so reopening by issue is safe.
  //
  // #747: the round gets its own bounded allowance instead of the original
  // attempt's remainder. The resumed session continues the run's own counter
  // — the attempt's spent turns travel with the session — so a ceiling of
  // `run.maxTurns` would leave the round only `run.maxTurns - run.turns`,
  // and returning a PR from a near-cap run could not finish. Open at the
  // run's ceiling, then raise it by the round's own allowance — the run's
  // ceiling again, capped by the project's hard ceiling — through the same
  // override path `omp-conductor extend` uses, so the raise is persisted on
  // the row (the run's budget stays explainable in status) while `turns`
  // keeps accumulating the run's true cost: nothing is reset.
  turnLimit = d.turnLimits.open(project.name, issue, runId, run.maxTurns);
  const revisionAllowance = Math.min(run.maxTurns, caps.workerMaxTurnsCeiling);
  const revisionCeiling = run.turns + revisionAllowance;
  const revisionRaise = d.turnLimits.extend(project.name, issue, revisionCeiling);
  if (revisionRaise.kind === "extended") {
    log(
      `#${issue} review round ${revision.round} turn ceiling ${run.maxTurns} → ${revisionCeiling} ` +
        `(+${revisionAllowance} round allowance on ${run.turns} spent)`,
    );
  }
  workerControl = d.workerControls.open(project.name, issue, runId);
  if (await settleStopBeforeSession()) return;
  if (await settleDrainBeforeSession()) return;

  try {
    // Reattach the run's own branch at the same per-issue path the run used:
    // a `pushed-green` settle removed the worktree, so provisioning is the
    // same continuation reattach as a normal re-claim. A capped/failed run's
    // tree was RETAINED at its terminal settle (`tree: "keep"`, #795) — the
    // reviewed PR is green, so the branch already holds the work and the kept
    // tree holds nothing the branch does not, unless its salvage failed. Clear
    // it the way restart recovery does (#692): re-attempt the salvage through
    // `tree: "remove"`, and refuse the removal if that still fails — the tree
    // is then the only copy of work and is never destroyed. A provisioning
    // failure restores the row — the PR is still green and still open, and
    // only the wake failed — and says so.
    try {
      if (run.worktree !== "" && existsSync(run.worktree)) {
        const keptSettlement = await settleWorktree({
          issue: run.issue,
          attempt: run.attempt,
          ending: `review round ${revision.round} resume`,
          worktree: run.worktree,
          branch: run.branch,
          publish: (branch) => pushRunBranch(project, { repo, runRepoPath: run.worktree, branch }),
          tree: "remove",
          mirrorPath,
        });
        if (keptSettlement.retained) {
          log(
            `#${issue} review round ${revision.round} retained worktree ${run.worktree}: ` +
              "its salvage failed, so the tree is the only copy of work and will not be removed",
          );
          store.updateRun(runId, {
            state: "pushed-green",
            endedAt: Date.now(),
            lastError: `review round ${revision.round} could not clear the run's retained worktree: ${keptSettlement.lines.join(" ")}`,
          });
          store.settleReviewRevision(revision.id, "skipped", Date.now());
          await safeEscalate(d, {
            tier: 1,
            project: project.name,
            issue,
            runId,
            summary: `#${issue} review round ${revision.round} could not be dispatched — the run's retained worktree is the only copy of work`,
            detail: [revision.prUrl, "", ...keptSettlement.lines].join("\n"),
          });
          return;
        }
      }
      const provisioned = await addRunRepo(repo, project.mirrorRoot, project.workspaceRoot, issue, branch);
      worktreePath = provisioned.path;
      runRepo = { repo, runRepoPath: worktreePath, branch };
    } catch (err) {
      const detail = errText(err);
      log(`#${issue} review round ${revision.round} cannot provision ${branch}: ${detail}`);
      // An operator stop that landed during provisioning wins over restoration —
      // it is a newer, explicit action, and the fences above already honoured
      // one that landed earlier.
      const pendingStop = workerControl?.requestedStop();
      if (pendingStop === undefined) {
        store.updateRun(runId, {
          state: "pushed-green",
          endedAt: Date.now(),
          lastError: `review round ${revision.round} could not provision the worktree: ${detail}`,
        });
        store.settleReviewRevision(revision.id, "skipped", Date.now());
        await safeEscalate(d, {
          tier: 1,
          project: project.name,
          issue,
          runId,
          summary: `#${issue} review round ${revision.round} could not be dispatched — worktree provisioning failed`,
          detail: [revision.prUrl, "", detail].join("\n"),
        });
      } else {
        recordOperatorStop(store, {
          project: project.name,
          issue,
          runId,
          inProgress,
          reason: pendingStop,
          patch: {
            endedAt: Date.now(),
            report: [
              `Operator stopped the review revision before its session started: ${pendingStop}`,
              `(worktree provisioning also failed: ${detail})`,
            ].join("\n"),
          },
        });
        store.settleReviewRevision(revision.id, "skipped", Date.now());
      }
      return;
    }
    if (await settleStopBeforeSession()) return;
    if (await settleDrainBeforeSession()) return;

    // The fleet-owned omp settings overlay, staged under the SAME session
    // directory the transcript lives in — a resumed session loads it exactly
    // as the original run did (#537).
    const ompSettingsFile = materializeOmpSettings(project, sessionDir);

    // The run's mutation channel (#126): same run id, same issue, same repo —
    // so `conductor_push` publishes the run's own branch and nothing else.
    verbListener = await listenVerbChannel(
      verbDeps(d),
      {
        kind: "run",
        path: verbSocketPath(ensureVerbSocketDir(stateDir()), `run-${String(issue)}`),
        project: project.name,
        role: "worker",
        runId,
        issue,
        repo,
        runRepoPath: worktreePath,
        branch,
      },
      {
        ...(d.verbPeerReader === undefined ? {} : { peerReader: d.verbPeerReader }),
      },
    );
    if (await settleStopBeforeSession()) return;
    if (await settleDrainBeforeSession()) return;

    // A review revision is a new worker process on the same run row. Clear the
    // prior process evidence before launch so a failed wake cannot inherit an
    // earlier round's harness route.
    store.updateRun(runId, { worktree: worktreePath, workerPid: null, terminalEvidence: null });

    // The row's own counters are cumulative across the revision: the revision
    // worker meters its own session from zero, so its deltas are added to the
    // totals the run already recorded.
    const baseTurns = run.turns;
    const baseSpend = run.spendUsd;
    const baseOutputTokens = run.outputTokens ?? 0;
    const baseReasoningTokens = run.reasoningTokens ?? 0;

    log(
      `#${issue} review round ${revision.round} → ${decision.launchMode} on ${branch} ` +
        `(${fresh ? "fresh session" : `resuming ${priorSessionFile ?? "(none)"}`}, model requested ` +
        `${decision.requestedModel ?? "harness default"}): ${decision.reason}`,
    );

    let workerStartedAt: number | undefined;
    let result: WorkerResult;
    try {
      const runAllowanceUsd = runSpendAllowanceUsd(caps);
      // Everything except the brief, the transcript decision and the model is
      // shared by both transports, and shared by construction rather than by
      // two lists that have to be kept equal: a correction that metered its
      // turns, spend, tokens or pauses differently from a resume would be a
      // second accounting path for one run row, which is exactly what #1047
      // forbids.
      const common = {
        cwd: worktreePath,
        caps,
        ...(repoSlug === undefined ? {} : { repoSlug }),
        maxTurns: () => turnLimit?.maxTurns() ?? run.maxTurns,
        // A review round meters its own session from zero and gets the same
        // allowance as any run (#851) — the round is real work with a real
        // cost, and an unbounded round would reopen the hole for any PR that
        // was ever returned for revision.
        ...(runAllowanceUsd === null ? {} : { maxSpendUsd: runAllowanceUsd }),
        onPauseControl: (control: WorkerPauseControl) => {
          workerSessionInstalled = true;
          workerControl?.install(control);
        },
        sessionDir,
        socketPath: join(sessionDir, "ipc.sock"),
        verbSocketPath: verbListener.path,
        onSpawn: (pid: number) => {
          workerStartedAt = Date.now();
          verbListener?.bindPid(pid);
          store.updateRun(runId, { workerPid: pid });
        },
        onChildLog: (line: string) => {
          log(`#${issue} ${line}`);
        },
        ...(ompSettingsFile === undefined ? {} : { ompSettingsFile }),
        releaseGrants: resolveReleaseGrants(project),
        onReleaseBlocked: workerReleaseBlockRecorder(project.name, issue, runId),
        onTurn: (n: number) => store.updateRun(runId, { turns: baseTurns + n }),
        onSpend: (usd: number) => store.updateRun(runId, { spendUsd: baseSpend + usd }),
        // A round runs on the SAME run, so its tokens add to the attempt's
        // totals exactly as its turns and spend do (#518) — resetting them would
        // make a long deliberation look like a fresh one.
        onTokens: (tokens: { output: number; reasoning: number }) =>
          store.updateRun(runId, {
            outputTokens: baseOutputTokens + tokens.output,
            reasoningTokens: baseReasoningTokens + tokens.reasoning,
          }),
        onKilled: () => {
          turnLimit?.close();
          turnLimit = undefined;
        },
        maySpawn: () => d.drain?.draining !== true,
      };
      if (fresh) {
        // #1046's transport: no `resume` — the option does not exist on it — a
        // brief built from durable evidence instead of a transcript, and the
        // model dispatch resolved TODAY rather than the one the rejected run
        // ran on. Same branch, same worktree, same PR, same reviewed head.
        const evidence = await correctionEvidenceFor(d, revision, run, repo, {
          branch,
          worktree: worktreePath,
          reason: decision.reason,
        });
        const correction = await runCorrectionWorker(
          {
            ...common,
            evidence,
            ...(decision.originSessionRef === undefined
              ? {}
              : { originSessionRef: decision.originSessionRef }),
            ...(decision.requestedModel === undefined ? {} : { model: decision.requestedModel }),
            onSessionFile: (f) => {
              store.updateRun(runId, { sessionFile: f });
            },
          },
          d.workerDeps,
        );
        result = correction;
        // The lineage is recorded only when the transport confirms it opened a
        // session of its own: `freshSession: false` means the harness handed
        // back the transcript we deliberately did not ask to continue, and
        // recording that as a correction lineage is #1046's named silent fake.
        if (correction.freshSession && correction.sessionRef !== undefined) {
          store.recordReviewCorrectionSession(
            project.name,
            revision.id,
            correction.sessionRef,
            correction.model,
          );
          log(
            `#${issue} review round ${revision.round} fresh correction session ${correction.sessionRef} ` +
              `ran on ${correction.model ?? "an unreported model"}` +
              `${correction.provider === undefined ? "" : ` (${correction.provider})`} ` +
              `(requested ${decision.requestedModel ?? "harness default"})`,
          );
        } else {
          log(
            `#${issue} review round ${revision.round} fresh correction did NOT open its own session ` +
              `(${correction.sessionRef ?? "no transcript reported"}): no correction lineage recorded`,
          );
        }
      } else {
        result = await runWorker(
          {
            ...common,
            brief: renderReviewRevisionPrompt(revision.findings, revision.round),
            // The same-session half of the transport: `continueRecent` on the
            // run's own session directory, never a fresh `run-<uuid>` transcript.
            resume: true,
            // The continuation stays on the model the green run used (#286);
            // a resume onto a NEW model string is the diagnosed #1035 failure,
            // and the fresh transport above is how a model change is honoured.
            ...(run.model === undefined ? {} : { model: run.model }),
            onSessionFile: (f) => {
              store.updateRun(runId, { sessionFile: f });
              // Same-file lineage is the proof the resume happened: a revision
              // that opened a different transcript fell back to a fresh session.
              if (f !== priorSessionFile) {
                log(
                  `#${issue} review round ${revision.round} resume fell back to a fresh session: opened ${f} ` +
                    `instead of ${priorSessionFile}`,
                );
              }
            },
          },
          d.workerDeps,
        );
      }
    } finally {
      turnLimit?.close();
      turnLimit = undefined;
    }

    if (result.modelFallbackMessage !== undefined) {
      log(`#${issue} review round ${revision.round} model fallback: ${result.modelFallbackMessage}`);
    }
    // What this round learned about the repo, into the fleet overlay the next
    // brief renders. A correction round is where the sharpest discoveries come
    // from — the worker has just been told what it got wrong — and a knowledge
    // write can never fail a settlement.
    recordWorkerDiscoveries(knowledgeRepoKey(repo), issue, result.discoveries);

    const verified: { state: RunState; reason?: string } =
      result.state === "pushed-green"
        ? await verifyPushedGreenClaim(tracker, result, {
            project: project.name,
            issue,
            runId,
            branch,
            repo: repoSlug ?? repo.name,
            store,
          })
        : { state: result.state };
    const state = verified.state;

    const settlement =
      state === "pushed-green" || state === "pushed-pending" || state === "merged"
        ? undefined
        : await settleWorktree({
            issue,
            attempt: run.attempt,
            ending:
              state === "blocked"
                ? "blocked for an operator decision"
                : state === "stopped"
                  ? `stopped by the operator: ${result.stoppedReason ?? "no reason recorded"}`
                  : endedBy(result.killedBy),
            worktree: worktreePath,
            branch,
            publish,
            ...(state === "failed" || state === "killed"
              ? ({ tree: "keep" } as const)
              : ({ tree: "remove", mirrorPath } as const)),
          });
    if (settlement === undefined) {
      const published = await publish(branch);
      if (!published.ok) {
        log(`#${issue} publish before removal failed, work is preserved in the mirror: ${published.stderr}`);
      }
      await removeWorktree(mirrorPath, worktreePath);
    }

    // The settlement report names the review round first (#692): a run that
    // reaches green again after a round must read as a revision outcome, not
    // as a fresh run's first push, and a round that failed must still say the
    // round — the state and classification below carry the failure itself.
    const finalReport = [
      `review round ${revision.round}: ${state}`,
      ...(verified.reason === undefined ? [] : ["", verified.reason]),
      result.report,
    ].join("\n");
    const latestRun = store.getRun(runId) ?? run;
    const terminalEvidence =
      state === "failed" || state === "killed"
        ? readTerminalEvidence(
            {
              sessionFile: result.sessionFile,
              startedAt: workerStartedAt ?? latestRun.startedAt,
              ...(latestRun.workerPid === undefined ? {} : { workerPid: latestRun.workerPid }),
            },
            d.harnessLogDir ?? harnessLogDir(),
          )
        : undefined;

    const terminalPatch: Partial<RunRecord> = {
      endedAt: Date.now(),
      turns: baseTurns + result.turns,
      spendUsd: baseSpend + result.spendUsd,
      provider429Count: result.provider429Count,
      ...(result.model === undefined ? {} : { resolvedModel: result.model }),
      ...(result.provider === undefined ? {} : { resolvedProvider: result.provider }),
      retryFallbacks: result.retryFallbacks,
      retryFallbackSucceeded: result.retryFallbackSucceeded,
      modelRecoveries: result.modelRecoveries,
      autoRetryCount: result.autoRetryCount,
      autoCompactionCount: result.autoCompactionCount,
      // The worker only reports these when it actually established them; a
      // failure whose report named no PR must not wipe what the row already
      // owns (#468).
      ...(result.prUrl === undefined ? {} : { prUrl: result.prUrl }),
      ...(result.headSha === undefined ? {} : { headSha: result.headSha }),
      sessionFile: result.sessionFile,
      ...(result.graphTools === undefined ? {} : { graphTools: result.graphTools }),
      ...(terminalEvidence === undefined ? {} : { terminalEvidence }),
      report: finalReport,
      ...settlement?.patch,
    };
    const terminalClassification =
      terminalEvidence === undefined
        ? undefined
        : classifyRun(
            {
              ...latestRun,
              ...terminalPatch,
              state,
              terminalEvidence,
            },
            {},
            caps,
          );
    if (terminalClassification?.cls === "model-empty-stop") {
      const recoveredAt = Date.now();
      store.updateRun(runId, {
        ...terminalPatch,
        state: "failed",
        failureClass: terminalClassification.cls,
        recoveryAction: terminalClassification.recovery,
        recoveredAt,
      });
      const continuations = store.continuationsFor(project.name, issue);
      if (hasContinuationBudget(continuations, caps.maxContinuationsPerIssue)) {
        // Keep the same review row, round, findings, session, branch and PR.
        // Restore the exact pushed-green origin state while moving this stop
        // into the explicit continuation counter; the next claim must not
        // depend on re-verifying a failed-state origin at an unchanged head.
        const chargedRun = store.getRun(runId) ?? latestRun;
        store.updateRun(runId, {
          ...terminalPatch,
          state: "pushed-green",
          failureClass: null,
          recoveryAction: null,
          recoveredAt: null,
          continuationCharges: (chargedRun.continuationCharges ?? 0) + 1,
          workerPid: null,
          lastError: terminalClassification.evidence,
        });
        store.requeueReviewRevision(revision.id);
        log(
          `#${issue} review round ${revision.round} re-queued after a provider empty-stop ` +
            `(${continuations}/${caps.maxContinuationsPerIssue} continuations): ${terminalClassification.evidence}`,
        );
        return;
      }

      store.settleReviewRevision(revision.id, "failed", recoveredAt);
      swapLabel(store, project.name, issue, inProgress, project.stateLabels.failed);
      const model = result.model ?? latestRun.resolvedModel ?? latestRun.model ?? "configured model";
      const provider = result.provider ?? latestRun.resolvedProvider ?? "provider";
      await safeEscalate(d, {
        tier: 1,
        project: project.name,
        issue,
        runId,
        summary: `#${issue} review round ${revision.round}: the model kept returning empty responses`,
        detail:
          `${provider}/${model} ended ${continuations} review session(s) with empty responses until ` +
          `the harness retry cap. The same review round was preserved while continuation budget remained; ` +
          `that budget is now exhausted at ${caps.maxContinuationsPerIssue}. No implementation failure or ` +
          `additional review round was charged.\n\n${terminalClassification.evidence}`,
      });
      log(`#${issue} review round ${revision.round} provider empty-stops exhausted the continuation budget`);
      return;
    }
    // #903: an infrastructure kill spends no review round.
    //
    // A dispatch that dies before the resumed session takes a turn — a host
    // permission fault, a spawn failure, a harness that could not start — did
    // none of the work a round exists to do, yet the pre-#903 path settled the
    // row `failed` with no retry. During the 2026-08-21/22 outage that turned
    // one broken mount into a review-ceiling deadlock: three such deaths
    // consumed the PR's three rounds, `conductor_pr_review` then refused any
    // further round, and the findings could never be addressed by a worker at
    // all. Nothing in the loop could recover it.
    //
    // So the round is returned to the pending set instead: the same row, the
    // same round number, the same findings, and the run restored to the exact
    // `pushed-green` state the verb recorded — which is what the restart
    // reconcile already does for a round a dying daemon interrupted. The next
    // dispatch pass resumes the same session. Bounded by
    // REVIEW_ROUND_INFRA_MAX_RETRIES and counted durably on the row, so a
    // permanently broken host escalates through the ordinary failed path
    // instead of retrying every tick forever.
    const infraKill =
      state === "failed" || state === "killed"
        ? reviewRoundNeverWorked({
            turns: result.turns,
            ...(result.prUrl === undefined ? {} : { prUrl: result.prUrl }),
            ...(result.headSha === undefined ? {} : { headSha: result.headSha }),
            ...(settlement?.patch?.salvageSha === undefined
              ? {}
              : { salvageSha: settlement.patch.salvageSha }),
          })
        : undefined;
    if (infraKill !== undefined) {
      const retry = store.retryReviewRevisionAfterInfra(revision.id, REVIEW_ROUND_INFRA_MAX_RETRIES);
      if (retry.kind === "requeued") {
        const detail =
          `review round ${revision.round} was killed before its session took a turn ` +
          `(${infraKill}); the round is re-queued, retry ${retry.retries}/${REVIEW_ROUND_INFRA_MAX_RETRIES}`;
        store.updateRun(runId, {
          ...terminalPatch,
          state: "pushed-green",
          // The transcript the round must still resume is the one the run
          // recorded, not whatever this dead dispatch opened: a session that
          // never took a turn has no lineage of its own to inherit.
          sessionFile: priorSessionFile,
          report: [`review round ${revision.round}: dispatch killed before turn 1`, "", detail].join("\n"),
          lastError: detail,
        });
        // No label swap and no escalation: nothing failed that a human can
        // act on yet, the issue keeps its in-progress label, and the next
        // dispatch pass picks the round up on its own.
        log(`#${issue} ${detail}`);
        return;
      }
      log(
        `#${issue} review round ${revision.round} exhausted its infra retries ` +
          `(${retry.retries}/${REVIEW_ROUND_INFRA_MAX_RETRIES}) — settling it the ordinary way`,
      );
    }
    if (state === "stopped") {
      recordOperatorStop(store, {
        project: project.name,
        issue,
        runId,
        inProgress,
        reason: result.stoppedReason ?? "no reason recorded",
        patch: terminalPatch,
      });
    } else {
      const sessionErr = state === "failed" || state === "killed" ? readSessionError(result.sessionFile) : undefined;
      const providerCredit = sessionErr === undefined ? undefined : providerCreditRefusal(sessionErr);
      const providerTransient =
        providerCredit !== undefined || sessionErr === undefined ? undefined : providerTransientFault(sessionErr);
      const lastError = completionLastError(providerCredit, providerTransient, verified.reason, sessionErr);
      store.updateRun(runId, {
        ...terminalPatch,
        state,
        ...(lastError === undefined ? {} : { lastError }),
      });
    }

    const outcome: ReviewRevisionOutcome =
      state === "pushed-green" ? "revised" : state === "pushed-pending" ? "pending" : "failed";
    store.settleReviewRevision(revision.id, outcome, Date.now());
    log(`#${issue} review round ${revision.round} ${state}${result.prUrl ? ` ${result.prUrl}` : ""}`);

    // Post-settle outcomes, mirroring handleIssue: a failed or killed revision
    // worker falls through to exactly the same label, requeue and escalation
    // handling as any failed run — budgets included — and a blocked one pages
    // the orchestrator for a decision.
    if (state === "stopped") {
      log(`#${issue} review round ${revision.round} stopped by operator: ${result.stoppedReason}`);
    } else if (state === "blocked") {
      swapLabel(store, project.name, issue, inProgress, project.stateLabels.blocked);
      wakeOrchestratorForBlockedRun(project.name, issue);
      await safeEscalate(d, {
        tier: 1,
        project: project.name,
        issue,
        runId,
        summary: `#${issue} is blocked on review round ${revision.round} and needs a decision`,
        detail: [`${result.prUrl ?? "(no PR URL)"}`, "", result.report].join("\n"),
      });
    } else if (state === "failed" || state === "killed") {
      const continuation = store.continuationsFor(project.name, issue);
      const continueTurns = shouldContinueAfterTurnsCap({
        killedBy: result.killedBy,
        prUrl: result.prUrl,
        headSha: result.headSha,
        salvageSha: settlement?.patch?.salvageSha,
        continuation,
        maxContinuations: caps.maxContinuationsPerIssue,
      });
      if (continueTurns) {
        // The same chokepoint the implementation path uses: one ordered pair
        // plus the dispatch wake, so a capped review round is picked up on an
        // immediate pass (#201, #1041).
        await swapToQueue(d, issue, inProgress);
        log(
          `#${issue} review round ${revision.round} turns-cap, continuation ` +
            `${continuation}/${caps.maxContinuationsPerIssue} — salvaged and re-queued`,
        );
        await safeEscalate(d, {
          tier: 1,
          project: project.name,
          issue,
          runId,
          summary: `#${issue} hit the turns cap on review round ${revision.round} — auto-requeued for continuation`,
          detail: [
            `${result.prUrl ?? "(no PR URL)"}`,
            "",
            "The queue label is back on; the next tick should reattach the branch",
            "and open a continuation brief. No failed label was applied.",
            "",
            result.report,
          ].join("\n"),
        });
      } else {
        swapLabel(store, project.name, issue, inProgress, project.stateLabels.failed);
        await safeEscalate(d, {
          tier: 1,
          project: project.name,
          issue,
          runId,
          summary: result.killedBy
            ? `#${issue} was killed on review round ${revision.round} by the ${result.killedBy} cap`
            : `#${issue} failed on review round ${revision.round}`,
          detail: [`${result.prUrl ?? "(no PR URL)"}`, "", result.report].join("\n"),
        });
      }
    }
  } catch (err) {
    // A crash anywhere after the worker ran settles like any run's dispatch
    // error: the row is terminal, the tree is kept, and the issue is relabelled
    // so nobody silently re-claims it.
    turnLimit?.close();
    turnLimit = undefined;
    const detail = errText(err);
    log(`#${issue} review round ${revision.round} errored: ${detail}`);
    const settlement =
      worktreePath === undefined
        ? undefined
        : await settleWorktree({
            issue,
            attempt: run.attempt,
            ending: "killed by a dispatch error",
            worktree: worktreePath,
            branch,
            publish,
            tree: "keep",
          });
    // A crash BEFORE the worker session was installed spent no round (#903):
    // nothing resumed, nothing was reviewed, and the round is re-queued for
    // the next pass instead of consuming one of the PR's three. Once the
    // session is installed the round belongs to the worker, and a crash after
    // that settles the ordinary way below — including the exhausted case, so
    // a dispatch that keeps throwing still reaches a human.
    if (!workerSessionInstalled) {
      const retry = store.retryReviewRevisionAfterInfra(revision.id, REVIEW_ROUND_INFRA_MAX_RETRIES);
      if (retry.kind === "requeued") {
        store.updateRun(runId, {
          state: "pushed-green",
          endedAt: Date.now(),
          sessionFile: priorSessionFile,
          lastError:
            `review round ${revision.round} could not be dispatched (${detail}); the round is re-queued, ` +
            `retry ${retry.retries}/${REVIEW_ROUND_INFRA_MAX_RETRIES}`,
          ...settlement?.patch,
        });
        log(
          `#${issue} review round ${revision.round} re-queued after a pre-session dispatch error — ` +
            `retry ${retry.retries}/${REVIEW_ROUND_INFRA_MAX_RETRIES}`,
        );
        return;
      }
      log(
        `#${issue} review round ${revision.round} exhausted its infra retries ` +
          `(${retry.retries}/${REVIEW_ROUND_INFRA_MAX_RETRIES}) — failing the round`,
      );
    }
    store.updateRun(runId, {
      state: "failed",
      endedAt: Date.now(),
      lastError: detail,
      ...settlement?.patch,
    });
    store.settleReviewRevision(revision.id, "failed", Date.now());
    swapLabel(store, project.name, issue, inProgress, project.stateLabels.failed);
    await safeEscalate(d, {
      tier: 1,
      project: project.name,
      issue,
      runId,
      summary: `#${issue} review round ${revision.round} could not be dispatched`,
      detail: [
        `${revision.prUrl}`,
        "",
        detail,
        ...(settlement?.lines ?? []),
      ].join("\n"),
    });
  } finally {
    turnLimit?.close();
    if (verbListener !== undefined) {
      try {
        await verbListener.close();
      } catch (err) {
        log(`#${issue} verb socket ${verbListener.path} did not close cleanly: ${errText(err)}`);
      }
      verbListener = undefined;
    }
    workerControl?.close();
    workerControl = undefined;
  }
}

/**
 * Restart recovery for in-flight review revisions (#692): a revision the
 * previous daemon claimed (`pushed-green` → `running`) and lost when it died
 * is restored to the exact state the orchestrator's verb left it in — same
 * run, same PR, same findings, same round, same session file — so the next
 * dispatch pass resumes the SAME OMP session rather than leaving the run to
 * the failure classifier.
 *
 * Runs AFTER the startup orphan sweep, on purpose: the sweep has already
 * salvaged the crashed revision's worktree to the branch (so no work is lost)
 * and marked the row `orphaned` — and this reconcile then restores that row
 * to `pushed-green` and re-queues the round before the first tick, so the
 * failure classifier never sees it. That ordering matters twice over: an
 * orphaned row would charge the continuation budget for a round that must
 * stay off it (#677), and the `review_revisions` row is the only place the
 * findings and round live. A revision the previous daemon never claimed (its
 * run is still `pushed-green`, its row still pending) needs no restoration —
 * the ordinary dispatch pass wakes it on the next tick, exactly as #677
 * already guarantees.
 *
 * A round whose run cannot be restored — the run is gone, settled terminal,
 * or its repo is no longer routed — fails closed instead: the revision is
 * settled `skipped` with the findings still on the row, ONE escalation names
 * what could not be restored, and no worker is started against a PR that may
 * already have one.
 */
export async function reconcileCrashedReviewRevisions(d: Deps): Promise<ReviewRevisionRecord[]> {
  const { project, store } = d;
  const revisions = store.unsettledReviewRevisions(project.name);
  if (revisions.length === 0) return [];
  const recovered: ReviewRevisionRecord[] = [];
  for (const revision of revisions) {
    const run = store.getRun(revision.runId);
    if (run === undefined) {
      log(
        `#${revision.issue} review round ${revision.round} cannot be restored across the restart: ` +
          `run ${revision.runId} no longer exists`,
      );
      store.settleReviewRevision(revision.id, "skipped", Date.now());
      await safeEscalate(d, {
        tier: 1,
        project: project.name,
        issue: revision.issue,
        runId: revision.runId,
        summary: `#${revision.issue} review round ${revision.round} could not be restored across the restart`,
        detail: [
          revision.prUrl,
          "",
          "The daemon restarted while this review round was in flight and the round cannot be re-dispatched:",
          `the run row ${revision.runId} no longer exists.`,
          "The round was settled skipped; the findings stay on this revision's durable row.",
          "Re-review the PR (it is still open and green) or merge it as it stands.",
        ].join("\n"),
      });
      continue;
    }
    // A revision the previous daemon never claimed: the run is still a
    // settled green row and the round is still pending, so the ordinary
    // dispatch pass wakes it on the next tick untouched. Same for a
    // capped/failed run whose review round is pending (#795): a `failed` /
    // `killed` row is terminal too — no process died on this round, the
    // revision was never claimed, and the ordinary dispatch pass claims it
    // exactly like a pending round on a pushed-green row.
    if (run.state === "pushed-green" || run.state === "failed" || run.state === "killed") continue;
    // The run a previous daemon claimed for this round and died on: the
    // orphan sweep just marked it `orphaned` (salvaging the tree to the
    // branch), so restore it to the reviewable state the verb recorded.
    const crashed =
      run.state === "claimed" || run.state === "running" || run.state === "orphaned";
    if (!crashed) {
      log(
        `#${revision.issue} review round ${revision.round} cannot be restored across the restart: run ` +
          `${revision.runId} is ${run.state}`,
      );
      store.settleReviewRevision(revision.id, "skipped", Date.now());
      await safeEscalate(d, {
        tier: 1,
        project: project.name,
        issue: revision.issue,
        runId: revision.runId,
        summary: `#${revision.issue} review round ${revision.round} could not be restored across the restart`,
        detail: [
          revision.prUrl,
          "",
          "The daemon restarted while this review round was in flight and the round cannot be re-dispatched:",
          `the run row is ${run.state}, not a state a review round resumes from.`,
          "The round was settled skipped; the findings stay on this revision's durable row.",
          "Re-review the PR (it is still open and green) or merge it as it stands.",
        ].join("\n"),
      });
      continue;
    }
    try {
      // The crashed revision's worktree — salvaged and published by the orphan
      // sweep that just ran, so the branch in the mirror has every committed
      // byte — must be gone before the next dispatch, or `addRunRepo` refuses
      // the reattach as possibly holding a previous attempt's work (#692).
      // `settleWorktree` with `tree: "remove"` re-attempts the salvage instead
      // of trusting it (a tree whose salvage failed is the only copy of work
      // and is NEVER removed), and refuses the removal if that re-attempt
      // fails — the fail-closed half of this recovery.
      if (run.worktree !== "" && existsSync(run.worktree)) {
        const repo = Object.values(project.routing.repos).find((candidate) => candidate.name === run.repo);
        if (repo === undefined) {
          log(
            `#${revision.issue} review round ${revision.round} cannot clear worktree ${run.worktree}: ` +
              `repo ${run.repo} is no longer routed — the tree may hold the only copy of work`,
          );
          store.settleReviewRevision(revision.id, "skipped", Date.now());
          await safeEscalate(d, {
            tier: 1,
            project: project.name,
            issue: revision.issue,
            runId: revision.runId,
            summary: `#${revision.issue} review round ${revision.round} could not be restored across the restart`,
            detail: [
              revision.prUrl,
              "",
              "The daemon restarted while this review round was in flight and the round cannot be re-dispatched:",
              `repo ${run.repo} is no longer routed, so the crashed revision's worktree cannot be published and removed.`,
              `The worktree is retained for recovery: ${run.worktree}`,
              "Recover it by hand (it may hold the round's last work), then re-review or merge the PR.",
            ].join("\n"),
          });
          continue;
        }
        const settlement = await settleWorktree({
          issue: run.issue,
          attempt: run.attempt,
          ending: "interrupted by a daemon restart during a review round",
          worktree: run.worktree,
          branch: run.branch,
          publish: (branch) => pushRunBranch(project, { repo, runRepoPath: run.worktree, branch }),
          tree: "remove",
          mirrorPath: mirrorPathFor(repo, project.mirrorRoot),
        });
        if (settlement.retained) {
          log(
            `#${revision.issue} review round ${revision.round} worktree ${run.worktree} retained: ` +
              "its salvage failed, so the tree is the only copy of work and will not be removed",
          );
          store.settleReviewRevision(revision.id, "skipped", Date.now());
          await safeEscalate(d, {
            tier: 1,
            project: project.name,
            issue: revision.issue,
            runId: revision.runId,
            summary: `#${revision.issue} review round ${revision.round} could not be restored across the restart`,
            detail: [
              revision.prUrl,
              "",
              "The daemon restarted while this review round was in flight and the round cannot be re-dispatched:",
              "the crashed revision's worktree could not be salvaged, so it is the only copy of work and was retained.",
              `Recover the tree by hand: ${run.worktree}`,
              "Then re-review the PR (still open and green) or merge it as it stands.",
            ].join("\n"),
          });
          continue;
        }
      }
    } catch (err) {
      // A settleWorktree throw (mirror unreachable, git refused) must not take
      // the whole startup reconcile down with one run's tree: fail this round
      // closed and let the next one try.
      const detail = errText(err);
      log(`#${revision.issue} review round ${revision.round} restore errored: ${detail}`);
      store.settleReviewRevision(revision.id, "skipped", Date.now());
      await safeEscalate(d, {
        tier: 1,
        project: project.name,
        issue: revision.issue,
        runId: revision.runId,
        summary: `#${revision.issue} review round ${revision.round} could not be restored across the restart`,
        detail: [revision.prUrl, "", detail].join("\n"),
      });
      continue;
    }
    // The exact state the verb recorded: a pushed-green row plus a pending
    // revision whose durable row still carries the findings, reviewed head,
    // round and target session. The next tick's dispatch pass claims and
    // wakes it exactly like a revision that was never dispatched.
    store.updateRun(revision.runId, {
      state: "pushed-green",
      endedAt: Date.now(),
      worktree: "",
      lastError: `review round ${revision.round} was restored across a daemon restart; the round is re-queued`,
    });
    store.requeueReviewRevision(revision.id);
    log(
      `#${revision.issue} review round ${revision.round} restored across restart: run ${revision.runId} ` +
        "is pushed-green again and the round is re-queued — the next dispatch pass resumes the same session",
    );
    recovered.push(revision);
  }
  return recovered;
}