/**
 * Taking one admitted issue to a running session, and closing the books when it
 * ends: the claim, the lease, the worktree, the launch, the settle.
 *
 * This is the module the whole dispatcher exists for, and it is drawn around a
 * single run's life rather than around a tick: `handleIssue` is called once per
 * admitted issue and owns every irreversible step in between, so the ordering
 * that makes a crash recoverable — provenance before label, label before
 * launch, lease before work — is readable in one place.
 *
 * The two in-process registries live here for the same reason. A turn-limit
 * controller and a worker-control slot are handles on a *running* session; they
 * are created by `runDaemon`, held by the HTTP surface, and consumed here, and
 * the only place that knows when a slot begins and ends is the launch path.
 *
 * The review-correction lane (`review.ts`) relaunches through the same helpers —
 * `swapLabel`, `endedBy`, `verifyPushedGreenClaim`, `shouldContinueAfterTurnsCap`,
 * `workerReleaseBlockRecorder` — and imports them from here. That direction is
 * deliberate and one-way: a correction is another launch of the same run, so it
 * borrows the launch path's rules rather than restating them.
 */
import { existsSync, mkdirSync, statSync } from "node:fs";
import { dirname, join } from "node:path";
import { hasContinuationBudget, runSpendAllowanceUsd } from "../admission.ts";
import { resolveReleaseGrants, stateDir } from "../config.ts";
import { oauthFenceVerdict } from "../credential-class.ts";
import { formatSettlementFlags, withDerivedChangedLine } from "../diff-flags.ts";
import { providerCreditRefusal, providerTransientFault } from "../failure-class.ts";
import { openWorkerPane, reportWorkerPaneState, retireWorkerPane, type WorkerPaneOutcome } from "../fleet.ts";
import { pushRunBranch, type RunRepoRef } from "../gitops.ts";
import { errText, log, safeEscalate } from "../log.ts";
import { DEFAULT_MODEL_FALLBACK_THRESHOLD, fallbackClause, providerFailureFacts, resolveDispatchModel } from "../model-fallback.ts";
import { materializeOmpSettings } from "../omp-settings.ts";
import { isPaused, setPaused } from "../pause.ts";
import { recordReleaseBlock, type GateShape, type ReleaseBlockContext } from "../release-policy.ts";
import { reliabilitySettlementLine } from "../reports.ts";
import { branchName, type Routed } from "../routing.ts";
import { collectSettlementFlags, reactToProviderCredit, readSessionError, recordOperatorStop, recordWorkerDiscoveries, settleWorktree, swapToQueue } from "../settlement.ts";
import type { EffectiveModel, FileLane, IssueComment, PrState, RunRecord, RunState, Store, Tracker } from "../types.ts";
import { listenVerbChannel, type VerbListener } from "../verbs/server.ts";
import { ensureVerbSocketDir, verbSocketPath } from "../verbs/socket.ts";
import { ORPHAN_RESUME_PROMPT, runWorker, type KilledBy, type WorkerPauseControl, type WorkerPausePhase, type WorkerResult } from "../worker.ts";
import { addRunRepo, mirrorPathFor, removeWorktree, worktreePathFor, type RunPublisher } from "../worktree.ts";
import { writeAdmissionAck } from "./ack.ts";
import { buildBrief, renderAnsweredBlockPrompt, renderDiscussion, resumeLaneBlock } from "./brief.ts";
import { completionLastError, exhaustedSessionReason, githubRepo, knowledgeRepoKey, repoSlug, verbDeps, type Deps, type TurnLimitController, type TurnLimitRegistry, type WorkerControlRegistry, type WorkerControlSlot } from "./deps.ts";
import { readDrain } from "./drain.ts";
import { workspaceOwnership } from "./panes.ts";
import { wakeOrchestratorForBlockedRun } from "./supervision.ts";

/**
 * Record a state-label swap for projection (#201). The add enqueues before the
 * remove — the reverse order would leave a window where the issue carries no
 * state label at all, which is exactly the shape `isEligible` treats as fresh
 * work. Synchronous: enqueueing is a local store write and cannot fail on the
 * tracker; the projector applies the pair in order and retries on refusal.
 */
export function swapLabel(store: Store, projectName: string, issue: number, from: string, to: string): void {
  store.enqueueLabelOps(projectName, [
    { issue, op: "add", label: to },
    { issue, op: "remove", label: from },
  ]);
}
/**
 * How a run's end is named — in the salvage commit, and to whoever reads it.
 * The whole clause, not a bare reason: a graceful block was not killed by
 * anything, and the commit subject is read during recovery.
 */
export function endedBy(killedBy: KilledBy | undefined): string {
  if (killedBy === "turns") return "killed by the turns cap";
  if (killedBy === "wallclock") return "killed by the wall-clock cap";
  if (killedBy === "spend") return "killed by the per-run spend cap";
  return "killed by a failed run";
}

/**
 * Authoritative live-run turn controls. Persistence happens synchronously
 * before the in-memory ceiling changes, so no turn event can interleave.
 */
export function createTurnLimitRegistry(
  persist: (runId: string, maxTurns: number) => void,
): TurnLimitRegistry {
  const active = new Map<string, { runId: string; maxTurns: number }>();
  const key = (project: string, issue: number): string => `${project}\0${issue}`;
  return {
    open(project, issue, runId, maxTurns) {
      const k = key(project, issue);
      if (active.has(k)) throw new Error(`#${issue} already has a live turn controller`);
      const entry = { runId, maxTurns };
      active.set(k, entry);
      return {
        maxTurns: () => entry.maxTurns,
        close: () => {
          if (active.get(k) === entry) active.delete(k);
        },
      };
    },
    extend(project, issue, maxTurns) {
      if (!Number.isSafeInteger(maxTurns) || maxTurns < 1) {
        throw new RangeError(`turn ceiling must be a positive integer, got ${maxTurns}`);
      }
      const entry = active.get(key(project, issue));
      if (entry === undefined) return { kind: "not-active" };
      if (maxTurns <= entry.maxTurns) {
        return { kind: "not-increase", runId: entry.runId, maxTurns: entry.maxTurns };
      }
      persist(entry.runId, maxTurns);
      entry.maxTurns = maxTurns;
      return { kind: "extended", runId: entry.runId, maxTurns };
    },
  };
}

/** Authoritative controls for sessions owned by this daemon process. */
export function createWorkerControlRegistry(): WorkerControlRegistry {
  interface Entry {
    project: string;
    issue: number;
    runId: string;
    control?: WorkerPauseControl;
    stopReason?: string;
    stopError?: string;
    finished: PromiseWithResolvers<void>;
    onPhase?: (phase: WorkerPausePhase) => void;
    /** Who asked for the live pause, and when — cleared on resume (#997). */
    pausedBy?: { source: string; at: number };
  }

  const active = new Map<string, Entry>();
  const key = (project: string, issue: number): string => `${project}\0${issue}`;
  return {
    open(project, issue, runId, onPhase) {
      const k = key(project, issue);
      if (active.has(k)) throw new Error(`#${issue} already has a live worker controller`);
      const entry: Entry = {
        project,
        issue,
        runId,
        finished: Promise.withResolvers<void>(),
        ...(onPhase === undefined ? {} : { onPhase }),
      };
      active.set(k, entry);
      return {
        install: (control) => {
          if (active.get(k) !== entry) return;
          entry.control = control;
          if (entry.stopReason === undefined) return;
          try {
            control.stop(entry.stopReason);
          } catch (err) {
            entry.stopError = err instanceof Error ? err.message : String(err);
          }
        },
        requestedStop: () => active.get(k) === entry ? entry.stopReason : undefined,
        close: () => {
          if (active.get(k) === entry) active.delete(k);
          entry.finished.resolve();
        },
      };
    },
    async pause(project, issue, source) {
      const entry = active.get(key(project, issue));
      if (entry?.control === undefined) return { kind: "not-active" };
      try {
        await entry.control.pause();
        const phase = entry.control.phase();
        entry.pausedBy = { source, at: Date.now() };
        entry.onPhase?.(phase);
        return { kind: "ok", runId: entry.runId, phase };
      } catch (err) {
        return {
          kind: "refused",
          runId: entry.runId,
          error: err instanceof Error ? err.message : String(err),
        };
      }
    },
    resume(project, issue) {
      const entry = active.get(key(project, issue));
      if (entry?.control === undefined) return { kind: "not-active" };
      try {
        entry.control.resume();
        const phase = entry.control.phase();
        delete entry.pausedBy;
        entry.onPhase?.(phase);
        return { kind: "ok", runId: entry.runId, phase };
      } catch (err) {
        return {
          kind: "refused",
          runId: entry.runId,
          error: err instanceof Error ? err.message : String(err),
        };
      }
    },
    async stop(project, issue, reason) {
      const entry = active.get(key(project, issue));
      if (entry === undefined) return { kind: "not-active" };
      if (entry.stopReason === undefined) {
        entry.stopReason = reason;
        if (entry.control !== undefined) {
          try {
            entry.control.stop(reason);
          } catch (err) {
            entry.stopReason = undefined;
            return {
              kind: "refused",
              runId: entry.runId,
              error: err instanceof Error ? err.message : String(err),
            };
          }
        }
      }
      await entry.finished.promise;
      if (entry.stopError !== undefined) {
        return { kind: "refused", runId: entry.runId, error: entry.stopError };
      }
      return { kind: "stopped", runId: entry.runId, reason: entry.stopReason! };
    },
    snapshot(project) {
      const workers: {
        issue: number;
        runId: string;
        phase: WorkerPausePhase;
        source?: string;
        pausedAtMs?: number;
      }[] = [];
      for (const entry of active.values()) {
        if (entry.project !== project || entry.control === undefined) continue;
        const phase = entry.control.phase();
        if (phase === "running") continue;
        workers.push({
          issue: entry.issue,
          runId: entry.runId,
          phase,
          ...(entry.pausedBy === undefined
            ? {}
            : { source: entry.pausedBy.source, pausedAtMs: entry.pausedBy.at }),
        });
      }
      return workers;
    },
  };
}

