/**
 * The privileged half of the mutation verbs (#126): every check, in the daemon.
 *
 * The child-side tool handler forwards arguments and renders an answer. It
 * evaluates no policy and writes no ledger entry, because it runs inside the
 * session process — anything it decided would be decided by code the model can
 * rewrite, and a session runs as the daemon's own user. Everything below runs on
 * the other side of the socket, which is what makes the checks real.
 *
 * Four rules shape the order of the checks, and each is here because prose
 * alone did not hold it on `main`:
 *
 *  - **Identity is the channel's, and the channel is bound to one process.**
 *    `project`, `run`, `issue` and role come from {@link VerbChannel} — which
 *    socket the call arrived on. No payload can name them, so a *request* to act
 *    as another run is unexpressible.
 *
 *    That alone was never authentication: every session runs as the daemon's own
 *    uid, so any of them could list the socket directory and connect to another
 *    channel — including the orchestrator's, which holds merge under
 *    `authority.merge: "orchestrator"` — and the ledger would have recorded the
 *    channel's role, attributing it to the orchestrator. A file mode cannot fix
 *    that, because the owner bits belong to the uid the session already has.
 *
 *    So each channel is **bound to the pid the daemon spawned for it**, and a
 *    connection from any other process is refused unanswered and logged. Until it
 *    is bound it refuses everything, because the socket necessarily exists before
 *    the child does. The residual is narrow and real: one uid can still `ptrace`
 *    or signal its siblings, so a determined session can interfere with the
 *    process that *is* bound — a much higher bar than connecting to a socket.
 *  - **Compared against the configured holder, never against one forbidden
 *    value.** `authority` has exactly two holders, so `!== "human"` would let a
 *    worker release. Merge and release both ask "is the caller the holder".
 *  - **Stop always wins.** Pause state, the config and the run's own row are
 *    re-read *inside* the verb, after the model decided to call it. A green
 *    check a second ago is not a green check now.
 *  - **Fail closed.** An unreadable config, an unknown run, a head nobody can
 *    resolve: refuse with a named reason. Never allow on doubt.
 */

import { randomUUID } from "node:crypto";
import { createServer, type Server, type Socket } from "node:net";
import { join } from "node:path";

import { resolvePolicy, resolveReleaseGrants, resolveReview, resolveSharedInstallAuthority } from "../config.ts";
import { wakeDispatch } from "../wake.ts";
// The mediated release is the drain's closing gesture (#791): a successful
// release ends this project's bounded release window, so the privileged half
// clears the record through the same operator surface the CLI uses.
import { cancelDrain } from "../fleet.ts";
import { effectiveLane, laneEcho, writeLaneSectionHeading } from "../admission.ts";
import { chainEntriesFromDiff, chainViolations } from "../chain-check.ts";
// The runner-infrastructure waiver a red check's own log can earn (#1043 lane).
// The worker keeps its two-red corrective budget in session, so this reply is
// the only place that waiver can be stated — and it is stated from the same
// classifier the daemon's own infra accounting uses, never a second copy.
import { runnerInfraFailure } from "../failure-class.ts";
import {
  repoSlugFor,
  type LaneFile,
  type LaneSource,
  type RunLaneProbe,
  type readBaseChain as readBaseChainType,
} from "../gitops.ts";
import { releaseRefusal } from "../release-policy.ts";
import { PR_LOOKUP_WINDOW_MS, REVISABLE_RUN_STATES, prReviewReadiness } from "../decisions.ts";
import { LIVE_STATES } from "../store.ts";
import { readyGate } from "../ready-gate.ts";
import { parseToSpecEvidence } from "../to-spec.ts";
import { DENIED_RELEASE_GRANTS } from "../types.ts";
import type {
  FileLane,
  IssueComment,
  IssueState,
  OpenCloser,
  PrState,
  PrVerification,
  ProjectConfig,
  ReleaseRequirement,
  ReleaseShape,
  RepoTarget,
  ResolvedGrants,
  ReviewReason,
  RunRecord,
  RunState,
  Store,
  Tracker,
  VerbLedgerEntry,
  VerbName,
  VerbRefusal,
} from "../types.ts";
import { failedRunIds, prUrlParts, GhPrMissingError } from "../tracker/github.ts";
import { parseVerbRequest, roleRefusal, VERB_SPECS, type VerbReply } from "./protocol.ts";
import {
  peerVerdict,
  secureBoundSocket,
  socketFd,
  unlinkStaleSocket,
  validateSocketPath,
  type PeerReader,
  type PeerVerdict,
  type SocketOwnership,
} from "./socket.ts";

/**
 * How long a merge lock may be held before another daemon may break it.
 *
 * Generous, because the window it covers is a real `gh pr merge` against a
 * repository with branch protection. Breaking it is safe rather than merely
 * tolerable: the second merge re-reads the live head before acting, so a stolen
 * lock cannot turn into a merge of something nobody looked at.
 */
export const MERGE_LOCK_STALE_MS = 10 * 60_000;

/** A hostile local client must not be able to make the daemon buffer forever. */
const MAX_REQUEST_BYTES = 64 * 1024;

/** A connection that opens and says nothing is a connection that is holding a slot. */
const REQUEST_TIMEOUT_MS = 30_000;

/**
 * Who is on the other end of one socket, decided by the daemon at dispatch.
 *
 * A discriminated union rather than a bag of optionals, because the two kinds
 * differ in exactly the way the verbs care about: a run channel names one run
 * and can name no other, and the orchestrator's names a project and no run at
 * all. Nothing in a payload can move a call from one to the other.
 */
export type VerbChannel =
  | {
      kind: "run";
      /** Absolute socket path. Also this channel's identity in the log. */
      path: string;
      project: string;
      role: "worker";
      runId: string;
      issue: number;
      /** The routed checkout, for the privileged push/PR half. */
      repo: RepoTarget;
      /** The run's own repository on disk. */
      runRepoPath: string;
      branch: string;
    }
  | {
      kind: "orchestrator";
      path: string;
      project: string;
      role: "orchestrator";
    };

export type ActionOutcome =
  | {
      ok: true;
      sha?: string;
      detail?: string;
      /** A reviewable version bump. New PRs return immediately; existing ones are re-checked and merged. */
      review?: { prUrl: string; created: boolean };
    }
  | { ok: false; stderr: string };

/** What one release asks the privileged half to do, after policy said yes. */
export interface ReleaseExecution {
  shape: ReleaseShape;
  repo: RepoTarget;
  tag?: string;
  /**
   * The exact commit a tag shape must point at (#695). Absent means "whatever
   * the released repo's default branch is at cut time", which is the behaviour
   * that stranded a tag every time the release window needed a merge: cutting
   * required a quiet window, reaching that window required merging the open PR,
   * and merging moved the branch past the tag that had just been cut. A pinned
   * commit removes the dependence on *when* the tag is cut; the privileged half
   * still refuses one that is not reachable from that branch.
   */
  sha?: string;
  artefact?: string;
  environment?: string;
}

/** What one install request asks the privileged half to do, after policy said yes. */
export interface InstallExecution {
  /** The published semver to pin all three surfaces to, exactly as npm has it. */
  version: string;
  /** Shared host authority and scope, carried into the detached report. */
  holder: ResolvedGrants["install"];
  projects: readonly string[];
}

/**
 * The privileged half, as an interface so every check above it is testable
 * without a repository, a credential or a network.
 *
 * `releasableShapes` is declared by the implementation rather than assumed by
 * the server: this daemon holds a GitHub credential, not an npm token and not a
 * deploy key, and a release shape it cannot actually cut must be refused by
 * name rather than attempted and reported as a command failure.
 */
export interface VerbActions {
  releasableShapes: readonly ReleaseShape[];
  push(run: { repo: RepoTarget; runRepoPath: string; branch: string }): Promise<ActionOutcome>;
  createPr(
    run: { repo: RepoTarget; runRepoPath: string; branch: string },
    opts: { title: string; body: string; base: string },
  ): Promise<{ ok: true; url: string } | { ok: false; stderr: string }>;
  updatePrBranch(prUrl: string): Promise<ActionOutcome>;
  updatePr(
    prUrl: string,
    fields: { title?: string; body?: string },
  ): Promise<{ ok: true } | { ok: false; stderr: string }>;
  mergePr(prUrl: string, headSha: string): Promise<ActionOutcome>;
  /**
   * Close one pull request, having first posted `comment` on it (#876).
   *
   * Both halves in one action because the ordering is the contract: a close that
   * lands while its explanation fails leaves a PR nobody can account for, so the
   * comment goes first and a failed comment means no close.
   */
  closePr(prUrl: string, comment: string): Promise<ActionOutcome>;
  release(execution: ReleaseExecution): Promise<ActionOutcome>;
  /**
   * Request the fleet upgrade itself to `version`: refuse when npm does not
   * expose it, then start the detached transient unit that performs it and
   * survives the restarts. The implementation owns the npm read and the
   * systemd-run spawn (#486) — the server only gates and records.
   */
  install(execution: InstallExecution): Promise<ActionOutcome>;
}

export interface VerbDeps {
  /**
   * The project config, **re-read per call**. A thunk rather than a value
   * because "unreadable config refuses the call" is only true if the call is
   * the thing that reads it: a config resolved once at boot cannot become
   * unreadable, and cannot pick up an operator's edit either.
   */
  project: () => ProjectConfig;
  /**
   * Every project served by the shared daemon, re-read from the same config.
   * Host-global mutations use this set rather than the calling socket's project.
   */
  projects: () => readonly ProjectConfig[];
  store: Store;
  tracker: Tracker;
  actions: VerbActions;
  /**
   * Why the fleet is stopped right now, or `undefined`. Called inside the verb,
   * after the model decided to call it. `hold` and `halt` both set the pause
   * sentinel alongside disarming ticks, so one read covers both.
   */
  fleetStop: () => string | undefined;
  /**
   * The epoch-ms timestamp the current pause began, or `undefined` when not
   * paused (or when the sentinel does not parse). Lets the fleet-stop gate tell
   * a mutating call from a run admitted *before* the pause — that run may
   * finish — apart from one admitted after, which must be refused. Parallels
   * {@link VerbDeps.fleetStop} by living behind the same per-call read.
   */
  pausedAt: () => number | undefined;
  log: (message: string) => void;
  now: () => number;
  /**
   * Read-only access to the ordered-migration-chain guard. A read seam, kept
   * off `VerbActions` (the mutation surface) on purpose: `prMergeVerb` consults
   * whether the merge would fork the chain, it never changes it (#227).
   */
  chain: { readBaseChain: typeof readBaseChainType };
  /**
   * Read-only access to one run's actual file lane (#925), kept off
   * {@link VerbDeps.actions} for the same reason {@link VerbDeps.chain} is:
   * `conductor_pr_recover` consults what a branch carries, it never changes it.
   *
   * Required rather than optional on purpose. This seam decides a fail-CLOSED
   * refusal, so "the field was not wired" must be a compile error rather than a
   * silently skipped check — an unwired guard that reports success is worse
   * than no guard, because nothing anywhere says the lane went unproven.
   */
  lane: { probeRunLane: RunLaneProbe };
}

/**
 * The labels a session may change on this project's issues.
 *
 * Two lists, not one filter, because the interesting refusal is the second:
 * `agent:in-progress` is a *real* label of this project that a session still
 * may not touch, since the loop's orphan detection is only trustworthy while
 * every lifecycle label on the tracker was written by the conductor (#26).
 * Telling a caller "not in the vocabulary" about a label it can plainly see on
 * the issue teaches it to try harder; telling it "that transition is the
 * daemon's" tells it to stop. The operator's park label is a third kind and a
 * harder line than either: it is not "not in the vocabulary" (setup creates it,
 * so that refusal would evaporate the day the vocabulary grows) and not the
 * daemon's either — it is the operator's, and naming that owner is the point
 * (#507). An orchestrator that could set or clear a park could hide work it
 * does not want to groom.
 */
export interface LabelVocabulary {
  grantable: string[];
  lifecycle: string[];
  /** Labels only the operator may change: the park gesture. */
  operator: string[];
}

export function labelVocabulary(p: ProjectConfig): LabelVocabulary {
  return {
    grantable: [
      p.queueLabel,
      ...Object.keys(p.routing.repos).map((name) => `${p.routing.labelPrefix}${name}`),
    ],
    lifecycle: [p.stateLabels.inProgress, p.stateLabels.blocked, p.stateLabels.failed],
    operator: [p.stateLabels.backlog],
  };
}

/** `owner/repo` and the number, from a fully-qualified issue URL. */
const ISSUE_URL = /^https?:\/\/[^\s/]+\/([^\s/]+\/[^\s/]+)\/issues\/(\d+)\/?$/;

/**
 * Parse an issue reference the caller fully qualified.
 *
 * Fully qualified for the reason `PR_URL` in the GitHub adapter is: `gh` also
 * accepts a bare number and resolves it against whatever repository the current
 * directory belongs to, so an unqualified reference is not refused — it is
 * answered, confidently, about some other repository's issue.
 */
export function issueRefFrom(url: string): { repo: string; issue: number } | undefined {
  const match = ISSUE_URL.exec(url);
  if (match === null) return undefined;
  const issue = Number.parseInt(match[2] ?? "", 10);
  return Number.isSafeInteger(issue) && issue > 0 ? { repo: match[1] ?? "", issue } : undefined;
}

/**
 * Whether a `verifyPr` refusal is the exact-head mismatch rather than a check
 * verdict.
 *
 * Matched on the adapter's own wording (`prVerificationFrom`, `tracker/
 * github.ts`), which already names both SHAs — reusing it is what #126 asks
 * for, and re-deriving the comparison here would be a second implementation of
 * the one rule that must not drift. Pinned by a test against that function's
 * real output, so a reworded reason fails a test rather than silently
 * downgrading a stale head into "checks not green".
 */
export function isHeadMismatch(reason: string): boolean {
  return reason.startsWith("PR head changed");
}

/**
 * Which declared release requirements are not satisfied.
 *
 * Pure, and fail-closed on anything it cannot settle: a requirement the daemon
 * cannot prove is a requirement that has not been met. `epic-children-closed`
 * is the honest example — nothing in a release request names an epic, so the
 * daemon cannot check it and says so rather than waving it through.
 */
export interface ReleaseFacts {
  /** Runs still occupying an issue: live workers and unmerged green PRs. */
  unsettledRuns: number;
  /** Of those, the ones with a pull request still open. */
  openPrs: number;
  /**
   * Of those, the ones with a live worker (claimed/running), as opposed to
   * finished pushes. Absent when the caller could not distinguish (the plain
   * refusal wording is used then).
   */
  liveWorkers?: number;
  /**
   * Of those, finished runs whose PR the tracker could not confirm merged yet
   * — the settle sweep either has not run or could not read GitHub. Distinct
   * from {@link liveWorkers} because only a live worker means the operator
   * must wait on a process; these clear by themselves on the next sweep.
   */
  unconfirmedMerges?: number;
  /**
   * The unsettled runs themselves, routed-repo and issue each, so the refusal
   * names what is blocking instead of a bare count (#603). Absent when the
   * caller could not gather them (the plain wording is used then).
   */
  blockingRuns?: { repo: string; issue: number }[];
  /** Queue depth, or `undefined` when the tracker could not be read. */
  queueDepth: number | undefined;
  /**
   * The project this release targets, so a runs-settled refusal can name that
   * project's own bounded drain command as the remedy (#791). Absent when the
   * caller cannot name one; the remedy then omits --project.
   */
  project?: string;
  /** Current live-head workflow verdict for the released routed repository. */
  baseCheck?: RunRecord["baseCheck"];
  /** Evidence attached to a current red verdict. */
  redBase?: string;
}

export function releaseRequirementRefusal(
  requires: readonly ReleaseRequirement[],
  facts: ReleaseFacts,
): string | undefined {
  for (const requirement of requires) {
    if ((requirement === "runs-settled" || requirement === "fleet-runs-settled") && facts.unsettledRuns > 0) {
      // The same blocker reads differently depending on WHY the row is still
      // active: a live worker is the operator's wait, an unconfirmed merge or
      // an unrecorded sweep clears by itself on the next settle pass. The
      // gate has the split; the message must hand it to the reader.
      const live = facts.liveWorkers ?? 0;
      const unconfirmed = facts.unconfirmedMerges ?? 0;
      // openPrs counts verifiably-open PRs *and* unread ones (fail-closed for
      // no-open-prs), so the verifiably-open sub-bucket is the difference.
      const open = Math.max(0, (facts.openPrs ?? 0) - unconfirmed);
      // Everything not named above: closed unmerged PRs and rows with no PR.
      const remaining = facts.unsettledRuns - live - unconfirmed - open;
      const parts: string[] = [];
      if (live > 0) parts.push(`${live} worker(s) still running`);
      if (open > 0) parts.push(`${open} pull request(s) still open`);
      if (unconfirmed > 0) parts.push(`${unconfirmed} run(s) whose PR merge the settle sweep has not yet recorded`);
      if (remaining > 0) parts.push(`${remaining} run(s) with no merged PR (closed or none yet)`);
      // Name the runs behind the count so the operator can see at a glance
      // whether this is the release's own work or someone else's (#603).
      const blocking =
        facts.blockingRuns === undefined || facts.blockingRuns.length === 0
          ? ""
          : ` — blocking: ${facts.blockingRuns.map((r) => `${r.repo} #${r.issue}`).join(", ")}`;
      // The bounded drain is the release-window remedy (#484): it pauses only
      // claiming while the settle pass keeps running, and expires on an
      // absolute deadline. It is the one command that reaches this gate
      // honestly — not hold, which stops every pass, and no queue-label
      // churn, which writes tracker state (#791).
      const drain =
        ". Settle the fleet with a bounded drain: " +
        `omp-conductor drain start${facts.project === undefined ? "" : ` --project ${facts.project}`} --until <deadline>` +
        " — claiming pauses until the deadline even once the fleet goes quiet (an emptied fleet opens" +
        " the window rather than ending the drain early), and admission resumes at the deadline on its own.";
      return (
        `policy.release.requires includes ${requirement} and ${facts.unsettledRuns} run(s) have not settled` +
        (parts.length === 0 ? "" : ` — ${parts.join("; ")}`) +
        blocking +
        drain
      );
    }
    if (requirement === "no-open-prs" && facts.openPrs > 0) {
      return `policy.release.requires includes no-open-prs and ${facts.openPrs} pull request(s) are still open`;
    }
    if (requirement === "queue-drained") {
      if (facts.queueDepth === undefined) {
        return "policy.release.requires includes queue-drained and the tracker could not be read; refusing rather than assuming the queue is empty";
      }
      if (facts.queueDepth > 0) {
        return `policy.release.requires includes queue-drained and ${facts.queueDepth} issue(s) are still queued`;
      }
    }
    if (requirement === "base-branch-green") {
      if (facts.baseCheck === "green") continue;
      if (facts.baseCheck === "pending") {
        return "policy.release.requires includes base-branch-green and the newest merge's base workflows are still pending";
      }
      if (facts.baseCheck === "red" || facts.baseCheck === "red-preexisting") {
        return (
          "policy.release.requires includes base-branch-green and the newest observed base branch is red" +
          (facts.redBase === undefined ? "" : `: ${facts.redBase}`)
        );
      }
      return "policy.release.requires includes base-branch-green and no green base-branch verdict is available; refusing rather than assuming the base is healthy";
    }
    if (requirement === "epic-children-closed") {
      return "policy.release.requires includes epic-children-closed, which nothing in a release request names an epic for. The daemon cannot settle it, so it refuses: take the requirement off this project's policy, or cut this release by hand.";
    }
  }
  return undefined;
}

