/**
 * Runs exactly one omp coding session for one issue, under caps this file
 * enforces itself.
 *
 * The caps are the point. A worker that is *asked* to respect a turn budget
 * will talk itself out of it around turn 80, so the ceiling lives out here:
 * count the turns, watch the clock, and abort the session. Likewise the final
 * state is parsed out of the worker's own report rather than trusted — an
 * unreadable report is a failure, because the alternative is unverified work
 * sliding into a merge queue.
 */

import { mkdirSync, readFileSync, readdirSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import { knowledgeSection } from "./knowledge.ts";
import { createSession, disposeSession, SessionAdmissionClosedError, type AgentSessionLike } from "./omp.ts";
import type { GateShape, ReleaseBlockContext } from "./release-policy.ts";
import { gateToSpecPayload, TO_SPEC_SCHEMA, type ToSpecFailure } from "./to-spec.ts";
import type { Caps, GraphToolsObservation, ResolvedGrants, RunState } from "./types.ts";

/** Structured evidence fields from the worker's final report. */
const PR_URL_PATTERN = /^pr:\s*(https:\/\/github\.com\/\S+\/pull\/\d+)\s*$/im;
const HEAD_SHA_PATTERN = /^head:\s*([0-9a-f]{40})\s*$/im;
const PUSHED_GREEN_PATTERN = /^state:\s*pushed-green\s*$/im;
const BLOCKED_PATTERN = /^state:\s*blocked\s*$/im;

/** Any explicit verdict line, whatever it claims. */
const STATE_LINE_PATTERN = /^state:\s*\S+\s*$/im;
/** GitHub PR URLs in unstructured prose; capture their canonical `owner/repo`. */
const GITHUB_PR_URL_PATTERN = /https:\/\/github\.com\/([^/\s]+\/[^/\s]+)\/pull\/\d+\b/gi;

/** `{{KEY}}` placeholders in a brief template. */
const PLACEHOLDER_PATTERN = /\{\{([A-Za-z0-9_]+)\}\}/g;

/**
 * The structured settlement a worker yields at the end of a run (#540).
 *
 * Fields decided here, not per project. The daemon stays the authority on
 * green — it verifies the PR itself — so this is the worker's *claim*,
 * cross-checked exactly as the prose report it replaces: a `green` claim
 * must carry the PR URL and the head SHA the worker actually watched go
 * green, and anything less fails closed.
 */
export interface WorkerSettlement {
  status: "green" | "blocked" | "failed";
  /** The run's pull request URL. Required with `status: "green"`. */
  prUrl?: string;
  /** The branch the PR was pushed from. */
  branch?: string;
  /** The 40-hex head SHA the green CI verdict was observed at. Required with `status: "green"`. */
  headSha?: string;
  /** What changed and why — the narrative a reviewer reads. */
  summary: string;
  /** What a decision or credential this run needed (blocked runs). */
  blockers?: string[];
  /** Commands run as evidence, exactly as executed. */
  proof?: string[];
  /**
   * Durable facts about the REPOSITORY this run learned — where a subsystem is
   * wired, which command actually proves a change, a test that looks like it
   * asserts something and does not (Phase 3 knowledge overlay). Appended to the
   * fleet's per-repo overlay at settlement and shown to every later run in the
   * same repo, which is why they must be about the repo and not about this
   * task: "I fixed the parser" helps nobody, "the parser's only caller is
   * cli.ts:route" saves the next run a search. Optional, and every settlement
   * that predates it stays valid.
   */
  discoveries?: string[];
}

/**
 * The JSON Schema form of {@link WorkerSettlement}, handed to the harness as
 * the session's `outputSchema` (#540).
 *
 * Used two ways, and both must stay honest to the same contract:
 * `requireYieldTool` puts the `yield` tool in front of the worker with this
 * schema as its description, and `outputSchemaMode: "permissive"` means a
 * violation is a fallback, never a lost report — the worker's text path must
 * still settle a run whose yield does not parse.
 */
export const WORKER_SETTLEMENT_SCHEMA = {
  $schema: "https://json-schema.org/draft/2020-12/schema",
  type: "object",
  description:
    "The run's settlement: what this worker claims it achieved. The daemon verifies the PR itself; " +
    "this is the claim, cross-checked like the settlement report it replaces.",
  additionalProperties: false,
  required: ["status", "summary"],
  properties: {
    status: {
      type: "string",
      enum: ["green", "blocked", "failed"],
      description:
        "green: pushed and the checks you watched are green. blocked: a decision, credential or " +
        "repo fact is missing. failed: the run could not complete.",
    },
    prUrl: { type: "string", description: "The run's pull request URL — required with status green." },
    branch: { type: "string", description: "The branch the PR was pushed from." },
    headSha: {
      type: "string",
      description: "The 40-character head SHA you observed green — required with status green.",
    },
    summary: { type: "string", description: "One short paragraph: what changed and why." },
    blockers: {
      type: "array",
      items: { type: "string" },
      description: "What is missing or uncertain, one item per blocker (blocked runs).",
    },
    proof: {
      type: "array",
      items: { type: "string" },
      description: "The commands you ran as evidence, exactly as executed.",
    },
    discoveries: {
      type: "array",
      items: { type: "string" },
      description:
        "Optional. Durable facts about THIS REPOSITORY that cost you turns to find and would cost " +
        "the next run the same: where a subsystem is wired, which command actually proves a change, " +
        "a test that looks like it asserts something and does not. Not a summary of your task — " +
        "these are shown to every later run in this repo, so leave the field out rather than " +
        "filling it with what you did.",
    },
  },
} as const;

function scheduleWallClock(callback: () => void, delayMs: number): () => void {
  const timer = setTimeout(callback, delayMs);
  return () => clearTimeout(timer);
}

/**
 * Which ceiling stopped a run. Only ever set alongside `state: "killed"`: the
 * turn counter caught a loop, the wall clock caught a session that was stuck
 * without spending turns, or the run reached the spend allowance admission
 * reserved for it (#851).
 */
export type KilledBy = "turns" | "wallclock" | "spend";

export type WorkerPausePhase = "running" | "pausing" | "paused";

export interface WorkerPauseControl {
  phase(): WorkerPausePhase;
  /** Cooperative park; resolves when the harness is idle. Rejects when the
   * run is settling, cap-killed, or already pausing/paused. */
  pause(): Promise<void>;
  /** Returns a parked session to work with a continuation prompt. Throws
   * when the phase is not `paused`. */
  resume(): void;
  /** Terminally ends this run for an operator-supplied reason. The run is
   * recorded as an administrative kill and cannot be resumed. */
  stop(reason: string): void;
}

/** What a resumed session is told. One literal so tests can pin it. */
export const RESUME_PROMPT =
  "The operator paused this session and has now resumed it. Continue exactly where you left off: " +
  "re-check the outcome of your last action before repeating it, then keep working your original " +
  "brief to the same report contract.";

/**
 * What an orphan-resumed worker is told instead of re-sending its brief. One
 * literal so tests can pin it (#536): the original brief is already in the
 * resumed transcript, and re-sending it is how a resumed worker ends up
 * re-doing the work it just did.
 */
export const ORPHAN_RESUME_PROMPT =
  "Your previous process was interrupted by a daemon restart. Review the transcript's final state before acting, " +
  "then continue exactly where you left off: re-check the outcome of your last action before repeating it, and keep " +
  "working your original brief to the same report contract.";

/**
 * What a review-revision worker is told instead of re-sending its brief (#677).
 *
 * Same rule as {@link ORPHAN_RESUME_PROMPT}: the original brief is already in
 * the resumed transcript, and re-sending it is how a resumed worker re-does
 * work it already did. What makes this a revision rather than a resume is the
 * deliverable contract: the PR stays open, the branch stays the run's own, and
 * the session ends only when the same final-report shape says the revised head
 * is green. One literal so tests can pin it.
 */
export const REVIEW_REVISION_PROMPT =
  "The orchestrator reviewed your green pull request and found blocking findings. Continue this same session " +
  "on this same run: same branch, same pull request — do not close or reopen it, and do not open another. " +
  "Address exactly the findings below, push with conductor_push, verify the checks with conductor_pr_status, " +
  "and finish with the same final-report contract as before (yield your settlement: status green, prUrl, headSha). " +
  "Your original brief is already in this transcript; redo only what the findings implicate.";

/**
 * The full opening prompt for one review revision round: the pinned continuation
 * text plus the orchestrator's findings verbatim. Exported so a unit test can
 * pin the exact handoff without standing up a session.
 */
export function renderReviewRevisionPrompt(findings: string, round: number): string {
  return `${REVIEW_REVISION_PROMPT}\n\nReview round ${round} — blocking findings:\n\n${findings}`;
}

/**
 * The template one fresh correction's brief is rendered from.
 *
 * Read per correction rather than cached, exactly like the dispatcher's own
 * `buildBrief`: editing the brief then takes effect on the next correction
 * instead of needing a daemon restart.
 */
const CORRECTION_BRIEF_PATH = join(import.meta.dir, "briefs", "correction.md");

/**
 * How much of the diff one fresh correction brief carries (#1046).
 *
 * The same bound, and the same reasoning, as {@link ADJUDICATION_DIFF_LIMIT}: a
 * correction needs the change it is correcting, not every generated line, and an
 * enormous PR must not crowd out the findings and criteria that follow it. The
 * marker says exactly what was withheld, so the session reads the rest off the
 * branch it already has checked out rather than assuming it saw everything.
 */
export const CORRECTION_DIFF_LIMIT = 60_000;

/**
 * Everything one fresh correction is told (#1046).
 *
 * Assembled by the caller from durable rows and live tracker reads, never from a
 * worker's own claims — and, unlike a resumed revision, this is ALL the session
 * gets. The exhausted implementation transcript is not replayed, so a field
 * missing here is context the correction simply does not have.
 */
export interface CorrectionEvidence {
  /** Canonical `owner/repo` of the code repo, or the routing name when the
   *  clone URL yields none. Also the key the knowledge overlay is read under. */
  repo: string;
  issue: number;
  issueTitle?: string;
  /** The issue's acceptance criteria, verbatim. */
  issueBody: string;
  /** The pull request this correction must push onto. Never a second PR. */
  prUrl: string;
  /** The exact head the findings were written against. */
  headSha: string;
  /** The branch and worktree that already exist for this run. */
  branch: string;
  worktree: string;
  /** Live check state at that head, as the tracker reported it. */
  checks: string;
  /** The diff at that head, bounded by {@link CORRECTION_DIFF_LIMIT}. */
  diff: string;
  round: number;
  /** The review-round ceiling, when the caller knows it. */
  maxRounds?: number;
  /** This round's blocking findings, delivered verbatim. */
  findings: string;
  /** Earlier rounds and what became of them. A finding without its outcome
   *  reads as an outstanding complaint even when it was fixed, which is how a
   *  correction session re-does work that already landed. */
  priorRounds: readonly AdjudicationRound[];
  /**
   * Why this round is a fresh session rather than a resume, in plain words —
   * "the implementation session was killed by the turns cap", "the fleet's
   * worker model changed after the rejected run". It goes in the brief because a
   * session that does not know its transcript is gone will look for it: #1035's
   * whole failure was correction rounds spent replaying an exhausted session,
   * and a session told nothing behaves as if it were still that one.
   */
  reason: string;
  /** The gate commands this correction must run before pushing. */
  proofCommands: readonly string[];
  /**
   * The fleet's accumulated repo knowledge, pre-rendered by
   * {@link knowledgeSection}. Omitted, it is read here from {@link repo} — the
   * overlay exists to reach every session launched against a repo, so it must
   * not depend on each caller remembering to pass it. Pass `""` to render none.
   */
  knowledge?: string;
}

/**
 * The full opening prompt for one fresh correction round (#1046).
 *
 * Exported so a test can pin the exact payload without standing up a session:
 * the brief IS the deliverable here. `REVIEW_REVISION_PROMPT` can be three
 * sentences because the session it prompts already holds the issue, the plan and
 * the diff; this one holds nothing, so every one of those facts has to arrive in
 * the text — and a test that only asserted "a session was launched" would pass on
 * an empty brief.
 */
export function renderCorrectionBrief(evidence: CorrectionEvidence): string {
  const template = readFileSync(CORRECTION_BRIEF_PATH, "utf8");
  const rounds =
    evidence.priorRounds.length === 0
      ? "_(none recorded — this is the first review round of this run)_"
      : evidence.priorRounds
          .map((round) => `Round ${String(round.round)} (${round.outcome ?? "never settled"}):\n${round.findings.trim()}`)
          .join("\n\n");
  const diff =
    evidence.diff.length <= CORRECTION_DIFF_LIMIT
      ? evidence.diff
      : `${evidence.diff.slice(0, CORRECTION_DIFF_LIMIT)}\n` +
        `[diff truncated at ${String(CORRECTION_DIFF_LIMIT)} of ${String(evidence.diff.length)} characters — ` +
        "read the remainder from your own checkout with `git diff origin/HEAD...HEAD`]";
  const proof =
    evidence.proofCommands.length === 0
      ? "_The project names no gate commands. Run the repo's own tests and say exactly what you ran._"
      : ["```sh", ...evidence.proofCommands, "```"].join("\n");
  // The overlay's own trailing blank line separates it from the heading that
  // follows; an absent overlay renders nothing at all, so the brief closes up
  // around it rather than leaving an empty section behind.
  const knowledge = evidence.knowledge ?? knowledgeSection(evidence.repo);
  return renderBrief(template, {
    ISSUE_NUMBER: String(evidence.issue),
    ISSUE_TITLE: evidence.issueTitle ?? "(title unavailable at dispatch)",
    REPO: evidence.repo,
    PR_URL: evidence.prUrl,
    HEAD_SHA: evidence.headSha,
    CHECKS: evidence.checks,
    WORKTREE: evidence.worktree,
    BRANCH: evidence.branch,
    ROUND:
      evidence.maxRounds === undefined
        ? String(evidence.round)
        : `${String(evidence.round)} of ${String(evidence.maxRounds)} allowed`,
    REASON: evidence.reason,
    FINDINGS: evidence.findings.trim(),
    ACCEPTANCE_CRITERIA: evidence.issueBody.trim(),
    PRIOR_ROUNDS: rounds,
    DIFF: diff,
    PROOF_COMMANDS: proof,
    KNOWLEDGE: knowledge,
  });
}

/**
 * What one fresh correction is launched with (#1046).
 *
 * Deliberately {@link WorkerOpts} minus two keys. `brief` is gone because this
 * transport renders its own from `briefs/correction.md` — a caller that could
 * pass a brief could pass the findings-only resume prompt, which is the failure
 * this exists to end. `resume` is gone because it is not a decision available
 * here: a "fresh correction" that resumes is #1046's named silent fake, and the
 * cheapest way to make it unavailable is to make it inexpressible.
 *
 * Everything else — worktree, caps, sockets, grants, meters, pause control — is
 * inherited unchanged, because a correction is an ordinary mutable worker run on
 * an existing branch. The caller owns the run's identity and passes it through
 * here: this transport creates no worktree, opens no pull request, and derives no
 * issue or run of its own.
 */
export interface CorrectionWorkerOpts extends Omit<WorkerOpts, "brief" | "resume"> {
  /** The evidence the brief is rendered from. */
  evidence: CorrectionEvidence;
  /**
   * The transcript of the session this correction replaces, when the caller
   * recorded one. Used for exactly one thing: proving the lineage actually
   * changed (see {@link CorrectionResult.freshSession}).
   */
  originSessionRef?: string;
}

/** What one fresh correction produced (#1046). */
export interface CorrectionResult extends WorkerResult {
  /**
   * The transcript this correction opened — the lineage #1045 persists as
   * `correctionSessionRef`. Absent when the session opened none.
   */
  sessionRef?: string;
  /**
   * Whether this really was a new session. False when the harness handed back
   * {@link CorrectionWorkerOpts.originSessionRef} — the exact silent fake #1046
   * names, a new model string over the old session file — and false when no
   * transcript was opened at all, because neither case proves a fresh lineage.
   * A caller must not record a fresh-correction provenance it did not get.
   */
  freshSession: boolean;
}

/**
 * Launch one fresh correction on an existing branch, worktree and pull request
 * (#1046).
 *
 * The failure this ends (#1035): a capped implementation session kept receiving
 * correction rounds by `resume: true` after it had exhausted its turns and after
 * the fleet had switched worker models, so every round was spent replaying a dead
 * transcript under a model that was no longer configured.
 *
 * So this composes {@link runWorker} — the same session primitive, the same caps,
 * the same verb surface — with exactly two things different: the brief is
 * self-contained ({@link renderCorrectionBrief}) and `resume` is never passed, so
 * the harness opens a new transcript under whatever model the caller resolved.
 * A second worker runtime would be a second set of cap, settlement and pause
 * semantics to keep in sync, which is how the two paths drift apart.
 *
 * What it deliberately does NOT do: nothing here touches the run's identity. The
 * branch, worktree, pull request, issue, run id and verb channel all arrive from
 * the caller and leave unchanged, so a correction cannot open a second PR, create
 * a second worktree, or start a new attempt against the issue.
 */
export async function runCorrectionWorker(
  o: CorrectionWorkerOpts,
  deps: RunWorkerDeps = { createSession },
): Promise<CorrectionResult> {
  const { evidence, originSessionRef, ...worker } = o;
  // No `resume` key exists on `worker` to forward — the type forbids one, so
  // `runWorker` takes its fresh-session path and the harness opens a new
  // transcript in the run's own session directory.
  const result = await runWorker({ ...worker, brief: renderCorrectionBrief(evidence) }, deps);
  const sessionRef = result.sessionFile;
  return {
    ...result,
    ...(sessionRef === undefined ? {} : { sessionRef }),
    freshSession: sessionRef !== undefined && sessionRef !== originSessionRef,
  };
}


/**
 * What the review-ceiling adjudicator is told (#877).
 *
 * Deliberately NOT a continuation of anything. `REVIEW_REVISION_PROMPT` resumes
 * the implementation session to change code; this opens a FRESH session, under a
 * stronger model role, to make one decision about code it did not write. A
 * relabelled resume of the same session and model is the named silent fake of
 * #873: a fourth nominal opinion from the author is not an adjudication.
 *
 * So the contract is a verdict, not a diff: it may not push, may not edit, and
 * ends by yielding `cleared` or `rejected` with reasons. One literal so a test
 * can pin the exact handoff.
 */
export const REVIEW_ADJUDICATION_PROMPT =
  "You are the final adjudicator for a pull request that has exhausted its review rounds. You did not write this " +
  "code and you are not fixing it: your entire job is one decision. Read the acceptance criteria, the diff, the " +
  "check state and every prior review finding below, then judge whether THIS head satisfies the issue. " +
  "Do not edit files, do not push, do not open or close a pull request, and do not ask for another review round — " +
  "there are none left. Finish by yielding your verdict: `cleared` if this head should merge as it stands, " +
  "`rejected` if it should not, with the specific reasons that decided it. Judge the code in front of you, not the " +
  "process that produced it.";

/**
 * The adjudicator's structured verdict (#877). Permissive like the worker
 * settlement schema: an invalid or absent yield falls back to the text path
 * rather than losing the decision, and the daemon settles the durable row from
 * the parsed object when there is one.
 */
export const ADJUDICATION_VERDICT_SCHEMA = {
  $schema: "https://json-schema.org/draft/2020-12/schema",
  type: "object",
  description:
    "The adjudicator's terminal decision about one exact pull-request head. There is no third option and no " +
    "further review round: `cleared` sends this head to the ordinary merge path, `rejected` sends it to the " +
    "conductor's autonomous disposition.",
  additionalProperties: false,
  required: ["verdict", "reasons"],
  properties: {
    verdict: {
      type: "string",
      enum: ["cleared", "rejected"],
      description: "cleared: this head should merge as it stands. rejected: it should not.",
    },
    reasons: {
      type: "string",
      description:
        "Why, specifically: the criteria met or unmet, and the evidence in the diff or checks that decided it.",
    },
    headSha: {
      type: "string",
      description: "The 40-character head you judged — echoed back so a moved head is detectable.",
    },
  },
} as const;

/** One prior review round as the adjudicator must see it (#877): what was asked
 *  for, and what became of it. 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 interface AdjudicationRound {
  round: number;
  findings: string;
  /** The revision's settled outcome (`revised`, `failed`, `skipped`, …), or
   *  absent for a round that never settled. */
  outcome?: string;
}

