/**
 * The durable seam between the detached fleet installer and the daemon that
 * verifies it (#486).
 *
 * `omp-conductor upgrade` has always been a transaction *inside* one process:
 * it paused, drained, installed, reloaded and verified, and the process that
 * verified was the process that restarted the fleet. The fleet-installs-itself
 * path replaces that wrapper, not the transaction: a detached transient unit
 * performs the install while `upgrade-journal.ts` records every surface it
 * touches, and the first tick of the post-restart daemon verifies against the
 * journal — a process that did not run the install, which is the only witness
 * worth trusting after the restarts.
 *
 * This module is a leaf on purpose. The daemon's privileged tick and the
 * upgrade engine both need the verifier, the request classification and the
 * transient-unit naming, and they already import each other — `upgrade.ts`
 * takes its pause helpers from `daemon.ts`, and the daemon runs the engine's
 * rollback. Putting the shared seams here instead of importing `upgrade.ts`
 * from the daemon keeps that graph acyclic; everything the verifier touches
 * that belongs to a host state is injected through `UpgradeVerifyDeps`, so the
 * only runtime imports are leaves (journal, config, store).
 */

import { readFileSync } from "node:fs";
import { loadConfig, stateDir } from "./config.ts";
import { DEFAULT_PORT } from "./lifecycle.ts";
import { dbPath, openStore } from "./store.ts";
import { appendJournal, readUpgradeJournal, upgradeJournalPath, type JournalSurfaces, type UpgradeCheck, type UpgradeJournalEntry } from "./upgrade-journal.ts";
import type { DoctorReport } from "./doctor.ts";
import type { FleetLayers } from "./status-render.ts";
import type { ReportDraft, ReportKind } from "./types.ts";

/** One command run to completion with captured output. The transient unit, the
 *  post-restart verifier and the rollback unit all run processes this way —
 *  one named construction site, so the seam tests can inject it everywhere. */
export interface UpgradeCommandResult {
  code: number;
  stdout: string;
  stderr: string;
}

export async function runCommand(
  command: string,
  args: readonly string[],
  /** Where to run it. Absent means this process's cwd, which is what every
   *  step but the bootstrap's source checks wants (#908). */
  cwd?: string,
): Promise<UpgradeCommandResult> {
  try {
    const child = Bun.spawn([command, ...args], {
      stdin: "ignore",
      stdout: "pipe",
      stderr: "pipe",
      env: process.env,
      ...(cwd === undefined ? {} : { cwd }),
    });
    const stdout = new Response(child.stdout).text();
    const stderr = new Response(child.stderr).text();
    const code = await child.exited;
    return { code, stdout: await stdout, stderr: await stderr };
  } catch (err) {
    return { code: 127, stdout: "", stderr: err instanceof Error ? err.message : String(err) };
  }
}

/**
 * The project a host-wide outcome report is filed under: the first in-scope
 * project, falling back to the configured first project, then the constant.
 */
export function reportProject(selectors: readonly (string | undefined)[]): string {
  for (const selector of selectors) {
    if (selector !== undefined) return selector;
  }
  try {
    const projects = loadConfig().projects;
    if (projects.length > 0) return projects[0]!.name;
  } catch {
    // no config: the fallback below stands
  }
  return "conductor";
}

/**
 * Hand a durable outbox row over in the detached context, where no tick is
 * alive to do it. The daemon's report outbox owns delivery from here (bounded
 * retries via the same store rows `omp-conductor report` writes), so a report
 * survives the process that enqueued it. Refusal to queue is logged and lost
 * — the journal on disk is the harder of the two records and is always kept.
 */
export function enqueueUpgradeReport(
  kind: ReportKind,
  project: string,
  summary: string,
  lines: string[],
  dedupeKey: string,
): void {
  try {
    const store = openStore(dbPath());
    try {
      store.enqueueReport({
        project,
        kind,
        body: [summary, "", ...lines].join("\n"),
        at: Date.now(),
        dedupeKey,
      });
    } finally {
      store.close();
    }
  } catch (err) {
    // logged by nothing: the journal carries the same outcome, and `status`
    // surfaces the reports block. A full disk must not loop here.
    process.stderr.write(`upgrade report could not be queued: ${err instanceof Error ? err.message : String(err)}\n`);
  }
}

// ---------------------------------------------------------------------------
// Classifying the journal
// ---------------------------------------------------------------------------

/** The state a pending request has reached, read off the journal. */
export type PendingUpgradeState =
  | { kind: "none" }
  | { kind: "closed"; phase: string }
  | { kind: "no-start" }
  | { kind: "awaiting-verification" }
  | { kind: "incomplete" }
  | { kind: "failed" };

/** Index of the newest request line, or -1 when none exists. */
function newestRequestIndex(entries: readonly UpgradeJournalEntry[]): number {
  for (let i = entries.length - 1; i >= 0; i--) {
    if (entries[i]!.kind === "request") return i;
  }
  return -1;
}