/** Everything one decided call writes down, whichever way it went. */
interface Verdict {
  reply: VerbReply;
  entry?: VerbLedgerEntry;
}

function record(
  deps: VerbDeps,
  channel: VerbChannel,
  verb: VerbName,
  args: Record<string, unknown>,
  outcome:
    | { decision: "allowed"; detail: string; sha?: string }
    | { decision: "refused"; refusal: VerbRefusal; detail: string },
  issue?: number,
): Verdict {
  // Reads are not ledgered. The ledger is the record of what a session *did* and
  // tried to do; a status poll every thirty seconds would bury the refusals it
  // exists to surface under its own noise.
  if (!VERB_SPECS[verb].mutating) {
    return {
      reply: {
        ok: outcome.decision === "allowed",
        verb,
        ...(outcome.decision === "refused" ? { refusal: outcome.refusal } : {}),
        text: outcome.detail,
      },
    };
  }
  const entry = deps.store.appendVerbLedger({
    project: channel.project,
    ...(channel.kind === "run" ? { runId: channel.runId } : {}),
    ...(issue === undefined ? {} : { issue }),
    verb,
    role: channel.role,
    args,
    decision: outcome.decision,
    ...(outcome.decision === "refused" ? { refusal: outcome.refusal } : {}),
    detail: outcome.detail,
    ...(outcome.decision === "allowed" && outcome.sha !== undefined ? { sha: outcome.sha } : {}),
  });
  return {
    entry,
    reply: {
      ok: outcome.decision === "allowed",
      verb,
      ...(outcome.decision === "refused" ? { refusal: outcome.refusal } : {}),
      text: outcome.detail,
      ...(outcome.decision === "allowed" && outcome.sha !== undefined ? { sha: outcome.sha } : {}),
    },
  };
}

/**
 * Decide and execute one verb call. Never throws: a thrown error inside a check
 * would close the connection with no ledger entry, which is exactly the shape
 * an attacker would like an unhandled case to have.
 */
export async function handleVerbCall(
  deps: VerbDeps,
  channel: VerbChannel,
  raw: unknown,
): Promise<VerbReply> {
  try {
    return (await decide(deps, channel, raw)).reply;
  } catch (err) {
    const detail = err instanceof Error ? err.message : String(err);
    deps.log(`verb call on ${channel.path} failed unexpectedly: ${detail}`);
    return { ok: false, refusal: "config-unreadable", text: `refused: the daemon errored (${detail}).` };
  }
}

async function decide(deps: VerbDeps, channel: VerbChannel, raw: unknown): Promise<Verdict> {
  const parsed = parseVerbRequest(raw);
  if (!parsed.ok) {
    // A payload whose verb never resolved cannot be filed under a verb. It is
    // still answered — the caller learns exactly what it sent wrong — but the
    // ledger stays a record of calls to real verbs rather than of noise.
    if (parsed.verb === undefined) {
      return { reply: { ok: false, refusal: parsed.refusal, text: parsed.detail } };
    }
    return record(deps, channel, parsed.verb, {}, { decision: "refused", refusal: parsed.refusal, detail: parsed.detail });
  }

  const { verb, args } = parsed.request;
  const spec = VERB_SPECS[verb];
  const issue = channel.kind === "run" ? channel.issue : undefined;
  const refuse = (refusal: VerbRefusal, detail: string, at = issue): Verdict =>
    record(deps, channel, verb, args, { decision: "refused", refusal, detail }, at);
  const allow = (detail: string, sha?: string, at = issue): Verdict =>
    record(deps, channel, verb, args, { decision: "allowed", detail, ...(sha === undefined ? {} : { sha }) }, at);

  // 1. Closed by default. Before anything else reads config or touches the
  //    network: a caller of the wrong kind learns nothing about this project.
  const wrongRole = roleRefusal(spec, channel.role);
  if (wrongRole !== undefined) return refuse("role-not-allowed", `refused: ${wrongRole}`);

  // 2. Fail closed on config. Read now, per call, so an operator's edit and an
  //    unreadable file both take effect on the next verb rather than the next
  //    daemon restart.
  let project: ProjectConfig;
  try {
    project = deps.project();
  } catch (err) {
    const why = err instanceof Error ? err.message : String(err);
    return refuse("config-unreadable", `refused: this project's config could not be read (${why}).`);
  }
  if (project.name !== channel.project) {
    return refuse(
      "config-unreadable",
      `refused: this socket belongs to project ${channel.project}, and the config now resolves ${project.name}.`,
    );
  }

  // 3. The run's own row, re-read. A worker whose run was killed, orphaned or
  //    settled between its last turn and this call is no longer holding
  //    anything, and a push from it would land work nothing is tracking. Read
  //    *before* the stop gate below, because the stop gate needs this row to
  //    tell an in-flight run (which may finish) from a post-pause call.
  let run: RunRecord | undefined;
  if (channel.kind === "run") {
    run = deps.store.getRun(channel.runId);
    if (run === undefined) {
      return refuse("run-not-live", `refused: run ${channel.runId} is no longer in the store.`);
    }
    if (!LIVE_STATES.some((state) => state === run?.state)) {
      return refuse("run-not-live", `refused: this run is ${run.state}, not live. Its mutations are over.`);
    }
  }

  // 4. Stop always wins for work-starting mutations, and is re-read here
  //    rather than remembered. Reads are exempt. A run admitted before the
  //    pause may finish its own mutations (#174). Orchestrator completion
  //    verbs resolve their target run inside the handler, where they can prove
  //    that run also predates the pause (#251). Release remains available:
  //    its authority, grant, and runs-settled gates still fail closed below.
  const orchestratorCompletion =
    channel.kind === "orchestrator" &&
    (verb === "conductor_pr_merge" ||
      verb === "conductor_pr_update_branch" ||
      verb === "conductor_pr_update" ||
      verb === "conductor_pr_review" ||
      // A clearance disposes of evidence against a PR a pre-pause run already
      // pushed (#913). It starts no work, and a pause that refused it would
      // simply relocate the deadlock it exists to break: `conductor_pr_merge`
      // stays available under a pause, so the one thing that unblocks it must
      // too.
      verb === "conductor_pr_review_clear" ||
      verb === "conductor_pr_recover" ||
      verb === "conductor_label");
  const stopped =
    spec.mutating && verb !== "conductor_release" && verb !== "conductor_install" && !orchestratorCompletion
      ? deps.fleetStop()
      : undefined;
  if (stopped !== undefined) {
    const at = deps.pausedAt();
    const admittedBeforePause = run !== undefined && at !== undefined && run.startedAt < at;
    if (!admittedBeforePause) {
      return refuse(
        "fleet-paused",
        `refused: ${stopped}. The pause refuses new claims and work-starting mutations. ` +
          "Completion verbs for runs admitted before the pause (conductor_pr_merge, conductor_pr_update_branch, " +
          "conductor_pr_update, conductor_pr_review, conductor_pr_review_clear, conductor_pr_recover, " +
          "conductor_label), conductor_release and conductor_install remain available.",
      );
    }
  }

  switch (verb) {
    case "conductor_push":
      return pushVerb(deps, channel, args, refuse, allow);
    case "conductor_pr_create":
      return prCreateVerb(deps, project, channel, args, refuse, allow);
    case "conductor_pr_update_branch":
      return prUpdateBranchVerb(deps, project, channel, args, refuse, allow);
    case "conductor_pr_update":
      return prUpdateVerb(deps, project, channel, args, refuse, allow);
    case "conductor_pr_merge":
      return prMergeVerb(deps, project, channel, args, refuse, allow);
    case "conductor_label":
      return labelVerb(deps, project, channel, args, refuse, allow);
    case "conductor_release":
      return releaseVerb(deps, project, channel, args, refuse, allow);
    case "conductor_install":
      return installVerb(deps, channel, args, refuse, allow);
    case "conductor_pr_review":
      return prReviewVerb(deps, project, channel, args, refuse, allow);
    case "conductor_pr_review_clear":
      return prReviewClearVerb(deps, project, channel, args, refuse, allow);
    case "conductor_pr_recover":
      return prRecoverVerb(deps, project, channel, args, refuse, allow);
    case "conductor_pr_status":
      return prStatusVerb(deps, project, channel, args, refuse, allow);
    case "conductor_ci_logs":
      return ciLogsVerb(deps, project, channel, args, refuse, allow);
  }
}

type Refuse = (refusal: VerbRefusal, detail: string, issue?: number) => Verdict;
type Allow = (detail: string, sha?: string, issue?: number) => Verdict;

/** A run channel, or the refusal for a verb that needs one. Narrows the union. */
function runChannel(
  channel: VerbChannel,
  refuse: Refuse,
): { ok: true; channel: Extract<VerbChannel, { kind: "run" }> } | { ok: false; verdict: Verdict } {
  if (channel.kind === "run") return { ok: true, channel };
  return {
    ok: false,
    verdict: refuse("role-not-allowed", "refused: this verb acts on a run, and this socket belongs to no run."),
  };
}

async function pushVerb(
  deps: VerbDeps,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  const scoped = runChannel(channel, refuse);
  if (!scoped.ok) return scoped.verdict;
  const run = scoped.channel;

  // The only ref this verb has. `ref` is an assertion the caller may make about
  // it, never a selector: a mismatch is refused rather than redirected, and
  // there is no `force` argument to reject because none is declared.
  const expected = `refs/heads/${run.branch}`;
  const ref = args["ref"];
  if (typeof ref === "string" && ref !== expected) {
    return refuse(
      "ref-not-run-branch",
      `refused: this run pushes ${expected} and nothing else. You named "${ref}". ` +
        "There is no force path and no way to name another ref: the daemon derives the ref from the run.",
    );
  }

  const outcome = await deps.actions.push({
    repo: run.repo,
    runRepoPath: run.runRepoPath,
    branch: run.branch,
  });
  if (!outcome.ok) {
    return refuse(
      "action-failed",
      `refused: the push was rejected. git said:\n${outcome.stderr}\n\n` +
        "This push is fast-forward only. A non-fast-forward rejection means the remote branch moved — " +
        "fetch and rebase in your own checkout, then push again.",
    );
  }
  return allow(`pushed ${expected}${outcome.sha === undefined ? "" : ` at ${outcome.sha}`}.`, outcome.sha);
}

async function prCreateVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  const scoped = runChannel(channel, refuse);
  if (!scoped.ok) return scoped.verdict;
  const run = scoped.channel;
  const base = run.repo.defaultBranch;

  const claimedBase = args["base"];
  if (typeof claimedBase === "string" && claimedBase !== base) {
    return refuse(
      "base-not-default-branch",
      `refused: ${run.repo.name} targets ${base}, and you named "${claimedBase}". ` +
        "The base is the repo's configured defaultBranch, not an argument.",
    );
  }
  const claimedHead = args["head"];
  if (typeof claimedHead === "string" && claimedHead !== run.branch) {
    return refuse(
      "ref-not-run-branch",
      `refused: this run's head is ${run.branch}, and you named "${claimedHead}".`,
    );
  }

  // #25's guard, asked of the tracker rather than the store: the store only
  // knows work this database recorded, and a restored or relocated state
  // directory presents finished work as an untouched queue item.
  const existing = deps.store.getRun(run.runId)?.prUrl;
  if (existing !== undefined) {
    return refuse("open-pr-exists", `refused: this run already opened ${existing}.`);
  }
  let closer: OpenCloser | undefined;
  try {
    closer = await deps.tracker.openCloserFor(run.issue);
  } catch (err) {
    const why = err instanceof Error ? err.message : String(err);
    return refuse(
      "open-pr-lookup-error",
      `refused: the tracker could not say whether #${run.issue} already has an open pull request (${why}). ` +
        "Refusing rather than opening a second one.",
    );
  }
  if (closer !== undefined) {
    // Adoption (#176): a continuation run is never bound to its issue's existing
    // PR — the old row owns the prUrl, and this new row's prUrl stayed NULL — so
    // when the open PR was opened from *this* run's branch, it is the same work.
    // Bind it to this run instead of refusing: pr_status then answers from the
    // live PR, and a merge can act on it. Branch mismatch (or an empty headRefName,
    // which is never identity) keeps the recorded refusal.
    if (closer.headRefName !== "" && closer.headRefName === run.branch) {
      deps.store.updateRun(run.runId, { prUrl: closer.url });
      return allow(
        `adopted ${closer.url} — already open for ${run.branch} and closes #${run.issue}; title/body arguments were not applied`,
      );
    }
    return refuse("open-pr-exists", `refused: ${closer.url} is already open and already closes #${run.issue}.`);
  }

  const title = args["title"];
  const body = args["body"];
  const outcome = await deps.actions.createPr(
    { repo: run.repo, runRepoPath: run.runRepoPath, branch: run.branch },
    { title: typeof title === "string" ? title : "", body: typeof body === "string" ? body : "", base },
  );
  if (!outcome.ok) {
    return refuse("action-failed", `refused: gh could not open the pull request:\n${outcome.stderr}`);
  }
  // Recorded now rather than at settlement: the merge verb resolves a run from
  // its PR URL, and a PR the store never learned about is one the orchestrator
  // cannot act on through any verb.
  deps.store.updateRun(run.runId, { prUrl: outcome.url });
  return allow(`opened ${outcome.url} (${run.branch} → ${base}).`);
}

/**
 * The run one pull request belongs to, within this project.
 *
 * How far back to look is the recent-history cutoff `status` already uses: a
 * pull request older than that is not one an orchestrator is mid-flight on.
 * The bound itself is shared with the review watch (`PR_LOOKUP_WINDOW_MS` in
 * `decisions.ts`), so verb and condition resolve ownership identically.
 */
function runForPr(deps: VerbDeps, project: string, prUrl: string): RunRecord | undefined {
  // Every attempt that recorded this PR, newest first — not `recentRuns`'
  // newest-attempt-per-issue view. The row a PR belongs to is the newest
  // attempt that owns it, and a requeued continuation that never inherited
  // the URL must not hide its predecessor's ownership (#434).
  return deps.store.runsForProjectPr(project, prUrl, deps.now() - PR_LOOKUP_WINDOW_MS)[0];
}

/**
 * Whether a PR URL names a repository this project routes.
 *
 * Orchestrator status / update-branch may act on open project PRs that no run
 * opened (workflow-generated release pins, etc.). Scope is the routed clone
 * slugs — not "any PR a run once recorded".
 */
function repoForPr(project: ProjectConfig, prUrl: string): RepoTarget | undefined {
  const parts = prUrlParts(prUrl);
  if (parts === undefined) return undefined;
  const slug = `${parts.owner}/${parts.repo}`;
  return Object.values(project.routing.repos).find((repo) => repoSlugFor(repo) === slug);
}

function prInProjectRouting(project: ProjectConfig, prUrl: string): boolean {
  return repoForPr(project, prUrl) !== undefined;
}

/** Refuse an orchestrator completion mutation that cannot prove a pre-pause run. */
function orchestratorPauseRefusal(
  deps: VerbDeps,
  channel: VerbChannel,
  target: RunRecord | undefined,
  refuse: Refuse,
): Verdict | undefined {
  if (channel.kind !== "orchestrator") return undefined;
  const stopped = deps.fleetStop();
  if (stopped === undefined) return undefined;
  const at = deps.pausedAt();
  if (target !== undefined && at !== undefined && target.startedAt < at) return undefined;
  const why =
    target === undefined
      ? "this issue has no recorded run admitted before it."
      : at === undefined
        ? "the pause timestamp cannot prove when this run started."
        : "this run started after it.";
  return refuse(
    "fleet-paused",
    `refused: ${stopped}. Under pause, this verb applies only to runs admitted before the pause; ${why}`,
    target?.issue,
  );
}