/** Everything the adjudicator is given (#877). Assembled by the caller from
 *  durable rows and live tracker reads, never from the worker's own claims. */
export interface AdjudicationEvidence {
  issue: number;
  issueTitle?: string;
  /** The issue body verbatim: the acceptance criteria are the standard being
   *  judged, so they are never summarised. */
  issueBody: string;
  prUrl: string;
  /** The exact head being judged. */
  headSha: string;
  /** The live check verdict at that head, as the tracker reported it. */
  checks: string;
  /** The PR's diff at that head. Bounded — see {@link renderAdjudicationBrief}. */
  diff: string;
  /** Every prior review round, oldest first. */
  rounds: readonly AdjudicationRound[];
  /** The review ceiling that was reached, so the adjudicator knows why it is
   *  the last word rather than one more opinion. */
  maxRounds: number;
}

/** How much of the diff one adjudication brief carries. A decision needs the
 *  change, not every generated line: a bound keeps one enormous PR from
 *  crowding out the findings and criteria that follow it, and the marker says
 *  exactly what was withheld so the adjudicator can ask for it rather than
 *  assuming it saw everything. */
export const ADJUDICATION_DIFF_LIMIT = 60_000;

/**
 * The full opening prompt for one adjudication: the pinned contract plus the
 * bounded evidence, in the order a decision needs it — what was asked for, what
 * is on the head now, what previous reviews said, and what happened to it.
 *
 * Exported so a test can pin the exact payload without standing up a session,
 * which is #877's requirement: the brief is the deliverable here, and a test
 * that only asserts a stored flag would pass on an empty one.
 */
export function renderAdjudicationBrief(evidence: AdjudicationEvidence): string {
  const rounds =
    evidence.rounds.length === 0
      ? ["(none recorded — this adjudication was admitted without prior rounds)"]
      : evidence.rounds.map(
          (round) =>
            `Round ${round.round} (${round.outcome ?? "never settled"}):\n${round.findings.trim()}`,
        );
  const diff =
    evidence.diff.length <= ADJUDICATION_DIFF_LIMIT
      ? evidence.diff
      : `${evidence.diff.slice(0, ADJUDICATION_DIFF_LIMIT)}\n` +
        `[diff truncated at ${ADJUDICATION_DIFF_LIMIT} of ${evidence.diff.length} characters — ` +
        "read the remainder from the pull request if the decision turns on it]";
  return [
    REVIEW_ADJUDICATION_PROMPT,
    "",
    `Issue #${evidence.issue}${evidence.issueTitle === undefined ? "" : `: ${evidence.issueTitle}`}`,
    `Pull request: ${evidence.prUrl}`,
    `Head under adjudication: ${evidence.headSha}`,
    `Checks at that head: ${evidence.checks}`,
    `Review rounds exhausted: ${evidence.rounds.length} of ${evidence.maxRounds} allowed`,
    "",
    "## Acceptance criteria (the issue, verbatim)",
    "",
    evidence.issueBody.trim(),
    "",
    "## Prior review findings, and what became of them",
    "",
    rounds.join("\n\n"),
    "",
    "## The diff at this head",
    "",
    diff,
  ].join("\n");
}