/**
 * The daemon half of a worker's `pushed-green` claim (#85's contact with
 * reality; #782's publication attribution).
 *
 * A yield is a transport, not proof of GitHub side effects: `claim` is
 * caller-supplied text and the worker adapter turns it into a green result
 * without the daemon ever having seen a mediated publication verb for it — the
 * shape that let the #777 incident guess a non-existent PR. The tracker read
 * at the end proves the PR is open and green at the exact head; the evidence
 * gate here proves that PR is *this run's own mediated work* first.
 *
 * Two things are asked, in cost order:
 *
 *  - the claimed URL must equal the one this run's row records, which is never
 *    a worker-supplied string at the moment a live run is verified —
 *    conductor_pr_create (create or adoption) writes it on the run, the daemon
 *    seeds it at claim from a terminal predecessor on the same branch (#434),
 *    and orchestrator-only recovery writes it for settled runs;
 *  - and the claimed head must have a mediated publisher: an allowed
 *    `conductor_push` on this exact run that published it on this run's branch
 *    (the verb only ever pushes `refs/heads/<branch>`), an allowed
 *    `conductor_pr_create` on this run for this exact PR, or — for a run that
 *    published nothing new — the live tip of this run's own branch.
 *
 * That last path is the one the row cannot supply. It used to be served by
 * comparing the claim against `run.prUrl`/`run.headSha`, and both are
 * worker-tainted upstream: terminal settlement writes a worker's reported pair
 * onto the row *before* any verification, and continuation inheritance
 * validates only the predecessor's metadata and open state — so a turn-capped
 * attempt that reported someone else's real green PR had that pair inherited
 * and re-presented as its own evidence. Asking the tracker for the branch tip
 * removes the worker from the loop entirely: `refs/heads/<branch>` is a ref
 * only a mediated `conductor_push` or a mediated `conductor_pr_update_branch`
 * can move, so its live commit is daemon provenance no reported string can
 * forge. It is also why a legitimate mediated base-branch update now passes:
 * that server-side merge creates a head no `conductor_push` ever published and
 * the row still carries the older one, which the recorded-pair test rejected.
 *
 * Binding to the *exact current run* is what makes an old attempt, a different
 * branch, or #806's orchestrator-only settled-run recovery invisible here: the
 * ledger query is run-scoped by `runId`, and recoveries store no runId at all.
 * That query asks for the run's complete history rather than the ledger's
 * newest rows — publication evidence sits at the *start* of a run, and a
 * review-revision round or a burst of refused mutations pushed it past the
 * default page, turning a verified push into a definitive false failure.
 *
 * A claim that fails this gate is a definitive failure — never a retryable
 * `pushed-pending` — because a guessed URL is not something a later tick is
 * waiting on. A claim the gate could not *read* is the opposite: an unreadable
 * branch tip is #781's transient outage, so it stays retryable rather than
 * burning an attempt on a flaky read.
 */
export async function verifyPushedGreenClaim(
  tracker: Pick<Tracker, "verifyPr" | "branchHead">,
  claim: Pick<WorkerResult, "prUrl" | "headSha">,
  publication: {
    project: string;
    issue: number;
    runId: string;
    /** The branch conductor routed this run onto. `conductor_push` publishes
     *  exactly `refs/heads/<branch>` and refuses any other ref. */
    branch: string;
    /** The identity `tracker.branchHead` reads the live tip with: the
     *  canonical `owner/repo` when the routed clone URL carries one, else the
     *  routed repository name. A tracker that cannot resolve it answers
     *  undefined, which stays retryable rather than definitive. */
    repo: string;
    store: Pick<Store, "verbLedger" | "getRun">;
  },
): Promise<{
  state: "pushed-green" | "pushed-pending" | "failed";
  reason?: string;
}> {
  const { prUrl, headSha } = claim;
  if (prUrl === undefined || headSha === undefined) {
    return { state: "failed", reason: "Worker did not report a PR URL and observed head SHA" };
  }
  // Exact current run, never "some publication for the issue": the query is
  // scoped to this runId plus project/issue, so a previous attempt's verbs and
  // the orchestrator's recovery verbs are invisible here. Unbounded on
  // purpose — see the note above about evidence ageing off the newest page.
  const ledger = publication.store.verbLedger(publication.project, {
    runId: publication.runId,
    issue: publication.issue,
    limit: Number.MAX_SAFE_INTEGER,
  });
  const run = publication.store.getRun(publication.runId);
  const branchRef = `refs/heads/${publication.branch}`;
  if (run?.prUrl !== prUrl) {
    return {
      state: "failed",
      reason:
        "Pushed-green claim has no mediated publication evidence: the claimed PR is not this run's " +
        "recorded PR",
    };
  }
  const pushedThisHead = ledger.some(
    (entry) =>
      entry.decision === "allowed" &&
      entry.verb === "conductor_push" &&
      entry.sha === headSha &&
      entry.detail.includes(branchRef),
  );
  // Bound to the claimed PR, not merely to "a create happened": both allowed
  // details name the URL they produced (`opened <url> …`, `adopted <url> …`),
  // so an unrelated create on this run cannot vouch for another PR.
  const createdHere = ledger.some(
    (entry) =>
      entry.decision === "allowed" &&
      entry.verb === "conductor_pr_create" &&
      entry.detail.includes(prUrl),
  );

  if (!pushedThisHead && !createdHere) {
    // Nothing this run published carries the claimed head, so the only
    // remaining evidence is the branch itself: a continuation or review round
    // that pushed nothing, or a head a mediated base-branch update produced.
    let tip: string | undefined;
    try {
      tip = await tracker.branchHead(publication.repo, publication.branch);
    } catch {
      tip = undefined;
    }
    if (tip === undefined) {
      return {
        state: "pushed-pending",
        reason: `Live head of ${branchRef} unavailable; retrying`,
      };
    }
    if (tip !== headSha) {
      return {
        state: "failed",
        reason:
          "Pushed-green claim has no mediated publication evidence: this run's conductor_push / " +
          `conductor_pr_create ledger does not cover the claimed head, and ${branchRef} is at ${tip}, ` +
          "not the claimed head",
      };
    }
  }

  const verification = await tracker.verifyPr(prUrl, headSha);
  if (verification === undefined) {
    return { state: "pushed-pending", reason: "GitHub PR verification unavailable; retrying" };
  }
  if (verification.status === "green") return { state: "pushed-green" };
  return {
    state: verification.status === "pending" ? "pushed-pending" : "failed",
    reason: verification.reason,
  };
}

/**
 * Whether a turns-cap kill is handed straight back to the queue.
 *
 * A turns cap consumes the independent continuation budget rather than an
 * implementation-failure attempt, so a run that ran out of turns mid-work is
 * requeued and the next tick reattaches its branch (#50).
 *
 * The artifact test is what keeps that from becoming a loop. A run that reached
 * the ceiling with no PR, no observed head and no salvage commit produced
 * nothing to continue *from*, and requeueing it spends another continuation on
 * the same spin — the seven rows on this fleet that burned $109 producing no
 * merged PR. That shape is `turn-cap-spinning`, which #132 says must escalate
 * with evidence instead; requeueing it here would settle the question before the
 * classifier ever saw the row, and the queue label would already be back on.
 *
 * Pure so the rule is testable without a worker session — the same split as
 * {@link settlementFor}.
 */
export function shouldContinueAfterTurnsCap(f: {
  killedBy?: string;
  prUrl?: string;
  headSha?: string;
  salvageSha?: string;
  continuation: number;
  maxContinuations: number;
}): boolean {
  if (f.killedBy !== "turns") return false;
  const hasArtifacts = f.prUrl !== undefined || f.headSha !== undefined || f.salvageSha !== undefined;
  if (!hasArtifacts) return false;
  return hasContinuationBudget(f.continuation, f.maxContinuations);
}