async function prUpdateBranchVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  const prUrl = String(args["prUrl"]);

  let target: RunRecord | undefined;
  if (channel.kind === "run") {
    const own = deps.store.getRun(channel.runId);
    if (own?.prUrl === undefined) {
      return refuse("pr-missing", "refused: this run has no pull request yet.");
    }
    if (own.prUrl !== prUrl) {
      return refuse(
        "pr-not-this-run",
        `refused: this socket belongs to the run for #${channel.issue}, whose pull request is ${own.prUrl}. ` +
          "A worker can only act on its own run's PR — the daemon resolves the run from the socket, not from what you sent.",
      );
    }
    target = own;
  } else {
    target = runForPr(deps, project.name, prUrl);
    if (target === undefined && !prInProjectRouting(project, prUrl)) {
      return refuse(
        "pr-not-this-run",
        `refused: ${prUrl} is not a pull request in ${project.name}'s routed repositories ` +
          `(${Object.values(project.routing.repos).map(repoSlugFor).join(", ") || "none"}).`,
      );
    }
  }

  const paused = orchestratorPauseRefusal(deps, channel, target, refuse);
  if (paused !== undefined) return paused;

  let state: PrState | undefined;
  try {
    state = await deps.tracker.prState(prUrl);
  } catch (err) {
    state = undefined;
    deps.log(`verb pr_update_branch could not read ${prUrl}: ${err instanceof Error ? err.message : String(err)}`);
  }
  if (state !== "open") {
    return refuse(
      "pr-not-open",
      state === undefined
        ? `refused: the tracker could not say whether ${prUrl} is open. Refusing rather than acting on an unknown state.`
        : `refused: ${prUrl} is ${state}, not open.`,
      target?.issue,
    );
  }

  const outcome = await deps.actions.updatePrBranch(prUrl);
  if (!outcome.ok) {
    return refuse("action-failed", `refused: gh could not update the branch:\n${outcome.stderr}`, target?.issue);
  }
  // The REST endpoint answers 202 Accepted: the refresh runs asynchronously, so
  // this call produces no head SHA and the reply/ledger must not claim one —
  // the head the action observed is the pre-update guard head, never a result
  // (#655 review). The caller re-reads PR and check state before merging.
  return allow(
    `requested a base-branch update of ${prUrl} from its observed head; PR and check state must be re-read before merging.`,
    undefined,
    target?.issue,
  );
}

async function prMergeVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {

  const prUrl = String(args["prUrl"]);
  const headSha = String(args["headSha"]);
  const reason = String(args["reason"]);
  // An explicit authorship claim: `<author> = "orchestrator"` states that no
  // run opened this PR — it is the caller's own work. The claim is the only
  // thing that can make a run-less PR eligible, and it unlocks nothing else:
  // the holder gate above, the routed-repo scope, the pause, the live-head
  // recheck, the green-checks gate and the ledger all still apply below.
  const authoredByOrchestrator = args["author"] === "orchestrator";
  const target = runForPr(deps, project.name, prUrl);
  const recovery =
    target === undefined
      ? project.recoveryMerges?.find((authorization) => authorization.prUrl === prUrl)
      : undefined;
  const routedRepo = repoForPr(project, prUrl);
  const authorizedRecovery =
    recovery !== undefined &&
    recovery.headSha.toLowerCase() === headSha.toLowerCase() &&
    recovery.reason === reason;

  // Ordinary calls obey the standing holder. A recovery entry is itself the
  // operator's one-PR grant to the orchestrator, so its exact URL/head/reason
  // match replaces that broad holder check without widening any other merge.
  const holder = project.authority.merge;
  if (holder !== channel.role && !authorizedRecovery) {
    return refuse(
      "authority-holder",
      `refused: merge authority is the ${holder}'s for ${project.name}, and this is a ${channel.role} session. ` +
        (holder === "human"
          ? "No session merges under this config. Report the PR as ready and stop."
          : "Ask the operator which session is meant to hold it."),
    );
  }
  if (authoredByOrchestrator && target !== undefined) {
    return refuse(
      "pr-not-this-run",
      `refused: ${prUrl} is a run-opened PR in ${project.name}; the orchestrator-authors claim is only for ` +
        "pull requests no run ever opened, and a run-owned PR is not claimable as the orchestrator's own work.",
    );
  }
  if (target === undefined && recovery === undefined) {
    if (!authoredByOrchestrator) {
      return refuse(
        "pr-not-this-run",
        `refused: ${prUrl} is not a pull request any run in ${project.name} opened. ` +
          "If the orchestrator authored it (no run opened it), re-call with --arg author=orchestrator; " +
          "a PR a run opened stays that run's.",
      );
    }
    if (routedRepo === undefined) {
      return refuse(
        "pr-not-this-run",
        `refused: ${prUrl} is not in ${project.name}'s routed repositories; the orchestrator-authors path cannot widen project scope.`,
      );
    }
  }
  if (target === undefined && routedRepo === undefined) {
    return refuse(
      "pr-not-this-run",
      `refused: ${prUrl} is not in ${project.name}'s routed repositories; a recovery authorization cannot widen project scope.`,
    );
  }
  if (recovery !== undefined && !authorizedRecovery) {
    return refuse(
      "recovery-authorization-mismatch",
      `refused: ${prUrl} has a recovery authorization, but not for this exact head and reason.`,
    );
  }
  const issue = target?.issue;

  if (!authorizedRecovery) {
    const paused = orchestratorPauseRefusal(deps, channel, target, refuse);
    if (paused !== undefined) return paused;
  }

  // Base-red-freeze gate (#283): a repo whose base is frozen must not accept
  // another merge, no matter how green this PR's own checks look. Deliberately
  // placed here — after target-repo identity resolution, before the merge lock
  // and every `gh` call — so the refusal is a verb-level assertion, not a
  // daemon-side flag the merge path could miss. Only the frozen repo is gated:
  // sibling repos keep merging. The freeze lifts mechanically on a green base
  // re-observation or via the operator's sanctioned `omp-conductor unfreeze
  // <repo>` — never by label surgery or a DB edit.
  const freezeRepo = target === undefined ? routedRepo?.name : target.repo;
  if (freezeRepo !== undefined) {
    const freeze = deps.store.baseFreeze(project.name, freezeRepo);
    if (freeze !== undefined && freeze.clearedAt === undefined) {
      return refuse(
        "base-red-freeze",
        `refused: merges to ${freezeRepo} are frozen because its base is red at ` +
          `${freeze.culpritSha.slice(0, 8)} (base-red-freeze). ${freeze.detail ?? ""} ` +
          "This is not a merge-lock or check verdict: the base that merges land on is broken, so " +
          "`conductor_pr_merge` for this repo is refused until the base is observed green again or the " +
          `operator lifts the freeze with \`omp-conductor unfreeze ${freezeRepo}\`. The override is the sanctioned ` +
          "path — never work around this refusal with label surgery or a DB edit.",
        issue,
      );
    }
  }

  // Exact-head review gate (#888): green checks say nothing about the durable
  // review lifecycle, and #886 merged through exactly that blind spot — CI
  // green at the live head while a review revision had settled `failed` on
  // that same head hours earlier. Consulted here, before the lock and every
  // `gh` call, from the same local rows the review verb writes: any round not
  // yet settled (queued, or dispatched and crashed mid-review) and any round
  // settled `failed` at exactly the requested head refuses the merge, unless a
  // `conductor_pr_review_clear` recorded at or after that round's own last
  // activity has dispositioned it (#913). A finding tied only to an older head
  // never appears here — clearance follows the exact head, so a corrected push
  // is never poisoned.
  // A head an adjudication decided against never merges (#876), whatever else
  // is green. The disposition normally closes the PR, so this rarely fires —
  // but "rarely" is not "never": a close that failed, or a race between the
  // disposition pass and this call, must not leave a rejected diff mergeable.
  // Read from the same durable row status renders, so the refusal and the board
  // cannot disagree.
  const decided = deps.store.reviewAdjudicationForHead(project.name, prUrl, headSha);
  if (decided !== undefined && decided.settledAt !== undefined && decided.state !== "cleared") {
    return refuse(
      "merge-blocked-by-review",
      `refused: the review-ceiling adjudication for ${prUrl} at ${headSha.slice(0, 12)} settled ` +
        `${decided.state}, not cleared${decided.evidence === undefined ? "" : ` — ${decided.evidence}`}. ` +
        "There is no further review round and no second adjudicator: this head does not merge. " +
        "Push a corrected head, which is adjudicated on its own merits.",
      issue,
    );
  }

  // The exact-head review-evidence gate (#888), whose reasoning the block above
  // this one continues from: any round not yet settled (queued, or dispatched
  // and crashed mid-review) and any round settled `failed` at exactly the
  // requested head refuses the merge, unless a `conductor_pr_review_clear`
  // recorded at or after that round's own last activity has dispositioned it.
  const blocking = deps.store.mergeBlockingReviews(project.name, prUrl, headSha);
  if (blocking.length > 0) {
    const b = blocking[0]!;
    const phase =
      b.settledAt !== undefined
        ? `settled ${b.outcome ?? "failed"} with unaddressed findings`
        : b.dispatchedAt !== undefined
          ? "was dispatched and never finished (crashed mid-review)"
          : "is queued and has not been addressed";
    const more = blocking.length > 1 ? ` (+${blocking.length - 1} further blocking round(s) at this head)` : "";
    // The advice names only clearances that exist (#913). Re-reviewing the
    // same head was the old wording's suggestion and it provably does not
    // work: `enqueueReviewRevision` appends to the pending row or opens a new
    // unsettled one, so a fresh round at an unchanged head is another blocking
    // round — and at the review-round ceiling there is no further round to
    // spend at all.
    return refuse(
      "merge-blocked-by-review",
      `refused: ${prUrl} cannot merge at ${headSha.slice(0, 12)} — review round ${b.round} at this exact head ` +
        `${phase}${more}. Green checks do not clear durable review findings, and neither does re-reviewing ` +
        "this head: a fresh `conductor_pr_review` at an unchanged head opens (or appends to) another " +
        "blocking round rather than superseding one. The two clearances that exist are a corrected head " +
        "pushed by the run — clearance follows the exact head, so a different head is not poisoned by this " +
        "one — or an explicit `conductor_pr_review_clear` for this exact head, recorded with its reason.",
      issue,
    );
  }

  // Active release-composition gate (#850): while the project has a release
  // declared as being assembled, only a PR named in that composition — or one
  // with an explicit recorded operator override — may merge into it. Placed
  // after identity resolution and before the merge lock and every tracker
  // call, so an unrelated green PR is refused without spending a gh round
  // trip: this is the mechanical form of "a patch contains no unrelated
  // merges", not a status warning. The override bypasses only this check —
  // authority, pause, base-red-freeze, exact-head, green-checks and
  // single-flight all still apply below.
  const composition = deps.store.activeReleaseComposition(project.name);
  if (composition !== undefined && !composition.allowedPrUrls.includes(prUrl)) {
    const overridden =
      deps.store.releaseCompositionOverride(project.name, composition.campaign, prUrl) !== undefined;
    if (!overridden) {
      const allowed =
        composition.allowedPrUrls.length === 0 ? "none yet" : composition.allowedPrUrls.join(", ");
      return refuse(
        "outside-active-release",
        `refused: ${project.name} is assembling release ${composition.campaign} and ${prUrl} is not in it ` +
          `(allowed so far: ${allowed}). Merging it would smuggle unrelated work into that release ` +
          "(outside-active-release). The guard clears only through `omp-conductor release-composition " +
          "complete|cancel`, or an explicit one-PR `omp-conductor release-composition override --pr <url> " +
          "--reason <why>` — this call's merge reason is not an override.",
        issue,
      );
    }
  }

  // Taken before any network call, so two concurrent callers contend here
  // rather than both spending a `gh` round trip and racing at the merge.
  const holderId = randomUUID();
  const lock = deps.store.acquireMergeLock(project.name, holderId, prUrl, deps.now(), MERGE_LOCK_STALE_MS);
  if (lock === undefined) {
    const live = deps.store.mergeLock(project.name);
    return refuse(
      "merge-in-flight",
      `refused: a merge is already in flight for ${project.name}` +
        (live === undefined ? "" : ` (${live.prUrl})`) +
        ". PRs land one at a time, each re-checked against its base: two agent merges at once is how they clobber each other.",
      issue,
    );
  }

  try {
    // Exact-head, re-read at mutation time. Any push since the caller observed
    // green invalidates that observation, and `verifyPr` is the one place that
    // comparison is implemented.
    let verification: PrVerification | undefined;
    try {
      verification = await deps.tracker.verifyPr(prUrl, headSha);
    } catch (err) {
      const why = err instanceof Error ? err.message : String(err);
      return refuse("head-unresolvable", `refused: the live head of ${prUrl} could not be read (${why}).`, issue);
    }
    if (verification === undefined) {
      return refuse(
        "head-unresolvable",
        `refused: the live head of ${prUrl} could not be resolved. Refusing rather than merging what you last saw.`,
        issue,
      );
    }
    if (verification.status !== "green") {
      const stale = verification.status === "failed" && isHeadMismatch(verification.reason);
      return refuse(
        stale ? "head-stale" : "checks-not-green",
        stale
          ? `refused: ${verification.reason}. You asked to merge ${headSha}; that is not what is on the branch now. ` +
            "Re-read the head, re-check the checks at it, and call again."
          : `refused: ${prUrl} is not mergeable at ${headSha} — ${verification.reason}.`,
        issue,
      );
    }

    const repoTarget =
      target === undefined
        ? routedRepo
        : Object.values(project.routing.repos).find((repo) => repo.name === target.repo);
    const chainDir = repoTarget?.migrations?.dir;
    if (chainDir !== undefined && repoTarget !== undefined) {
      // Checked against the base tip at merge time, not against a stale advisory:
      // the author hit two merged PRs leaving main with two Alembic heads (#227).
      const diff = await deps.tracker.prDiff(prUrl).catch(() => undefined);
      if (diff === undefined) {
        return refuse(
          "chain-unreadable",
          `refused: the diff of ${prUrl} could not be read, so the migration chain in ${chainDir} cannot be verified. Failing closed: a wrong merge here is a silent production schema drift (#227).`,
          issue,
        );
      }
      const { added, changed, deleted } = chainEntriesFromDiff(diff.files, chainDir);
      if (added.length + changed.length + deleted.length > 0) {
        if (diff.truncated) {
          return refuse(
            "chain-unreadable",
            `refused: the diff of ${prUrl} was truncated and it touches ${chainDir} — the chain cannot be verified from a partial diff.`,
            issue,
          );
        }
        const base = await deps.chain.readBaseChain(project, repoTarget, chainDir);
        if (!base.ok) {
          return refuse(
            "chain-unreadable",
            `refused: could not read ${chainDir} at the tip of ${repoTarget.defaultBranch} (${base.stderr}).`,
            issue,
          );
        }
        const violations = chainViolations({ base: base.entries, added, changed, deleted });
        if (violations.length > 0) {
          return refuse(
            "chain-conflict",
            `refused: merging ${prUrl} would corrupt the ordered migration chain in ${chainDir} — ${violations.join("; ")}. Rebase onto the current tip, regenerate the migration, and call again.`,
            issue,
          );
        }
      }
    }

    const outcome = await deps.actions.mergePr(prUrl, headSha);
    if (!outcome.ok) {
      return refuse("action-failed", `refused: gh could not merge:\n${outcome.stderr}`, issue);
    }
    return allow(
      `merged ${prUrl} at ${headSha} (${verification.reason})${
        recovery === undefined ? "." : " under its exact operator recovery authorization."
      }`,
      outcome.sha ?? headSha,
      issue,
    );
  } finally {
    deps.store.releaseMergeLock(project.name, holderId);
  }
}

async function labelVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  const ref = issueRefFrom(String(args["issueUrl"]));
  if (ref === undefined) {
    return refuse(
      "malformed-argument",
      `refused: "${String(args["issueUrl"])}" is not a full issue URL. A bare number resolves against whatever ` +
        "repository the daemon happens to be in, which is how a label lands on the wrong project's issue.",
    );
  }
  if (ref.repo !== project.tracker.repo) {
    return refuse(
      "malformed-argument",
      `refused: ${ref.repo} is not this project's tracker (${project.tracker.repo}).`,
    );
  }

  const paused = orchestratorPauseRefusal(
    deps,
    channel,
    deps.store.latestRun(project.name, ref.issue),
    refuse,
  );
  if (paused !== undefined) return paused;

  const label = String(args["label"]);
  const vocabulary = labelVocabulary(project);
  if (vocabulary.operator.includes(label)) {
    // Harder than the lifecycle refusal: add and remove alike, and never
    // learned away by growing the vocabulary. The park is the operator's
    // gesture — an orchestrator that can park work can hide work it does not
    // want to groom (#507).
    return refuse(
      "label-is-operator-owned",
      `refused: ${label} is the operator's park label and only the operator may set or clear it. ` +
        "No session — orchestrator or worker — may add or remove it; use the tracker directly.",
      ref.issue,
    );
  }
  if (vocabulary.lifecycle.includes(label)) {
    return refuse(
      "label-is-lifecycle",
      `refused: ${label} is a lifecycle label. Those transitions stay the daemon's — orphan detection is only ` +
        "trustworthy while every state label on the tracker was written by the conductor. Use omp-conductor unblock.",
      ref.issue,
    );
  }
  if (!vocabulary.grantable.includes(label)) {
    return refuse(
      "label-not-in-vocabulary",
      `refused: ${label} is not in ${project.name}'s own vocabulary. It declares ${vocabulary.grantable.join(", ")}.`,
      ref.issue,
    );
  }

  const action = args["action"] === "remove" ? "remove" : "add";
  const outcome = `${action === "add" ? "added" : "removed"} ${label} on #${ref.issue}.`;
  // Adding the queue label is the promotion: the author is asserting the file
  // lane right now, so the verb echoes the one admission will enforce — the
  // parsed file list, or the explicit fail-open note (#724). A clearly
  // delimited write-lane section that parsed nothing REFUSES the promotion
  // instead of failing open (#825): the echo must never claim fail-open for a
  // heading that plainly tried to declare one, and failing open beside
  // overlapping work is the exact gap the interlock exists to close. Best
  // effort still holds for reads: an unreadable issue or thread can never
  // block the promotion itself (the label is the point), so a failed read
  // falls back to the plain message.
  let echo: string | undefined;
  let malformed: string | undefined;
  if (action === "add" && label === project.queueLabel) {
    const lane = await laneForEcho(deps, ref.issue);
    if (lane.readable) {
      malformed = lane.malformed;
      echo = laneEcho(lane.lane);
    }
    // The durable grooming verdict is an admission contract only while the
    // dispatch brief still matches it (#1036): a stale issue body once
    // promoted beside a PROMOTABLE verdict whose fileLane had moved on, and
    // the run dispatched on the stale lane. So before any tracker mutation or
    // daemon wake, a promotable row must verify against the same parse the
    // echo reports — exact in both directions, order-insensitive. The syntax
    // refusal below keeps precedence: fix the section first, then the contract.
    if (malformed === undefined) {
      const mismatch = promotionBriefRefusal(deps, project, ref.issue, lane);
      if (mismatch !== undefined) {
        return refuse("promotion-brief-mismatch", mismatch, ref.issue);
      }
    }
  }
  if (malformed !== undefined) {
    return refuse(
      "file-lane-unparseable",
      `refused: ${label} was not added — #${ref.issue} declares a write-lane section (${malformed}) but no ` +
        "path-like files parsed under it, so admission would fail open beside overlapping work. " +
        'Name the files as backticked bullets directly under the heading (e.g. "- `omp/src/a.ts` — what ' +
        'changes"), or remove the heading, then add the queue label again.',
      ref.issue,
    );
  }
  // The queue label is the fleet's lock, so it must not ride the GraphQL
  // surface a provider outage just broke: `gh issue edit` is a GraphQL
  // mutation in gh 2.86, and #642 comment 4 measured a label withdrawal
  // refused by a GraphQL 503 leaving two overlapping issues claimable. The
  // tracker's label methods are core REST with noop classification for the
  // already-holding end state — the guarded path.
  try {
    if (action === "add") {
      await deps.tracker.addLabel(ref.issue, label);
    } else {
      await deps.tracker.removeLabel(ref.issue, label);
    }
  } catch (err) {
    const why = err instanceof Error ? err.message : String(err);
    return refuse("action-failed", `refused: the tracker rejected the label change:\n${why}`, ref.issue);
  }

  // Promotion provenance (#1041): the tick audits what reached the queue
  // unattended, and "who queued this" is only answerable if every promotion
  // path stamps it — a verdict promoted by hand that carried no actor would
  // read in the audit exactly like one the daemon promoted. Stamped after the
  // mutation committed, so a refused or failed label change leaves no
  // provenance, and latched inside the store: a re-add of a label the issue
  // already carries returns false and never restamps.
  if (action === "add" && label === project.queueLabel) {
    deps.store.markGroomingPromoted(project.name, ref.issue, Date.now(), "orchestrator");
  }

  // A promotion that committed has made an issue claimable *now*, so the
  // resident daemon is poked rather than leaving the candidate for the next
  // scheduled pass (#878): three claimable issues once sat against `workers
  // 0 / 3` for five minutes, which reads from outside exactly like a stalled
  // queue. Only the queue label earns it — no other label changes what is
  // dispatchable — and only after the tracker mutation above committed, so a
  // refused or failed change never pokes anything. The pass it prompts applies
  // every hold, drain, lane and budget gate as usual.
  const woken =
    action === "add" && label === project.queueLabel ? await wakeDispatch(project.name) : undefined;

  return allow(
    [echo === undefined ? outcome : `${outcome} File lane: ${echo}.`, woken]
      .filter((part) => part !== undefined)
      .join(" "),
    undefined,
    ref.issue,
  );
}