/** Newest terminal outcome in one request's progression, without copying it. */
function newestOutcome(
  progression: readonly UpgradeJournalEntry[],
): UpgradeJournalEntry | undefined {
  for (let i = progression.length - 1; i >= 0; i--) {
    const entry = progression[i]!;
    if (entry.kind === "outcome") return entry;
  }
  return undefined;
}

/**
 * Classify one journal: no request, a request whose outcome is already
 * terminal, or a request still in flight — with the phase evidence that says
 * how far it got. An outcome is terminal: the verifier writes it last, after
 * the durable report is already queued, so a later tick reads it and does
 * nothing rather than re-reporting.
 *
 * The journal is append-only across upgrades, so the newest request is the one
 * under test: an older request's terminal outcome must not close or shadow a
 * newer request, and only the newest request's progression decides its state.
 */
export function classifyUpgrade(
  entries: readonly UpgradeJournalEntry[],
): PendingUpgradeState {
  const start = newestRequestIndex(entries);
  if (start === -1) return { kind: "none" };
  const progression = entries.slice(start);
  const outcome = newestOutcome(progression);
  if (outcome !== undefined) {
    return { kind: "closed", phase: outcome.phase ?? "outcome" };
  }
  const phases = progression.filter((entry) => entry.kind === "phase");
  if (phases.length === 0) return { kind: "no-start" };
  if (phases.some((entry) => entry.phase === "awaiting-verification" && entry.ok)) {
    return { kind: "awaiting-verification" };
  }
  if (phases.some((entry) => entry.ok === false)) {
    return { kind: "failed" };
  }
  return { kind: "incomplete" };
}

/** Terminal outcomes that prove no broken or mixed install remains active. */
const RESUMABLE_OUTCOMES = new Set([
  "already-current",
  "recovered",
  "recovery-acknowledged",
  "verified",
]);
/** Outcomes that require a durable host-wide safety fence. */
const RECOVERY_HOLD_OUTCOMES = new Set([
  "rollback-failed",
  "rollback-requested",
  "rollback-unavailable",
  "rolled-back",
]);
/** Closed rollback outcomes whose installed surfaces must be re-verified. */
const RECOVERY_CHECK_OUTCOMES = new Set(["rollback-failed", "rolled-back"]);

/** One pending request plus the snapshot its verifier needs. */
export interface PendingUpgradeRequest {
  version: string;
  gitHead?: string;
  pauseSince?: number;
  initialPaused?: boolean;
  pauseKey?: string;
  selectors: readonly (string | undefined)[];
  installHolder?: UpgradeJournalEntry["installHolder"];
  installProjects?: readonly string[];
  initial?: UpgradeJournalEntry["initial"];
  configBackup?: string;
  previous?: JournalSurfaces;
  /**
   * Epoch-ms when the install began (the journal snapshot's write time): the
   * deadline the live orchestrator session must have restarted after (#832).
   */
  reloadAfterMs?: number;
}

/** The newest request plus everything journaled after it: the view one
 *  transaction owns. Older closed upgrades stay in the journal but do not
 *  contribute identity or phases to the one in flight. */
export function requestProgression(
  entries: readonly UpgradeJournalEntry[],
): readonly UpgradeJournalEntry[] | undefined {
  const start = newestRequestIndex(entries);
  if (start === -1) return undefined;
  return entries.slice(start);
}

export function pendingUpgradeRequest(
  entries: readonly UpgradeJournalEntry[],
): PendingUpgradeRequest | undefined {
  const progression = requestProgression(entries);
  if (progression === undefined) return undefined;
  const request = progression[0]!;
  if (request.ok === false) return undefined;
  const snapshot = progression.find((entry) => entry.kind === "snapshot");
  const pausePhase = progression.find(
    (entry) => entry.kind === "phase" && entry.phase === "paused",
  );
  const snapshotAt = snapshot === undefined ? undefined : Date.parse(snapshot.at);
  return {
    version: request.version ?? "unknown",
    gitHead: request.gitHead,
    initialPaused: snapshot?.initialPaused,
    pauseKey: snapshot?.pauseKey,
    selectors: (snapshot?.selectors ?? request.selectors ?? []).map((selector) => selector ?? undefined),
    pauseSince: pausePhase?.pauseSince,
    initial: snapshot?.initial,
    configBackup: snapshot?.configBackup,
    previous: snapshot?.previous,
    installHolder: request.installHolder,
    installProjects: request.installProjects,
    ...(snapshotAt !== undefined && Number.isFinite(snapshotAt) ? { reloadAfterMs: snapshotAt } : {}),
  };
}

function reportProjects(request: PendingUpgradeRequest): readonly (string | undefined)[] {
  return request.installProjects ?? request.selectors;
}

function installAttribution(request: PendingUpgradeRequest): string {
  const projects =
    reportProjects(request).length === 0
      ? "all configured projects"
      : reportProjects(request).map((selector) => selector ?? "(default)").join(", ");
  return `Host-global install holder: ${request.installHolder ?? "unknown"}; affected projects: ${projects}.`;
}