export interface WorkerOpts {
  brief: string;
  cwd: string;
  caps: Caps;
  /** Canonical `owner/repo` identity used to scope prose-only PR URLs. */
  repoSlug?: string;
  /**
   * Directory the harness writes this run's transcript into — a directory, not
   * a file. The SDK takes no `sessionFile` input, so naming a path here would
   * only name one nothing ever writes to. Omitted, the harness picks its own
   * location; either way the real path comes back on {@link WorkerResult}.
   */
  sessionDir?: string;
  /**
   * Continue the most recent transcript in `sessionDir` instead of opening a
   * blank session (#536). The daemon sets it only for an orphan-clean
   * continuation, after verifying the prior transcript and worktree still
   * exist — the harness's own `continueRecent` is the backstop, and a silent
   * fallback to a fresh session is surfaced by the `sessionFile` lineage
   * compare at the dispatch site, never left quiet.
   */
  resume?: boolean;
  /**
   * Model pattern for this session, in omp's model/role syntax. Omitted leaves
   * the harness to pick, which is what an unconfigured project wants.
   */
  model?: string;
  /**
   * Absolute path to the fleet-owned omp settings overlay (#537): the YAML the
   * daemon materialised from the project's `ompSettings` map (plus the retry
   * keys derived from `modelFallbacks`, which is where #539's staging lives)
   * under the run's session directory. Forwarded to `createSession`, which
   * loads it through `Settings.init({ configFiles: [<path>] })`. Absent, no
   * settings are staged and dispatch is byte-for-byte what it is today.
   */
  ompSettingsFile?: string;
  /**
   * Effective per-shape release grants for this session. A worker is refused
   * every shape whatever they say — see {@link SessionRole} — so this is passed
   * for the audit trail and the wording of the refusal, not for permission.
   */
  releaseGrants?: ResolvedGrants;
  /** Durable audit sink for rejected release/deploy calls. */
  onReleaseBlocked?: (shape: GateShape, context: ReleaseBlockContext) => void;
  /** The child's pid, the instant it exists. See {@link VerbListener.bindPid}. */
  onSpawn?: (pid: number) => void;
  /**
   * The worker's Herdr representation (#840): opened from the exact child pid,
   * released when that child is gone.
   *
   * Passed straight through to `createSession`, and deliberately not built here
   * — a worker run knows its issue and attempt, but which workspace a pane
   * belongs in, and whether one should exist at all, is the daemon's to decide.
   */
  pane?: { open(pid: number): void; release(): void };
  /** Control socket for that child, beside the run's own session directory. */
  socketPath?: string;
  /**
   * The run's conductor verb socket (#126) — its only route to a push, a PR or
   * any other mutation. Distinct from `socketPath`: that is the daemon's
   * control channel to the child, this is the child's mediated way back out.
   */
  verbSocketPath?: string;
  /** Where the session child's stdout/stderr go. Defaults to this process's stderr. */
  onChildLog?: (line: string) => void;
  /**
   * Reads the effective ceiling at each turn boundary. Omitted, the configured
   * startup cap remains fixed for the run.
   */
  maxTurns?: () => number;
  /**
   * Estimated-USD ceiling for this session, from the allowance admission
   * reserved for the run (#851). Checked as each cost-bearing message lands,
   * because that is the only moment a response's cost is known — so the
   * declared overshoot is exactly one completed model response, never zero.
   * Omitted means the fleet has no spend cap and nothing to enforce.
   */
  maxSpendUsd?: number;
  /** Synchronous cap latch; called before the session abort begins. */
  onKilled?: (by: KilledBy) => void;
  onTurn?: (n: number) => void;
  /** Cumulative USD spend, reported as each cost-bearing message finishes. */
  onSpend?: (usd: number) => void;
  /**
   * Cumulative output and reasoning tokens for this run (#518).
   *
   * Reported alongside the spend it is derived from, so a reader can tell a run
   * deliberating expensively in wall clock but cheaply in dollars from one that
   * is idle or stuck — the difference that decides whether to intervene.
   */
  onTokens?: (tokens: { output: number; reasoning: number }) => void;
  /**
   * The transcript path, handed over the moment the session opens it rather
   * than at the end with {@link WorkerResult.sessionFile}. Both report the same
   * path; only this one arrives while there is still something to watch, which
   * is what `omp-conductor tail` attaches to. Never called for a session that
   * opened no transcript — there is no path to report, and an empty string
   * would be a path that fails to open rather than an absence.
   */
  onSessionFile?: (path: string) => void;
  /**
   * Installs the run's operator pause controller the moment the session
   * exists; absent = no pause surface (tests, one-shot callers).
   */
  onPauseControl?: (control: WorkerPauseControl) => void;
  /**
   * Pre-spawn admission gate, forwarded to `createSession` (#374): the daemon
   * closes it the moment its stop begins, so a stop that lands while the
   * session socket is binding refuses the spawn and the run settles `stopped`
   * instead of `failed`. Absent, the spawn always proceeds.
   */
  maySpawn?: () => boolean;
}

/**
 * Session creation plus the clock seam needed to prove wall-clock behavior
 * without sleeping. Production callers use the defaults.
 */
export interface RunWorkerDeps {
  createSession: typeof createSession;
  now?: () => number;
  schedule?: (callback: () => void, delayMs: number) => () => void;
}

export interface WorkerResult {
  state: RunState;
  prUrl?: string;
  headSha?: string;
  turns: number;
  spendUsd: number;
  /**
   * The run's settlement text. Today this is the worker's own last words; a
   * run that yielded a structured {@link WorkerSettlement} instead carries the
   * canonical rendering of that parsed object (#540) — the narrative lives in
   * its `summary`, transcribed here field for field rather than regexed out of
   * prose.
   */
  report: string;
  /** In-session HTTP 429 responses the session recorded (stopReason "error",
   *  errorStatus 429), counted as the messages streamed in. A healthy run
   *  reports 0; a run the harness retried through a barrel of rate limits
   *  carries the number, which is what distinguishes provider-capacity (the
   *  provider was throttling all along) from an ordinary failure (#573). */
  provider429Count: number;
  /**
   * The model that actually wrote this run's messages, read from
   * `AssistantMessage.model` on the newest assistant `message_end`. Present
   * even for a run that never failed over — it is the durable answer to "which
   * model wrote this" (#535 slice 1, from the message field rather than the
   * payload-free `model_changed` event). Absent only when no assistant message
   * carried a model.
   */
  model?: string;
  /** The provider that wrote them, read from the same `AssistantMessage.provider`. */
  provider?: string;
  /** Every within-run model fallback the harness applied
   *  (`retry_fallback_applied`), newest first. The `to` target is what a
   *  settlement report names when a run swapped providers mid-run. */
  retryFallbacks: { from: string; to: string }[];
  /** `retry_fallback_succeeded` events: within-run fallbacks the harness
   *  confirmed recovered on. */
  retryFallbackSucceeded: number;
  /** Assistant messages whose `retryRecovery.recovery === "model"` — the durable
   *  transcript record of a within-run model swap. */
  modelRecoveries: number;
  /** `auto_retry_start` events: in-session provider retries the harness ran. */
  autoRetryCount: number;
  /** `auto_compaction_start` events: in-session context compactions. */
  autoCompactionCount: number;
  killedBy?: KilledBy;
  /** Present only when an operator terminally stopped this run. */
  stoppedReason?: string;
  /**
   * Transcript the session actually opened, absent if it opened none. Recorded
   * per run because it is the only readable evidence left once the worktree is
   * cleaned up.
   */
  sessionFile?: string;
  /**
   * Set when the harness could not honour {@link WorkerOpts.model} and used
   * another. Carried out rather than swallowed: a run that quietly read dumber
   * is otherwise indistinguishable from a run that was merely unlucky.
   */
  modelFallbackMessage?: string;
  /**
   * The code-graph session observation (#726): what the session's own registry
   * held at start, read off the session exactly like `modelFallbackMessage`.
   * Absent means the session surface did not record one — never "graph tools
   * absent", which is the `present: false` truth value. A dispatched run that
   * records it lets the doctor tell "the model ignored a tool it had" from
   * "the tool was missing" apart.
   */
  graphTools?: GraphToolsObservation;
  /**
   * The repo facts this run reported in its structured settlement
   * ({@link WorkerSettlement.discoveries}), verbatim and unjudged — the
   * daemon appends them to the repo's knowledge overlay at settlement.
   *
   * Surfaced here rather than parsed back out of {@link report}: the rendered
   * report is prose for a human, and re-deriving a list from it would be a
   * second, weaker contract beside the schema the worker actually answered.
   * Absent whenever the run yielded no structured settlement or named nothing
   * — never an empty array, which would read as "the fleet learned nothing"
   * rather than "the worker was not asked".
   */
  discoveries?: string[];
}

/**
 * Fill `{{KEY}}` placeholders from `vars`.
 *
 * An unknown key is left verbatim rather than blanked: a brief that reads
 * `Fix {{ISSUE}}` is obviously broken to whoever reads the transcript, whereas
 * `Fix undefined` looks like an instruction.
 */
export function renderBrief(template: string, vars: Record<string, string>): string {
  return template.replace(PLACEHOLDER_PATTERN, (placeholder, key: string) => {
    // Own properties only, so `{{constructor}}` cannot reach up the prototype.
    if (!Object.hasOwn(vars, key)) return placeholder;
    const value = vars[key];
    return value === undefined ? placeholder : value;
  });
}

/**
 * Read the run's structured outcome evidence from its final report.
 *
 * A textual `pushed-green` claim is not success by itself. The exact state line
 * must carry both a PR URL and the head SHA observed after CI; the daemon then
 * asks the tracker to verify those facts independently. Missing or malformed
 * evidence fails closed.
 *
 * A parsed {@link WorkerSettlement} wins over the text whenever it is present
 * (#540): the schema is the contract now, and a misleadingly regular-looking
 * prose block must not out-vote the object the worker actually yielded. The
 * text path is unchanged underneath, so a run that never yields settles
 * exactly as it did before.
 */