/**
 * The effective file lane admission will enforce for one issue, as the
 * one-line echo (#724), plus the malformed write-lane section marker (#825):
 * the body plus the whole comment thread, the same inputs `effectiveLane`
 * reads at admission. `readable` says whether the issue and thread were read
 * at all — the echo stays best-effort (an unreadable issue never blocks its
 * own promotion, #724), but the promotion-brief gate (#1036) must be able to
 * tell "no lane declared" from "could not look", because behind a durable
 * `promotable` verdict the second is a fail-closed refusal rather than
 * feedback.
 */
async function laneForEcho(deps: VerbDeps, issue: number): Promise<{
  readable: boolean;
  lane: FileLane | undefined;
  malformed: string | undefined;
  /** What the gate re-reads: the same body, thread and labels this parse used.
   *  Absent exactly when `readable` is false. */
  read: { body: string; comments: IssueComment[]; labels: readonly string[] } | undefined;
}> {
  let body: string;
  let comments: IssueComment[];
  let labels: readonly string[];
  try {
    const row = await deps.tracker.getIssue(issue);
    if (row === undefined) return { readable: false, lane: undefined, malformed: undefined, read: undefined };
    body = row.body;
    labels = row.labels;
    comments = await deps.tracker.listComments(issue);
  } catch {
    return { readable: false, lane: undefined, malformed: undefined, read: undefined };
  }
  const read = { body, comments, labels };
  const lane = effectiveLane(body, comments);
  if (lane !== undefined) return { readable: true, lane, malformed: undefined, read };
  // No declaration parsed anywhere. If a clearly delimited write-lane section
  // exists anyway, the section tried to declare and failed — the echo must
  // refuse, never claim fail-open.
  const heading =
    writeLaneSectionHeading(body) ??
    comments.map((c) => writeLaneSectionHeading(c.body)).find((h) => h !== undefined);
  return { readable: true, lane: undefined, malformed: heading, read };
}

/**
 * The promotion gate (#1036, #1041): a durable `promotable` grooming verdict is
 * an admission contract only while the issue still matches it, so a
 * queue-label add must prove the match before any tracker mutation or daemon
 * wake. Returns the refusal detail, or `undefined` when promotion may proceed:
 * no grooming row, a non-`promotable` row, or a verdict that passes the gate.
 *
 * The judgement itself is {@link readyGate}, not this function. It used to be
 * a lane comparison written here, which was correct and also the wrong place:
 * since #1041 the daemon promotes passing verdicts unattended, and two copies
 * of "is this ready" drift the moment one of them learns something. So this is
 * the verb-shaped wrapper — recover the result, hand the gate the same body,
 * thread and labels the echo parsed, and render whatever it refuses for. The
 * gate names every missing element at once: a promotion that fails four checks
 * costs one round trip, not four.
 *
 * Fail-closed by design: evidence behind a `promotable` row that does not
 * recover as a strict PROMOTABLE result (`parseToSpecEvidence`, the same
 * restart round-trip the selection side trusts), or an issue that cannot be
 * read to compare at all, refuses the promotion rather than dispatching on an
 * unverifiable brief. The gate never edits the issue — it names what is
 * missing and points at the verdict's `proposedBrief`; correcting the dispatch
 * brief is an explicit orchestrator act.
 */
function promotionBriefRefusal(
  deps: VerbDeps,
  project: ProjectConfig,
  issue: number,
  read: { readable: boolean; read: { body: string; comments: IssueComment[]; labels: readonly string[] } | undefined },
): string | undefined {
  const row = deps.store.grooming(project.name, issue);
  if (row?.verdict !== "promotable") return undefined;
  const result = parseToSpecEvidence(row.evidence);
  if (result === undefined || result.verdict !== "PROMOTABLE") {
    return (
      `refused: ${project.queueLabel} was not added — #${issue}'s durable grooming row reads promotable, but its ` +
      "evidence does not recover as a strict PROMOTABLE to-spec result, so promotion cannot prove the dispatch " +
      "brief still matches what was groomed. Re-groom the issue so a fresh verdict replaces the row, then add the label again."
    );
  }
  // `"unread"` rather than an empty thread when the read failed: behind a
  // promotable row "could not look" is a refusal, not an empty lane (#1036).
  const verdict = readyGate({
    result,
    issueBody: read.read?.body ?? "",
    comments: read.readable && read.read !== undefined ? read.read.comments : "unread",
    labels: read.read?.labels ?? [],
    project,
  });
  if (verdict.ok) return undefined;
  return (
    `refused: ${project.queueLabel} was not added — #${issue} carries a durable PROMOTABLE grooming verdict, but ` +
    `the promotion is not ready:\n${verdict.missing.map((miss) => `- ${miss}`).join("\n")}\n` +
    "The durable verdict is the admission contract: apply the verdict's proposedBrief to the issue (its " +
    "## Exact write lane included) and fix every element above, then add the label again. The gate never " +
    "rewrites the issue itself."
  );
}

async function releaseVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  // Same rule as merge, written twice on purpose. A worker never reaches here
  // — the role gate refused it — and an orchestrator still has to *be* the
  // configured holder.
  const holder = project.authority.release;
  if (holder !== channel.role) {
    return refuse(
      "authority-holder",
      `refused: release authority is the ${holder}'s for ${project.name}, and this is a ${channel.role} session.` +
        (holder === "human" ? " No session releases under this config." : ""),
    );
  }

  const shape = args["shape"] as ReleaseShape;
  const grants = resolveReleaseGrants(project);
  const granted = releaseRefusal(grants, channel.role, shape);
  if (granted !== undefined) {
    return refuse("release-not-granted", `refused: ${granted.reason}`);
  }

  const repoName = String(args["repo"]);
  const repo: RepoTarget | undefined = project.routing.repos[repoName];
  if (repo === undefined) {
    return refuse(
      "malformed-argument",
      `refused: ${project.name} routes no repository called "${repoName}". It has ${Object.keys(project.routing.repos).join(", ")}.`,
    );
  }

  const policy = resolvePolicy(project);
  const artefact = args["artefact"];
  const environment = args["environment"];
  if (shape === "package-publish") {
    if (typeof artefact !== "string" || !policy.release.artefacts.includes(artefact)) {
      return refuse(
        "release-target-not-declared",
        `refused: ${project.name} declares ${policy.release.artefacts.length === 0 ? "no artefacts" : `artefacts ${policy.release.artefacts.join(", ")}`}, ` +
          `and you named ${JSON.stringify(artefact)}. An undeclared artefact is one nobody authorised publishing.`,
      );
    }
  }
  if (shape === "deploy") {
    if (typeof environment !== "string" || !policy.release.environments.includes(environment)) {
      return refuse(
        "release-target-not-declared",
        `refused: ${project.name} declares ${policy.release.environments.length === 0 ? "no environments" : `environments ${policy.release.environments.join(", ")}`}, ` +
          `and you named ${JSON.stringify(environment)}. An undeclared environment is one nobody authorised deploying to.`,
      );
    }
  }

  const active = deps.store.activeRuns(project.name);
  const wantsQueue = policy.release.requires.includes("queue-drained");
  let queueDepth: number | undefined;
  if (wantsQueue) {
    try {
      queueDepth = (await deps.tracker.listReady()).length;
    } catch {
      queueDepth = undefined;
    }
  }
  const wantsBase = policy.release.requires.includes("base-branch-green");
  const health = wantsBase
    ? deps.store.baseHealth(project.name).find((row) => row.repo === repoName)
    : undefined;

  // `runs-settled` (and `no-open-prs`) read the PR fact, not the row (#482):
  // a `pushed-green` row whose PR has actually merged is settled work, and a
  // merged PR is not an open one. The settle sweep records the same fact on
  // its next tick; the gate must not wait an extra tick the operator reached
  // this release by holding. Fail-closed stays fail-closed: a row whose PR
  // state cannot be read is still unsettled, and an unknown PR is still
  // treated as open.
  //
  // The run facts are scoped to the released repo (#603): a run in another
  // routed repo cannot invalidate a tag on this one, so it must not gate this
  // release. A genuinely suite-wide shape — a pin or manifest that consumes
  // several repos — opts back into project-wide strictness through the named
  // `fleet-runs-settled` requirement, which widens the scope to every active
  // run. `no-open-prs` shares the scope: the branch being released is this
  // repo's, so an open PR elsewhere is not "against" it either.
  const fleetWide = policy.release.requires.includes("fleet-runs-settled");
  const wantsRunFacts =
    policy.release.requires.includes("runs-settled") ||
    fleetWide ||
    policy.release.requires.includes("no-open-prs");
  const scoped = fleetWide ? active : active.filter((r) => r.repo === repoName);
  let unsettledRuns = scoped.length;
  let openPrs = scoped.filter((r) => r.prUrl !== undefined).length;
  let liveWorkers: number | undefined;
  let unconfirmedMerges: number | undefined;
  let blockingRuns: { repo: string; issue: number }[] | undefined;
  if (wantsRunFacts) {
    let unsettled = 0;
    let open = 0;
    let live = 0;
    let unconfirmed = 0;
    const blocking: { repo: string; issue: number }[] = [];
    for (const run of scoped) {
      if (run.state === "claimed" || run.state === "running") {
        unsettled += 1;
        live += 1;
        blocking.push({ repo: run.repo, issue: run.issue });
        // A live worker's PR is not yet confirmable; count it as open, the
        // same fail-closed answer the old row-count gave it.
        if (run.prUrl !== undefined) open += 1;
        continue;
      }
      // pushed-pending / pushed-green: the worker is finished; the PR decides.
      if (run.prUrl === undefined) {
        unsettled += 1; // nothing to read; the sweep cannot settle it either
        blocking.push({ repo: run.repo, issue: run.issue });
        continue;
      }
      let pr: PrState | undefined;
      try {
        pr = await deps.tracker.prState(run.prUrl);
      } catch {
        pr = undefined;
      }
      if (pr === "merged") continue; // settled in fact; not an open PR
      unsettled += 1;
      blocking.push({ repo: run.repo, issue: run.issue });
      if (pr === "open") open += 1;
      else if (pr === undefined) {
        // Cannot confirm the merge (or its absence): fail closed, and count
        // the row as an unconfirmed merge. Also open — an unread PR may be
        // open — so `no-open-prs` keeps its fail-closed refusal.
        open += 1;
        unconfirmed += 1;
      }
      // "closed": the row still owes a failure write from the sweep; blocks
      // exactly as it did before, the message bucket covers it.
    }
    unsettledRuns = unsettled;
    openPrs = open;
    liveWorkers = live;
    unconfirmedMerges = unconfirmed;
    blockingRuns = blocking;
  }
  const unmet = releaseRequirementRefusal(policy.release.requires, {
    unsettledRuns,
    openPrs,
    ...(liveWorkers === undefined ? {} : { liveWorkers }),
    ...(unconfirmedMerges === undefined ? {} : { unconfirmedMerges }),
    ...(blockingRuns === undefined ? {} : { blockingRuns }),
    queueDepth,
    project: project.name,
    ...(health === undefined ? {} : { baseCheck: health.verdict }),
    ...(health?.verdict === "red" && health.detail !== undefined
      ? { redBase: health.detail }
      : {}),
  });
  if (unmet !== undefined) return refuse("release-not-granted", `refused: ${unmet}`);

  if (!deps.actions.releasableShapes.includes(shape)) {
    return refuse(
      "release-shape-not-executable",
      `refused: this daemon cannot cut a ${shape}. It holds a GitHub credential and nothing else — no npm token, ` +
        "no deploy key — and that is deliberate. It can do " +
        `${deps.actions.releasableShapes.join(", ")}. Escalate this one to your operator.`,
    );
  }

  const tag = args["tag"];
  const sha = args["sha"];
  // A pinned commit is only meaningful for the two tag shapes; anywhere else it
  // would be a decorative argument, which is exactly the silent fake this
  // guards against. Refused rather than ignored, so the caller learns.
  if (typeof sha === "string" && shape !== "git-tag" && shape !== "git-push-tags") {
    return refuse(
      "malformed-argument",
      `refused: sha pins a tag to one commit and only git-tag and git-push-tags cut tags; you asked for ${shape}.`,
    );
  }
  if (typeof sha === "string" && !FULL_OR_ABBREVIATED_SHA.test(sha)) {
    return refuse(
      "malformed-argument",
      `refused: ${JSON.stringify(sha)} is not a commit id. Pass the exact hex sha the tag must point at.`,
    );
  }
  const outcome = await deps.actions.release({
    shape,
    repo,
    ...(typeof tag === "string" ? { tag } : {}),
    ...(typeof sha === "string" ? { sha } : {}),
    ...(typeof artefact === "string" ? { artefact } : {}),
    ...(typeof environment === "string" ? { environment } : {}),
  });
  if (!outcome.ok) {
    return refuse("action-failed", `refused: the release command failed:\n${outcome.stderr}`);
  }
  // The bounded drain is a release window: it stays latched until the release
  // work it fences has actually completed, then clears so admission resumes
  // without a second operator action (#791). Every refusal and failed action
  // above returned without touching it. The release-window boundary is the
  // *terminal* act: a version-bump preparation — whether opened, already
  // prepared, or merged — explicitly leaves git-tag pending (review #1 on
  // #823), so clearing there would admit a run that makes that follow-up
  // git-tag fail runs-settled and forces a second drain. Only the final
  // cut/publish/deploy allow below clears the drain.
  if (shape === "version-bump-pr") {
    if (outcome.review === undefined) {
      return allow(
        `release version for ${repo.name} is already prepared${outcome.detail === undefined ? "" : ` — ${outcome.detail}`}.`,
        outcome.sha,
      );
    }
    if (outcome.sha === undefined) {
      return refuse("action-failed", "refused: version-bump-pr returned no exact pull-request head.");
    }
    if (outcome.review.created) {
      return allow(
        `opened release preparation ${outcome.review.prUrl} at ${outcome.sha}; wait for its checks, then call version-bump-pr again.`,
        outcome.sha,
      );
    }

    const holderId = randomUUID();
    const lock = deps.store.acquireMergeLock(
      project.name,
      holderId,
      outcome.review.prUrl,
      deps.now(),
      MERGE_LOCK_STALE_MS,
    );
    if (lock === undefined) {
      const live = deps.store.mergeLock(project.name);
      return refuse(
        "merge-in-flight",
        `refused: a merge is already in flight for ${project.name}` +
          (live === undefined ? "" : ` (${live.prUrl})`) +
          ". Release preparation waits for the project's single merge slot.",
      );
    }
    try {
      let verification: PrVerification | undefined;
      try {
        verification = await deps.tracker.verifyPr(outcome.review.prUrl, outcome.sha);
      } catch (err) {
        const why = err instanceof Error ? err.message : String(err);
        return refuse(
          "head-unresolvable",
          `refused: the live head of release preparation ${outcome.review.prUrl} could not be read (${why}).`,
        );
      }
      if (verification === undefined) {
        return refuse(
          "head-unresolvable",
          `refused: the live head of release preparation ${outcome.review.prUrl} could not be resolved.`,
        );
      }
      if (verification.status !== "green") {
        const stale = verification.status === "failed" && isHeadMismatch(verification.reason);
        return refuse(
          stale ? "head-stale" : "checks-not-green",
          `refused: release preparation ${outcome.review.prUrl} is not mergeable at ${outcome.sha} — ${verification.reason}.`,
        );
      }

      const merged = await deps.actions.mergePr(outcome.review.prUrl, outcome.sha);
      if (!merged.ok) {
        return refuse("action-failed", `refused: gh could not merge the release preparation:\n${merged.stderr}`);
      }
      return allow(
        `merged release preparation ${outcome.review.prUrl} at ${outcome.sha}; git-tag remains refused until the live version matches.`,
        merged.sha ?? outcome.sha,
      );
    } finally {
      deps.store.releaseMergeLock(project.name, holderId);
    }
  }
  cancelDrain(project.name);
  return allow(
    `cut ${shape} for ${repo.name}${outcome.detail === undefined ? "" : ` — ${outcome.detail}`}.`,
    outcome.sha,
  );
}