export interface UpgradeRecoveryStatus {
  failedVersion: string;
  phase: string;
  journalPath: string;
  verificationOwed: string;
}

/**
 * The unresolved safety hold represented by the newest journal request.
 * Status reads this same lifecycle verdict as the verifier; it never guesses
 * from the pause reason alone.
 */
export function upgradeRecoveryStatus(
  entries: readonly UpgradeJournalEntry[],
  root = stateDir(),
): UpgradeRecoveryStatus | undefined {
  const request = pendingUpgradeRequest(entries);
  const state = classifyUpgrade(entries);
  if (
    request === undefined ||
    state.kind !== "closed" ||
    !RECOVERY_HOLD_OUTCOMES.has(state.phase)
  ) {
    return undefined;
  }
  const expected = request.previous;
  const surfaces =
    expected === undefined
      ? "the pre-upgrade surface snapshot is missing"
      : [
          `CLI=${expected.cliVersion}`,
          `omp=${expected.ompVersion ?? "absent"}`,
          `herdr=${expected.herdrSource ?? "absent"}`,
        ].join(", ");
  const verificationOwed =
    state.phase === "rollback-requested"
      ? `detached rollback completion, then installed-surface verification against ${surfaces}`
      : `installed-surface verification against ${surfaces}`;
  return {
    failedVersion: request.version,
    phase: state.phase,
    journalPath: upgradeJournalPath(root),
    verificationOwed,
  };
}

/**
 * Record the operator's explicit choice to release an unresolved upgrade
 * recovery fence. `resume` calls this before removing an upgrade-owned pause,
 * so the append-only journal distinguishes acknowledgement from verification.
 */
export function acknowledgeUpgradeRecovery(root = stateDir()): boolean {
  const entries = readUpgradeJournal(root);
  const recovery = upgradeRecoveryStatus(entries, root);
  if (recovery === undefined) return false;
  appendJournal(root, {
    at: new Date().toISOString(),
    kind: "outcome",
    phase: "recovery-acknowledged",
    ok: true,
    version: recovery.failedVersion,
    detail: `operator resumed dispatch with ${recovery.verificationOwed} still owed`,
  });
  return true;
}

// ---------------------------------------------------------------------------
// The live-orchestrator session attestation (#832)
// ---------------------------------------------------------------------------

/**
 * The live orchestrator surface one project's `/healthz` carries: the mode the
 * fleet runs, and — when the daemon hosts the session itself — the extension
 * version that session process actually LOADED. `undefined` means the daemon
 * carried no attestation at all: it predates the contract, or the body is
 * unreadable, and that is "not attested", never "the newest".
 */
export interface LiveOrchestratorFacts {
  mode: "embedded" | "external" | "failed";
  loaded?: string;
}

/**
 * Parse one project's `orchestrator` block off a daemon `/healthz` body.
 * Absent or malformed answers are `undefined` — the caller decides what
 * missing evidence means (for a post-upgrade daemon: unprovable reload).
 */
export function orchestratorFactsFromHealth(
  body: string | undefined,
  project: string,
): LiveOrchestratorFacts | undefined {
  if (body === undefined) return undefined;
  let parsed: unknown;
  try {
    parsed = JSON.parse(body);
  } catch {
    return undefined;
  }
  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return undefined;
  const projects = Reflect.get(parsed, "projects");
  if (!Array.isArray(projects)) return undefined;
  const entry = projects.find(
    (candidate) =>
      candidate !== null &&
      typeof candidate === "object" &&
      !Array.isArray(candidate) &&
      Reflect.get(candidate as object, "project") === project,
  );
  if (entry === undefined) return undefined;
  const raw = Reflect.get(entry as object, "orchestrator");
  if (raw === null || typeof raw !== "object" || Array.isArray(raw)) return undefined;
  const record = raw as Record<string, unknown>;
  const mode = record["mode"];
  if (mode !== "embedded" && mode !== "external" && mode !== "failed") return undefined;
  const loaded = record["loaded"];
  return {
    mode,
    ...(typeof loaded === "string" ? { loaded } : {}),
  };
}

/**
 * Epoch-ms when one pid started, read off `/proc` — the live-process fact an
 * external (pane-owned) orchestrator attestation is built from. `undefined`
 * when the answer cannot be read; absence is never "it restarted".
 */
export function processStartTimeMs(pid: number): number | undefined {
  try {
    const stat = readFileSync(`/proc/${String(pid)}/stat`, "utf8");
    // Field 22 (`starttime`) counts clock ticks since boot; the comm field can
    // contain `)`, so split after the last one. Dropping the `pid (comm)`
    // prefix shifts every index by three, so full-field 22 is `fields[19]` —
    // `fields[21]` would be `rss`, an ever-present constant that never times a
    // live restart. CLK_TCK is 100 on every Linux host herdr targets.
    const fields = stat.slice(stat.lastIndexOf(")") + 2).split(" ");
    const startTicks = Number(fields[19]);
    const boot = readFileSync("/proc/stat", "utf8");
    const bootSeconds = Number(/^btime\s+(\d+)/m.exec(boot)?.[1]);
    if (!Number.isFinite(startTicks) || !Number.isFinite(bootSeconds)) return undefined;
    return (bootSeconds + startTicks / 100) * 1000;
  } catch {
    return undefined;
  }
}