export function deriveResult(
  report: string,
  repoSlug?: string,
  structured?: WorkerSettlement,
): {
  state: RunState;
  prUrl?: string;
  headSha?: string;
} {
  if (structured !== undefined && (structured.status === "green" || structured.status === "blocked" || structured.status === "failed")) {
    const prUrl = structured.prUrl;
    const headSha = structured.headSha?.toLowerCase();
    if (structured.status === "green" && prUrl !== undefined && headSha !== undefined) {
      return { state: "pushed-green", prUrl, headSha };
    }
    return {
      state: structured.status === "blocked" ? "blocked" : "failed",
      ...(prUrl === undefined ? {} : { prUrl }),
      ...(headSha === undefined ? {} : { headSha }),
    };
  }
  const structuredPrUrl = PR_URL_PATTERN.exec(report)?.[1];
  const headSha = HEAD_SHA_PATTERN.exec(report)?.[1]?.toLowerCase();
  if (PUSHED_GREEN_PATTERN.test(report) && structuredPrUrl !== undefined && headSha !== undefined) {
    return { state: "pushed-green", prUrl: structuredPrUrl, headSha };
  }

  let prUrl = structuredPrUrl;
  if (prUrl === undefined && repoSlug !== undefined) {
    const expectedRepo = repoSlug.toLowerCase();
    for (const match of report.matchAll(GITHUB_PR_URL_PATTERN)) {
      if (match[1]?.toLowerCase() === expectedRepo) prUrl = match[0];
    }
  }

  const state: RunState = BLOCKED_PATTERN.test(report) ? "blocked" : "failed";
  return {
    state,
    ...(prUrl === undefined ? {} : { prUrl }),
    ...(headSha === undefined ? {} : { headSha }),
  };
}

/**
 * Parse a worker's structured settlement out of one assistant message's
 * content (#540).
 *
 * The worker calls the `yield` tool with `{ result: { data: <settlement> } }`
 * and the harness records that call as a `toolCall` content block on the
 * assistant message — the same content `reportText` flattens. Parsed
 * permissively on purpose: `outputSchemaMode` is permissive, so a yield the
 * schema rejected must still fall back to the text path rather than vanish.
 * The acceptance bar mirrors the schema's own `required`: a recognizable
 * `status` and a string `summary`; every other field is adopted when it has
 * the right type and dropped otherwise. Exported so the run loop's precedence
 * (yield over prose) is pinned by a unit test.
 */
export function structuredSettlement(content: unknown): WorkerSettlement | undefined {
  if (!Array.isArray(content)) return undefined;
  let parsed: WorkerSettlement | undefined;
  for (const block of content) {
    if (field(block, "type") !== "toolCall") continue;
    if (field(block, "name") !== "yield") continue;
    const args = field(block, "arguments");
    if (args === null || typeof args !== "object") continue;
    const result = field(args, "result");
    if (result === null || typeof result !== "object") continue;
    const data = field(result, "data");
    if (data === null || typeof data !== "object" || Array.isArray(data)) continue;
    const record = data as Record<string, unknown>;
    const status = record.status;
    if (status !== "green" && status !== "blocked" && status !== "failed") continue;
    const summary = record.summary;
    if (typeof summary !== "string") continue;
    const settlement: WorkerSettlement = { status, summary };
    for (const key of ["prUrl", "branch", "headSha"] as const) {
      const value = record[key];
      if (typeof value === "string" && value !== "") settlement[key] = value;
    }
    for (const key of ["blockers", "proof", "discoveries"] as const) {
      const value = record[key];
      if (Array.isArray(value) && value.every((item) => typeof item === "string")) {
        settlement[key] = value;
      }
    }
    parsed = settlement;
  }
  return parsed;
}

/**
 * The adjudicator's parsed verdict (#877), or `undefined` when it yielded
 * nothing usable.
 *
 * Same shape of parse as {@link structuredSettlement}, and separate from it on
 * purpose: an adjudication is a decision, not a settlement, and letting the two
 * share a parser would let a worker-shaped yield read as a verdict.
 */
export function structuredVerdict(content: unknown): AdjudicationVerdict | undefined {
  if (!Array.isArray(content)) return undefined;
  let parsed: AdjudicationVerdict | undefined;
  for (const block of content) {
    if (field(block, "type") !== "toolCall") continue;
    if (field(block, "name") !== "yield") continue;
    const args = field(block, "arguments");
    if (args === null || typeof args !== "object") continue;
    const result = field(args, "result");
    if (result === null || typeof result !== "object") continue;
    const data = field(result, "data");
    if (data === null || typeof data !== "object" || Array.isArray(data)) continue;
    const record = data as Record<string, unknown>;
    const verdict = record.verdict;
    if (verdict !== "cleared" && verdict !== "rejected") continue;
    const reasons = record.reasons;
    if (typeof reasons !== "string" || reasons.trim() === "") continue;
    const decided: AdjudicationVerdict = { verdict, reasons };
    const headSha = record.headSha;
    if (typeof headSha === "string" && headSha !== "") decided.headSha = headSha;
    parsed = decided;
  }
  return parsed;
}

/** One adjudicator's decision as the daemon receives it (#877). */
export interface AdjudicationVerdict {
  verdict: "cleared" | "rejected";
  reasons: string;
  /** The head the adjudicator says it judged, when it echoed one back. The
   *  daemon compares it against the head it dispatched: a mismatch is a
   *  stale-head settle, never a merge. */
  headSha?: string;
}

/** What one adjudication run produced (#877). */
export interface AdjudicationResult {
  /** The parsed verdict, absent when the session ended without yielding one —
   *  which settles the durable row `failed`, never `cleared`. */
  verdict?: AdjudicationVerdict;
  /** The session's last words, kept whether or not a verdict parsed: for a
   *  failed adjudication this text IS the evidence. */
  report: string;
  /** What actually ran it, for the durable provenance (#875). */
  model?: string;
  provider?: string;
  turns: number;
  spendUsd: number;
}

export interface AdjudicatorOpts {
  /** The rendered brief from {@link renderAdjudicationBrief}. */
  brief: string;
  /** A directory the session may read: everything it needs is in the brief, so
   *  this is deliberately not a run worktree — an adjudicator has no branch. */
  cwd: string;
  /** The model pattern the configured adjudicator role resolved to, in omp's
   *  model/role syntax. Omitted leaves the harness to choose, which is what an
   *  unconfigured project gets. */
  model?: string;
  sessionDir?: string;
  ompSettingsFile?: string;
  socketPath?: string;
  onSpawn?: (pid: number) => void;
  onChildLog?: (line: string) => void;
  maySpawn?: () => boolean;
  /** Hard turn ceiling for the decision. A verdict is one read and one answer;
   *  a session still going after this is not deciding. */
  maxTurns: number;
}

/**
 * Run one bounded adjudication (#877).
 *
 * A FRESH session under the configured role — never a resume of the
 * implementation worker, which is the whole point of escalating at the ceiling:
 * a fourth opinion from the same session and model is not an adjudication.
 *
 * Read-only by construction, and that is enforced by what is NOT passed:
 * no `verbSocketPath`, so the conductor mutation verbs (`conductor_push`,
 * `conductor_pr_create`, everything downstream) are unreachable from this
 * session — it cannot push, open, close or merge anything. No release grants
 * either. Its only output is the verdict it yields.
 */
export async function runAdjudicator(
  o: AdjudicatorOpts,
  deps: RunWorkerDeps = { createSession },
): Promise<AdjudicationResult> {
  let session: AgentSessionLike;
  try {
    session = await deps.createSession({
      cwd: o.cwd,
      ...(o.sessionDir === undefined ? {} : { sessionDir: o.sessionDir }),
      ...(o.model === undefined ? {} : { model: o.model }),
      ...(o.ompSettingsFile === undefined ? {} : { ompSettingsFile: o.ompSettingsFile }),
      ...(o.socketPath === undefined ? {} : { socketPath: o.socketPath }),
      ...(o.onSpawn === undefined ? {} : { onSpawn: o.onSpawn }),
      ...(o.onChildLog === undefined ? {} : { onChildLog: o.onChildLog }),
      ...(o.maySpawn === undefined ? {} : { maySpawn: o.maySpawn }),
      // Worker role: structured file tools stay inside `cwd` and no release
      // grant can reach this session. Deliberately no `verbSocketPath` — see
      // the doc comment: an adjudicator decides, it never mutates.
      role: "worker",
      outputSchema: ADJUDICATION_VERDICT_SCHEMA,
      outputSchemaMode: "permissive",
      requireYieldTool: true,
    });
  } catch (err) {
    // The pre-spawn admission gate closed, or the harness could not start. Both
    // are "no verdict", and the caller settles the durable row accordingly —
    // never as a clear.
    return {
      report: err instanceof Error ? err.message : String(err),
      turns: 0,
      spendUsd: 0,
    };
  }

  let turns = 0;
  let spendUsd = 0;
  let report = "";
  let verdict: AdjudicationVerdict | undefined;
  let model: string | undefined;
  let provider: string | undefined;
  const finished = Promise.withResolvers<void>();
  session.on("turn_start", () => {
    turns += 1;
    if (turns > o.maxTurns) session.abort();
  });
  session.on("message_end", (event) => {
    const message = field(event, "message");
    if (field(message, "role") !== "assistant") return;
    const content = field(message, "content");
    const parsed = structuredVerdict(content);
    if (parsed !== undefined) verdict = parsed;
    const text = reportText(content);
    if (text !== "") report = text;
    const usage = field(message, "usage");
    const cost = usage === null ? null : field(usage, "cost");
    const total = cost === null ? null : field(cost, "total");
    if (typeof total === "number") spendUsd += total;
    const messageModel = field(message, "model");
    if (typeof messageModel === "string" && messageModel !== "") model = messageModel;
    const messageProvider = field(message, "provider");
    if (typeof messageProvider === "string" && messageProvider !== "") provider = messageProvider;
  });
  session.on("agent_end", (event) => {
    // Same rule as the worker path: a non-terminal `agent_end` means the
    // harness will resume the session, so completing there would read a
    // mid-flight snapshot as the verdict.
    if (shouldComplete(event as { isTerminal?: boolean })) finished.resolve();
  });
  session.on("session_exit", () => finished.resolve());
  try {
    await session.prompt(o.brief);
    await finished.promise;
  } finally {
    await disposeSession(session);
  }
  return {
    ...(verdict === undefined ? {} : { verdict }),
    report,
    ...(model === undefined ? {} : { model }),
    ...(provider === undefined ? {} : { provider }),
    turns,
    spendUsd,
  };
}

/**
 * The highest turn count any to-spec pass that finished used, measured across
 * both projects' grooming ledger on 2026-08-25 (#1064): of the fourteen
 * groom sessions ever run, the eight that finished verdicts used 10–19 turns
 * (BLOCKED ×6, PROMOTABLE ×1, NEEDS DECOMPOSITION ×1), while every session
 * killed at the ceiling was 21+ turns in when the abort fired. The ceiling
 * and its measurement live here, in one place: the ceiling is the measured
 * max plus one turn of headroom.
 */
const TO_SPEC_MEASURED_FINISHED_MAX_TURNS = 19;

/**
 * The turn ceiling for one to-spec grooming scout (#1041, #1064).
 *
 * Half the adjudicator's 40, and for the opposite reason: an adjudication is
 * handed its whole evidence in the brief (issue, diff, checks, every prior
 * round) and spends its turns reasoning, while a scout is handed a candidate
 * and must go and READ — the source repo at a ref, the entry points, the
 * existing tests, the later work that may have retired the premise. That is
 * search, and search is where a session bounded only by a worker's budget
 * quietly spends a worker's budget. One turn above the highest measured
 * finishing pass is enough for a bounded repo walk plus the structured
 * answer; a scout still going after them has stopped grooming one candidate
 * and started exploring the repository, which is exactly the cost the
 * mechanical selection bounds everywhere else.
 */
export const TO_SPEC_MAX_TURNS = TO_SPEC_MEASURED_FINISHED_MAX_TURNS + 1;