/**
 * The install is host-global: one package tree and daemon serve every project.
 * Authority is therefore the consensus holder across the configured project
 * set, never whichever project happened to invoke the verb (#1018). Mixed
 * holders refuse before version lookup or transient-unit creation.
 */
async function installVerb(
  deps: VerbDeps,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  let projects: readonly ProjectConfig[];
  try {
    projects = deps.projects();
  } catch (err) {
    const why = err instanceof Error ? err.message : String(err);
    return refuse("config-unreadable", `refused: shared install authority could not be read (${why}).`);
  }
  if (projects.length === 0) {
    return refuse("config-unreadable", "refused: shared install authority has no configured projects.");
  }
  const authority = resolveSharedInstallAuthority(projects);
  const holderDetail = authority.entries.map((entry) => `${entry.project}=${entry.holder}`).join(", ");
  if (authority.holder === undefined) {
    return refuse(
      "release-not-granted",
      `refused: conductor_install is host-global, but configured install holders conflict (${holderDetail}). ` +
        "No dispatch pause or transient install unit was created.",
    );
  }
  const globalHolder = authority.holder;
  const granted = releaseRefusal(
    { ...DENIED_RELEASE_GRANTS, install: globalHolder },
    channel.role,
    "install",
  );
  if (granted !== undefined) {
    return refuse(
      "release-not-granted",
      `refused: ${granted.reason} Shared-host holders: ${holderDetail}.`,
    );
  }
  const affectedProjects = authority.entries.map((entry) => entry.project);

  const version = args["version"];
  if (typeof version !== "string" || version.trim().length === 0) {
    return refuse(
      "malformed-argument",
      `refused: conductor_install needs a version, got ${JSON.stringify(version)}. ` +
        "Name the exact published semver, e.g. 0.15.11.",
    );
  }

  const outcome = await deps.actions.install({
    version: version.trim(),
    holder: globalHolder,
    projects: affectedProjects,
  });
  if (!outcome.ok) {
    return refuse("action-failed", `refused: the install request failed:\n${outcome.stderr}`);
  }
  const unit = outcome.detail;
  return allow(
    `install requested by host-global holder ${globalHolder}: omp-conductor@${version} will be installed ` +
      `for projects ${affectedProjects.join(", ")} by a detached unit` +
      `${unit === undefined ? "" : ` (${unit})`} that survives the restart. ` +
      "The first tick after the fleet restarts verifies version, /healthz, ticks, pane and doctor, " +
      "and the outcome is reported through the durable outbox.",
  );
}

/**
 * The one read verb, and the reason a worker reports a PR's real state rather
 * than its own recollection.
 *
 * Reuses `Tracker.verifyPr` in read mode, so the answer is literally the merge
 * gate's own verdict rather than a second opinion assembled here — including
 * the moved-head case, whose reason already names both shas. A worker that
 * quotes this has read a fact; a worker that says "should be green" has not.
 * Read mode also reports a merged or closed PR as a fact (`ok: true`) instead
 * of the merge gate's `expected OPEN` refusal, so an orchestrator can confirm
 * release evidence on a just-merged PR (#389).
 */
async function prStatusVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  const asked = args["prUrl"];
  let prUrl: string;
  let issue: number | undefined;

  if (channel.kind === "run") {
    // A worker reads its own run's PR and no other. Omitting `prUrl` is the
    // normal path; naming one is an assertion, checked, never a selector.
    const own = deps.store.getRun(channel.runId);
    if (own?.prUrl === undefined) {
      return refuse("pr-missing", "refused: this run has no pull request yet. Open one with conductor_pr_create.");
    }
    if (typeof asked === "string" && asked !== own.prUrl) {
      return refuse(
        "pr-not-this-run",
        `refused: this socket belongs to the run for #${channel.issue}, whose pull request is ${own.prUrl}. ` +
          "A worker reads only its own run's PR — the daemon resolves the run from the socket.",
      );
    }
    prUrl = own.prUrl;
    issue = channel.issue;
  } else {
    // A read resolves any syntactically valid PR URL — nothing to gate. A
    // release-preparation PR the workflow opened (never a run row, and not
    // necessarily in `routing.repos`) still has to be confirmable before the
    // orchestrator cuts a release (#389). `runForPr` only decides which issue
    // to record on the ledger; `verifyPr`'s `PR_URL` check rejects a URL that
    // is not a pull request.
    if (typeof asked !== "string") {
      return refuse("malformed-argument", "refused: conductor_pr_status needs prUrl when it has no run to infer one from.");
    }
    const target = runForPr(deps, project.name, asked);
    prUrl = asked;
    issue = target?.issue;
  }

  const askedHead = args["headSha"];
  const expectedHead = typeof askedHead === "string" ? askedHead : undefined;
  let verification: PrVerification | undefined;
  try {
    // Read mode: a merged or closed PR is a fact to report, not the merge
    // gate's `expected OPEN` refusal (#389). The mutating verbs never pass
    // this, so their open-PR gate is untouched.
    verification = await deps.tracker.verifyPr(prUrl, expectedHead, { read: true });
  } catch (err) {
    verification = undefined;
    deps.log(`verb pr_status could not read ${prUrl}: ${err instanceof Error ? err.message : String(err)}`);
  }
  if (verification === undefined) {
    return refuse(
      "head-unresolvable",
      `refused: ${prUrl} could not be read. That is not "green" and not "red" — wait and ask again, ` +
        "and report blocked rather than guessing if it stays unreadable.",
      issue,
    );
  }
  return allow(`${prUrl} at ${verification.headSha}: ${verification.status} — ${verification.reason}`, undefined, issue);
}

/**
 * How much failing-job log text one `conductor_ci_logs` reply may carry.
 *
 * Characters rather than bytes: `gh run view --log-failed` output is
 * effectively ASCII, so this is the ~32 KB posture the daemon's own log reads
 * already work at, and a character bound cannot end mid-multibyte-sequence the
 * way a byte slice can. It bounds the log text; the few framing lines are
 * small and constant.
 */
export const CI_LOGS_MAX_CHARS = 32 * 1024;

/** The smallest slice of one failed job worth showing. A PR whose matrix
 *  exploded into forty failed jobs must not answer with forty useless
 *  fragments — better to show the first few in full and say how many were left
 *  out. */
const CI_LOGS_MIN_JOB_CHARS = 2 * 1024;

/**
 * How many of the head's failing workflow runs are read, and how many attempts
 * of each.
 *
 * Both exist because every attempt of every failing job is a separate guarded
 * `gh` call, and one pathologically re-run pull request must not turn a worker's
 * single diagnostic call into a hundred of them. The attempt bound is the
 * tracker's own contract (`runFailedAttemptLogs` refuses rather than presenting
 * attempts 1..N as the whole history), so exceeding it surfaces as the honest
 * read failure below, never as a partial log the worker would read as complete.
 */
const CI_LOGS_MAX_RUNS = 5;
const CI_LOGS_MAX_ATTEMPTS = 10;

/**
 * One bounded rendering of the failing jobs' logs.
 *
 * Pure and exported so the bound is tested directly rather than inferred from a
 * verb reply. Empty chunks are dropped: `runFailedAttemptLogs` deliberately
 * emits one per failed job even when its failed steps printed nothing, which
 * the daemon's all-infra guard needs and a reader does not.
 *
 * Each shown job gets a fair share of the budget, so one enormous test job
 * cannot starve the sibling that actually explains the failure. An over-share
 * job keeps its head *and* its tail — the head carries the step banner and the
 * runner's own setup sentences, the tail carries the failure — with the elision
 * stated in the middle rather than left to be guessed at.
 *
 * No redaction is applied, deliberately and consistently with every other place
 * these same chunks are read (escalations, the historical infra repair): the
 * text is GitHub's own log, which Actions has already masked its registered
 * secrets out of. A second, weaker masking here would only suggest a guarantee
 * this does not make.
 */
export function boundCiLogs(chunks: readonly string[], maxChars = CI_LOGS_MAX_CHARS): string {
  const failing = chunks.filter((chunk) => chunk.trim() !== "");
  if (failing.length === 0) return "";
  const share = Math.max(CI_LOGS_MIN_JOB_CHARS, Math.floor(maxChars / failing.length));
  const room = Math.max(1, Math.floor(maxChars / share));
  const shown = failing.slice(0, room);
  const parts = shown.map((chunk, i) => {
    // Head a quarter, tail the rest: the head carries the step banner and the
    // runner's own setup sentences, the tail carries the failure.
    const head = Math.floor(share / 4);
    const body =
      chunk.length <= share
        ? chunk
        : `${chunk.slice(0, head)}\n--- ${chunk.length - share} characters elided ---\n` +
          chunk.slice(chunk.length - (share - head));
    return `--- failing job ${i + 1} of ${failing.length} ---\n${body}`;
  });
  if (shown.length < failing.length) {
    parts.push(
      `--- ${failing.length - shown.length} further failing job(s) not shown: this reply is bounded to ` +
        `${maxChars} characters of log. Re-run this call after fixing the ones above, or read the rest in the ` +
        "Actions UI. ---",
    );
  }
  return parts.join("\n");
}

/**
 * The second read verb: what CI actually printed when it went red (#1043 lane).
 *
 * A worker that has to fix a red check otherwise has three bad options — guess
 * from the check name, re-run the whole suite locally hoping to reproduce a
 * runner-only failure, or report blocked. This hands it the failing jobs' own
 * failed steps through the tracker read the daemon already uses for its infra
 * classification, so there is exactly one transport for "what did CI say" and
 * the worker reads the same evidence the classifier does.
 *
 * Same resolution as {@link prStatusVerb}: a worker reads its own run's PR and
 * no other; an orchestrator names one. `headSha` is an assertion, never a
 * selector — logs from a superseded commit are worse than no logs, because they
 * describe a failure the current head may not even have.
 *
 * An unreadable log is a refusal, never an empty success. "No failing jobs" and
 * "the log could not be read" invite opposite next moves, and a verb that
 * spelled them the same way would teach a worker to declare a red check fixed
 * because nothing came back.
 */
async function ciLogsVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  const asked = args["prUrl"];
  let prUrl: string;
  let issue: number | undefined;

  if (channel.kind === "run") {
    const own = deps.store.getRun(channel.runId);
    if (own?.prUrl === undefined) {
      return refuse("pr-missing", "refused: this run has no pull request yet, so it has no CI to read.");
    }
    if (typeof asked === "string" && asked !== own.prUrl) {
      return refuse(
        "pr-not-this-run",
        `refused: this socket belongs to the run for #${channel.issue}, whose pull request is ${own.prUrl}. ` +
          "A worker reads only its own run's CI — the daemon resolves the run from the socket.",
      );
    }
    prUrl = own.prUrl;
    issue = channel.issue;
  } else {
    if (typeof asked !== "string") {
      return refuse("malformed-argument", "refused: conductor_ci_logs needs prUrl when it has no run to infer one from.");
    }
    prUrl = asked;
    issue = runForPr(deps, project.name, asked)?.issue;
  }

  const parts = prUrlParts(prUrl);
  if (parts === undefined) {
    return refuse("malformed-argument", `refused: ${prUrl} is not a pull request URL.`, issue);
  }
  const repo = `${parts.owner}/${parts.repo}`;

  const askedHead = args["headSha"];
  const expectedHead = typeof askedHead === "string" ? askedHead : undefined;
  let verification: PrVerification | undefined;
  try {
    // Read mode, exactly as `conductor_pr_status`: a merged or closed PR is a
    // fact worth reading logs against, not the merge gate's `expected OPEN`
    // refusal.
    verification = await deps.tracker.verifyPr(prUrl, expectedHead, { read: true });
  } catch (err) {
    verification = undefined;
    deps.log(`verb ci_logs could not read ${prUrl}: ${err instanceof Error ? err.message : String(err)}`);
  }
  if (verification === undefined) {
    return refuse(
      "head-unresolvable",
      `refused: ${prUrl} could not be read, so there is no head to pin these logs to. Wait and ask again, ` +
        "and report blocked rather than guessing if it stays unreadable.",
      issue,
    );
  }
  if (isHeadMismatch(verification.reason)) {
    return refuse(
      "head-stale",
      `refused: ${verification.reason}. These logs would describe a commit that is no longer the head — ` +
        "read the current head with conductor_pr_status and ask again for that one.",
      issue,
    );
  }

  const runIds = failedRunIds(await deps.tracker.checkConclusions(prUrl));
  if (runIds.length === 0) {
    // Honest and distinguishable from an unreadable log: the check rollup was
    // read and named no non-green run. `verification.status` says which kind of
    // "nothing to diagnose" this is, because a pending head is worth asking
    // again about and a green one is not.
    return allow(
      `${prUrl} at ${verification.headSha}: no failing workflow run to read logs from — ${verification.reason}. ` +
        (verification.status === "pending"
          ? "Checks are still running; poll conductor_pr_status and ask again if one goes red."
          : "Nothing here needs diagnosing."),
      undefined,
      issue,
    );
  }

  const chunks: string[] = [];
  for (const runId of runIds.slice(0, CI_LOGS_MAX_RUNS)) {
    let logs: string[] | undefined;
    try {
      logs = await deps.tracker.runFailedAttemptLogs(
        repo,
        `https://github.com/${repo}/actions/runs/${runId}`,
        CI_LOGS_MAX_ATTEMPTS,
      );
    } catch (err) {
      logs = undefined;
      deps.log(`verb ci_logs could not read run ${runId} of ${repo}: ${err instanceof Error ? err.message : String(err)}`);
    }
    if (logs === undefined) {
      // Fail loudly. The tracker answers `undefined` for a read it could not
      // make *or* for an attempt history past the bound; either way the chunks
      // gathered so far are not the evidence set, and handing them over as if
      // they were is how a worker concludes the wrong thing about a red check.
      return refuse(
        "action-failed",
        `refused: the failing-job logs of ${repo} run ${runId} could not be read (an unreadable job log, a ` +
          `revoked token, or more than ${CI_LOGS_MAX_ATTEMPTS} re-run attempts to gather). That is not "no ` +
          `failures": ${prUrl} is ${verification.status} at ${verification.headSha}. Read the run in the Actions ` +
          "UI, or ask again — this call mutates nothing.",
        issue,
      );
    }
    chunks.push(...logs);
  }

  const rendered = boundCiLogs(chunks);
  const omitted =
    runIds.length > CI_LOGS_MAX_RUNS
      ? ` (${CI_LOGS_MAX_RUNS} of ${runIds.length} non-green workflow runs read)`
      : "";
  if (rendered === "") {
    // Every failed job's failed steps printed nothing readable. Determinate —
    // the reads succeeded — and still not a diagnosis, so it says which run to
    // look at rather than pretending the PR is fine.
    return allow(
      `${prUrl} at ${verification.headSha} is ${verification.status}, but its failing jobs printed no log ` +
        `output${omitted}. Read the run itself: https://github.com/${repo}/actions/runs/${runIds[0]}`,
      undefined,
      issue,
    );
  }

  // The two-red corrective budget is a rule the *worker* keeps, in session,
  // from its brief — nothing in the daemon counts a run's reds, so nothing in
  // the daemon can waive one on the worker's behalf. This reply is therefore
  // the only surface where "that red was the runner, not your diff" can become
  // true, and the brief's promise of an infra exemption is kept here or nowhere.
  //
  // Classified from the chunks already fetched: no second read, no store write.
  // The verdict line goes *above* the log, because a worker that has just been
  // handed 32k of log must not have to reach the end to learn the red was free.
  const classified = chunks.filter((chunk) => chunk.trim() !== "").map((chunk) => runnerInfraFailure(chunk));
  const wordings = new Set(classified.filter((match): match is string => match !== undefined));
  let note = "";
  if (wordings.size > 0 && classified.every((match) => match !== undefined)) {
    note =
      `RUNNER INFRASTRUCTURE: every failing job's own log names the runner, not your diff — ` +
      `${[...wordings].map((w) => `"${w}"`).join(", ")}. This red does not spend one of your two corrective ` +
      "pushes. Do not change the diff for it and do not push a fix: wait, poll conductor_pr_status again, and " +
      "if it is still red stop and report with that wording quoted — the dispatcher re-runs infrastructure " +
      "checks itself and does not charge the attempt.\n\n";
  } else if (wordings.size > 0) {
    // Mixed evidence: the same rule the daemon's own classifier applies — one
    // real failure outranks any number of infra ones. Said out loud rather than
    // left silent, because a worker reading a 429 in job 2 would otherwise
    // claim a waiver the classifier never granted.
    note =
      `MIXED: ${wordings.size} of the failing jobs name runner infrastructure ` +
      `(${[...wordings].map((w) => `"${w}"`).join(", ")}), but at least one failed for a reason that is not ` +
      "infrastructure. A real failure outranks any number of infra ones, so this red counts: fix the " +
      "non-infrastructure failure below.\n\n";
  }
  // Nothing matched: silence. An unclassifiable red is charged, and a reply that
  // hedged here would teach a worker to argue for a waiver from any red log.
  return allow(
    `${prUrl} at ${verification.headSha}: ${verification.status} — ${verification.reason}${omitted}\n\n${note}${rendered}`,
    undefined,
    issue,
  );
}