/** The omp processes herdr claims for the fleet pane resolved to start times,
 *  or why they could not be read. The probe seam keeps the upgrade hermetic:
 *  the pure verdict below compares the starts against the install deadline. */
export type PaneOmpProbe = { starts: readonly number[] } | { problem: string };

/**
 * What a live pane session must prove after an upgrade: every omp process the
 * fleet pane currently runs must have STARTED at or after the instant the
 * install began — a pane process still running from before the install loaded
 * the pre-upgrade extension, and an empty pane is exactly the stale-or-missing
 * reload the upgrade must not claim success over. `undefined` when the pane
 * proves the reload.
 */
export function paneRestartProblem(starts: readonly number[], reloadAfterMs: number): string | undefined {
  const stale = starts.find((startedAt) => startedAt < reloadAfterMs);
  if (stale !== undefined) {
    const at = new Date(stale).toISOString();
    const deadline = new Date(reloadAfterMs).toISOString();
    return `the fleet pane is still running the omp process it started at ${at} — before the install began at ${deadline}; its loaded conductor extension predates this upgrade`;
  }
  if (starts.length === 0) {
    return "no live omp process is claimed by the fleet pane — the orchestrator session did not reload";
  }
  return undefined;
}

/**
 * What one project's live session must prove after the reloads. `undefined`
 * is the pass; a string is the problem that must fail the transaction.
 *
 *  - `embedded` — the daemon hosts the session: the live process ATTESTED the
 *    extension version it loaded at start, so the loaded version must equal
 *    the target release. A session that survived the upgrade answers its old
 *    version and fails here.
 *  - `failed` — the embedded session failed to start: no live session exists
 *    to have reloaded, and the upgrade must not claim one.
 *  - `external` — the fleet pane owns the session and the daemon cannot
 *    attest its loaded code, so the reload is proven from the pane's live
 *    process start times (via the pane probe): every omp process the pane
 *    runs must have started at/after the install began. A herdr restart that
 *    never happened (`herdrRestarted` false) leaves the pane untouched and
 *    this leg unattestable — the transaction skips it rather than failing a
 *    pane nothing was supposed to reload.
 */
export function sessionReloadProblem(o: {
  facts: LiveOrchestratorFacts | undefined;
  targetVersion: string;
  reloadAfterMs: number;
  herdrRestarted: boolean;
  pane: PaneOmpProbe | undefined;
}): string | undefined {
  if (o.facts === undefined) {
    return "the restarted daemon did not attest its orchestrator session — /healthz carried no orchestrator fact; the live extension's code cannot be proven";
  }
  if (o.facts.mode === "embedded") {
    if (o.facts.loaded !== o.targetVersion) {
      return `the live orchestrator session loaded omp-conductor ${o.facts.loaded ?? "(unattested)"}, not ${o.targetVersion} — it was not reloaded by this upgrade`;
    }
    return undefined;
  }
  if (o.facts.mode === "failed") {
    return "the embedded orchestrator session failed to start after the restart — there is no live session on the new extension";
  }
  // external: the pane owns the session; the daemon cannot attest the pane's
  // loaded code, so the transaction proves the reload from the pane's live
  // process start times — a pane process that survived the herdr restart
  // loaded the pre-upgrade extension.
  if (!o.herdrRestarted) return undefined;
  if (o.pane === undefined) {
    return "the fleet pane could not be probed — its reload cannot be proven";
  }
  if ("problem" in o.pane) return `the fleet pane could not be probed: ${o.pane.problem}`;
  return paneRestartProblem(o.pane.starts, o.reloadAfterMs) ?? undefined;
}

/**
 * Every injectable process/state seam the post-restart verifier touches.
 *
 * The verifier runs in the daemon, so everything that belongs to the daemon's
 * live state (the pause sentinel, the layers, the outbox, the health probe)
 * is injected rather than imported — the daemon wires its own functions, and
 * a test drives the whole pass with fakes.
 */