/** What one to-spec scout produced (#1041). */
export interface ToSpecScoutResult {
  /**
   * The agent's answer, verbatim and unjudged — the structured payload it
   * yielded (through the repair round when the first answer failed), or a
   * fully conforming block recovered from its own text when it never yielded
   * (#1064). This string is `recordToSpecGrooming`'s `input`, and nothing
   * here reads a field of it beyond what the repair round must probe:
   * conductor re-validates every result through the strict parser at
   * persistence time, so a runner that judged would be a second, weaker
   * contract. Empty when the session produced no answer at all — the
   * narration stays in `report`, and the pass records `no-answer`.
   */
  raw: string;
  /** The session's last words, kept whether or not it answered: for a scout
   *  that produced nothing this text IS the evidence of what went wrong, and
   *  it is recorded as the refusal's context — never parsed as a verdict
   *  (#1064). */
  report: string;
  /**
   * How `raw` arrived (#1064): through the `yield` tool, or recovered from
   * the session's text because it never yielded (`"text"` is the record's own
   * admission that the answer did not come through the contract), or `"none"`
   * when the pass produced no answer at all. Absent on results written before
   * #1064; callers read a non-empty `raw` as a yield then.
   */
  via?: "yield" | "text" | "none";
  /** Whether the one bounded repair round produced the carried answer
   *  (#1064): the first answer parsed but violated the contract, the
   *  validation error was handed back, and this raw is the re-answer. */
  repaired?: boolean;
  /** Whether the turn ceiling aborted the session before it finished — the
   *  cap-kill shape of the no-answer class (#1064). */
  killedAtCeiling?: boolean;
  /** What actually ran it, for the durable provenance (#875). */
  model?: string;
  provider?: string;
  turns: number;
  spendUsd: number;
}

export interface ToSpecScoutOpts {
  /** The rendered `briefs/to-spec.md` for exactly one candidate. */
  brief: string;
  /** A directory the session may read: everything about the candidate is in
   *  the brief, and a scout has no branch — so this is the session directory,
   *  never a run worktree. */
  cwd: string;
  /** The model pattern `project.groomRole` resolved to, in omp's model/role
   *  syntax. Omitted leaves the harness to choose. */
  model?: string;
  sessionDir?: string;
  ompSettingsFile?: string;
  socketPath?: string;
  onSpawn?: (pid: number) => void;
  onChildLog?: (line: string) => void;
  maySpawn?: () => boolean;
  /** Hard turn ceiling; the daemon passes {@link TO_SPEC_MAX_TURNS}. */
  maxTurns: number;
}

/**
 * Run one bounded to-spec grooming pass (#1041, #1064).
 *
 * The adjudicator's twin, and read-only by the same construction: no
 * `verbSocketPath`, so no conductor mutation verb — no label change, no
 * promotion, no push — is reachable from this session. That absence is the
 * whole safety argument for letting the daemon launch grooming unattended:
 * the scout cannot queue the candidate it is grooming, only describe it. The
 * queue label is added later, by the daemon, and only through the ready gate.
 *
 * Deliberate differences from {@link runAdjudicator}. It carries
 * {@link TO_SPEC_SCHEMA} with `outputSchemaMode: "strict"` — the to-spec
 * contract has always been invoked strictly (#772), and a permissive schema
 * would let a half-filled result through the harness only for the parser to
 * refuse it as malformed one layer later, spending the batch to learn nothing.
 * And the answer is carried UNJUDGED: `recordToSpecGrooming` is the single
 * persistence path and the only validator. The one exception is the repair
 * round (#1064): a payload that parses but violates the contract is probed
 * with the same collector parser (`gateToSpecPayload`), the validation error
 * is prompted back into the session, and the re-answer is carried instead —
 * a complete, source-verified verdict must not be discarded over a key name.
 * A probe that refuses a payload is stale narration, never a verdict.
 *
 * What comes back goes beyond a plain string:
 *
 *  - `raw` is ONLY the structured answer — the yield payload, or a fully
 *    conforming block recovered from the session's text when it never
 *    yielded. The runner never substitutes narration for the answer, so a
 *    session that produced no payload returns `raw: ""` and the pass records
 *    `no-answer` rather than grading prose as JSON.
 *  - `via` says how the answer arrived; a text-recovery carries `"text"` so
 *    the durable record admits the answer did not come through the yield
 *    contract even though its verdict may be used.
 *  - `killedAtCeiling` names the turn-ceiling kill, which is what makes the
 *    no-answer shape legible instead of a silent failure.
 */
export async function runToSpecScout(
  o: ToSpecScoutOpts,
  deps: RunWorkerDeps = { createSession },
): Promise<ToSpecScoutResult> {
  let session: AgentSessionLike;
  try {
    session = await deps.createSession({
      cwd: o.cwd,
      ...(o.sessionDir === undefined ? {} : { sessionDir: o.sessionDir }),
      ...(o.model === undefined ? {} : { model: o.model }),
      ...(o.ompSettingsFile === undefined ? {} : { ompSettingsFile: o.ompSettingsFile }),
      ...(o.socketPath === undefined ? {} : { socketPath: o.socketPath }),
      ...(o.onSpawn === undefined ? {} : { onSpawn: o.onSpawn }),
      ...(o.onChildLog === undefined ? {} : { onChildLog: o.onChildLog }),
      ...(o.maySpawn === undefined ? {} : { maySpawn: o.maySpawn }),
      // Worker role, and deliberately no `verbSocketPath` — see the doc
      // comment: a scout reads and answers, it never mutates.
      role: "worker",
      outputSchema: TO_SPEC_SCHEMA,
      outputSchemaMode: "strict",
      requireYieldTool: true,
    });
  } catch (err) {
    // The pre-spawn admission gate closed, or the harness could not start.
    // Both are "no answer", and the caller settles the grooming row on the
    // failure text rather than leaving the in-flight marker to expire.
    return {
      raw: "",
      report: err instanceof Error ? err.message : String(err),
      via: "none",
      turns: 0,
      spendUsd: 0,
    };
  }

  let turns = 0;
  let spendUsd = 0;
  let report = "";
  let yielded = "";
  let model: string | undefined;
  let provider: string | undefined;
  let killedAtCeiling = false;
  // Re-armed for the repair round: the completion resolvers serve one prompt
  // at a time, and the repair prompt completes its own (turn-bounded) window.
  let finished = Promise.withResolvers<void>();
  session.on("turn_start", () => {
    turns += 1;
    if (turns > o.maxTurns) {
      killedAtCeiling = true;
      session.abort();
    }
  });
  session.on("message_end", (event) => {
    const message = field(event, "message");
    if (field(message, "role") !== "assistant") return;
    const content = field(message, "content");
    const payload = yieldedPayload(content);
    if (payload !== undefined) yielded = payload;
    const text = reportText(content);
    if (text !== "") report = text;
    const usage = field(message, "usage");
    const cost = usage === null ? null : field(usage, "cost");
    const total = cost === null ? null : field(cost, "total");
    if (typeof total === "number") spendUsd += total;
    const messageModel = field(message, "model");
    if (typeof messageModel === "string" && messageModel !== "") model = messageModel;
    const messageProvider = field(message, "provider");
    if (typeof messageProvider === "string" && messageProvider !== "") provider = messageProvider;
  });
  session.on("agent_end", (event) => {
    // Same rule as every other session runner: a non-terminal `agent_end`
    // means the harness will resume, so completing there would read a
    // mid-flight snapshot as the answer.
    if (shouldComplete(event as { isTerminal?: boolean })) finished.resolve();
  });
  session.on("session_exit", () => finished.resolve());

  /**
   * What the session produced after one prompt, and whether that answer is a
   * single re-answer away from the contract (#1064). The newest yield wins
   * and is carried unjudged; without a yield, a fully conforming fenced block
   * from the text is carried as the answer (marked `via: "text"`), a
   * repairable block is NOT carried but its failure is reported so the repair
   * round can run, and prose is never carried — narration is context, not an
   * answer.
   */
  const inspect = (): {
    payload: string;
    via: "yield" | "text" | "none";
    failure: ToSpecFailure | undefined;
  } => {
    if (yielded !== "") {
      const gate = gateToSpecPayload(yielded);
      return {
        payload: yielded,
        via: "yield",
        failure: gate.state === "repairable" ? gate.failure : undefined,
      };
    }
    const gate = gateToSpecPayload(report);
    if (gate.state === "conforming") return { payload: gate.body, via: "text", failure: undefined };
    if (gate.state === "repairable") return { payload: "", via: "none", failure: gate.failure };
    return { payload: "", via: "none", failure: undefined };
  };

  let repaired = false;
  try {
    await session.prompt(o.brief);
    await finished.promise;
    let answer = inspect();
    // One bounded repair round (#1064): a payload that parsed but violated
    // the contract is handed back once with the validation error before any
    // refusal is persisted. The same turn ceiling still bounds the session —
    // a cap kill means the session is gone and there is nothing to prompt.
    if (answer.failure !== undefined && !killedAtCeiling) {
      repaired = true;
      finished = Promise.withResolvers<void>();
      try {
        await session.prompt(toSpecRepairPrompt(answer.failure.detail));
        await finished.promise;
        answer = inspect();
      } catch {
        // The session could not take a repair prompt (it had already
        // exited). Whatever the first round produced stands, unjudged; the
        // persistence layer refuses it, with the payload kept in the row.
      }
    }
    return {
      raw: answer.payload,
      report,
      via: answer.via,
      ...(repaired ? { repaired: true } : {}),
      ...(killedAtCeiling ? { killedAtCeiling: true } : {}),
      ...(model === undefined ? {} : { model }),
      ...(provider === undefined ? {} : { provider }),
      turns,
      spendUsd,
    };
  } finally {
    await disposeSession(session);
  }
}

/**
 * The one repair round's prompt (#1064): what failed and which field, plus
 * the yield contract the corrected answer must come through. Bounded by the
 * same remaining turn ceiling as the original pass.
 */
function toSpecRepairPrompt(detail: string): string {
  return (
    "Your grooming verdict was received but failed validation and was NOT recorded.\n\n" +
    `Validation error: ${detail}\n\n` +
    "This is your one repair round. Yield the complete, corrected verdict payload through the yield tool — " +
    "an answer written as plain text is not an answer."
  );
}

/**
 * The `yield` call's payload, serialized, or `undefined` when this message
 * carried none (#1041).
 *
 * Transport, not decoding: it walks to `arguments.result.data` and stringifies
 * whatever object is there without reading one field of it. That is the
 * difference from {@link structuredVerdict} and {@link structuredSettlement},
 * which validate because their callers act on the parse — here the strict
 * to-spec parser at persistence time is the only validator, and a shape this
 * function rejected would be a verdict silently downgraded to "no answer".
 */
function yieldedPayload(content: unknown): string | undefined {
  if (!Array.isArray(content)) return undefined;
  let payload: string | undefined;
  for (const block of content) {
    if (field(block, "type") !== "toolCall") continue;
    if (field(block, "name") !== "yield") continue;
    const result = field(field(block, "arguments"), "result");
    const data = field(result, "data");
    if (data === null || data === undefined || typeof data !== "object") continue;
    payload = JSON.stringify(data);
  }
  return payload;
}

/**
 * Render a parsed {@link WorkerSettlement} as the run's stored report (#540).
 *
 * A worker that yields structured output may write little or no prose after
 * the yield call, and the prose shape is no longer the contract — so the
 * report the store keeps for such a run is this canonical rendering of the
 * parsed object itself: the same fields a reviewer would have had to regex
 * out of prose, derived from the object, field for field. The daemon's own
 * postfixes — the PR-diff `changed:` line and the reliability sentence — are
 * appended to it exactly as they are to a prose report.
 */