/**
 * Update one pull request's title and/or body — nothing else.
 *
 * Same resolution as {@link prStatusVerb}: a worker edits its own run's PR (and
 * only that one), an orchestrator names one this project's runs opened.
 * `conductor_pr_update` exists so the floor's own edits go through the same
 * checks and the same ledger as every other mutation, instead of a raw
 * `gh pr edit` from a session that already disclosed it would do exactly that
 * (#176).
 */
async function prUpdateVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  const asked = args["prUrl"];
  let prUrl: string;
  let issue: number | undefined;

  if (channel.kind === "run") {
    const own = deps.store.getRun(channel.runId);
    if (own?.prUrl === undefined) {
      return refuse("pr-missing", "refused: this run has no pull request yet. Open one with conductor_pr_create.");
    }
    if (typeof asked === "string" && asked !== own.prUrl) {
      return refuse(
        "pr-not-this-run",
        `refused: this socket belongs to the run for #${channel.issue}, whose pull request is ${own.prUrl}. ` +
          "A worker edits only its own run's PR.",
      );
    }
    prUrl = own.prUrl;
    issue = channel.issue;
  } else {
    if (typeof asked !== "string") {
      return refuse("malformed-argument", "refused: conductor_pr_update needs prUrl when it has no run to infer one from.");
    }
    const target = runForPr(deps, project.name, asked);
    if (target === undefined) {
      return refuse("pr-not-this-run", `refused: ${asked} is not a pull request any run in ${project.name} opened.`);
    }

    const paused = orchestratorPauseRefusal(deps, channel, target, refuse);
    if (paused !== undefined) return paused;
    prUrl = asked;
    issue = target.issue;
  }

  const title = args["title"];
  const body = args["body"];
  if (!(typeof title === "string" && title !== "") && !(typeof body === "string" && body !== "")) {
    return refuse(
      "malformed-argument",
      "refused: conductor_pr_update changes nothing unless at least one of title or body is a non-empty string.",
    );
  }

  const outcome = await deps.actions.updatePr(prUrl, {
    ...(typeof title === "string" && title !== "" ? { title } : {}),
    ...(typeof body === "string" && body !== "" ? { body } : {}),
  });
  if (!outcome.ok) {
    return refuse("action-failed", `refused: gh could not update the pull request:\n${outcome.stderr}`, issue);
  }
  return allow(`updated ${prUrl}.`, undefined, issue);
}

/**
 * Return one green, run-owned pull request to its worker with blocking
 * findings (#677): the transport that replaces the manual close-PR →
 * comment-on-issue → unblock → continuation dance.
 *
 * Everything is recorded durably BEFORE anything is woken: the findings, the
 * exact reviewed head, the round number, and the target run/session. The wake
 * itself is the daemon's next dispatch pass (the verb has no handle on the
 * worker machinery, and the CLI path must behave exactly like the embedded
 * one), and the run row is reused — same branch, same PR, same attempt, same
 * session directory — so a revision round never creates a new attempt and
 * never touches the failed-attempt or continuation budgets.
 *
 * A second orchestrator axis finishing against the same fixed head before the
 * first revision dispatches folds its finding into the pending revision
 * (#786) rather than being refused: one round, one row, one worker attempt,
 * and the resumed session reads both findings. The fold is only for the same
 * reviewed head; a pending revision at a different head is a review of a
 * moved PR and is refused, as is anything once the revision has dispatched.
 *
 * The duplicate in-flight guard therefore has two halves: while a revision is
 * queued but not yet dispatched, the pending `review_revisions` row folds a
 * same-head finding into itself atomically inside `enqueueReviewRevision` and
 * refuses a different head; once dispatched, the run row itself is `running`
 * (claimed from a revisable terminal state by the dispatch pass), which this
 * verb refuses by name — an already-running worker's prompt is never silently
 * changed.
 */

/**
 * The run states a review revision may start from (#795, #1101).
 *
 * A revision round resumes the exact run whose row owns the PR, so the
 * revisable states are exactly the terminal runs that own the named PR: a
 * settled `pushed-green` row, a `failed` / `killed` row — a run that capped or
 * failed *after* pushing a green PR — and, since #1101, a `stopped` row that
 * had already pushed when the operator stopped it. Ownership is proved by the
 * selection itself (`runsForProjectPr` matches rows that recorded the PR), so
 * a stopped run that never pushed can never reach this gate. Without the
 * stopped leg a stopped-with-green-PR run was both unreviewable (this verb)
 * and unreplaceable (the open-PR dispatch guard) at once — closing the PR by
 * hand was the only exit, which is exactly what #1096 hit.
 *
 * Closed on purpose: a live row (`running` / `claimed`) is already doing its
 * own work, a `pushed-pending` PR is not green yet, and a `blocked` /
 * `orphaned` / `merged` row is not work returned for revision. The set itself
 * is the shared `REVISABLE_RUN_STATES` in `decisions.ts` — the one definition
 * the review verb and the `pr-review-ready` watch gate on, so the two can
 * never drift (#844).
 */

async function prReviewVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  const prUrl = String(args["prUrl"]);
  const headSha = String(args["headSha"]);
  const findings = String(args["findings"]);
  const reason = String(args["reason"]) as ReviewReason;

  // The shared review-readiness gate (#844, #870): the newest project-owned
  // run that recorded the PR — a stopped duplicate is transparent to that
  // selection (#870) — in a revisable state; the same predicate the
  // `pr-review-ready` watch reads, so the verb and the condition resolve
  // ownership and revisability identically. A PR no run of this project
  // opened (or one outside the routed repos) cannot be returned to a worker
  // this project can resume.
  const readiness = prReviewReadiness(deps.store, project.name, prUrl, deps.now());
  if (readiness.kind === "no-owner") {
    return refuse(
      "pr-not-this-run",
      `refused: ${prUrl} is not a pull request any run in ${project.name} opened. ` +
        "A review revision acts on a run-owned PR only.",
    );
  }
  const target = readiness.run;
  if (!prInProjectRouting(project, prUrl)) {
    return refuse(
      "pr-not-this-run",
      `refused: ${prUrl} is not in ${project.name}'s routed repositories ` +
        `(${Object.values(project.routing.repos).map(repoSlugFor).join(", ") || "none"}).`,
      target.issue,
    );
  }
  const issue = target.issue;

  if (findings.trim() === "") {
    return refuse(
      "malformed-argument",
      "refused: conductor_pr_review needs a non-empty findings string — an empty revision returns the worker nothing to fix.",
      issue,
    );
  }

  // The revisable state is a run that pushed a green PR and is not live: a
  // settled `pushed-green` row, or a `failed` / `killed` row — a run that
  // capped or failed after the push (#795). The exact-head verification below
  // still re-reads the PR and insists it is green at the reviewed SHA, so
  // admitting a capped row here never reviews a PR that stopped being green.
  // Any other state is a worker in flight (the original run or an earlier
  // revision) or a run whose PR no longer waits on revision — the detail names
  // the state field that produced the refusal.
  if (readiness.kind !== "ready") {
    const live = target.state === "running" || target.state === "claimed";
    return refuse(
      "review-in-flight",
      `refused: run ${target.id} is ${target.state}, not a settled green run — ${
        live
          ? `a worker (the original run or an earlier revision) is still live on ${prUrl}; wait for it to settle before returning it.`
          : `a review revision starts from a run that pushed a green PR, and this one is ${target.state}.`
      }`,
      issue,
    );
  }

  const paused = orchestratorPauseRefusal(deps, channel, target, refuse);
  if (paused !== undefined) return paused;

  // Exact-head, re-read at request time: the orchestrator reviewed a specific
  // head, and a PR that moved since is not the thing it reviewed. Same verdict
  // the merge gate uses, so "green at headSha" here is the same fact the merge
  // would have required.
  let verification: PrVerification | undefined;
  try {
    verification = await deps.tracker.verifyPr(prUrl, headSha);
  } catch (err) {
    const why = err instanceof Error ? err.message : String(err);
    return refuse(
      "head-unresolvable",
      `refused: the live head of ${prUrl} could not be read (${why}).`,
      issue,
    );
  }
  if (verification === undefined) {
    return refuse(
      "head-unresolvable",
      `refused: the live head of ${prUrl} could not be resolved. Refusing rather than returning a PR you did not re-read.`,
      issue,
    );
  }
  if (verification.status !== "green") {
    const stale = verification.status === "failed" && isHeadMismatch(verification.reason);
    return refuse(
      stale ? "head-stale" : "checks-not-green",
      stale
        ? `refused: ${verification.reason}. You reviewed ${headSha}; that is not what is on the branch now. Re-read the head and re-review.`
        : `refused: ${prUrl} is not green at ${headSha} — ${verification.reason}. A revision is returned to a green PR only.`,
      issue,
    );
  }

  // The hard bound (#678): the round ceiling is the dispatcher's, not the
  // orchestrator's to re-read from prose. A PR may be returned at most
  // `maxRounds` times per lifecycle; at the ceiling this refuses, and the
  // refusal states the two non-actions that replace a further round — leave
  // the PR open, record the unresolved findings and escalate once. The same
  // count `latestReviewRound` drives the renderer's `review-revision N`, so
  // the bound and the visible round can never disagree.
  //
  // A finding that folds into an already-pending revision opens no round, so
  // the ceiling never binds an append — only a genuine new round counts. The
  // store is the single authority on the fold; this read exists only to
  // decide whether the ceiling gates this call.
  const review = resolveReview(project);
  const pendingForRun = deps.store.pendingReviewForRun(project.name, target.id);
  if (pendingForRun === undefined && deps.store.latestReviewRound(project.name, target.id) >= review.maxRounds) {
    // The ceiling no longer ends in an operator task (#873, #932). There are no
    // worker rounds left — that part is unchanged and is the whole point of a
    // ceiling — so what escalates is the AUTOMATION: this admits one bounded
    // adjudication under the project's configured stronger role, and the
    // daemon's next dispatch pass launches it.
    //
    // The head is the one already verified live and green above, so a moved head
    // has refused (`head-stale`) before reaching here: an adjudication is never
    // opened against a head nobody re-read. The findings being submitted travel
    // with it — they are the most relevant single input the adjudicator gets,
    // and dropping them on the floor would make it re-derive (or miss) exactly
    // the concern that provoked the escalation.
    const admitted = deps.store.openReviewAdjudication({
      project: project.name,
      issue,
      prUrl,
      headSha,
      role: review.adjudicator,
      findings,
      requestedAt: deps.now(),
    });
    if (admitted.kind === "refused") {
      return refuse(
        "review-in-flight",
        `refused: an adjudication is already live for ${prUrl} at ${admitted.record.headSha} ` +
          `(${admitted.record.state}), and this call names ${headSha}. That PR moved while it was being ` +
          "adjudicated; wait for the live adjudication to settle before returning a finding at the new head.",
        issue,
      );
    }
    // Idempotent by construction: a repeated call at the same head is told the
    // row it already has, never given a second adjudicator.
    return allow(
      admitted.kind === "created"
        ? `review round ceiling reached (${review.maxRounds}) — recorded a final adjudication for ${prUrl} at ` +
            `${headSha} under the "${review.adjudicator}" model role. The daemon launches it on its next dispatch ` +
            "pass; do not review, merge, close or repair this PR by hand."
        : `an adjudication for ${prUrl} at ${headSha} already exists (${admitted.record.state}) under the ` +
            `"${admitted.record.role}" role — nothing further to record. The daemon owns it from here.`,
      undefined,
      issue,
    );
  }

  // The run row is re-read synchronously, immediately before the durable
  // enqueue and with no await in between: the daemon's dispatch pass
  // (`pushed-green` → `running`) cannot land inside a single synchronous
  // block, so a revision that dispatched while the live-head verification
  // above was in flight is refused here rather than racing the worker that
  // now owns the PR — an already-running worker's prompt is never silently
  // changed.
  const fresh = deps.store.getRun(target.id);
  if (fresh === undefined || REVISABLE_RUN_STATES[fresh.state] !== true) {
    return refuse(
      "review-in-flight",
      `refused: run ${target.id} is no longer revisable on ${prUrl} (now ${fresh?.state ?? "gone"}) — ` +
        "the pending revision has dispatched to a worker. " +
        "Wait for it to settle, then return the finding as its own later review round.",
      issue,
    );
  }

  // Durable, and decided atomically against a concurrent same-head request:
  // the same transaction that records a new round also folds a same-head
  // finding into the one already pending, and the merge never overwrites.
  const round = deps.store.latestReviewRound(project.name, target.id) + 1;
  const enqueued = deps.store.enqueueReviewRevision({
    project: project.name,
    runId: target.id,
    issue,
    prUrl,
    headSha,
    findings,
    round,
    reason,
    ...(fresh.sessionFile === undefined ? {} : { sessionFile: fresh.sessionFile }),
    requestedAt: deps.now(),
  });
  if (enqueued.kind === "refused") {
    return refuse(
      "review-in-flight",
      `refused: a review revision is already pending for ${prUrl} (run ${target.id}) at a different head — ` +
        "amend nothing, because that round reviewed a moved PR. Wait for it to dispatch and settle, then return the finding as its own later review round.",
      issue,
    );
  }
  const revision = enqueued.record;
  return allow(
    enqueued.kind === "created"
      ? `recorded review round ${revision.round} for ${prUrl} at ${headSha}; the daemon will resume run ${target.id}'s session` +
          `${fresh.sessionFile === undefined ? "" : ` (${fresh.sessionFile})`} on its next dispatch pass.`
      : `appended a finding to the pending review round ${revision.round} for ${prUrl} at ${headSha}; ` +
          "the daemon will resume run " +
          `${target.id}'s session${fresh.sessionFile === undefined ? "" : ` (${fresh.sessionFile})`} on its next dispatch pass ` +
          "and the resumed worker reads both findings.",
    undefined,
    issue,
  );
}

/**
 * Record that the durable review findings standing at one exact head are
 * settled, so the #888 merge gate stops refusing that head (#913).
 *
 * The gate it feeds had no reachable clearance at an unchanged head. Its own
 * refusal advised a fresh `conductor_pr_review`, and that provably does not
 * work: `enqueueReviewRevision` either appends the finding to the pending row
 * or inserts a new unsettled one, and both block. The only other clearances
 * are a corrected head — which needs a worker — or a round settling
 * `skipped`/`revised`, which nothing an orchestrator can call does. A PR at
 * the review-round ceiling was therefore permanently unmergeable at that head:
 * a gate that can deadlock a release with no way out.
 *
 * This is a separate verb rather than an argument on the merge, following
 * #850's composition override exactly: a merge call's own `reason` is not an
 * override, because an override that rides inside the act it authorizes leaves
 * no separable record of the judgement. Here the judgement is its own ledger
 * row, its own material event and its own append-only table row.
 *
 * Three gates, and each exists against a specific way of faking it:
 *  - **ownership**, through the same `prReviewReadiness` the review verb uses,
 *    but *without* its revisable-state requirement: the run legitimately sits
 *    `pushed-green` here, and demanding revisability would make the clearance
 *    unreachable for exactly the PRs it exists for;
 *  - **exact head**, re-read live. Green is deliberately NOT required —
 *    clearing findings and passing checks are different questions, and the
 *    merge gate re-checks green itself immediately before merging;
 *  - **non-vacuous**: a clearance at a head where nothing blocks is refused,
 *    so one can never be banked "just in case" against a finding that has not
 *    been written yet. Combined with the store's timestamp bound, a clearance
 *    can only ever dispose of evidence that already existed when it was taken.
 */
async function prReviewClearVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  const prUrl = String(args["prUrl"]);
  const headSha = String(args["headSha"]);
  const reason = String(args["reason"]);

  // Ownership only: the newest project-owned run that recorded this PR. A
  // `not-revisable` readiness is the *expected* answer — the run pushed green
  // and is waiting on the merge — so only `no-owner` refuses here.
  const readiness = prReviewReadiness(deps.store, project.name, prUrl, deps.now());
  if (readiness.kind === "no-owner") {
    return refuse(
      "pr-not-this-run",
      `refused: ${prUrl} is not a pull request any run in ${project.name} opened. ` +
        "A review clearance disposes of evidence recorded against a run-owned PR only.",
    );
  }
  const target = readiness.run;
  if (!prInProjectRouting(project, prUrl)) {
    return refuse(
      "pr-not-this-run",
      `refused: ${prUrl} is not in ${project.name}'s routed repositories ` +
        `(${Object.values(project.routing.repos).map(repoSlugFor).join(", ") || "none"}).`,
      target.issue,
    );
  }
  const issue = target.issue;

  if (reason.trim() === "") {
    return refuse(
      "malformed-argument",
      "refused: conductor_pr_review_clear needs a non-empty reason — the reason IS the audit trail for " +
        "merging a head a review round refused, and an unexplained clearance is indistinguishable from " +
        "the blanket override this verb exists not to be.",
      issue,
    );
  }

  const paused = orchestratorPauseRefusal(deps, channel, target, refuse);
  if (paused !== undefined) return paused;

  // Exact head, read live. `verification.headSha` is the live head read in the
  // same request, so the comparison is against what is on the branch now — not
  // against the status verdict, which is why a red or pending PR is cleared
  // just as well as a green one. The merge gate is what insists on green.
  let verification: PrVerification | undefined;
  try {
    verification = await deps.tracker.verifyPr(prUrl, headSha);
  } catch (err) {
    const why = err instanceof Error ? err.message : String(err);
    return refuse("head-unresolvable", `refused: the live head of ${prUrl} could not be read (${why}).`, issue);
  }
  if (verification === undefined) {
    return refuse(
      "head-unresolvable",
      `refused: the live head of ${prUrl} could not be resolved. Refusing rather than clearing findings ` +
        "against a head nobody re-read.",
      issue,
    );
  }
  if (verification.headSha.toLowerCase() !== headSha.toLowerCase()) {
    return refuse(
      "head-stale",
      `refused: ${prUrl} is at ${verification.headSha}, not ${headSha}. A clearance names the exact head ` +
        "whose findings it disposes of; the branch moved, so those findings no longer stand between this PR " +
        "and a merge. Re-read the head and re-review it.",
      issue,
    );
  }

  // Non-vacuous. Read after the head check on purpose: if the caller named a
  // head that moved, the honest refusal is the stale head, not "nothing to
  // clear".
  const blocking = deps.store.mergeBlockingReviews(project.name, prUrl, headSha);
  if (blocking.length === 0) {
    return refuse(
      "review-clearance-vacuous",
      `refused: nothing blocks merging ${prUrl} at ${headSha.slice(0, 12)} — there is no review evidence ` +
        "at this head to dispose of. A clearance is a disposition of specific findings, never a standing " +
        "pass recorded ahead of them: a later round at this head would not be cleared by it anyway.",
      issue,
    );
  }

  const cleared = deps.store.recordReviewClearance({
    project: project.name,
    prUrl,
    headSha,
    by: "orchestrator",
    reason,
    at: deps.now(),
  });
  deps.store.recordMaterialEvent({
    project: project.name,
    category: "review-clearance",
    summary:
      `#${issue} review findings cleared at ${headSha.slice(0, 12)} ` +
      `(round(s) ${blocking.map((row) => row.round).join(", ")}): ${reason}`,
    evidence: prUrl,
    occurredAt: cleared.at,
    recordedAt: cleared.at,
  });
  return allow(
    `cleared ${blocking.length} blocking review round(s) (${blocking.map((row) => row.round).join(", ")}) ` +
      `for ${prUrl} at ${headSha}: ${reason}. conductor_pr_merge may now proceed at this exact head; every ` +
      "other gate still applies, and a finding recorded after this clearance blocks again.",
    headSha,
    issue,
  );
}