export interface UpgradeVerifyDeps {
  projectName: string;
  /** The journal directory; defaults to the conductor state dir. */
  journalRoot?: string;
  /** Sink for journal lines; defaults to the state-dir journal. */
  journal?: (entry: UpgradeJournalEntry) => void;
  run: (command: string, args: readonly string[]) => Promise<UpgradeCommandResult>;
  layers(project?: string): FleetLayers;
  health(port: number): Promise<{ ok: boolean; body?: string }>;
  doctor(projectName: string): Promise<DoctorReport>;
  /** Read all three package/plugin identities from the live host. */
  surfaces(): Promise<JournalSurfaces>;
  /** The durable outbox enqueue — one row, daemon-owned delivery. */
  enqueue(draft: ReportDraft): void;
  /** Read one pause sentinel (who set it, when) — daemon-owned state. */
  pause(project?: string): { source: string; reason?: string; since: number } | undefined;
  /** Lift the pause sentinel for a project, clearing the legacy global too. */
  resume(project?: string): void;
  /** Escalate a project pause to the host-global safety fence. */
  holdGlobal(): void;
  /** Remove only the exact dedicated recovery fence observed by this pass. */
  releaseGlobalRecovery(expectedSince: number | undefined): void;
  /** Start the detached rollback unit for a version. */
  launchRollback(
    version: string,
  ): Promise<{ ok: true; unit: string } | { ok: false; stderr: string }>;
  /**
   * The herdr session the fleet pane lives in — the pane-side name for
   * resolving the external orchestrator's live processes (#832).
   */
  herdrSession: string;
  /**
   * The live omp processes the fleet pane claims, resolved to their start
   * times — the evidence an external orchestrator's reload is judged on
   * (#832). `{ problem }` means the pane surface answered unreadably; the
   * upgrade must not invent a pass from an unreadable pane.
   */
  probePaneOmp(session: string): Promise<PaneOmpProbe>;
  log(message: string): void;
  now?(): number;
}

/** What one verify pass did with the pending request. */
export interface UpgradeVerifyResult {
  handled:
    | "none"
    | "already-closed"
    | "verified"
    | "recovered"
    | "recovery-owed"
    | "aborted"
    | "rollback-requested"
    | "rollback-unavailable";
  version?: string;
  detail?: string;
}

/**
 * The first-tick verification for a fleet-initiated install (#486).
 *
 * The installing process journals progress and stops; whichever process comes
 * back — the restarted daemon's first tick — reads that journal and confirms
 * independently, then closes the transaction on the evidence: the installed
 * version matches the request, `/healthz` answers, ticks and pane match the
 * layers the install began under, and `doctor` has no failing finding.
 *
 * Every action is ordered for a crash: the durable outbox row goes in before
 * the terminal journal outcome, and the outcome's dedupe key makes the
 * at-least-once re-enqueue a no-op — so a tick that died mid-close re-runs
 * the whole pass on the next tick without doubling the report or the rollback
 * unit. The rollback unit is spawned at most once: a `rollback` phase marker
 * journaled with its outcome is the crash guard, and only a marker that
 * records a failed spawn (nothing was started) may retry.
 */