export function renderSettlement(settlement: WorkerSettlement): string {
  const lines: string[] = [`status: ${settlement.status}`];
  if (settlement.prUrl !== undefined) lines.push(`pr: ${settlement.prUrl}`);
  if (settlement.branch !== undefined) lines.push(`branch: ${settlement.branch}`);
  if (settlement.headSha !== undefined) lines.push(`head: ${settlement.headSha}`);
  if (settlement.summary !== "") lines.push("", settlement.summary);
  if (settlement.blockers !== undefined && settlement.blockers.length > 0) {
    lines.push("", "blockers:");
    for (const blocker of settlement.blockers) lines.push(`  - ${blocker}`);
  }
  if (settlement.proof !== undefined && settlement.proof.length > 0) {
    lines.push("", "proof:");
    for (const item of settlement.proof) lines.push(`  - ${item}`);
  }
  // Rendered like the other list fields so the stored report shows what was
  // appended to the repo's knowledge overlay: an entry a reader disagrees with
  // is traceable to the run that claimed it.
  if (settlement.discoveries !== undefined && settlement.discoveries.length > 0) {
    lines.push("", "discoveries:");
    for (const item of settlement.discoveries) lines.push(`  - ${item}`);
  }
  return lines.join("\n");
}

/**
 * Did this report state a verdict at all?
 *
 * The discriminator between "the worker said something about its outcome" and
 * "the worker said something else entirely". A finished worker woken by its own
 * stale polling timers emits neither `state: pushed-green` nor `state: blocked`
 * — it emits prose — and that is the only case where an earlier verdict is
 * still the run's real answer.
 */
export function hasVerdictLine(report: string): boolean {
  return STATE_LINE_PATTERN.test(report);
}

/**
 * Is this `agent_end` the end of the run?
 *
 * `isTerminal: false` means the harness will resume the session for maintenance
 * or async delivery, so the messages so far are a snapshot rather than a
 * result: completing there truncates the worker mid-flight and reports a
 * partial run as final. An absent field is terminal — that is what older
 * harness builds send, and treating it as non-terminal would hang every run.
 *
 * Exported because it is the one seam that lets this rule be tested without
 * standing up an SDK session.
 */
export function shouldComplete(event: { isTerminal?: boolean }): boolean {
  return event.isTerminal !== false;
}

/**
 * Drive one session to completion, a cap, or a failure. Never throws for a
 * failed run — a rejected `prompt()` is reported as `state: "failed"` so the
 * dispatcher's retry/escalate logic has one shape to reason about.
 */