/**
 * The run states a missing-PR recovery may act on (#806).
 *
 * Recovery exists for a run whose worker finished and recorded an exact head
 * but whose PR never materialised — exactly the states the settlement sweep
 * owns: `pushed-pending` (the #777/#809 ghosts) and `pushed-green`. Those
 * rows sit in the store's active set, so the sweep verifies the recovered
 * PR's checks on its next tick (`pushed-pending` → `pushed-green`) and
 * review/merge settlement resumes normally. A terminal row outside this set —
 * failed/killed/stopped/orphaned/blocked — has its own prescribed
 * reconciliation (continuation or review revision, operator stop, restart
 * reconciliation, unblock), and receiving a PR would strand it: nothing would
 * ever verify, review or merge it.
 *
 * Closed on purpose, like {@link REVISABLE_RUN_STATES}: recovery may never
 * widen into a state the daemon's own sweep does not revisit.
 */
const RECOVERABLE_RUN_STATES: Record<string, true> = {
  "pushed-pending": true,
  "pushed-green": true,
};

/** A complete commit SHA, as current workers record it. */
const FULL_HEAD_SHA = /^[0-9a-f]{40}$/i;

/**
 * A commit id a release may pin a tag to (#695): full, or a git-legal
 * abbreviation of at least 7 hex — the same lower bound the recovery path uses,
 * because shorter prefixes are ambiguous. Syntax only; whether the commit
 * exists and is on the released branch is the privileged half's answer, which
 * needs the mirror.
 */
const FULL_OR_ABBREVIATED_SHA = /^[0-9a-f]{7,40}$/i;

/**
 * A legacy abbreviated commit SHA (#836): the `git rev-parse --short` output
 * an older worker generation recorded instead of the full SHA. Git defaults
 * to 7 hex characters and lengthens the abbreviation when 7 is ambiguous, so
 * a legacy record is always at least 7 hex; 4–6 hex are git-legal
 * abbreviations but cannot be legacy records, and at that length the prefix
 * evidence is too weak for an identity recovery to rest on. Anything shorter
 * than git's own abbreviation minimum (4) is not a usable object id at all —
 * ambiguous, and it fails closed.
 */
const ABBREVIATED_HEAD_SHA = /^[0-9a-f]{7,39}$/i;

/**
 * The durable PR field's worker settlement sentinel (#866). A run row whose
 * settlement recorded "no PR observed yet" carries the literal `pending`
 * instead of a URL. It asserts nothing about any GitHub pull request — there
 * is no PR to read — so recovery normalizes it to absence and re-proves the
 * no-duplicate shape before opening a PR, exactly like a row that recorded no
 * URL at all. Querying GitHub with it would only produce the tracker's
 * fail-closed `undefined`, which recovery would misread as an unreadable-but-
 * real PR and refuse as retryable forever.
 */
const PR_PENDING_SENTINEL = "pending";

/**
 * Open (or adopt) the missing pull request for a settled run whose branch was
 * preserved (#806).
 *
 * A worker that pushed and recorded an exact head can still leave the run
 * spent: the PR it claimed never existed (a guessed URL — #779), or the run
 * settled before a PR was ever opened. Nothing else in the verb surface can
 * act then — the mediated PR verbs resolve the PR from a URL some run
 * recorded (`conductor_pr_review` refuses "not a PR any run opened"), the
 * worker-only create verb is gone with the worker, and the package-floor
 * fallback is a fresh clone and re-push, which republishes code. This is the
 * strictly smaller act: publish the pull request the run's work already
 * earned, at the exact head the daemon recorded.
 *
 * Every identity the worker's own `conductor_pr_create` read off its live
 * channel is re-proven here in its absence:
 *  - the run is terminal — no worker is live on it (`claimed`/`running`
 *    refuse, and a merged row is resolved, not recoverable);
 *  - the routed repository is still routed, and the branch comes from the run
 *    row, never from an argument — `branch`/`base`/`headSha` are assertions
 *    exactly as on the worker verb, and a disagreement is a refusal, never a
 *    redirect;
 *  - the remote branch still exists at the recorded identity: a 40-hex head
 *    must equal the live branch head exactly, and a legacy abbreviated head
 *    (7–39 hex, #836) may only canonicalize against the fixed routed branch —
 *    the live branch head must begin with it, and that canonical object is
 *    the identity compared everywhere after, never an argument;
 *  - the issue still wants the work: a closed issue, a merged row or a merged
 *    recorded PR is "resolved" and refuses; a recorded PR a human closed
 *    without merging refuses as `pr-not-open`;
 *  - no PR already represents that head: an open closer on the run's branch
 *    at the recorded head is adopted (idempotent — the row gains the URL it
 *    never recorded, so review/merge verbs can act on it), an open closer at
 *    a different head or on another branch refuses, and a recorded PR that is
 *    still open is that same recovery's answer.
 *
 * The PR describes itself from durable state only — no title/body arguments,
 * because nothing here may publish a headline a session did not already
 * settle. A definitively missing recorded PR (a corroborated 404 — the #779
 * guess) is replaced by the new one; an unreadable recorded PR, unreadable
 * branch head or unreadable closer list fails closed and stays retryable.
 */
async function prRecoverVerb(
  deps: VerbDeps,
  project: ProjectConfig,
  channel: VerbChannel,
  args: Record<string, unknown>,
  refuse: Refuse,
  allow: Allow,
): Promise<Verdict> {
  const ref = issueRefFrom(String(args["issueUrl"]));
  if (ref === undefined) {
    return refuse(
      "malformed-argument",
      `refused: "${String(args["issueUrl"])}" is not a full issue URL. A bare number resolves against whatever ` +
        "repository the daemon happens to be in, which is how a recovered PR lands on the wrong project's issue.",
    );
  }
  if (ref.repo !== project.tracker.repo) {
    return refuse(
      "malformed-argument",
      `refused: ${ref.repo} is not this project's tracker (${project.tracker.repo}).`,
    );
  }

  // The run is the newest attempt for the issue — the row that occupies the
  // issue's slot, and exactly the row `unblock`/`worker stop` name by issue
  // number. Recovery never re-implements dispatch's per-issue choice.
  const target = deps.store.latestRun(project.name, ref.issue);
  if (target === undefined) {
    return refuse(
      "recovery-no-run",
      `refused: ${project.name} has no recorded attempt for #${ref.issue}, so there is no run branch to recover a ` +
        "pull request for. Check the issue's status before trying again.",
    );
  }
  const issue = target.issue;

  if (LIVE_STATES.some((state) => state === target.state)) {
    return refuse(
      "recovery-run-live",
      `refused: run ${target.id} for #${issue} is ${target.state} — a worker is still live on it. ` +
        "Recovery opens the missing PR only for a settled run; wait for it to settle (or stop the worker) first.",
      issue,
    );
  }
  if (target.state === "merged") {
    return refuse(
      "recovery-issue-resolved",
      `refused: run ${target.id} for #${issue} already merged — the work landed and there is no missing PR to recover.`,
      issue,
    );
  }
  if (RECOVERABLE_RUN_STATES[target.state] !== true) {
    const guidance =
      target.state === "failed" || target.state === "killed"
        ? "A failed or killed run's branch continues through `omp-conductor unblock` (continuation), or through a review revision once it owns a PR."
        : target.state === "stopped"
          ? "The operator ended this run; recovery does not reopen stopped work."
          : target.state === "orphaned"
            ? "Restart reconciliation owns orphaned runs."
            : "Answer the run's question and unblock the issue.";
    return refuse(
      "recovery-unsettled-state",
      `refused: run ${target.id} for #${issue} is ${target.state}, a state the settlement sweep does not own — a ` +
        `recovered PR would never be verified, reviewed or merged. ${guidance}`,
      issue,
    );
  }

  const paused = orchestratorPauseRefusal(deps, channel, target, refuse);
  if (paused !== undefined) return paused;

  const routed = Object.values(project.routing.repos).find((repo) => repo.name === target.repo);
  if (routed === undefined) {
    return refuse(
      "recovery-repo-unrouted",
      `refused: run ${target.id} was routed to repository ${target.repo}, which ${project.name} no longer routes ` +
        `(${Object.values(project.routing.repos).map(repoSlugFor).join(", ") || "none"}). ` +
        "A recovery PR cannot be created for a repository the daemon has no entry for; re-route the repo first.",
      issue,
    );
  }

  // The recorded head the preserved branch must still carry. Absent, the
  // recovery cannot prove it would publish this run's work rather than
  // whatever the branch tip happens to be — fail closed on the missing
  // record. Legacy rows recorded git's default abbreviation (`rev-parse
  // --short`, 7+ hex) instead of the full SHA; those stay eligible and are
  // canonicalized against the fixed routed branch below (#836).
  const recordedHead = target.headSha;
  if (recordedHead === undefined) {
    return refuse(
      "recovery-unrecorded-head",
      `refused: run ${target.id} for #${issue} records no head to verify its branch against. ` +
        "Recovery publishes only the head the run itself recorded.",
      issue,
    );
  }
  const isFullHead = FULL_HEAD_SHA.test(recordedHead);
  const isAbbreviatedHead = !isFullHead && ABBREVIATED_HEAD_SHA.test(recordedHead);
  if (!isFullHead && !isAbbreviatedHead) {
    return refuse(
      "recovery-unrecorded-head",
      `refused: run ${target.id} for #${issue} records ${JSON.stringify(recordedHead)}, which is neither a full ` +
        "40-hex SHA nor an unambiguous legacy abbreviation of one (7–39 hex). Recovery publishes only the head " +
        "the run itself recorded.",
      issue,
    );
  }

  // Assertions, never selectors — the mirror of `conductor_pr_create`'s
  // head/base claims. A disagreement means the caller is pointing somewhere
  // the run row does not own; the daemon derives everything from the row.
  const claimedBranch = args["branch"];
  if (typeof claimedBranch === "string" && claimedBranch !== target.branch) {
    return refuse(
      "ref-not-run-branch",
      `refused: run ${target.id}'s recorded branch is ${target.branch}, and you named "${claimedBranch}". ` +
        "Recovery derives the head branch from the run; it cannot be pointed at another branch.",
      issue,
    );
  }
  const claimedBase = args["base"];
  if (typeof claimedBase === "string" && claimedBase !== routed.defaultBranch) {
    return refuse(
      "base-not-default-branch",
      `refused: ${routed.name} targets ${routed.defaultBranch}, and you named "${claimedBase}". ` +
        "The base is the configured defaultBranch, not an argument.",
      issue,
    );
  }
  const claimedHead = args["headSha"];
  if (typeof claimedHead === "string" && claimedHead.toLowerCase() !== recordedHead.toLowerCase()) {
    return refuse(
      "recovery-head-mismatch",
      `refused: run ${target.id} recorded head ${recordedHead}, and you asserted "${claimedHead}". ` +
        "The exact recorded head is the daemon's, not a selector.",
      issue,
    );
  }

  // Closed issues are resolved issues: a merge closed it, or a human closed it
  // without merging. Either way it does not want a recovered PR — and an
  // unreadable state is not a green light: "could not tell" must never open a
  // PR beside (or for) an issue that may already be closed, so it fails
  // closed and stays retryable like every sibling read in this verb.
  let issueState: IssueState | undefined;
  try {
    issueState = await deps.tracker.issueState(issue);
  } catch (err) {
    issueState = undefined;
    deps.log(`conductor_pr_recover could not read #${issue} state: ${err instanceof Error ? err.message : String(err)}`);
  }
  if (issueState === undefined) {
    return refuse(
      "issue-state-unreadable",
      `refused: the tracker could not say whether #${issue} is open. An unreadable state is not a green light — ` +
        "retry before recovering, so a recovered PR is never opened beside (or for) a closed issue.",
      issue,
    );
  }
  if (issueState === "closed") {
    return refuse(
      "recovery-issue-resolved",
      `refused: #${issue} is closed. The work already merged or a human declined it; recovery does not reopen either.`,
      issue,
    );
  }

  // The branch/head verification: the routed repository's branch, read live,
  // must still sit at the exact head this run recorded. This is the proof the
  // PR to be created publishes verified work and not something that replaced
  // it. An unreadable branch stays retryable — that is "could not tell", never
  // a green light.
  const identity = repoSlugFor(routed);
  let liveBranchHead: string | undefined;
  try {
    liveBranchHead = await deps.tracker.branchHead(identity, target.branch);
  } catch (err) {
    liveBranchHead = undefined;
    deps.log(
      `conductor_pr_recover could not read ${identity}/${target.branch} head: ${err instanceof Error ? err.message : String(err)}`,
    );
  }
  if (liveBranchHead === undefined) {
    return refuse(
      "head-unresolvable",
      `refused: the live head of ${identity}/${target.branch} could not be read. That is a transient or unknown ` +
        "state, not a green light — retry before recovering, and check the branch still exists.",
      issue,
    );
  }

  // The canonical object, decided from the fixed routed branch only (#836). A
  // full 40-hex record keeps its exact-identity behavior: the branch must
  // still sit at that object. A legacy abbreviated record can never be used
  // as an identity on its own — the live branch head must literally begin
  // with the abbreviation, and that full head becomes the identity every
  // later check compares against (PR adoption, no-duplicate proof, the PR's
  // own description). Nothing resolves the abbreviation anywhere else, and
  // never from caller input: `headSha` remains an assertion of the recorded
  // value, so no argument can appoint a different object.
  const canonicalHead: string = isFullHead ? recordedHead : liveBranchHead;
  if (isAbbreviatedHead) {
    if (!FULL_HEAD_SHA.test(liveBranchHead)) {
      return refuse(
        "head-unresolvable",
        `refused: the live head of ${identity}/${target.branch} is ${liveBranchHead}, not a full commit SHA, so the ` +
          `recorded abbreviation ${recordedHead} cannot be canonicalized against it. That is an unknown state, not a ` +
          "green light — retry before recovering.",
        issue,
      );
    }
    if (!liveBranchHead.toLowerCase().startsWith(recordedHead.toLowerCase())) {
      return refuse(
        "recovery-head-mismatch",
        `refused: ${identity}/${target.branch} now points at ${liveBranchHead}, which does not begin with the ` +
          `recorded run head ${recordedHead}. A legacy abbreviated head canonicalizes only against a branch that ` +
          "still carries it; the branch was replaced, force-updated or continued elsewhere — a PR created now would " +
          "publish work the daemon did not verify at that head. Do not force; reconcile what the branch should be first.",
        issue,
      );
    }
  } else if (liveBranchHead.toLowerCase() !== recordedHead.toLowerCase()) {
    return refuse(
      "recovery-head-mismatch",
      `refused: ${identity}/${target.branch} now points at ${liveBranchHead}, not the recorded run head ${recordedHead}. ` +
        "The branch was replaced, force-updated or continued elsewhere — a PR created now would publish work the daemon " +
        "did not verify at that head. Do not force; reconcile what the branch should be first.",
      issue,
    );
  }

  // The no-duplicate-proof: any OPEN PR that already closes the issue. The
  // run's own branch at the recorded head is the run's PR (whoever opened it —
  // the operator's manual recovery is PR #805 for run #777) and is adopted,
  // idempotently; the same branch at a different head, or another branch with
  // the issue, each refuse rather than duplicate.
  let closers: OpenCloser[];
  try {
    closers = await deps.tracker.openClosersFor(issue);
  } catch (err) {
    const why = err instanceof Error ? err.message : String(err);
    return refuse(
      "open-pr-lookup-error",
      `refused: the tracker could not say whether #${issue} already has an open pull request (${why}). ` +
        "Refusing rather than opening a second one.",
      issue,
    );
  }
  const sameBranch = closers.find(
    (closer) => closer.headRefName !== "" && closer.headRefName === target.branch && closer.repo === identity,
  );
  if (sameBranch !== undefined) {
    let live: string | undefined;
    try {
      live = await deps.tracker.prHead(sameBranch.url);
    } catch (err) {
      live = undefined;
      deps.log(
        `conductor_pr_recover could not read ${sameBranch.url} head: ${err instanceof Error ? err.message : String(err)}`,
      );
    }
    if (live === undefined) {
      return refuse(
        "head-unresolvable",
        `refused: ${sameBranch.url} is open on ${target.branch}, but its head could not be read. ` +
          "Retry before deciding whether to adopt it or treat it as a conflict.",
        issue,
      );
    }
    if (live.toLowerCase() === canonicalHead.toLowerCase()) {
      const already = deps.store.getRun(target.id)?.prUrl === sameBranch.url;
      if (!already) deps.store.updateRun(target.id, { prUrl: sameBranch.url });
      return allow(
        `recovered ${sameBranch.url} — an open pull request already exists at run head ${canonicalHead} ` +
          `(${target.branch}); ${
            already
              ? "it is already this run's recorded PR"
              : `recorded it on run ${target.id} so review and merge verbs can act on it`
          }.`,
        undefined,
        issue,
      );
    }
    return refuse(
      "recovery-head-mismatch",
      `refused: ${sameBranch.url} is open on ${target.branch} but at ${live}, not the recorded run head ${canonicalHead}. ` +
        "That open PR superseded the run's recorded head; a second PR would duplicate it.",
      issue,
    );
  }
  if (closers.length > 0) {
    const other = closers[0]!;
    return refuse(
      "open-pr-exists",
      `refused: ${other.url} is already open and already closes #${issue} (head ${other.headRefName || "unknown"} in ${
        other.repo || "unknown"
      }). Recovery opens a PR only for the run's own preserved branch, and this issue already has open work.`,
      issue,
    );
  }

  // The durable PR field is normalized and validated before any recorded-PR
  // lookup (#866). A real recorded PR is a pull-request URL the tracker can
  // address — the `prUrlParts` shape, the same guard the tracker's own
  // `prState` applies. The `pending` settlement sentinel is not a URL: it
  // records the run believed a PR was in flight but never observed one, so it
  // is normalized to absence and recovery proceeds like a run with no
  // recorded PR. Any other non-URL value is corruption — there is nothing for
  // the tracker to read, and had it been queried the fail-closed `undefined`
  // would read as an unreadable-but-real PR that can never clear — so it
  // refuses by name instead of hanging the run on a retry.
  const recordedPrUrl = target.prUrl;
  const pendingSentinel = recordedPrUrl === PR_PENDING_SENTINEL;
  if (recordedPrUrl !== undefined && !pendingSentinel && prUrlParts(recordedPrUrl) === undefined) {
    return refuse(
      "recorded-pr-corrupt",
      `refused: run ${target.id} for #${issue} records prUrl ${JSON.stringify(recordedPrUrl)}, which is not a pull ` +
        "request URL. A non-URL durable prUrl is a corrupt run record, not an unreadable PR — recovery will not " +
        "guess a GitHub pull request from it. Clear or correct the run's prUrl, then retry.",
      issue,
    );
  }

  if (!pendingSentinel && target.prUrl !== undefined) {
    // The run believes it has a PR. Before creating one, the recorded URL is
    // decided: still open → it is the recovery's answer (idempotent); merged →
    // the work landed; closed → a human declined it; definitively missing →
    // the #779 guess, replaced by a real PR below.
    let recorded: PrState | "missing" | undefined;
    try {
      recorded = await deps.tracker.prState(target.prUrl);
    } catch (err) {
      recorded = err instanceof GhPrMissingError ? "missing" : undefined;
      if (recorded !== "missing") {
        deps.log(
          `conductor_pr_recover could not read recorded PR ${target.prUrl}: ${err instanceof Error ? err.message : String(err)}`,
        );
      }
    }
    if (recorded === "open") {
      // An open recorded PR is the recovery's answer only when it actually
      // represents the recorded head — otherwise it is some other PR the row
      // once believed in, and answering "already open, nothing to create"
      // would leave the run's own head unpublished.
      let recordedLive: string | undefined;
      try {
        recordedLive = await deps.tracker.prHead(target.prUrl);
      } catch (err) {
        recordedLive = undefined;
        deps.log(
          `conductor_pr_recover could not read ${target.prUrl} head: ${err instanceof Error ? err.message : String(err)}`,
        );
      }
      if (recordedLive === undefined) {
        return refuse(
          "head-unresolvable",
          `refused: ${target.prUrl} is open, but its head could not be read. ` +
            "Retry before deciding whether it is this run's PR or another one.",
          issue,
        );
      }
      if (recordedLive.toLowerCase() === canonicalHead.toLowerCase()) {
        return allow(
          `recovered ${target.prUrl} — this run's recorded pull request is already open at the recorded head; nothing to create.`,
          undefined,
          issue,
        );
      }
      return refuse(
        "recovery-head-mismatch",
        `refused: the run records ${target.prUrl}, which is open at ${recordedLive}, not the recorded run head ${canonicalHead}. ` +
          "That PR is not this run's head; recovery cannot claim it. Match the PR to the head by hand, or close it and recover afresh.",
        issue,
      );
    }
    if (recorded === "merged") {
      return refuse(
        "recovery-issue-resolved",
        `refused: the run's recorded PR ${target.prUrl} already merged — the work landed; there is no missing PR to recover.`,
        issue,
      );
    }
    if (recorded === "closed") {
      return refuse(
        "pr-not-open",
        `refused: the run's recorded PR ${target.prUrl} was closed without merging. ` +
          "Recovery does not resurrect a PR a human declined; reopen it by hand if that was a mistake.",
        issue,
      );
    }
    if (recorded === undefined) {
      return refuse(
        "head-unresolvable",
        `refused: the run records PR ${target.prUrl}, but its state could not be read. ` +
          "Retry rather than creating a second PR next to one that may exist.",
        issue,
      );
    }
    // `missing`: the recorded URL provably never existed (a #779 guess). Fall
    // through and replace it with a real PR at the verified head.
  }

  // The lane gate (#925), and the last check before anything is published.
  //
  // #899 released a worker-free branch's files: its worker is finished, nothing
  // is writing through it, and holding its whole diff forever idled the fleet.
  // The consequence is that by the time recovery runs, another worker may
  // legitimately hold those same files — and publishing this frozen branch then
  // puts an older diff into review, and into the merge queue, beside a live
  // worker still rewriting it. That is the one reactivation nothing else
  // catches: the merge gate reads checks and heads, never who else is mid-edit.
  //
  // This half fails CLOSED, unlike admission's interlock, and the asymmetry is
  // the point. Admission runs unattended on every tick, so refusing there on an
  // unreadable probe would stall the queue on a transient. Recovery is one
  // deliberate call whose refusal is legible and retryable: it costs a retry and
  // names the file, the holder and when to try again. Between "publish
  // silently on an unproven lane" and "say what could not be read", only the
  // second is defensible for a call that ends in a public pull request.
  const leases = deps.store.leasedRuns(project.name).filter((run) => run.repo === target.repo && run.issue !== issue);
  if (leases.length > 0) {
    const base = routed.defaultBranch;
    const mirror = join(project.mirrorRoot, `${target.repo}.git`);
    const laneOf = async (run: RunRecord): Promise<LaneFile[]> =>
      deps.lane.probeRunLane({
        worktree: run.worktree,
        baseRef: `refs/remotes/origin/${base}`,
        ...(run.branch === "" ? {} : { branchRef: `refs/heads/${run.branch}` }),
        mirror,
      });
    let carried: LaneFile[];
    try {
      carried = await laneOf(target);
    } catch (err) {
      return refuse(
        "recovery-lane-unprovable",
        `refused: the files preserved branch ${target.branch} carries could not be read ` +
          `(${err instanceof Error ? err.message : String(err)}), and ${leases.length} live mutation lease(s) are ` +
          `writing in ${identity}. Recovery will not publish a branch whose overlap with live work is unknown — ` +
          "retry, or wait for those runs to settle.",
        issue,
      );
    }
    // A branch that carries nothing cannot collide with anything: no read of
    // the leaseholders is needed, and none is made.
    if (carried.length > 0) {
      const wanted = new Set(carried.map(({ file }) => file));
      for (const run of leases) {
        let held: LaneFile[];
        try {
          held = await laneOf(run);
        } catch (err) {
          return refuse(
            "recovery-lane-unprovable",
            `refused: the lane of live run ${run.id} (#${run.issue}, ${run.state}) could not be read ` +
              `(${err instanceof Error ? err.message : String(err)}), so recovery cannot prove ${target.branch} does ` +
              "not overlap work in flight. Retry, or wait for that run to settle.",
            issue,
          );
        }
        // The run's own persisted declaration counts too: a leaseholder admitted
        // on a lane it has not written yet still owns those files, exactly as
        // admission's occupancy reads it (#744).
        const occupies = [
          ...held.map(({ file, source }) => ({ file, source: source as LaneSource | "declared" })),
          ...(run.lane?.files ?? []).map((file) => ({ file, source: "declared" as const })),
        ];
        const collision = occupies.find(({ file }) => wanted.has(file));
        if (collision !== undefined) {
          return refuse(
            "recovery-lane-occupied",
            `refused: ${collision.file} is held by run ${run.id} for #${run.issue} (${run.state}, ${collision.source}), ` +
              `and preserved branch ${target.branch} carries it. Publishing now would put this frozen diff into ` +
              "review beside a worker still rewriting that file. The lease releases when that run settles — recover " +
              "then, or land its work first.",
            issue,
          );
        }
      }
    }
  }

  // The PR's own voice comes from durable state only: the issue title and the
  // run's settlement report. The closing keyword must name the TRACKER
  // repository — the issue lives there, not in the routed code repo, and
  // `openClosersFor` asks the tracker's GraphQL `closedByPullRequestsReferences`
  // whether an OPEN PR already closes it. A keyword aimed at the code repo
  // would close an unrelated same-numbered issue there, leave the tracker
  // issue open, and hide this PR from the duplicate guard on the next call.
  let issueTitle: string | undefined;
  try {
    issueTitle = (await deps.tracker.getIssue(issue))?.title;
  } catch (err) {
    issueTitle = undefined;
    deps.log(`conductor_pr_recover could not read #${issue} title: ${err instanceof Error ? err.message : String(err)}`);
  }
  const body = [
    `Closes ${project.tracker.repo}#${issue}`,
    "",
    `Recovered pull request for run ${target.id} (attempt ${target.attempt}).`,
    `The run ended ${target.state} with no pull request; this PR opens its preserved branch ${target.branch} at the ` +
      `exact recorded head ${canonicalHead} through the orchestrator's conductor_pr_recover verb.`,
    ...(target.report === undefined ? [] : ["", "Worker report:", "", target.report]),
  ].join("\n");
  // The worktree of a settled run is long removed; `openRunPr` resolves
  // everything from the remote identities (repo slug, branch, base) and never
  // opens the path — it is carried only to satisfy the shared action surface.
  const outcome = await deps.actions.createPr(
    { repo: routed, runRepoPath: target.worktree, branch: target.branch },
    {
      title: issueTitle ?? `recover ${target.branch} (${project.tracker.repo} #${issue})`,
      body,
      base: routed.defaultBranch,
    },
  );
  if (!outcome.ok) {
    return refuse("action-failed", `refused: gh could not open the recovered pull request:\n${outcome.stderr}`, issue);
  }
  deps.store.updateRun(target.id, { prUrl: outcome.url });
  return allow(
    `opened ${outcome.url} for ${target.branch} at ${canonicalHead} (${identity} → ${routed.defaultBranch}), ` +
      "recorded on the run so review/merge settlement resumes.",
    undefined,
    issue,
  );
}