export async function verifyPendingUpgrade(deps: UpgradeVerifyDeps): Promise<UpgradeVerifyResult> {
  const root = deps.journalRoot ?? stateDir();
  const now = deps.now ?? Date.now;
  const entries = readUpgradeJournal(root);
  const progression = requestProgression(entries);
  if (progression === undefined) return { handled: "none" };
  const request = pendingUpgradeRequest(entries);
  if (request === undefined) return { handled: "none" };
  const state = classifyUpgrade(entries);

  const journal = (entry: Omit<UpgradeJournalEntry, "at">, at = now()): void => {
    const write = deps.journal ?? ((line: UpgradeJournalEntry) => appendJournal(root, line));
    write({ ...entry, version: entry.version ?? request.version, at: new Date(at).toISOString() });
  };

  // ---------------------------------------------------------------- terminal
  if (state.kind === "closed") {
    // A closed failure is still a safety hold: rollback-requested means the
    // rollback unit has not landed yet, while rollback-unavailable and
    // rollback-failed mean the host may contain mixed surfaces. Only outcomes
    // that prove the transaction safe may release its pause.
    const outcome = newestOutcome(progression);
    const parsedOutcomeAt = outcome === undefined ? Number.NaN : Date.parse(outcome.at);
    const outcomeAt = Number.isFinite(parsedOutcomeAt) ? parsedOutcomeAt : undefined;
    if (RECOVERY_CHECK_OUTCOMES.has(state.phase)) {
      return verifyRecoveredUpgrade(deps, request, state.phase, outcomeAt, journal, now);
    }
    if (RECOVERY_HOLD_OUTCOMES.has(state.phase)) deps.holdGlobal();
    if (RESUMABLE_OUTCOMES.has(state.phase) && outcomeAt !== undefined) {
      resumeUpgradePause(deps, request, outcomeAt);
    }
    return { handled: "already-closed", version: request.version, detail: state.phase };
  }

  // ------------------------------------------------------------ never started
  if (state.kind === "no-start") {
    // The unit died before journaling a single phase. Nothing was installed,
    // but no phase recorded the exact pause instance either. Close the request
    // without guessing that an upgrade-owned sentinel still belongs to it.
    const abortedAt = now();
    deps.enqueue({
      project: reportProject(reportProjects(request)),
      kind: "tier2",
      body: [
        `Fleet upgrade to omp-conductor@${request.version} never started (journal: ${upgradeJournalPath(root)}).`,
        installAttribution(request),
        "The detached install unit died before touching any surface.",
        "Dispatch may still hold an unidentifiable install pause; check `status` and clear it explicitly with `omp-conductor resume`.",
        `Re-run the request, or install by hand with \`omp-conductor upgrade --to ${request.version}\`.`,
      ].join("\n"),
      at: abortedAt,
      dedupeKey: `upgrade:${request.version}:aborted`,
    });
    journal({
      kind: "outcome",
      phase: "aborted",
      ok: false,
      detail: "requested upgrade never started a phase",
    }, abortedAt);
    return { handled: "aborted", version: request.version, detail: "no phase was ever journaled" };
  }

  // ----------------------------------------------- incomplete or failed mid-run
  if (state.kind === "incomplete" || state.kind === "failed") {
    // The installer died mid-transaction or journaled a failure it could not
    // roll back itself. Surfaces may be mixed; the fixed answer is the
    // snapshot rollback, detached, plus a tier-2 page naming the journal.
    deps.log(
      `upgrade to ${request.version} left an ${state.kind} journal — triggering the detached rollback`,
    );
    return triggerUpgradeRollback(deps, root, journal, request, progression, now(), "mid-flight failure");
  }

  // ------------------------------------------------------- awaiting verification
  const checks: UpgradeCheck[] = [];
  await runUpgradeChecks(deps, request, checks);
  const failed = checks.find((check) => !check.ok);
  if (failed !== undefined) {
    return triggerUpgradeRollback(
      deps,
      root,
      journal,
      request,
      progression,
      now(),
      `verification failed on ${failed.name} (${failed.detail ?? "no detail"})`,
    );
  }

  // Good. Persist the evidence before releasing anything: if this process dies
  // after the outcome, the next tick can finish the idempotent resume.
  const verifiedAt = now();
  deps.enqueue({
    project: reportProject(reportProjects(request)),
    kind: "material",
    body: [
      `omp-conductor@${request.version} is installed and verified on the first tick after the restart.`,
      installAttribution(request),
      "",
      ...checks.map((check) => `${check.ok ? "ok" : "FAIL"}  ${check.name}${check.detail === undefined ? "" : ` — ${check.detail}`}`),
      "",
      "Dispatch restored to its prior state.",
    ].join("\n"),
    at: verifiedAt,
    dedupeKey: `upgrade:${request.version}:verified`,
  });
  journal({ kind: "outcome", phase: "verified", ok: true, checks }, verifiedAt);
  const paused = deps.pause(request.pauseKey);
  resumeUpgradePause(deps, request, verifiedAt);
  if (paused !== undefined && deps.pause(request.pauseKey) === undefined) {
    deps.log(`upgrade verified — dispatch resumed (${request.pauseKey ?? "global"})`);
  }
  return { handled: "verified", version: request.version };
}

async function verifyRecoveredUpgrade(
  deps: UpgradeVerifyDeps,
  request: PendingUpgradeRequest,
  failedPhase: string,
  failedOutcomeAt: number | undefined,
  journal: (entry: Omit<UpgradeJournalEntry, "at">, at?: number) => void,
  now: () => number,
): Promise<UpgradeVerifyResult> {
  const globalPause = deps.pause();
  const globalRecoverySince =
    globalPause?.source === "upgrade-recovery" ? globalPause.since : undefined;
  const checks: UpgradeCheck[] = [];
  const expected = request.previous;
  if (expected === undefined) {
    checks.push({ name: "installed-surfaces", ok: false, detail: "pre-upgrade snapshot is missing" });
  } else {
    try {
      const actual = await deps.surfaces();
      for (const [name, wanted, found] of [
        ["cli", expected.cliVersion, actual.cliVersion],
        ["omp", expected.ompVersion, actual.ompVersion],
        ["herdr", expected.herdrSource, actual.herdrSource],
      ] as const) {
        checks.push({
          name: `surface:${name}`,
          ok: wanted === found,
          detail: `expected ${wanted ?? "absent"}; found ${found ?? "absent"}`,
        });
      }
    } catch (err) {
      checks.push({
        name: "installed-surfaces",
        ok: false,
        detail: err instanceof Error ? err.message : String(err),
      });
    }
  }
  const failed = checks.find((check) => !check.ok);
  if (failed !== undefined) {
    // A recovery verdict owns the host-wide gate. Re-establish it on every
    // owed pass so a legacy failed journal or a lost sentinel cannot admit.
    deps.holdGlobal();
    return {
      handled: "recovery-owed",
      version: request.version,
      detail: `${failedPhase}; ${failed.name}: ${failed.detail ?? "failed"}`,
    };
  }

  const recoveredAt = now();
  deps.enqueue({
    project: reportProject(reportProjects(request)),
    kind: "material",
    body: [
      `The failed upgrade to omp-conductor@${request.version} is recovered.`,
      installAttribution(request),
      "",
      ...checks.map((check) => `ok  ${check.name} — ${check.detail ?? "matched"}`),
      "",
      "Every installed surface matches the durable pre-upgrade snapshot; dispatch restored to its prior state.",
    ].join("\n"),
    at: recoveredAt,
    dedupeKey: `upgrade:${request.version}:recovered`,
  });
  journal({
    kind: "outcome",
    phase: "recovered",
    ok: true,
    detail: `verified recovery after ${failedPhase}`,
    checks,
  }, recoveredAt);
  resumeUpgradePause(deps, request, failedOutcomeAt);
  deps.releaseGlobalRecovery(globalRecoverySince);
  return { handled: "recovered", version: request.version, detail: failedPhase };
}