export async function runWorker(
  o: WorkerOpts,
  deps: RunWorkerDeps = { createSession },
): Promise<WorkerResult> {
  const { workerWallClockMs } = o.caps;
  const maxTurns = o.maxTurns ?? (() => o.caps.workerMaxTurns);
  const now = deps.now ?? Date.now;
  const schedule = deps.schedule ?? scheduleWallClock;

  let session: AgentSessionLike;
  // The per-issue reviewer brief (#542): when the staged omp settings turn the
  // advisor on, drop a WATCHDOG.md rendered from this brief's own acceptance
  // criteria into the worktree before the session exists — the advisor's
  // watchdog discovery runs at session start, rooted at cwd, and a file that
  // appears afterwards is a reviewer flying blind. The managed exclude block
  // ignores that exact path (never `.omp/`), so it cannot reach the PR diff or
  // a salvage commit. Best-effort: a reviewer brief is advisory, and a staging
  // failure must not cost the run.
  if (o.resume === undefined && o.ompSettingsFile !== undefined) {
    stageIssueWatchdog(o.cwd, o.brief, o.ompSettingsFile);
  }
  try {
    session = await deps.createSession({
      cwd: o.cwd,
      ...(o.sessionDir === undefined ? {} : { sessionDir: o.sessionDir }),
      ...(o.model === undefined ? {} : { model: o.model }),
      ...(o.resume === undefined ? {} : { resume: o.resume }),
      // The fleet-owned omp settings overlay (#537): carry the staged overlay
      // path to the session so it loads the project's omp settings. Absent,
      // nothing is staged and the harness discovers settings as it does today.
      ...(o.ompSettingsFile === undefined ? {} : { ompSettingsFile: o.ompSettingsFile }),
      // Prevention half of #24: as a worker, structured file tools cannot leave
      // this worktree, and no release grant can ever reach this session (#122).
      role: "worker",
      ...(o.releaseGrants === undefined ? {} : { releaseGrants: o.releaseGrants }),
      ...(o.onReleaseBlocked === undefined ? {} : { onReleaseBlocked: o.onReleaseBlocked }),
      ...(o.onSpawn === undefined ? {} : { onSpawn: o.onSpawn }),
      ...(o.pane === undefined ? {} : { pane: o.pane }),
      ...(o.socketPath === undefined ? {} : { socketPath: o.socketPath }),
      ...(o.verbSocketPath === undefined ? {} : { verbSocketPath: o.verbSocketPath }),
      ...(o.onChildLog === undefined ? {} : { onChildLog: o.onChildLog }),
      ...(o.maySpawn === undefined ? {} : { maySpawn: o.maySpawn }),
      // The structured settlement contract (#540): the worker's `yield` tool
      // validates its `data` payload against this schema, permissively — an
      // invalid or absent yield falls back to the text path, never a lost
      // report. Always-on for workers: one schema, and no per-project shape.
      outputSchema: WORKER_SETTLEMENT_SCHEMA,
      outputSchemaMode: "permissive",
      requireYieldTool: true,
    });
  } catch (err) {
    // The pre-spawn gate closed (#374): a daemon stop landed while the session
    // socket was binding, so no child was ever created. That is a stop, not a
    // failure — reporting `failed` here would charge an attempt against a
    // worker the shutdown refused to start. `onSpawn` was never called.
    if (err instanceof SessionAdmissionClosedError) {
      return {
        state: "stopped",
        turns: 0,
        spendUsd: 0,
        provider429Count: 0,
        retryFallbacks: [],
        retryFallbackSucceeded: 0,
        modelRecoveries: 0,
        autoRetryCount: 0,
        autoCompactionCount: 0,
        report: "",
        stoppedReason: "daemon shutdown began before the worker session started",
      };
    }
    throw err;
  }

  // Before the first turn, not after the last: a caller that only learns the
  // transcript path from the result learns it once the run it wanted to watch
  // is already over.
  if (session.sessionFile !== undefined) o.onSessionFile?.(session.sessionFile);

  // Every exit below reports the session's own facts the same way: the
  // transcript it actually opened, and any model downgrade it announced. Read at
  // return time so a session that materialises either late is still reported
  // honestly.
  // The within-run reliability surface this worker now records (#539): the
  // resolved model/provider plus the fallback/retry/compaction events. Layered
  // last, at return time, so every exit path reports the same shape without
  // each spelling the metrics out by hand.
  type ReliabilityKeys =
    | "retryFallbacks"
    | "retryFallbackSucceeded"
    | "modelRecoveries"
    | "autoRetryCount"
    | "autoCompactionCount";
  // Folded once, on whichever exit path actually runs: every return below
  // passes through `withSessionFacts`, so the advisor's separately-recorded
  // spend (#542) is added to exactly one result and reported to the daemon
  // exactly once.
  let advisorSpendFolded = false;
  const withSessionFacts = (
    result: Omit<WorkerResult, ReliabilityKeys>,
  ): Omit<WorkerResult, ReliabilityKeys> => {
    const { sessionFile, modelFallbackMessage, graphTools } = session;
    if (!advisorSpendFolded) {
      advisorSpendFolded = true;
      // Advisor turns live in their own `__advisor*.jsonl` and never surface
      // as primary `message_end`s, so without this read the run's spend — and
      // therefore `caps.dailySpendUsd` — silently under-counts real provider
      // consumption (the #46 shape). Read at the very end: the advisor's
      // review of the final turn may still be landing.
      const extra = advisorSpendUsd(sessionFile);
      if (extra > 0) {
        spendUsd += extra;
        o.onSpend?.(spendUsd);
      }
    }
    return {
      ...result,
      // Override whatever the exit path spelled: its literal captured the
      // pre-fold `spendUsd`, and the advisor fold happened after.
      spendUsd,
      ...(sessionFile === undefined ? {} : { sessionFile }),
      ...(modelFallbackMessage === undefined ? {} : { modelFallbackMessage }),
      ...(graphTools === undefined ? {} : { graphTools }),
    };
  };

  // The count fields always travel (0 for a clean run, so an absent field can
  // never be misread); the resolved model/provider only when some assistant
  // message actually carried them.
  const withMetrics = (result: Omit<WorkerResult, ReliabilityKeys>): WorkerResult => ({
    ...result,
    ...(resolvedModel === undefined ? {} : { model: resolvedModel }),
    ...(resolvedProvider === undefined ? {} : { provider: resolvedProvider }),
    retryFallbacks,
    retryFallbackSucceeded,
    modelRecoveries,
    autoRetryCount,
    autoCompactionCount,
  });

  let turns = 0;
  let spendUsd = 0;
  // Cumulative output/reasoning tokens (#518), beside the spend they are read
  // from: the share of one against the other is what makes an expensive-in-time,
  // cheap-in-dollars run legible instead of looking barely started.
  let outputTokens = 0;
  let reasoningTokens = 0;
  let provider429Count = 0;
  let report = "";
  // The newest structured settlement the worker yielded (#540). A terminal
  // `yield` ends the run, so anything parsed here is the run's own claim, and
  // any later message is stale-polling chatter, never a newer verdict.
  let structured: WorkerSettlement | undefined;
  // Which model/provider actually wrote the newest assistant message. Last
  // assistant message wins: that is the durable answer even for a run that
  // never failed over (#535 slice 1, read off the message field which is where
  // the resolved model actually lives).
  let resolvedModel: string | undefined;
  let resolvedProvider: string | undefined;
  // Harness reliability surface (#539): within-run provider failover and the
  // retry/compaction activity that surrounds it, all of it events this worker
  // does not yet subscribe to but that the run row and settlement report want.
  let retryFallbacks: { from: string; to: string }[] = [];
  let retryFallbackSucceeded = 0;
  let modelRecoveries = 0;
  let autoRetryCount = 0;
  let autoCompactionCount = 0;
  // The newest COMPLETE `pushed-green` verdict this session emitted. Tracked
  // apart from `report` because `report` is deliberately the newest non-empty
  // text — a run cut off mid-sentence must still report what it said last —
  // and a worker that finishes, then wakes on its own stale polling timers,
  // overwrites its own verdict with "nothing to resume" chatter. veltro#406
  // burned 180/180 turns that way and charged an attempt against a PR that
  // merged (#217).
  let claim: { prUrl: string; headSha: string } | undefined;
  let killedBy: KilledBy | undefined;
  let stoppedReason: string | undefined;
  // Canceler for the armed wall clock; invoked on every exit path below.
  let cancelWallClock: (() => void) | undefined;
  let wallClockRemainingMs = workerWallClockMs;
  let wallClockArmedAt = now();
  let pausePhase: WorkerPausePhase = "running";
  let resumeWaiter: PromiseWithResolvers<string> | undefined;
  // A terminal agent_end or cap has settled the run.
  let done = false;
  // Resolved by the first terminal `agent_end`, and by every cap kill. Only
  // ever awaited when the harness has already said it is not finished.
  const { promise: settled, resolve: settle } = Promise.withResolvers<void>();
  // Resolves whenever a pause request must interrupt the existing
  // non-terminal agent_end wait. Re-armed after its resume prompt is consumed.
  let pauseRequested = Promise.withResolvers<void>();
  // Set by a non-terminal `agent_end`: the harness will resume this session.
  let resuming = false;

  const clearWallClock = () => {
    if (cancelWallClock === undefined) return;
    cancelWallClock();
    cancelWallClock = undefined;
  };

  const armWallClock = () => {
    wallClockArmedAt = now();
    cancelWallClock = schedule(() => kill("wallclock"), wallClockRemainingMs);
  };

  const kill = (by: KilledBy) => {
    if (killedBy !== undefined || stoppedReason !== undefined) return;
    killedBy = by;
    o.onKilled?.(by);
    clearWallClock();
    session.abort();
    // An aborted session may never reach a terminal `agent_end`. The cap is the
    // outcome now, so nothing may still be waiting for one.
    done = true;
    settle();
  };

  const takeResumePrompt = async (): Promise<string | undefined> => {
    const waiter = resumeWaiter;
    if (waiter === undefined) return undefined;
    const prompt = await Promise.race([
      waiter.promise,
      settled.then(() => undefined),
    ]);
    if (resumeWaiter === waiter) {
      resumeWaiter = undefined;
      pauseRequested = Promise.withResolvers<void>();
    }
    return killedBy !== undefined || stoppedReason !== undefined || done ? undefined : prompt;
  };

  o.onPauseControl?.({
    phase: () => pausePhase,
    pause: async () => {
      if (killedBy !== undefined || done) {
        throw new Error("the run is settling; nothing left to pause");
      }
      if (pausePhase !== "running") throw new Error(`the worker is already ${pausePhase}`);
      if (resumeWaiter !== undefined) {
        throw new Error("the worker resume is still starting");
      }
      pausePhase = "pausing";
      resumeWaiter = Promise.withResolvers<string>();
      pauseRequested.resolve();
      // Bank the remaining wall clock before the abort: a slow drain must not
      // be cap-killed mid-park (#238 acceptance 3).
      wallClockRemainingMs = Math.max(
        1_000,
        wallClockRemainingMs - (now() - wallClockArmedAt),
      );
      clearWallClock();
      try {
        await session.park();
      } catch (err) {
        pausePhase = "running";
        if (killedBy !== undefined || done) {
          resumeWaiter = undefined;
          throw err;
        }
        // The abort may already have unwound prompt() even though park itself
        // failed. Wake that path with the same defensive continuation rather
        // than leaving the worker hung on an orphaned waiter.
        armWallClock();
        resumeWaiter?.resolve(RESUME_PROMPT);
        throw err;
      }
      if (killedBy !== undefined || done) {
        pausePhase = "running";
        resumeWaiter = undefined;
        throw new Error("the run settled while pausing");
      }
      pausePhase = "paused";
    },
    resume: () => {
      if (pausePhase !== "paused") {
        throw new Error(
          pausePhase === "pausing"
            ? "still pausing — wait until it reports paused"
            : "the worker is not paused",
        );
      }
      pausePhase = "running";
      armWallClock();
      resumeWaiter?.resolve(RESUME_PROMPT);
    },
    stop: (reason) => {
      if (killedBy !== undefined || done) {
        throw new Error("the run is settling; nothing left to stop");
      }
      stoppedReason = reason;
      clearWallClock();
      done = true;
      settle();
      session.abort();
    },
  });

  session.on("turn_start", () => {
    // The documented watchdog signal, and the honest one: `turn_start` fires
    // exactly once per turn, whereas one turn can emit several assistant
    // `message_end`s and would burn the cap on a run that is behaving.
    turns += 1;
    o.onTurn?.(turns);
    if (turns > maxTurns()) kill("turns");
  });

  session.on("message_end", (event) => {
    const message = field(event, "message");
    if (field(message, "role") !== "assistant") return;
    const content = field(message, "content");
    // Keep the newest non-empty assistant text: whatever the worker said last
    // is its report, whether it finished cleanly or was cut off.
    const text = reportText(content);
    if (text !== "") {
      report = text;
    }
    // The structured settlement (#540): the `yield` tool call rides the same
    // message's content as a `toolCall` block. Newest yield wins; the
    // derivation itself prefers it over prose, so a yield and a misleadingly
    // regular-looking text block on the same message agree in its favour.
    const yielded = structuredSettlement(content);
    if (yielded !== undefined) structured = yielded;
    const stated = deriveResult(text, o.repoSlug, yielded);
    if (stated.state === "pushed-green" && stated.prUrl !== undefined && stated.headSha !== undefined) {
      claim = { prUrl: stated.prUrl, headSha: stated.headSha };
    }

    // Real cost lives on assistant messages as `usage.cost.total` (live hermes
    // transcripts, 2026-08-07). The earlier agent_end.telemetry path never
    // fired, so every run recorded $0 and the daily cap was theater (#46).
    // Tokens ride the same usage block as the cost, so this is one read of a
    // field already in the transcript (#518) — no new call, no polling.
    const tokens = outputTokensFromMessage(message);
    if (tokens !== undefined) {
      outputTokens += tokens.output;
      reasoningTokens += tokens.reasoning;
      o.onTokens?.({ output: outputTokens, reasoning: reasoningTokens });
    }

    const cost = costUsdFromMessage(message);
    if (cost !== undefined) {
      spendUsd += cost;
      o.onSpend?.(spendUsd);
      // The live half of the spend bound (#851). The daily cap alone stopped
      // the fleet on the NEXT dispatch pass, so one run could cross it by its
      // whole cost before anything objected — on 2026-08-21 a single run spent
      // an estimated $25.16 against a $25.00 daily cap unopposed. A run that
      // reaches the allowance reserved for it is killed here, through the same
      // latch the turn and wall-clock ceilings use, so it settles through the
      // ordinary salvage path with its work preserved.
      if (o.maxSpendUsd !== undefined && spendUsd >= o.maxSpendUsd) kill("spend");
    }

    // Count the provider rate limits the harness retried in-session (#573). A
    // run that drowns in 429s records `stopReason:"error", errorStatus:429`
    // dozens of times and never surfaces one as `lastError` — omp swallowed
    // every retry — so `unknown` and the provider failover chain (#286) never
    // see them. Counted here, alongside spend, so the classifier can tell
    // "the provider was throttling the whole run" from an ordinary failure.
    if (provider429FromMessage(message)) provider429Count += 1;

    // The resolved model and provider live on the message, not on any event
    // (#539). Last assistant message wins, which is the run's durable answer
    // even when it never failed over.
    const model = field(message, "model");
    if (typeof model === "string" && model !== "") resolvedModel = model;
    const provider = field(message, "provider");
    if (typeof provider === "string" && provider !== "") resolvedProvider = provider;
    // A within-run model swap the harness persisted into the transcript rather
    // than only emitting as a transient event. Recovery kind "model" is the
    // durable spelling of the same thing `retry_fallback_applied` says.
    if (field(field(message, "retryRecovery"), "recovery") === "model") modelRecoveries += 1;
  });

  // The harness reliability events the run row and settlement report now want
  // (#539): within-run model fallback, and the retry/compaction activity that
  // surrounds a throttled provider. Each is a session event (AgentSessionEvent)
  // carrying the fields this worker reads — a fallback's from→to pair, the
  // confirmed recoveries, and the auto-retry/compaction attempt counts.
  session.on("retry_fallback_applied", (event) => {
    const from = field(event, "from");
    const to = field(event, "to");
    if (typeof from === "string" && typeof to === "string") {
      retryFallbacks = [{ from, to }, ...retryFallbacks];
    }
  });
  session.on("retry_fallback_succeeded", () => {
    retryFallbackSucceeded += 1;
  });
  session.on("auto_retry_start", () => {
    autoRetryCount += 1;
  });
  session.on("auto_compaction_start", () => {
    autoCompactionCount += 1;
  });

  session.on("agent_end", (event) => {
    // Fallback for harnesses that only attach cost on the terminal event.
    const estimated = field(field(field(event, "telemetry"), "cost"), "estimatedUsd");
    if (
      spendUsd === 0 &&
      typeof estimated === "number" &&
      Number.isFinite(estimated) &&
      estimated > 0
    ) {
      // Prefer message totals when both exist — do not double-count a run that
      // already accumulated per-message costs.
      spendUsd = estimated;
      o.onSpend?.(spendUsd);
    }

    // Anything that is not literally `false` — including garbage or nothing at
    // all — is a finished run.
    const isTerminal = field(event, "isTerminal");
    if (shouldComplete(typeof isTerminal === "boolean" ? { isTerminal } : {})) {
      if (pausePhase !== "running") return;
      done = true;
      settle();
      return;
    }
    resuming = true;
  });

  // A stuck session spends no turns, so turns alone cannot detect it. The
  // callback does not drop the handle itself: every exit runs `clearWallClock()`
  // exactly once instead, and clearing an already-fired handle is a documented
  // no-op — cheaper than assuming a fired timer holds nothing.
  armWallClock();

  try {
    let next: string | undefined = o.brief;
    while (next !== undefined) {
      try {
        await session.prompt(next);
      } catch (cause) {
        // Our own abort surfaces here on some paths: a cap kill (existing
        // behavior) or an operator park (new). Anything else is a real failure.
        if (killedBy === undefined && stoppedReason === undefined && resumeWaiter === undefined) throw cause;
      }
      next = undefined;
      if (killedBy !== undefined) break;
      if (resumeWaiter === undefined && resuming && !done) {
        // A non-terminal agent_end ordinarily waits for the harness to finish
        // later. A pause request must wake that wait so resume can prompt the
        // same session instead of hanging behind the old settlement promise.
        await Promise.race([settled, pauseRequested.promise]);
      }
      if (killedBy !== undefined) break;
      if (resumeWaiter !== undefined) {
        next = await takeResumePrompt();
        if (next === undefined) break;
        continue;
      }
    }
  } catch (cause) {
    // Our own abort surfaces here on some paths; that is a kill, not a crash.
    if (killedBy === undefined && stoppedReason === undefined) {
      const detail = cause instanceof Error ? cause.message : String(cause);
      return withMetrics(withSessionFacts({
        state: "failed",
        turns,
        spendUsd,
        provider429Count,
        report: report === "" ? detail : report,
      }));
    }
  } finally {
    done = true;
    // Runs on every exit, including the early return above: a live timer keeps
    // the dispatcher process alive long after the run it was guarding.
    clearWallClock();
    try {
      await disposeSession(session);
    } catch {
      // Teardown noise must not overwrite the run's actual outcome.
    }
  }

  if (stoppedReason !== undefined) {
    return withMetrics(withSessionFacts({
      state: "stopped",
      turns,
      spendUsd,
      provider429Count,
      report,
      stoppedReason,
      ...(claim === undefined ? {} : { prUrl: claim.prUrl, headSha: claim.headSha }),
    }));
  }

  if (killedBy !== undefined) {
    // The PR and head are facts the session already established, so they
    // survive the kill. Without them `shouldContinueAfterTurnsCap` sees no
    // artifacts and charges an implementation attempt for a cap kill that had
    // real work to continue from.
    return withMetrics(withSessionFacts({
      state: "killed",
      turns,
      spendUsd,
      provider429Count,
      report,
      killedBy,
      ...(claim === undefined ? {} : { prUrl: claim.prUrl, headSha: claim.headSha }),
    }));
  }

  // A structured yield is the run's settlement: it wins over the last text —
  // the worker may have written nothing after the yield call, and the stored
  // report is the canonical rendering of the parsed object, not prose the
  // regex has to recover (#540).
  if (structured !== undefined) {
    return withMetrics(withSessionFacts({
      ...deriveResult(report, o.repoSlug, structured),
      turns,
      spendUsd,
      provider429Count,
      report: renderSettlement(structured),
      // The only exit that can carry them: a killed or unparseable run yielded
      // no settlement, and inventing an empty list there would tell the
      // knowledge overlay that the fleet asked and learned nothing.
      ...(structured.discoveries === undefined || structured.discoveries.length === 0
        ? {}
        : { discoveries: structured.discoveries }),
    }));
  }

  // An explicit later verdict always wins: a worker that pushed green and then
  // stopped to ask a question means the question. The earlier claim is only
  // restored when the last thing said was not a verdict at all.
  if (claim !== undefined && !hasVerdictLine(report)) {
    return withMetrics(withSessionFacts({ state: "pushed-green", ...claim, turns, spendUsd, provider429Count, report }));
  }
  return withMetrics(withSessionFacts({
    ...deriveResult(report, o.repoSlug),
    turns,
    spendUsd,
    provider429Count,
    report,
  }));
}