// ------------------------------------------------------------------ the listener

export interface VerbListener {
  path: string;
  ownership: SocketOwnership;
  /**
   * Name the process this channel belongs to.
   *
   * Called with the child's pid the moment the daemon has spawned it, and the
   * channel refuses every connection until it is: the socket has to exist before
   * the child can connect to it, so there is a window between `listen` and the
   * spawn, and a session that connected in that window would be one nobody had
   * identified. `Bun.spawn` returns the pid synchronously, so the window closes
   * before the child has finished booting.
   */
  bindPid(pid: number): void;
  close(): Promise<void>;
}

/**
 * Refusing to bind is refusing to dispatch. Thrown rather than returned so no
 * caller can accidentally continue past it into a run with no boundary.
 */
export class VerbSocketRefusal extends Error {
  constructor(message: string) {
    super(message);
    this.name = "VerbSocketRefusal";
  }
}

export interface ListenOptions {
  peerReader?: PeerReader;
  daemonUid?: number;
  /**
   * The identity the channel is secured to — the worker account for a worker
   * run channel (#798). When set, the bound socket is chowned to that uid/gid
   * (the worker child, and only it, must be able to connect) and the peer
   * verdict expects that uid on the wire. Absent, the channel stays
   * daemon-owned and daemon-uid as today (orchestrator, tests).
   */
  channelOwner?: { uid: number; gid: number };
  /** Injected in tests; the real one chowns, which needs privilege. */
  secure?: typeof secureBoundSocket;
}

/**
 * Bind one channel's socket, after proving nobody else could have tampered with
 * where it is going.
 *
 * The validation is not defensive tidiness: a socket under a directory a third
 * party can write is a socket a third party can unlink and re-bind, and the
 * daemon would then be handing a run's verbs to whoever won that race. A failed
 * check refuses dispatch rather than degrading, because the degraded mode here
 * is "no boundary at all, silently".
 */
export async function listenVerbChannel(
  deps: VerbDeps,
  channel: VerbChannel,
  opts: ListenOptions = {},
): Promise<VerbListener> {
  const daemonUid = opts.daemonUid ?? process.getuid?.() ?? 0;
  const problem = validateSocketPath(channel.path, daemonUid);
  if (problem !== undefined) {
    throw new VerbSocketRefusal(
      `dispatch refused: ${problem.message}. The verb socket layout requires every component to be ` +
        "owned by the daemon (or root), free of symlinks, and unwritable by anyone else.",
    );
  }
  // Only the daemon ever unlinks these, and it has just proven the parent is
  // not writable by anyone else — which is what makes unlink-then-bind safe
  // here and a race in a world-writable directory.
  unlinkStaleSocket(channel.path);

  // Unbound until the daemon names the child. Read through a closure rather than
  // captured by value so the connection handler sees the binding that exists when
  // a connection arrives, not the one that existed when the server was created.
  let boundPid: number | undefined;

  const expectedUid = opts.channelOwner?.uid ?? process.getuid?.() ?? 0;

  const server: Server = createServer((socket) => {
    handleConnection(deps, channel, socket, opts.peerReader, () => boundPid, expectedUid);
  });
  server.on("error", (err) => {
    deps.log(`verb socket ${channel.path} errored: ${err.message}`);
  });

  await new Promise<void>((resolve, reject) => {
    server.once("error", reject);
    server.listen(channel.path, () => {
      server.off("error", reject);
      resolve();
    });
  });

  // A worker-run channel belongs to the worker: the socket is chowned to it so
  // the session child can connect, and the ownership line in the startup banner
  // says so rather than guessing.
  const secure = opts.secure ?? ((path: string) => secureBoundSocket(path, { owner: opts.channelOwner }));
  const ownership = secure(channel.path);

  return {
    path: channel.path,
    ownership,
    bindPid: (pid) => {
      boundPid = pid;
    },
    close: async () => {
      await new Promise<void>((resolve) => {
        server.close(() => {
          resolve();
        });
      });
      unlinkStaleSocket(channel.path);
    },
  };
}

function handleConnection(
  deps: VerbDeps,
  channel: VerbChannel,
  socket: Socket,
  peerReader: PeerReader | undefined,
  boundPid: () => number | undefined,
  expectedUid: number,
): void {
  const fd = socketFd(socket);
  const peer = fd === undefined || peerReader === undefined ? undefined : peerReader(fd);
  const expectedPid = boundPid();
  // Refused until the daemon has named the process this channel belongs to.
  //
  // The socket must exist before the child can connect to it, so there is a real
  // window between `listen` and the spawn — and any session already running shares
  // this daemon's uid, so one of them could reach a freshly bound orchestrator
  // channel in that window and be authorised as the orchestrator. Nothing about
  // the socket distinguishes them; only the pid does (#163).
  const verdict: PeerVerdict =
    expectedPid === undefined
      ? {
          ok: false,
          peerUid: peer?.uid ?? -1,
          expectedUid,
          detail:
            "this channel has not been bound to a process yet, so the caller cannot be identified — " +
            "this socket is 0600 and its owner is the only uid that can reach it",
        }
      : peerVerdict({ uid: expectedUid, pid: expectedPid }, peer);
  if (!verdict.ok) {
    // Not a client error, and deliberately not answered: a caller who is not
    // who the socket was allocated to gets no reply to calibrate against.
    deps.log(
      `IMPERSONATION on ${channel.path}: ${verdict.detail}. Connection closed without a reply. ` +
        `This socket belongs to ${channel.kind === "run" ? `run ${channel.runId} (#${channel.issue})` : "the orchestrator"} in ${channel.project}.`,
    );
    // `end`, not `destroy`: the caller gets no reply either way, but a clean
    // close means it learns *now* instead of sitting out its whole timeout. An
    // impersonation attempt that also hangs the honest client behind it is a
    // denial of service the refusal handed out for free.
    socket.end();
    return;
  }

  socket.setTimeout(REQUEST_TIMEOUT_MS, () => {
    socket.destroy();
  });

  let buffer = "";
  let answered = false;
  const answer = async (raw: unknown): Promise<void> => {
    if (answered) return;
    answered = true;
    const reply = await handleVerbCall(deps, channel, raw);
    socket.end(`${JSON.stringify(reply)}\n`);
  };

  socket.on("data", (chunk) => {
    if (answered) return;
    buffer += chunk.toString("utf8");
    if (buffer.length > MAX_REQUEST_BYTES) {
      answered = true;
      socket.end(
        `${JSON.stringify({ ok: false, refusal: "malformed-argument", text: `refused: a verb request may not exceed ${MAX_REQUEST_BYTES} bytes.` })}\n`,
      );
      return;
    }
    const newline = buffer.indexOf("\n");
    if (newline < 0) return;
    void answer(parseJson(buffer.slice(0, newline)));
  });
  socket.on("end", () => {
    if (!answered && buffer.trim().length > 0) void answer(parseJson(buffer));
  });
  socket.on("error", () => {
    socket.destroy();
  });
}

/** A body that is not JSON is `undefined`, which the parser refuses by name. */
function parseJson(text: string): unknown {
  try {
    return JSON.parse(text);
  } catch {
    return undefined;
  }
}