/**
 * Lift the install's own pause sentinel, and only that one. `pauseKey` is the
 * sentinel the engine paused; a fixed `initialPaused` true means the fleet was
 * already held and the install never paused it, so nothing is lifted. A pause
 * that an operator re-created (source changed) is left standing.
 *
 * New journals carry the exact pause creation instant, so identical
 * provenance can never make a newer transaction look owned. `notAfter` is the
 * compatibility bound for older journals that predate that identity field:
 * only a pause that existed before the safe terminal outcome may be released.
 */
function resumeUpgradePause(
  deps: UpgradeVerifyDeps,
  request: PendingUpgradeRequest,
  notAfter?: number,
): void {
  if (request.initialPaused === true) return;
  const owned = deps.pause(request.pauseKey);
  const sameTransaction =
    request.pauseSince !== undefined
      ? owned !== undefined &&
        owned.since >= request.pauseSince &&
        (notAfter === undefined || owned.since <= notAfter)
      : notAfter !== undefined && owned !== undefined && owned.since <= notAfter;
  if (owned?.source === "upgrade" && sameTransaction) {
    deps.resume(request.pauseKey);
  }
}

/**
 * Doctor findings that only a privileged `setup host` can clear — the systemd
 * units (and the recovery pair they name), and the herdr pane-shell key. A
 * release that re-renders any of those templates fails them on every host
 * whose `setup host` has not been re-run, so they are not rollback-worthy
 * regression evidence: the upgrade is deliberately unable to perform the
 * privileged install, and rolling back over drift it cannot clear would
 * strand the fleet on the previous release with the same outstanding `setup
 * host` (#598). The verifier escalates them instead — the doctor check's
 * detail and the verify report name the drift and its fix.
 */
const PRIVILEGED_INSTALL_DRIFT = new Set(["systemd-unit", "systemd-recovery", "herdr-resume"]);

/**
 * The independent checks the returning process runs. Each compares a live
 * fact with the journal's record of the world the install began in:
 *
 *  - `version` — the installed CLI (all three surfaces are pinned to one
 *    release by the engine, so the CLI read stands for the set);
 *  - `ticks` / `pane` / `herdr` — delta against the pre-install layers, the
 *    same predicate the in-process upgrade used, so a fleet that was armed
 *    stays armed and a live pane stays live;
 *  - `health` — the daemon that came back answers `/healthz`;
 *  - `doctor` — no failing finding beyond the privileged-install drift
 *    ({@link PRIVILEGED_INSTALL_DRIFT}) only `setup host` can clear, which is
 *    escalated rather than rolled back.
 */
export async function runUpgradeChecks(
  deps: UpgradeVerifyDeps,
  request: PendingUpgradeRequest,
  checks: UpgradeCheck[],
): Promise<void> {
  const version = await deps.run("omp-conductor", ["--version"]);
  checks.push({
    name: "version",
    ok: version.code === 0 && version.stdout.trim() === request.version,
    detail: version.code === 0 ? version.stdout.trim() : version.stderr.trim() || `exit ${version.code}`,
  });

  const layers = deps.layers(deps.projectName);
  const before = request.initial;
  if (before !== undefined) {
    checks.push({ name: "ticks", ok: layers.ticks === before.ticks, detail: `${before.ticks} → ${layers.ticks}` });
    checks.push({ name: "pane", ok: layers.pane === before.pane, detail: `${before.pane} → ${layers.pane}` });
    if (before.herdr === "active") {
      checks.push({ name: "herdr", ok: layers.herdr === "active", detail: layers.herdr });
    }
  }
  // One probe feeds both the health check and the session attestation below:
  // the body the daemon answered is the same read the session check parses.
  const health = layers.daemon.running
    ? await liveHealth(deps, layers.daemon.port ?? DEFAULT_PORT)
    : undefined;
  checks.push({
    name: "health",
    ok: health?.ok === true,
    detail: layers.daemon.running ? undefined : "no daemon record",
  });

  // The live-orchestrator session (#832): a successful install must leave the
  // orchestrator process executing the newly installed extension, and package
  // identities on disk say nothing about what the running session loaded. The
  // restarted daemon attests its hosted session's loaded version on /healthz;
  // an external (pane-owned) session is proven from the pane's live process
  // start times. A reload that cannot be proven fails the transaction — the
  // verifier rolls back rather than report a success the fleet is not running.
  const healthBody = health?.body;
  const herdrRestarted = request.initial?.herdr === "active";
  const sessionRequired = herdrRestarted || (request.initial?.dispatch ?? "running") !== "stopped";
  let sessionProblem = "";
  if (sessionRequired && (layers.daemon.running || herdrRestarted)) {
    const pane = await deps.probePaneOmp(deps.herdrSession).catch(
      (err): PaneOmpProbe => ({
        problem: err instanceof Error ? err.message : String(err),
      }),
    );
    sessionProblem =
      sessionReloadProblem({
        facts: orchestratorFactsFromHealth(healthBody, deps.projectName),
        targetVersion: request.version,
        reloadAfterMs: request.reloadAfterMs ?? Date.now(),
        herdrRestarted,
        pane,
      }) ?? "";
  }
  checks.push({
    name: "session",
    ok: sessionProblem === "",
    ...(sessionProblem === "" ? {} : { detail: sessionProblem }),
  });

  const doctor = await deps.doctor(deps.projectName);
  const failing = doctor.findings.filter((finding) => finding.status === "fail");
  const drift = failing.filter((finding) => PRIVILEGED_INSTALL_DRIFT.has(finding.id));
  const regression = failing.filter((finding) => !PRIVILEGED_INSTALL_DRIFT.has(finding.id));
  checks.push({
    name: "doctor",
    ok: regression.length === 0,
    detail:
      regression.length > 0
        ? `${regression[0]!.id}: ${regression[0]!.summary}`
        : drift.length > 0
          ? `status ${doctor.status}; ${drift.map((f) => f.id).join(", ")} — privileged-install drift, run \`omp-conductor setup host\` from the fleet account`
          : `status ${doctor.status}`,
  });
}