/** Bind a worker's run identity to every release-policy block it emits. */
export function workerReleaseBlockRecorder(
  project: string,
  issue: number,
  runId: string,
  root = stateDir(),
): (shape: GateShape, context: ReleaseBlockContext) => void {
  return (shape, context) =>
    recordReleaseBlock(
      project,
      "worker",
      shape,
      {
        ...context,
        issue,
        runId,
      },
      root,
    );
}

/**
 * The exact PR a new claim inherits from its terminal predecessor (#434).
 *
 * `recentRuns` shows only the newest attempt per issue, and the mediated verbs
 * resolve a PR's owner through the store — so the moment an orphan-clean
 * requeue creates its continuation row, the predecessor's green PR (recorded
 * on the older row) stops resolving and every mediated verb refuses it as "not
 * a pull request any run opened". Seeding the continuation row with the same
 * PR at claim time closes that window: ownership never depends on a worker
 * rediscovering and re-reporting the PR.
 *
 * The checks are deliberately the exact counterpart of the store query's scope:
 * the claim's project/issue/repo/branch must all match the predecessor, the
 * predecessor must be terminal (a live row is still doing its own work, and a
 * claim beside one is already a double dispatch), and the tracker must confirm
 * the PR is still open — a closed or merged predecessor PR is settled by the
 * sweep against the row that owns it, never carried onto a fresh attempt. An
 * unreadable tracker fails closed: the new row simply starts without PR
 * ownership, exactly as before #434, and adoption or the worker's own
 * pr_create recovers the link.
 */
export async function inheritedPrForContinuation(
  tracker: Tracker,
  prior: RunRecord | undefined,
  claim: { project: string; issue: number; repo: string; branch: string },
): Promise<{ prUrl: string; headSha?: string } | undefined> {
  if (prior === undefined || prior.prUrl === undefined) return undefined;
  // Terminal only. A live predecessor owns its PR outright and is not done
  // with it; inheriting would launder a double dispatch's row.
  if (prior.state === "claimed" || prior.state === "running") return undefined;
  if (prior.project !== claim.project || prior.issue !== claim.issue) return undefined;
  if (prior.repo !== claim.repo || prior.branch !== claim.branch) return undefined;
  let state: PrState | undefined;
  try {
    state = await tracker.prState(prior.prUrl);
  } catch {
    return undefined;
  }
  if (state !== "open") return undefined;
  return { prUrl: prior.prUrl, ...(prior.headSha === undefined ? {} : { headSha: prior.headSha }) };
}

/**
 * Which continuation, if any, may resume its predecessor's session instead of
 * dispatching a cold brief.
 *
 * Two shapes qualify, and they resume differently:
 *
 *  - `orphan-clean` (#536): a daemon death interrupted a healthy worker.
 *    `reconcileOrphanedRuns` KEEPS the worktree (salvage commit included) and
 *    the transcript is file-backed, so the next attempt gets its own memory
 *    back whole: the same worktree, the same session directory, `resume: true`.
 *    The old dispatch built a fresh `run-<uuid>` session and re-read the repo
 *    from zero — the exact rediscovery orphan-clean spent turns on in #535.
 *  - `answered-block` (Phase 3 inner loop): the worker asked a question, parked,
 *    and an operator answered. Its tree was removed at the blocked settle
 *    because the branch already held the work, so this one REATTACHES the branch
 *    the ordinary way and resumes only the session — the transcript is the
 *    expensive part, and today's cold brief throws away a whole investigation
 *    to deliver one answer.
 *
 * Everything else dispatches cold, and that is not laziness: a cap-killed or
 * otherwise failed worker was killed for cause, and orphan-dirty is held
 * precisely because the worktree is the only copy.
 *
 * An answered block additionally has to pass #1047's health test. Resuming a
 * session that is out of turns or that the provider abandoned buys a session
 * that dies the same way with the answer unread, which is the #1035 shape in a
 * different lane. The orphan-clean case deliberately keeps its own rule set: it
 * is defined by an interruption rather than an ending, and #536's contract is
 * that such a run continues.
 *
 * The checks here are what the daemon can prove cheaply before the claim (the
 * transcript is present and non-empty; the worktree, where the mode needs one);
 * the harness's own `continueRecent` is the backstop, and a corrupt transcript
 * it silently falls back from is surfaced loudly by the `sessionFile` lineage
 * compare at the dispatch site rather than left quiet.
 */
export type ResumeMode = "orphan-clean" | "answered-block";

export function continuationResumeVerdict(
  prior: RunRecord | undefined,
): { kind: "resume"; prior: RunRecord; mode: ResumeMode } | { kind: "fresh"; reason?: string } {
  const mode: ResumeMode | undefined =
    prior === undefined
      ? undefined
      : prior.state === "orphaned" && prior.failureClass === "orphan-clean"
        ? "orphan-clean"
        : prior.state === "blocked"
          ? "answered-block"
          : undefined;
  if (prior === undefined || mode === undefined) return { kind: "fresh" };
  const what = mode === "orphan-clean" ? "the orphaned attempt" : "the blocked attempt";
  if (prior.sessionFile === undefined) {
    return { kind: "fresh", reason: `${what} recorded no transcript` };
  }
  try {
    if (!existsSync(prior.sessionFile)) {
      return { kind: "fresh", reason: `transcript ${prior.sessionFile} is gone` };
    }
    if (statSync(prior.sessionFile).size === 0) {
      return { kind: "fresh", reason: `transcript ${prior.sessionFile} is empty` };
    }
  } catch (err) {
    return { kind: "fresh", reason: `transcript ${prior.sessionFile} is unreadable (${errText(err)})` };
  }
  if (mode === "orphan-clean" && (prior.worktree === "" || !existsSync(prior.worktree))) {
    return { kind: "fresh", reason: `worktree ${prior.worktree} is gone` };
  }
  if (mode === "answered-block") {
    const exhausted = exhaustedSessionReason(prior);
    if (exhausted !== undefined) {
      return { kind: "fresh", reason: `${what}'s session cannot continue: ${exhausted}` };
    }
  }
  return { kind: "resume", prior, mode };
}

/**
 * One attempt at one issue, from claim to terminal state. Everything is inside
 * a single try/catch so that a bad issue costs its own run and nothing else.
 */