/**
 * USD cost from one assistant message's `usage.cost` block.
 *
 * Prefer `total` when present; otherwise sum the component fields the live
 * harness emits (input/output/cacheRead/cacheWrite). Exported so a unit test
 * can pin the shape without standing up a session.
 */
export function costUsdFromMessage(message: unknown): number | undefined {
  const usage = field(message, "usage");
  const cost = field(usage, "cost");
  if (cost === null || typeof cost !== "object") return undefined;
  const total = field(cost, "total");
  if (typeof total === "number" && Number.isFinite(total) && total >= 0) return total;
  let sum = 0;
  let any = false;
  for (const key of ["input", "output", "cacheRead", "cacheWrite"] as const) {
    const v = field(cost, key);
    if (typeof v === "number" && Number.isFinite(v) && v >= 0) {
      sum += v;
      any = true;
    }
  }
  return any ? sum : undefined;
}

/**
 * Output and reasoning token counts from one assistant message (#518).
 *
 * Both come off the same `usage` block the cost is already read from, so this
 * adds no provider call and no polling — it reads a field the transcript
 * already carries. Absent or unparseable counts return `undefined` rather than
 * zero, because "this provider reported no tokens" and "this turn produced no
 * output" invite opposite readings, and a zero would make the share arithmetic
 * quietly wrong.
 *
 * `reasoning` is counted as part of `output` by every provider that reports it,
 * which is what makes the share meaningful: 96% reasoning on run 42f73f52 was
 * the whole answer to "why has this been working 42 minutes for $0.05", and it
 * was recorded and unshown.
 */
export function outputTokensFromMessage(
  message: unknown,
): { output: number; reasoning: number } | undefined {
  const tokens = field(field(message, "usage"), "tokens");
  if (tokens === null || typeof tokens !== "object") return undefined;
  const count = (key: string): number | undefined => {
    const v = field(tokens, key);
    return typeof v === "number" && Number.isFinite(v) && v >= 0 ? v : undefined;
  };
  const output = count("output");
  if (output === undefined) return undefined;
  // A provider that reports no reasoning is reporting zero reasoning, which is
  // a true share of 0% — distinct from an absent output count above.
  return { output, reasoning: count("reasoning") ?? 0 };
}

/**
 * Is this assistant message a provider HTTP 429 the harness recorded mid-run?
 *
 * Live transcripts mark a rate-limited turn with `stopReason:"error"`,
 * `errorStatus:429` and a message like `429 Provider returned error`, and the
 * harness's in-session auto-retry usually swallows it — the run carries on and
 * the 429 never surfaces as `lastError` (#573). Exported so a unit test can pin
 * the signature without standing up a session; the count itself distinguishes a
 * run that drowned in them from a run that hit one and recovered.
 */
export function provider429FromMessage(message: unknown): boolean {
  if (field(message, "stopReason") !== "error") return false;
  return field(message, "errorStatus") === 429;
}

/**
 * Read one property off an unvalidated harness event. The event union lives in
 * the peer dependency, so the worker narrows the handful of fields it reads
 * instead of importing types it cannot see at build time.
 */
function field(source: unknown, key: string): unknown {
  if (source === null || typeof source !== "object") return undefined;
  return Reflect.get(source, key);
}

/** Advisor transcripts are recorded under this reserved stem beside the session's. */
const ADVISOR_TRANSCRIPT_PREFIX = "__advisor";

/**
 * Whether a staged omp-settings overlay (the file `materializeOmpSettings`
 * writes under the run's session directory) turns the omp advisor on.
 *
 * The overlay is conductor's own YAML — written by `yaml.stringify` over a map
 * this package controls — so this is deliberately not a general YAML parser:
 * it scans for an `advisor:` mapping and asks whether one of its direct
 * `enabled:` keys reads the literal `true`. Handles both the block form the
 * overlay is written in (`advisor:\n  enabled: true`) and an inline flow map
 * (`advisor: { enabled: true }`); everything else — `enabled: false`, no
 * `advisor` key, an absent overlay — answers false. The staged settings are
 * what the harness itself resolves, so this is the same truth the session
 * acts on, not a parallel decode. Exported so the staging decision in
 * {@link runWorker} is pinned by a unit test.
 */
export function overlayEnablesAdvisor(overlayText: string): boolean {
  const lines = overlayText.split(/\r?\n/);
  for (let i = 0; i < lines.length; i++) {
    const match = /^(\s*)advisor\s*:\s*(.*)$/.exec(lines[i] ?? "");
    if (match === null) continue;
    if (/\benabled\s*:\s*true\b/.test(match[2] ?? "")) return true;
    const indent = match[1] ?? "";
    for (let j = i + 1; j < lines.length; j++) {
      const line = lines[j];
      if (line === undefined || line.trim() === "") continue;
      // A line at or shallower than `advisor`'s own indent ends its block.
      if (!line.startsWith(`${indent} `) && !line.startsWith(`${indent}\t`)) break;
      if (/^\s*enabled\s*:\s*true\s*$/.test(line)) return true;
    }
  }
  return false;
}

/** A heading that begins a brief's acceptance section, whatever the brief's casing. */
const ACCEPTANCE_HEADING = /^#{1,4}\s+acceptance\s+criteri(?:a|on)\s*$/im;

/**
 * Render the per-issue reviewer brief (#542): the acceptance section of one
 * worker brief, wrapped so the omp advisor checks the primary against *this
 * run's own* criteria rather than generic taste. Returns `undefined` when the
 * brief carries no acceptance section — nothing useful to review against.
 *
 * The section runs from the first heading mentioning acceptance criteria to
 * the next heading of any level (or the end of the brief). Exported so a unit
 * test can pin the rendering without standing up a session.
 */
export function renderIssueWatchdog(brief: string): string | undefined {
  const lines = brief.split(/\r?\n/);
  let start = -1;
  for (let i = 0; i < lines.length; i++) {
    if (ACCEPTANCE_HEADING.test(lines[i] ?? "")) {
      start = i + 1;
      break;
    }
  }
  if (start === -1) return undefined;
  const body: string[] = [];
  for (let i = start; i < lines.length; i++) {
    const line = lines[i];
    if (line === undefined || /^#{1,4}\s/.test(line)) break;
    body.push(line);
  }
  const trimmed = body.join("\n").replace(/\n{3,}/g, "\n\n").trim();
  if (trimmed === "") return undefined;
  return (
    "# Issue watchdog (rendered from this run's brief)\n\n" +
    "This run's own acceptance criteria. Review the worker against these — not generic taste: " +
    "a criterion is met only when the transcript or workspace shows it actually verified, and a " +
    "worker claiming completion without evidence is a concern at least.\n\n" +
    "## Acceptance criteria\n\n" +
    `${trimmed}\n`
  );
}

/**
 * Stage the per-issue `WATCHDOG.md` into the worktree when the staged omp
 * settings turn the advisor on (#542).
 *
 * The omp advisor's watchdog discovery is cwd-rooted, so per-issue guidance
 * cannot ride the out-of-tree settings overlay: it must live inside the
 * worktree, at `<cwd>/.omp/WATCHDOG.md`, which the managed exclude block
 * ignores by that exact path (never the `.omp/` directory) so it can never
 * reach the diff a worker ships or a salvage commit. Best-effort on purpose:
 * a reviewer brief is advisory, and a worktree too broken to take it will fail
 * the run on its own terms. Exported so the runWorker staging decision is
 * testable without a session.
 */
export function stageIssueWatchdog(cwd: string, brief: string, overlayFile: string): boolean {
  let overlayText: string;
  try {
    overlayText = readFileSync(overlayFile, "utf8");
  } catch {
    return false;
  }
  if (!overlayEnablesAdvisor(overlayText)) return false;
  const watchdog = renderIssueWatchdog(brief);
  if (watchdog === undefined) return false;
  try {
    const dir = join(cwd, ".omp");
    mkdirSync(dir, { recursive: true });
    writeFileSync(join(dir, "WATCHDOG.md"), watchdog);
    return true;
  } catch {
    return false;
  }
}

/**
 * Total USD cost of one run's advisor turns (#542).
 *
 * Advisor turns are recorded to a separate `__advisor*.jsonl` beside the
 * session transcript — the primary session never sees them as `message_end`s —
 * so without this a run with an advisor under-counts `spendUsd` and the daily
 * cap is theater, the #46 failure mode. Reads every advisor transcript in the
 * session's advisor directory and sums the same `usage.cost` blocks
 * {@link costUsdFromMessage} reads off primary messages: one spelling, so the
 * two cannot drift. Unreadable or absent transcripts bill nothing — the run
 * with no advisor stays a zero, and a corrupt advisor log must not crash the
 * settlement.
 */
export function advisorSpendUsd(sessionFile: string | undefined): number {
  if (sessionFile === undefined || !sessionFile.endsWith(".jsonl")) return 0;
  // The harness records advisor transcripts in the directory named after the
  // primary transcript stem (`<dir>/<stem>/__advisor*.jsonl`); slicing the
  // suffix is exactly what the harness's own cost loader does.
  const dir = sessionFile.slice(0, -".jsonl".length);
  let names: string[];
  try {
    names = readdirSync(dir);
  } catch {
    return 0;
  }
  let total = 0;
  for (const name of names) {
    if (!name.startsWith(ADVISOR_TRANSCRIPT_PREFIX) || !name.endsWith(".jsonl")) continue;
    let text: string;
    try {
      text = readFileSync(join(dir, name), "utf8");
    } catch {
      continue;
    }
    for (const line of text.split("\n")) {
      if (line.trim() === "") continue;
      let entry: unknown;
      try {
        entry = JSON.parse(line);
      } catch {
        continue;
      }
      if (field(entry, "type") !== "message") continue;
      const message = field(entry, "message");
      if (field(message, "role") !== "assistant") continue;
      const cost = costUsdFromMessage(message);
      if (cost !== undefined) total += cost;
    }
  }
  return total;
}

/** Flatten an assistant message's content blocks to their plain text. */
/**
 * The newest assistant text, flattened out of whatever block shape the harness
 * sent. Exported for ./setup-probe.ts, which collects a probe's answer the same
 * way a worker's report is collected — one spelling, so the two cannot drift.
 */
export function reportText(content: unknown): string {
  if (typeof content === "string") return content.trim();
  const blocks: readonly unknown[] = Array.isArray(content) ? content : [];
  const parts: string[] = [];
  for (const block of blocks) {
    if (field(block, "type") !== "text") continue;
    const text = field(block, "text");
    if (typeof text === "string") parts.push(text);
  }
  return parts.join("\n").trim();
}