async function liveHealth(
  deps: UpgradeVerifyDeps,
  port: number,
): Promise<{ ok: boolean; body?: string }> {
  try {
    return await deps.health(port);
  } catch (err) {
    return { ok: false, body: err instanceof Error ? err.message : String(err) };
  }
}

/** Spawn the detached rollback once and page the evidence, whatever the cause. */
async function triggerUpgradeRollback(
  deps: UpgradeVerifyDeps,
  root: string,
  journal: (entry: Omit<UpgradeJournalEntry, "at">) => void,
  request: PendingUpgradeRequest,
  progress: readonly UpgradeJournalEntry[],
  at: number,
  why: string,
): Promise<UpgradeVerifyResult> {
  // Every install surface is host-wide even when the request named one
  // project. Once recovery is required, the safety fence must gate every
  // project served by the shared daemon.
  deps.holdGlobal();
  // The marker is the spawn guard: present-and-ok means a rollback unit is
  // already on its way and a crash after the spawn must not spawn a second
  // one. Present-and-failed means nothing started, so a retry is a first
  // attempt, not a double. Scoped to this request's progression — an older
  // upgrade's rollback marker must not suppress a new one.
  const marker = progress.find((entry) => entry.kind === "phase" && entry.phase === "rollback");
  let spawned: { ok: true; unit: string } | { ok: false; stderr: string };
  if (marker?.ok === true) {
    spawned = { ok: true, unit: marker.unit ?? "omp-conductor-upgrade-rollback" };
  } else {
    spawned = await deps.launchRollback(request.version);
    journal({
      kind: "phase",
      phase: "rollback",
      ok: spawned.ok,
      ...(spawned.ok ? { unit: spawned.unit } : { detail: spawned.stderr }),
    });
  }
  if (!spawned.ok) {
    // Nothing was spawned, so the next tick may legitimately retry a first
    // attempt. Page with the journal path — the fleet may be half-upgraded
    // and must not be quietly mixed.
    journal({ kind: "outcome", phase: "rollback-unavailable", ok: false, detail: spawned.stderr });
    deps.enqueue({
      project: reportProject(reportProjects(request)),
      kind: "tier2",
      body: [
        `Fleet upgrade to omp-conductor@${request.version} is INCOMPLETE (${why}) and its detached rollback could not start.`,
        installAttribution(request),
        `Reason: ${spawned.stderr}`,
        "",
        "Surfaces may be at mixed versions; do not resume dispatch.",
        `Restore manually from ${upgradeJournalPath(root)}.`,
      ].join("\n"),
      at,
      dedupeKey: `upgrade:${request.version}:rollback-unavailable`,
    });
    return { handled: "rollback-unavailable", version: request.version, detail: spawned.stderr };
  }
  journal({ kind: "outcome", phase: "rollback-requested", ok: false, detail: why });
  deps.enqueue({
    project: reportProject(reportProjects(request)),
    kind: "tier2",
    body: [
      `Fleet upgrade to omp-conductor@${request.version} did not verify (${why}) and is being rolled back by ${spawned.unit}.`,
      installAttribution(request),
      "Dispatch stays paused until the rollback lands and a later tick verifies the old version.",
      "",
      `Journal: ${upgradeJournalPath(root)}`,
    ].join("\n"),
    at,
    dedupeKey: `upgrade:${request.version}:rollback-requested`,
  });
  return { handled: "rollback-requested", version: request.version, detail: `${why}; rollback unit ${spawned.unit}` };
}