export async function handleIssue(
  d: Deps,
  r: Routed,
  attempt: number,
  admittedLane?: FileLane,
  admittedModel?: EffectiveModel,
): Promise<void> {
  const { project, caps, tracker, store } = d;
  const issue = r.issue.number;
  const branch = branchName(r.issue);
  const inProgress = project.stateLabels.inProgress;

  let claimed = false;
  let run: RunRecord | undefined;
  // Hoisted out of the try so the catch path can still name and save the tree:
  // a crash mid-dispatch is one of the ends whose uncommitted work has to be
  // salvaged too, and it is the path least likely to have committed first.
  const mirrorPath = mirrorPathFor(r.repo, project.mirrorRoot);
  let worktreePath: string | undefined;
  let turnLimit: TurnLimitController | undefined;
  let workerControl: WorkerControlSlot | undefined;
  let workerSessionInstalled = false;
  // The run's Herdr representation (#840), hoisted for the same reason the verb
  // listener is: whoever tears the run down has to hand the pane's lifecycle
  // authority back, and a representation that outlives its child is a worker the
  // workspace still shows as live.
  let workerPane: Extract<WorkerPaneOutcome, { kind: "tracked" }> | undefined;
  // Herdr orders lifecycle reports by `seq`, so this run owns one counter and
  // every report takes the next value (#842). A shared or restarted counter would
  // let a late report overwrite a newer state — the pane would then show `working`
  // for a worker that has already blocked.
  let paneSeq = 0;
  const nextPaneSeq = (): number => (paneSeq += 1);
  /**
   * The injected Herdr runner, when this daemon carries one (#1035); every
   * representation call below shares it so tests see one argv stream. The
   * ownership adapter is always present in production — the store-backed
   * half of workspace discovery, so what a Herdr restart forgets, the
   * durable row remembers (#1035 review).
   */
  const herdr = {
    ...(d.herdrRun === undefined ? {} : { run: d.herdrRun }),
    ownership: workspaceOwnership(d.store),
  };
  /**
   * Project one authoritative transition onto the pane (#842).
   *
   * The input is always a durable run transition or a typed session event —
   * never the pane's own output, which is display and nothing more. A failure is
   * logged, not raised: the workspace lagging is not worth failing a run over.
   */
  const projectPaneState = (state: "working" | "idle" | "blocked" | "unknown", message?: string): void => {
    if (workerPane === undefined) return;
    const reported = reportWorkerPaneState(workerPane.paneId, workerPane.label, state, {
      ...herdr,
      seq: nextPaneSeq(),
      ...(message === undefined ? {} : { message }),
    });
    if (!reported.ok) log(`#${issue} herdr pane state ${state} not reported: ${reported.reason}`);
  };
  // The run's own repository. Hoisted for the same reason `worktreePath` is —
  // the catch and finally paths have to publish the branch.
  let runRepo: RunRepoRef | undefined;
  // #126: the run's own verb socket. Hoisted like the run repo, because the catch
  // and finally paths have to close it — a socket outliving its run is a
  // channel nobody is authenticating any more.
  let verbListener: VerbListener | undefined;

  /**
   * Publishes the run's branch on the privileged side: run repo → mirror →
   * GitHub, fast-forward only. The dispatcher performs every network git
   * operation for a run so the settlement record and the branch cannot
   * disagree (#126); this is the only route a worker's commits take out, and it
   * is also what stops a per-run repository from being the *only* copy when the
   * tree is removed.
   */
  const publish: RunPublisher = async () => {
    if (runRepo === undefined) return { ok: false, stderr: "the run repository was never provisioned" };
    return pushRunBranch(project, runRepo);
  };

  /**
   * A stop can arrive after the run is claimed but before `runWorker` exposes
   * its controller. Settle that run here instead of turning an operator action
   * into a dispatch failure or making the HTTP request wait for a session that
   * will never exist.
   */
  const settleStopBeforeSession = async (): Promise<boolean> => {
    const reason = workerControl?.requestedStop();
    if (reason === undefined || run === undefined || workerSessionInstalled) return false;
    turnLimit?.close();
    turnLimit = undefined;
    const settlement =
      worktreePath === undefined
        ? undefined
        : await settleWorktree({
            issue,
            attempt,
            ending: `stopped by the operator: ${reason}`,
            worktree: worktreePath,
            branch,
            publish,
            tree: "remove",
            mirrorPath,
          });
    recordOperatorStop(store, {
      project: project.name,
      issue,
      runId: run.id,
      inProgress,
      reason,
      patch: {
        endedAt: Date.now(),
        turns: run.turns,
        spendUsd: run.spendUsd,
        worktree: worktreePath ?? run.worktree,
        report: [
          "Operator stopped the run before its worker session started.",
          `Reason: ${reason}`,
          ...(settlement?.lines ?? []),
        ].join("\n"),
        ...settlement?.patch,
      },
    });
    log(`#${issue} stopped by operator before its worker session started: ${reason}`);
    return true;
  };

  /**
   * The daemon-wide stop fence can land after the claim but before `runWorker`
   * exposes its controller — the same window `settleStopBeforeSession` closes
   * for a per-run operator stop, entered from the shutdown path instead of the
   * pause control. Settle identically (close the claim, never launch) so the
   * drain waits only for workers admitted before the stop request, and the
   * shutdown cannot extend its own workload (#374).
   */
  const settleDrainBeforeSession = async (): Promise<boolean> => {
    if (d.drain?.draining !== true || run === undefined || workerSessionInstalled) return false;
    turnLimit?.close();
    turnLimit = undefined;
    const settlement =
      worktreePath === undefined
        ? undefined
        : await settleWorktree({
            issue,
            attempt,
            ending: "daemon shutdown began while the run was being prepared",
            worktree: worktreePath,
            branch,
            publish,
            tree: "remove",
            mirrorPath,
          });
    recordOperatorStop(store, {
      project: project.name,
      issue,
      runId: run.id,
      inProgress,
      reason: "daemon shutdown began after the claim",
      patch: {
        endedAt: Date.now(),
        turns: run.turns,
        spendUsd: run.spendUsd,
        worktree: worktreePath ?? run.worktree,
        report: [
          "Daemon shutdown began while this run was still being prepared; the worker",
          "never started and the claim was closed so the shutdown could complete.",
          ...(settlement?.lines ?? []),
        ].join("\n"),
        ...settlement?.patch,
      },
    });
    log(`#${issue} not launched: daemon shutdown began after the claim`);
    return true;
  };

  /**
   * The launch half of the credential-class fence (#852).
   *
   * Admission already held every candidate while a required provider billed to
   * the wrong credential, but a grant can be disabled in the window *between*
   * that check and this spawn — a refresh failing, an operator revoking, a token
   * expiring — and that window is exactly what the incident's acceptance
   * criterion names. So the same question is asked again here, as the last thing
   * before the session exists, and the run is closed rather than launched.
   *
   * Settled the same way the stop and drain fences settle: the claim closes, the
   * tree is salvaged, nothing is launched. It is deliberately NOT a failed
   * attempt against the issue's budget — the issue is not what is wrong, the
   * host's credentials are, and charging an attempt for it would exhaust an
   * issue while the operator re-authenticates.
   */
  const settleCredentialClassBeforeSession = async (): Promise<boolean> => {
    const required = project.requireOauthProviders ?? [];
    if (required.length === 0 || run === undefined || workerSessionInstalled) return false;
    const fence = await oauthFenceVerdict(
      required,
      d.probeCredentialClass ??
        (async (provider) => ({
          ok: false,
          reason: `no credential probe is wired for ${provider}`,
        })),
    );
    if (fence.ok) return false;
    turnLimit?.close();
    turnLimit = undefined;
    const settlement =
      worktreePath === undefined
        ? undefined
        : await settleWorktree({
            issue,
            attempt,
            ending: "not launched: a required subscription credential is unavailable",
            worktree: worktreePath,
            branch,
            publish,
            tree: "remove",
            mirrorPath,
          });
    recordOperatorStop(store, {
      project: project.name,
      issue,
      runId: run.id,
      inProgress,
      reason: `required subscription credential unavailable: ${fence.reason}`,
      patch: {
        endedAt: Date.now(),
        turns: run.turns,
        spendUsd: run.spendUsd,
        worktree: worktreePath ?? run.worktree,
        report: [
          "This run was closed instead of launched: a provider this project requires to bill",
          "to its subscription would have billed to something else.",
          fence.reason,
          ...(settlement?.lines ?? []),
        ].join("\n"),
        ...settlement?.patch,
      },
    });
    log(`#${issue} not launched: ${fence.reason}`);
    return true;
  };

  try {
    // The claim-side of the stop fence (#374): the run row is the boundary the
    // shutdown drain waits on, so the claim itself refuses once the daemon is
    // draining. The tick re-checks before the dispatch tail; this covers the
    // same fence for any caller that reaches `handleIssue` without one, and
    // makes "no post-stop run rows" a property of the claim, not of its caller.
    if (d.drain?.draining === true) {
      log(`#${issue} not claimed: daemon is draining`);
      return;
    }

    // The claim-side of the pause fence (#651 review #3): the tick's pause
    // gate sits above routing, so a tick that passed that gate before an
    // operator wrote the freeze can still be mid-routing when the fence lands
    // — it would claim after the setup barrier's acknowledgement and before
    // its first mutation. The claim itself re-checks the pause and refuses,
    // writing the durable admission acknowledgement the barrier waits for.
    // A claim observed under the fence is the last admission boundary there
    // is: nothing may create a run row beside a held fleet.
    if (isPaused(d.project.name)) {
      try {
        writeAdmissionAck(d.project.name);
      } catch (err) {
        log(`admission acknowledgement write failed: ${errText(err)}`);
      }
      log(`#${issue} not claimed: dispatch is paused at claim time`);
      return;
    }

    // The claim-side of the project drain fence (#484): the tick's gate sits
    // above routing, so a drain created after that gate can still land
    // mid-pass — the claim re-checks and refuses while a fresh drain is in
    // force. An invalid record fails closed the same way. The read here is
    // observational on purpose: the claim refuses but must not consume the
    // marker, or the first refused claim would unbind the rest of its own
    // pass; the tick's consumeDrain removes it on the next pass, so no claim
    // path can ever be blocked permanently by it.
    const claimDrain = readDrain(d.project.name);
    if (claimDrain.kind === "active" || claimDrain.kind === "error") {
      log(
        claimDrain.kind === "error"
          ? `#${issue} not claimed: drain record invalid (${claimDrain.problem})`
          : `#${issue} not claimed: project drain in effect`,
      );
      return;
    }

    // Claim on the STORE first, before anything that can fail. The run row —
    // not the label — is the crash-safe guard against double dispatch: rows
    // are local, written before any network call, and the startup orphan
    // sweep marks process-less claimed/running rows `orphaned`, so a daemon
    // dying mid-claim leaves a row a human can triage instead of a label only
    // the orchestrator may touch. The in-progress label is a write-behind
    // projection of that row: enqueued here, flushed by the post-admission
    // projection pass in the same tick on the healthy path, and — while
    // pending — kept effective off the ready set by the routing overlay, so a
    // crash between the row and its label cannot double-dispatch either
    // (#201).
    // Read before this attempt's own row exists, so `latestRun` still means the
    // attempt whose work this one inherits.
    const prior = store.latestRun(project.name, issue);
    const priorSalvage = prior?.salvageSha;
    // The failover decision is made from the store the same way admission is:
    // counts of this issue's own rows, so the switch is sticky per chain and a
    // fresh issue always starts on the primary (#286). Read before the claim
    // writes this attempt's row, which would otherwise break the streak.
    const chainFacts = providerFailureFacts(store.runsForIssue(project.name, issue));
    // Resolved before the claim, not after provisioning: the model is also
    // part of the #536 resume decision, which has to be made before the
    // dispatch shape (fresh provision vs kept worktree) is chosen. With no
    // `modelFallbacks` configured this is the primary model — or none, for an
    // unconfigured project — and today's dispatch is byte for byte what it
    // has always been.
    const chainConfigured = (project.modelFallbacks?.length ?? 0) > 0;
    // A `Model:` declaration admission resolved for this candidate (#535) is
    // this issue's workerModel: the orchestrator names a tier when it
    // promotes, and dispatch launches on that selector exactly as if the
    // project's `workerModel` were the declared value. Absent a declaration,
    // today's `project.workerModel` is unchanged, and the failover chain
    // (#286) keeps its semantics in both cases — it is the same resolution,
    // one different primary.
    const declaredModel = admittedModel?.model;
    // #807: a chain that already bought its one model escalation dispatches on
    // that stronger selector, outranking both the declaration and the project
    // default — the settlement that wrote the marker did so precisely because
    // the previous tier spun to a cap with nothing to show. Read from the
    // store, so it is sticky to this issue's chain and a fresh issue (no
    // marker) resolves exactly as it always has. The provider chain still
    // layers on top: an escalated run whose provider then aborts fails over
    // normally, because escalation only ever changes the primary.
    const escalation = store.modelEscalation(project.name, issue);
    const primaryModel = escalation?.model ?? declaredModel ?? project.workerModel;
    const choice = resolveDispatchModel({
      workerModel: primaryModel,
      modelFallbacks: project.modelFallbacks,
      threshold: project.modelFallbackThreshold ?? DEFAULT_MODEL_FALLBACK_THRESHOLD,
      streak: chainFacts.streak,
    });
    const clause =
      fallbackClause(choice, chainFacts, primaryModel) ??
      (escalation === undefined
        ? undefined
        : `on ${escalation.model} — this chain's one model escalation, after ${escalation.failureClass}`);

    // Whether this attempt continues its predecessor's session instead of
    // reading the repo from zero (#536, and the answered-block half of the
    // Phase 3 inner loop). Decided here, before this attempt's row exists,
    // because `prior` is still the attempt whose work this one inherits and the
    // whole dispatch shape follows the verdict.
    let resuming: RunRecord | undefined;
    let resumeMode: ResumeMode | undefined;
    const verdict = continuationResumeVerdict(prior);
    if (verdict.kind === "resume") {
      const resumePrior = verdict.prior;
      // The continuation must stay on the model the interrupted session was
      // using; a chain that now resolves differently dispatches fresh rather
      // than quietly continuing on another model, which would smear one
      // attempt's work across two models (#286 attribution) — and, on the
      // review side, is the diagnosed #1035 failure itself.
      if (resumePrior.model !== undefined && choice.model !== undefined && resumePrior.model !== choice.model) {
        log(
          `#${issue} attempt ${attempt} not resumed: attempt ${resumePrior.attempt} ran on ${resumePrior.model} but dispatch ` +
            `now resolves ${choice.model} — fresh dispatch`,
        );
      } else {
        resuming = resumePrior;
        resumeMode = verdict.mode;
        log(
          `#${issue} attempt ${attempt} continuing session of attempt ${resumePrior.attempt} (${verdict.mode}) → ` +
            `transcript ${resumePrior.sessionFile}, ` +
            `${verdict.mode === "orphan-clean" ? "worktree kept" : "branch reattached"}`,
        );
      }
    } else if (verdict.reason !== undefined) {
      log(`#${issue} attempt ${attempt} not resumed: ${verdict.reason} — fresh dispatch`);
    }

    run = store.createRun({
      project: project.name,
      issue,
      repo: r.repo.name,
      branch,
      worktree: "",
      state: "claimed",
      attempt,
      turns: 0,
      spendUsd: 0,
      maxTurns: caps.workerMaxTurns,
      startedAt: Date.now(),
      // #567: the orphan-clean attempt this claim continues, when the verdict
      // above fired. `undefined` for a fresh dispatch — the store maps that to
      // NULL, so a fresh row simply never carries the field.
      resumedFromRunId: resuming?.id,
      // #744: the file-lane declaration admission resolved for this candidate
      // is persisted on the row, so lane occupancy survives across dispatch
      // passes — a later pass knows what this run *intends* to touch, not only
      // what it has touched so far. This is the exact `Admission.lane` the gate
      // enforced and the brief rendered, never a re-parse. Absent for a run
      // with no declaration (fail open), exactly as it was admitted.
      lane: admittedLane,
      // #851: the allowance admission reserved for this run, recorded on the
      // row before anything launches. Derived from the same caps the gate
      // read, so the reservation the pass enforced and the one the row holds
      // are one value; `undefined` (NULL) when the fleet has no spend cap at
      // all, which is the truth rather than "reserved nothing".
      spendReservedUsd: runSpendAllowanceUsd(caps) ?? undefined,
    });
    // #434: carry a terminal predecessor's open PR onto the continuation row.
    // The claim itself stays synchronous — `handleIssue` claims at the top of
    // its dispatch burst, the window the shutdown fence and its tests rely on
    // (#374) — so the tuple-exact, tracker-verified inheritance lands here,
    // still before any provisioning, network call or verb socket. The new row
    // is `recentRuns`' newest attempt for the issue the moment it exists, so
    // without this a requeue hides the green PR the orphaned attempt opened
    // until a worker happens to rediscover it.
    const inherited = await inheritedPrForContinuation(tracker, prior, {
      project: project.name,
      issue,
      repo: r.repo.name,
      branch,
    });
    if (inherited !== undefined && prior !== undefined) {
      store.updateRun(run.id, {
        prUrl: inherited.prUrl,
        ...(inherited.headSha === undefined ? {} : { headSha: inherited.headSha }),
      });
      log(`#${issue} inherited PR ${inherited.prUrl} from attempt ${prior.attempt} (${prior.state})`);
    }
    const maxTurns = run.maxTurns;
    const turnOverride = maxTurns > caps.workerMaxTurns ? maxTurns : undefined;
    const runId = run.id;
    if (turnOverride !== undefined) {
      log(
        `#${issue} claimed with turn override ${turnOverride} ` +
          `(base ${caps.workerMaxTurns})`,
      );
    }
    store.enqueueLabelOps(project.name, [{ issue, op: "add", label: inProgress }]);
    claimed = true;
    turnLimit = d.turnLimits.open(project.name, issue, runId, maxTurns);
    // The pane follows the phase the registry actually reached (#842) — never a
    // guess made at the call site, and never the pane's own output read back.
    workerControl = d.workerControls.open(project.name, issue, runId, (phase) => {
      projectPaneState(
        phase === "running" ? "working" : "idle",
        phase === "running" ? "resumed" : `${phase} by the operator`,
      );
    });
    if (await settleStopBeforeSession()) return;
    if (await settleDrainBeforeSession()) return;
    // The credential fence, twice, for the same reason the stop fence is
    // consulted more than once (#852). Here it is the cheap one: a run whose
    // required subscription credential is already unavailable is closed before
    // anything clones a mirror or cuts a worktree, because provisioning for a
    // launch that cannot happen is pure waste. The second call, immediately
    // before the spawn, is the one that closes the admission-to-session window.
    if (await settleCredentialClassBeforeSession()) return;


    // A run's tree is <workspaceRoot>/<issue> and addRunRepo refuses to reuse
    // an existing path, so a retry — or a tree kept from a failed attempt — has
    // to be cleared first. Both helpers are pure path math, and removeWorktree
    // tolerates a mirror or tree that is not there yet. An ORPHAN-CLEAN resume
    // skips the whole dance: the orphaned run's tree is the work to continue
    // (its salvage commit is already on the branch), and re-cloning it from
    // the mirror would be exactly the rediscovery this feature exists to skip.
    // An answered-block resume does NOT skip it: that run's tree was removed at
    // its blocked settle because the branch already held the work, so it
    // reattaches the branch exactly as any continuation does and resumes only
    // the session.
    let provisioned: Awaited<ReturnType<typeof addRunRepo>> | undefined;
    if (resuming !== undefined && resumeMode === "orphan-clean") {
      worktreePath = resuming.worktree;
      runRepo = { repo: r.repo, runRepoPath: worktreePath, branch };
    } else {
      await removeWorktree(mirrorPath, worktreePathFor(project.workspaceRoot, issue));
      if (await settleStopBeforeSession()) return;
      if (await settleDrainBeforeSession()) return;
      const provisionedTree = await addRunRepo(
        r.repo,
        project.mirrorRoot,
        project.workspaceRoot,
        issue,
        branch,
      );
      provisioned = provisionedTree;
      worktreePath = provisionedTree.path;
      runRepo = { repo: r.repo, runRepoPath: worktreePath, branch };
    }
    if (await settleStopBeforeSession()) return;
    if (await settleDrainBeforeSession()) return;

    // The SDK names the transcript itself, so the daemon supplies the parent
    // directory and learns the real path back from the result. Inventing one
    // here would put a file that never gets written into an escalation.
    //
    // Per run rather than one shared directory, so one run's transcript cannot
    // be truncated or replaced by the next — except for a resumed attempt,
    // which deliberately reuses the interrupted session's directory so the
    // SDK's `continueRecent(cwd, dir)` picks up that transcript and keeps
    // writing it.
    const runTreeRoot = stateDir();
    const sessionDir =
      resuming === undefined
        ? join(runTreeRoot, "sessions", `run-${String(runId)}`)
        : dirname(resuming.sessionFile!);
    mkdirSync(sessionDir, { recursive: true });

    // The fleet-owned omp settings overlay (#537): the project's `ompSettings`
    // map (plus the retry keys derived from `modelFallbacks`, #539's staging
    // half) materialised to YAML under the run's session directory — never
    // inside the worktree, whose diff is the PR a worker ships. Rewritten on
    // every attempt, so a config edit takes effect on the next dispatch and a
    // resumed attempt reuses the kept session dir with the *current* config.
    // Absent `ompSettings` and `modelFallbacks`, no file is written, nothing
    // is passed, and dispatch is byte-for-byte today's.
    const ompSettingsFile = materializeOmpSettings(project, sessionDir);

    // ---- the run's mutation channel (#126) -------------------------------
    // A shared, daemon-owned 0711 parent with one 0600 socket per run, never a
    // per-run *directory*: a directory owned by the run principal would hand
    // back the power to unlink a sibling's socket and bind an impostor in its
    // place, which is the whole thing the layout buys. `listenVerbChannel`
    // validates every component before binding and throws rather than
    // degrading, so a tampered path refuses this dispatch instead of running
    // the worker with a channel nobody can vouch for.
    verbListener = await listenVerbChannel(
      verbDeps(d),
      {
        kind: "run",
        // `runTreeRoot`, not the private state dir: the slot has to *connect*
        // to this socket, and a 0700 ancestor makes it unreachable however
        // correct the socket's own mode is. `ensureVerbSocketDir` still owns
        // the parent and still runs its anti-rebind checks there (#126).
        path: verbSocketPath(ensureVerbSocketDir(runTreeRoot), `run-${String(issue)}`),
        project: project.name,
        role: "worker",
        runId,
        issue,
        repo: r.repo,
        runRepoPath: worktreePath,
        branch,
      },
      {
        ...(d.verbPeerReader === undefined ? {} : { peerReader: d.verbPeerReader }),
      },
    );
    if (await settleStopBeforeSession()) return;
    if (await settleDrainBeforeSession()) return;

    store.updateRun(runId, { worktree: worktreePath, state: "running" });

    // Where this attempt goes, and (when the failover fired) the clause that
    // makes it attributable. `chainConfigured`/`choice`/`clause` were resolved
    // before the claim — the resume verdict had to be made before the dispatch
    // shape was chosen (#536) — so only the record write lives here.
    // Recorded before the launch, so even a run killed mid-flight leaves the
    // model it chose on its row. Only a chain-configured project writes the
    // column: absent `modelFallbacks` must preserve today's rows byte for byte.
    // An escalated chain (#807) writes it too — the selector it dispatched on
    // is the whole provenance of that recovery, and its harness-resolved model
    // lands beside it in `resolvedModel` when the run settles.
    if ((chainConfigured || escalation !== undefined) && choice.model !== undefined) {
      store.updateRun(runId, { model: choice.model });
    }

    log(
      `#${issue} attempt ${attempt}${clause === undefined ? "" : ` ${clause}`} → ${r.repo.name} ${branch}` +
        (provisioned?.reattached ? " (continuation: reattached existing branch)" : ""),
    );

    // The continuation notice replaces the brief for a resumed attempt (#536).
    // The original brief is already in the resumed transcript; re-sending it is
    // how a resumed worker ends up re-doing the work it just did. Everything
    // below the brief is fresh-dispatch-only, exactly as today — except the
    // file lane admission enforced for this continuation, which rides the
    // notice itself: the retained transcript's brief may show an earlier
    // declaration, and a worker continuing under a stale lane is the collision
    // the gate exists to stop (#608).
    //
    // An answered block resumes for a different reason and so carries something
    // different: the investigation is in the transcript, but the ANSWER it
    // parked for is not — it arrived as a comment afterwards. So the discussion
    // travels with the notice, rendered by the same budgeted renderer the cold
    // brief uses, and an unreadable thread degrades to a named absence rather
    // than silently reading as "nobody answered".
    let brief: string;
    if (resuming !== undefined && resumeMode === "answered-block") {
      let answered: IssueComment[] | "unread";
      try {
        answered = await tracker.listComments(issue);
      } catch (err) {
        log(`#${issue} issue comments unreadable at dispatch; the continuation will say so: ${errText(err)}`);
        answered = "unread";
      }
      brief =
        renderAnsweredBlockPrompt(issue, renderDiscussion(answered, admittedLane)) +
        (admittedLane === undefined ? "" : resumeLaneBlock(admittedLane));
    } else if (resuming !== undefined) {
      brief =
        ORPHAN_RESUME_PROMPT + (admittedLane === undefined ? "" : resumeLaneBlock(admittedLane));
    } else {
      // The discussion is rendered at dispatch so a worker never depends on a
      // runtime `gh` read to see the orchestrator's grooming (#517). The read is
      // best-effort, but its failure is not silent: an unreadable tracker names
      // itself in the brief's Discussion section instead of reading as "no
      // comments" — the exact confusion this fix removes.
      let comments: IssueComment[] | "unread";
      try {
        comments = await tracker.listComments(issue);
      } catch (err) {
        log(`#${issue} issue comments unreadable at dispatch; the brief will say so: ${errText(err)}`);
        comments = "unread";
      }

      brief = await buildBrief(project, r, branch, worktreePath, {
        continuation: provisioned?.reattached === true,
        defaultBranch: r.repo.defaultBranch,
        ...(provisioned?.reattached === true && priorSalvage !== undefined
          ? { salvagedSha: priorSalvage }
          : {}),
        comments,
        lane: admittedLane,
        model: admittedModel,
        host: d.host,
      });
    }
    if (await settleStopBeforeSession()) return;
    if (await settleDrainBeforeSession()) return;
    // Last, so it is the newest fact anything has about the credentials — a
    // grant disabled since admission is caught here rather than paid for.
    if (await settleCredentialClassBeforeSession()) return;

    const repoSlug = githubRepo(r.repo.cloneUrl);


    let result: WorkerResult;
    try {
      const runAllowanceUsd = runSpendAllowanceUsd(caps);
      result = await runWorker({
        brief,
        cwd: worktreePath,
        caps,
        ...(repoSlug === undefined ? {} : { repoSlug }),
        maxTurns: () => turnLimit?.maxTurns() ?? maxTurns,
        // #851: the allowance this run reserved at admission is also its live
        // ceiling. Without it the reservation would only bound how many runs
        // start, not what one of them spends, and the day's total would be
        // unbounded again.
        ...(runAllowanceUsd === null ? {} : { maxSpendUsd: runAllowanceUsd }),
        onPauseControl: (control) => {
          workerSessionInstalled = true;
          workerControl?.install(control);
        },
        sessionDir,
        ...(resuming === undefined ? {} : { resume: true }),
        // The session's control socket, under the daemon's own state directory —
        // a child process of the daemon reaches it directly.
        socketPath: join(sessionDir, "ipc.sock"),
        verbSocketPath: verbListener.path,
        // Names this run's channel the instant the child exists. Until this
        // fires the channel refuses every connection, because a session that
        // reached it first would be one nobody had identified — and every
        // session shares this daemon's uid, so the socket alone cannot tell
        // them apart (#163).
        onSpawn: (pid) => {
          verbListener?.bindPid(pid);
        },
        // The workspace representation of this exact child (#840). Best effort by
        // construction: a worker that cannot be *shown* is still a worker, so a
        // failure is logged with its reason and the run proceeds. What a failure
        // should mean for the launch — fail closed, or a named degraded state —
        // is #841's policy and is deliberately not decided here. Nothing is
        // claimed silently: either the pane id is logged, or the reason is.
        pane: {
          open: (pid) => {
            // Adopt before creating (#842): a run that already carries a pane
            // identity gets that pane reported against, never a second one. The
            // durable row is what makes this survive the process that opened it,
            // and duplicate-prevention is exactly what it buys — a re-entered
            // launch for one run must not leave two panes claiming it.
            const recorded = store.getRun(runId);
            if (recorded?.paneId !== undefined && recorded.paneLabel !== undefined) {
              workerPane = { kind: "tracked", paneId: recorded.paneId, label: recorded.paneLabel, pid };
              store.updateRun(runId, { workerPid: pid });
              const state = reportWorkerPaneState(recorded.paneId, recorded.paneLabel, "working", {
                ...herdr,
                seq: nextPaneSeq(),
              });
              log(
                state.ok
                  ? `#${issue} herdr pane ${recorded.paneId} re-reported for pid ${pid}`
                  : `#${issue} herdr pane ${recorded.paneId} could not be re-reported: ${state.reason}`,
              );
              return;
            }
            const outcome = openWorkerPane(
              {
                project: project.name,
                issue,
                attempt,
                runId,
                pid,
                ...(recorded?.sessionFile === undefined ? {} : { sessionFile: recorded.sessionFile }),
              },
              herdr,
            );
            if (outcome.kind === "tracked") {
              workerPane = outcome;
              // Durable before it is announced: a pane the store does not know
              // about is a pane a restart cannot reconcile.
              store.updateRun(runId, {
                workerPid: pid,
                paneId: outcome.paneId,
                paneLabel: outcome.label,
                // Cleared, because there is now a pane: a leftover reason beside a
                // tracked run would keep `status` saying degraded forever (#841).
                paneUnavailable: null,
              });
              log(`#${issue} herdr pane ${outcome.paneId} (${outcome.label}) tracks pid ${pid}`);
            } else {
              // Named, never silent (#841): the run proceeds untracked, and the
              // reason is durable so `status` can say the fleet is running blind
              // rather than looking identical to a fleet with no workers.
              store.updateRun(runId, { workerPid: pid, paneUnavailable: outcome.reason });
              log(`#${issue} no herdr pane: ${outcome.reason}`);
            }
          },
          release: () => {
            if (workerPane === undefined) return;
            // Retirement, not release alone (#1035): a settled run keeps no
            // representation. The close reaches only the follower; the
            // authoritative child was never in the pane to be signalled.
            const retired = retireWorkerPane(workerPane.paneId, workerPane.label, {
              ...herdr,
              seq: nextPaneSeq(),
            });
            if (!retired.ok) log(`#${issue} herdr pane retirement failed: ${retired.reason}`);
            workerPane = undefined;
          },
        },
        onChildLog: (line) => {
          log(`#${issue} ${line}`);
        },
        ...(choice.model === undefined ? {} : { model: choice.model }),
        // The fleet-owned omp settings overlay (#537): the staged YAML the
        // session loads through `Settings.init({ configFiles: [<path>] })` —
        // the project's `ompSettings` map plus the within-run failover keys
        // #581 staged directly (the project's own chain, not an empty default).
        // Absent both, nothing is staged, today's dispatch byte for byte.
        ...(ompSettingsFile === undefined ? {} : { ompSettingsFile }),
        releaseGrants: resolveReleaseGrants(project),
        onReleaseBlocked: workerReleaseBlockRecorder(project.name, issue, runId),
        onTurn: (n) => store.updateRun(runId, { turns: n }),
        onSpend: (usd) => store.updateRun(runId, { spendUsd: usd }),
        // #518: recorded live, not at settlement — the moment this explains is
        // forty minutes before the wall-clock cap fires, so a number only a
        // finished run carries would answer the question too late.
        onTokens: (tokens) =>
          store.updateRun(runId, { outputTokens: tokens.output, reasoningTokens: tokens.reasoning }),
        onKilled: () => {
          turnLimit?.close();
          turnLimit = undefined;
        },
        // Recorded the moment the session opens its transcript, not when the run
        // ends: `omp-conductor tail` resolves an issue to a file through this row,
        // and a path written at completion is a path nobody can follow live. The
        // completion-time update below writes the same value again, harmlessly.
        onSessionFile: (f) => {
          store.updateRun(runId, { sessionFile: f });
          // #536: the harness's own `continueRecent` can still fall back to a
          // blank session (corrupt transcript, nothing to continue) — and a
          // blank session in the "resumed" worktree is indistinguishable from
          // today's dispatch unless the downgrade is named. Same-file lineage
          // is the proof the resume happened: the resumed run must keep
          // writing the orphaned attempt's transcript.
          if (resuming !== undefined && f !== resuming.sessionFile) {
            log(
              `#${issue} attempt ${attempt} resume fell back to a fresh session: opened ${f} ` +
                `instead of the orphaned attempt's ${resuming.sessionFile}`,
            );
          }
        },
        // The last fence (#374): every pre-launch settle check above has
        // passed, but the stop can still land while the session socket is
        // binding inside `createSession`. This gate is re-checked there,
        // immediately before the child spawn, and closes the run as stopped
        // instead of launching a worker the shutdown would have to wait for.
        maySpawn: () => d.drain?.draining !== true,
      }, d.workerDeps);
    } finally {
      // This is the authoritative settlement edge for `extend`: close before
      // PR verification or terminal row writes can leave stale `running` state.
      turnLimit?.close();
      turnLimit = undefined;
    }

    // A configured model the harness could not honour means this run was done by
    // a different model than the operator chose. Logged per run, because it is
    // the only place that fact is still attached to the issue it affected.
    if (result.modelFallbackMessage !== undefined) {
      log(`#${issue} model fallback: ${result.modelFallbackMessage}`);
    }
    // What this run learned about the repo, into the fleet overlay the next
    // brief renders: the entry point it actually had to find, the gate that
    // really proved the change, the fake that looked like a test. Reported by
    // the worker in its own settlement, so nothing is inferred — and a
    // knowledge write can never fail a settlement.
    recordWorkerDiscoveries(knowledgeRepoKey(r.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 ?? r.repo.name,
            store,
          })
        : { state: result.state };
    const state = verified.state;

    // Read before the row is written so `lastError` carries the provider's own
    // message — it names the fix and the URL, which no classification can.
    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);

    // The other half of not believing a worker about its own run (#128). The
    // claim being audited is `state: pushed-green`, so the audit runs on the
    // worker's claim rather than on what verification made of it: a claim that
    // GitHub then contradicted is exactly the report whose details matter
    // most. Advisory throughout — nothing below reads `flags` when deciding
    // `state`, and the ordering here makes that checkable.
    const audit =
      result.state === "pushed-green"
        ? await collectSettlementFlags(tracker, {
            prUrl: result.prUrl,
            issueText: `${r.issue.title}\n${r.issue.body}`,
            // The claimed-proof check compares the PR's Verified commands
            // against what this run's session actually recorded.
            sessionFile: result.sessionFile,
            // The effective file lane admission resolved for this run at
            // dispatch — the same value the brief rendered, pre-dispatch
            // comment declarations included (#608, #744). The audit flags a
            // diff that escapes it, so a widened lane is named on evidence
            // rather than found by reading the PR's file list by hand (#739).
            lane: admittedLane,
          })
        : undefined;
    if (result.state === "pushed-green" && audit?.truncated) {
      log(`#${issue} settlement audit read a truncated PR diff`);
    }
    const auditLines =
      audit === undefined
        ? []
        : formatSettlementFlags(audit.flags, { truncated: audit.truncated });
    // The report's file list is derived from the PR's own diff, not retyped
    // from the worker's memory of the session: the narrative stays the
    // worker's, the disclosure becomes the diff's (#488). A diff that could
    // not be read leaves the worker's text untouched and the audit's
    // `changed-line-missing` flag says so.
    // The within-run reliability sentence, appended where a human reads it
    // (#584): which model the run finished on and whether it swapped
    // mid-flight. Empty (undefined) for a clean run, so a run that never
    // retried, swapped or compacted keeps today's settlement report byte for
    // byte — the "additive" claim asserted rather than assumed.
    const reliabilityLine = reliabilitySettlementLine({
      resolvedModel: result.model,
      resolvedProvider: result.provider,
      retryFallbacks: result.retryFallbacks,
      retryFallbackSucceeded: result.retryFallbackSucceeded,
      modelRecoveries: result.modelRecoveries,
      autoRetryCount: result.autoRetryCount,
      autoCompactionCount: result.autoCompactionCount,
    });

    const settlementReport = [
      audit?.changedLine === undefined
        ? result.report
        : withDerivedChangedLine(result.report, audit.changedLine),
      ...(reliabilityLine === undefined ? [] : ["", reliabilityLine]),
    ].join("\n");

    const finalReport = [
      ...(verified.reason === undefined ? [] : [verified.reason, ""]),
      ...(auditLines.length === 0 ? [] : [...auditLines, ""]),
      settlementReport,
    ].join("\n");

    // What becomes of the tree, decided once, before any label or page. A
    // `pushed-*` run is the only end that does not salvage: its deliverable is
    // already on a remote branch, whatever is left loose in the tree is by the
    // worker's own account not part of it, and appending a WIP commit would
    // turn the green PR this daemon just verified red. Every other end may
    // contain work, so it is salvaged before the tree's final fate is decided.
    const settlement =
      state === "pushed-green" || state === "pushed-pending" || state === "merged"
        ? undefined
        : await settleWorktree({
            issue,
            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) {
      // A `pushed-*` end does not salvage — its deliverable is already on a
      // remote branch — but its repository still goes away here, so the branch
      // is published first for the same reason `settleWorktree` does it: the
      // run repo is no longer a view of the mirror, and anything it holds that
      // never reached the mirror dies with the directory.
      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);
    }

    const terminalPatch: Partial<RunRecord> = {
      endedAt: Date.now(),
      turns: result.turns,
      spendUsd: result.spendUsd,
      // The count of in-session provider 429s the worker metered live, so a
      // run the provider throttled into the ground carries its own diagnosis
      // instead of landing `unknown` — the classifier reads it straight off
      // this column (#573).
      provider429Count: result.provider429Count,
      // The within-run harness reliability surface #581 collected, persisted
      // now that settlement owns the row (#584). The resolved model/provider
      // only travel when some assistant message carried them (the run recorded
      // no model, or the worker never established one); the count fields always
      // travel, 0 for a clean run, so an absent column can never be read as a
      // quiet fleet. Written for every terminal state, clean or not.
      ...(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 kill
      // or a settle whose report named no PR must not wipe what a verb recorded
      // earlier in the same run (#468). The sink in `updateRun` skips undefined
      // too, but a settle should not even build such a patch.
      ...(result.prUrl === undefined ? {} : { prUrl: result.prUrl }),
      ...(result.headSha === undefined ? {} : { headSha: result.headSha }),
      sessionFile: result.sessionFile,
      // The code-graph session observation (#726): what the run's own session
      // registry held at start, persisted with the rest of the run's facts.
      // Absent only when the session surface did not record one.
      ...(result.graphTools === undefined ? {} : { graphTools: result.graphTools }),
      // Every terminal state persists the worker's report — with the `changed:`
      // file list derived from the PR's diff where one could be read — not
      // just a green push: a stopped attempt's partial report is still part of
      // its audit trail.
      report: settlementReport,
      ...settlement?.patch,
      ...(audit === undefined || audit.flags.length === 0
        ? {}
        : { settlementFlags: audit.flags }),
    };
    if (state === "stopped") {
      recordOperatorStop(store, {
        project: project.name,
        issue,
        runId,
        inProgress,
        reason: result.stoppedReason ?? "no reason recorded",
        patch: terminalPatch,
      });
    } else {
      const lastError = completionLastError(
        providerCredit,
        providerTransient,
        verified.reason,
        sessionErr,
      );
      store.updateRun(runId, {
        ...terminalPatch,
        state,
        ...(lastError === undefined ? {} : { lastError }),
      });
    }

    const salvaged = settlement?.lines ?? [];

    // The run's reliability news, surfaced where the tick digest reads it
    // (#584): the digest is model-authored but consumes the store's material
    // ledger, so a run that swapped mid-flight or rode out a throttled
    // provider lands one event the digest can name. Clean runs record nothing
    // here, so a quiet fleet's digest is unchanged.
    if (reliabilityLine !== undefined) {
      store.recordMaterialEvent({
        project: project.name,
        category: "reliability",
        summary: `#${issue} ${reliabilityLine}`,
        evidence: `${r.issue.title}\n${r.issue.url}\n\n${reliabilityLine}`,
        occurredAt: Date.now(),
        recordedAt: Date.now(),
      });
    }

    if (state === "stopped") {
      log(`#${issue} stopped by operator on attempt ${attempt}: ${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 attempt ${attempt} and needs a decision`,
        detail: [`${r.issue.title}`, r.issue.url, "", ...salvaged, "", result.report].join("\n"),
      });
    } else if (state === "failed" || state === "killed") {
      // The row is already `killed`, so this count includes the segment that
      // just ended.
      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 (providerCredit !== undefined) {
        await reactToProviderCredit({ project: d.project, escalate: (e) => d.escalate(e), isPaused, setPaused }, issue, providerCredit, result.sessionFile);
        await swapToQueue(d, issue, inProgress);
      } else if (continueTurns) {
        // One ordered pair through the shared chokepoint: the in-progress
        // removal before the queue add, exactly the order the projector will
        // apply them in (#201) — and the wake that comes with it, so a
        // salvaged continuation is claimed on an immediate pass instead of
        // waiting out the interval (#1041).
        await swapToQueue(d, issue, inProgress);
        log(
          `#${issue} turns-cap on run ${attempt}, 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 attempt ${attempt} — auto-requeued for continuation`,
          detail: [
            `${r.issue.title}`,
            r.issue.url,
            ...salvaged,
            `Session: ${result.sessionFile ?? "(no transcript)"}`,
            "",
            "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,
          // The dedup key includes the summary, so the attempt number is what
          // lets a genuine second failure page again while a tick that keeps
          // seeing the same dead issue stays quiet.
          summary: result.killedBy
            ? `#${issue} was killed on attempt ${attempt} by the ${result.killedBy} cap`
            : `#${issue} failed on attempt ${attempt}`,
          detail: [
            `${r.issue.title}`,
            r.issue.url,
            ...salvaged,
            `Session: ${result.sessionFile ?? "(no transcript)"}`,
            "",
            finalReport,
          ].join("\n"),
        });
      }
    } else {
      // A verified or still-pending PR keeps the in-progress label until its
      // checks or merge settle, preventing another worker from duplicating it.
      log(`#${issue} ${state}${result.prUrl ? ` ${result.prUrl}` : ""}`);
      // A green run's report is the only one nothing ever delivers: it raises no
      // escalation, and the row it settles into says nothing about what the
      // worker wrote. That is fine while the report is believed — and it is
      // exactly wrong once something has stopped believing it. A flagged
      // settlement is the one case where a `pushed-green` run has news, and it
      // has it *now*, while the PR is still open and nobody has merged it (#128).
      //
      // Tier 1, because the judgement is the orchestrator's: it can read the
      // diff, it can ask, and it can promote to a human. The state is
      // deliberately unchanged either way — the run below this line settles
      // `pushed-green` whether it was flagged or not.
      //
      // Findings only. A diff too large to read in full is noted in the log and
      // in the report, but it is not news for a human: paging "0 flags, could
      // not read it all" is exactly the kind of line that teaches an
      // orchestrator to skim past this escalation the next time it carries one.
      const found = audit?.flags ?? [];
      if (found.length > 0) {
        await safeEscalate(d, {
          tier: 1,
          project: project.name,
          issue,
          runId,
          summary:
            `#${issue} settled ${state} on attempt ${attempt} with ` +
            `${found.length} settlement audit flag(s)`,
          detail: [
            `${r.issue.title}`,
            r.issue.url,
            result.prUrl ?? "(no PR URL)",
            "",
            "The run is NOT blocked and its state is unchanged. These are advisory",
            "findings about the worker's own account of its work — judge them, then",
            "merge, ask, or close as you would have anyway.",
            "",
            finalReport,
          ].join("\n"),
        });
      }
    }
  } catch (err) {
    // Dispatch setup can fail after the controller opens but before runWorker's
    // inner settlement guard exists. Latch it before any terminal write or await.
    turnLimit?.close();
    turnLimit = undefined;
    if (await settleStopBeforeSession()) return;
    const detail = errText(err);
    log(`#${issue} errored: ${detail}`);
    // A crash lands anywhere, including mid-edit in a tree holding the only
    // copy of real work. Nothing else on this path so much as looks at it.
    const settlement =
      worktreePath === undefined
        ? undefined
        : await settleWorktree({
            issue,
            attempt,
            ending: "killed by a dispatch error",
            worktree: worktreePath,
            branch,
            publish,
            tree: "keep",
          });
    if (run) {
      store.updateRun(run.id, {
        state: "failed",
        endedAt: Date.now(),
        lastError: detail,
        ...settlement?.patch,
      });
    }
    if (claimed) {
      // Leaving the issue stuck as in-progress would hide it from both the
      // queue and the human, so relabel even on the error path.
      // The failure-path relabel cannot throw: it is a local outbox write, and
      // the projector retries until the tracker takes it (#201).
      swapLabel(store, project.name, issue, inProgress, project.stateLabels.failed);
    }

    const salvaged = settlement?.lines ?? [];
    await safeEscalate(d, {
      tier: 1,
      project: project.name,
      issue,
      runId: run?.id,
      summary: `#${issue} could not be dispatched on attempt ${attempt}`,
      detail: salvaged.length === 0 ? detail : [detail, "", ...salvaged].join("\n"),
    });
    // The worktree, if one was created, is deliberately left in place: this is
    // a failure path, and whatever it still held is now a commit on the branch.
  } finally {
    turnLimit?.close();
    // The run is over, so its channel is too. Closed here rather than beside
    // the session so the crash path closes it as well: a listener left bound
    // after its run settled is a socket whose `run-not-live` check is the only
    // thing standing between a stale child and a push.
    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;
  }
}