import { existsSync, readFileSync } from "node:fs";
import { homedir } from "node:os";
import { dirname, join } from "node:path";
import { backupTimestamp, copyToUniqueBackup } from "./backups.ts";
import { inspectBriefLayout, type BriefLayout } from "./brief-upgrade.ts";
import { pauseInstance, setPaused, statusSnapshot, type AdmissionAckRecord } from "./daemon.ts";
import {
  fleetLayers,
  resolveHerdrSession,
  telegramStateDir,
} from "./fleet.ts";
import { herdrPaneOmpStarts } from "./fleet.ts";
import type { DispatchLayer, FleetLayers } from "./status-render.ts";
import { DEFAULT_PORT, healthCheck, livingDaemon, restartDaemon } from "./lifecycle.ts";
import { configBackupDir, configPath, findProject, loadConfig, resolveCaps, stateDir, writeConfigRaw } from "./config.ts";
import { renderBriefForProject } from "./setup.ts";
import {
  planHostRuntime,
  totalConfiguredWorkers,
  type HostRuntimePlan,
} from "./setup-host.ts";
import {
  appendJournal,
  readUpgradeJournal,
  upgradeJournalPath,
  type UpgradeCheck,
  type UpgradeJournalEntry,
  type InstallSnapshotManagement,
} from "./upgrade-journal.ts";
import {
  enqueueUpgradeReport,
  orchestratorFactsFromHealth,
  paneRestartProblem,
  pendingUpgradeRequest,
  processStartTimeMs,
  reportProject,
  requestProgression,
  runCommand,
  sessionReloadProblem,
  type PaneOmpProbe,
  type PendingUpgradeRequest,
  type UpgradeCommandResult,
} from "./upgrade-verify.ts";

const PACKAGE = "omp-conductor";
const HERDR_PLUGIN = "herdr-conductor";
const HERDR_SOURCE = "TerrifiedBug/conductor/herdr";

/** The package root inside the repo, for a bootstrap install by git ref. */
const PACKAGE_SOURCE = "TerrifiedBug/conductor/omp";

/** A 40-hex commit: the only identity shape a bootstrap accepts. A branch or a
 *  tag is a moving target, and "the exact build I verified" is the whole point. */
const COMMIT_SHA = /^[0-9a-f]{40}$/i;
const HERDR_UNIT = "herdr-fleet.service";
const DRAIN_POLL_MS = 5_000;
const RECOVERY_POLL_MS = 2_000;
const RECOVERY_ATTEMPTS = 30;

export interface UpgradeOptions {
  version?: string;
  /**
   * Bootstrap from an exact source/build identity instead of a published semver
   * (#908) — the path that exists because a reliability release cannot be cut
   * when the *installed* conductor is the thing that is broken: publishing
   * needs working workers, and the workers are what is broken.
   *
   * `sha` is the whole identity (a 40-hex commit, never a range), and `source`
   * is a checkout of exactly that commit — its only job is to be the tree the
   * package's own checks run against before anything is installed. The install
   * itself is by git ref, so the identity is globally attestable rather than
   * dependent on one machine's directory.
   *
   * Mutually exclusive with {@link version}: two identity kinds in one call is
   * an ambiguity, not a merge.
   */
  bootstrap?: { sha: string; source: string };
  project?: string;
  /**
   * Run as the detached fleet installer (#486): journal every surface the
   * transaction touches, skip the in-process verification and leave dispatch
   * paused — the first tick of the post-restart daemon verifies version,
   * /healthz, ticks, pane and doctor, and closes or rolls back on that
   * evidence. Only the transient unit's `upgrade-install` entry sets this.
   */
  detached?: boolean;
}

export interface UpgradeResult {
  previousVersion: string;
  version: string;
  gitHead: string;
  alreadyCurrent: boolean;
  dispatch: DispatchLayer;
}

export interface UpgradeDeps {
  run(command: string, args: readonly string[]): Promise<UpgradeCommandResult>;
  /**
   * A command run in a named working directory (#908): the bootstrap's checks
   * run inside the source tree, and `run` is deliberately cwd-free everywhere
   * else so no other step can depend on where it was invoked from.
   */
  runIn(cwd: string, command: string, args: readonly string[]): Promise<UpgradeCommandResult>;
  /** Read one file as text. Injected so a bootstrap's manifest read is
   *  testable without a checkout on disk. */
  readFile(path: string): string;
  snapshot(project?: string): { liveWorkers: number };
  layers(project?: string): FleetLayers;
  brief(project?: string): { kind: BriefLayout["kind"]; current: boolean };
  /**
   * Every configured project name, in config order. One daemon serves all of
   * them (the generated unit's ExecStart carries no `--project`), so a
   * transaction that restarts that daemon has to know the whole set rather
   * than the one project a caller happened to name (#389).
   */
  projectNames(): readonly string[];
  /**
   * The daemon the transaction targets, when one is running. `generation`
   * identifies the exact instance (pid + start time) so a later restart cannot
   * stop an instance created after the request began (#377).
   */
  daemonIdentity(): { running: boolean; project?: string; generation?: string };
  /**
   * The durable pause sentinel this fleet currently holds, as an *instance*,
   * read from one selected file (the per-project sentinel first): who set it,
   * why, and when that sentinel was created. Two pauses with identical
   * provenance are still different instances if `since` differs, so a drain
   * can tell its own sentinel from one a later actor re-created after a
   * resume lifted it — even when a legacy global sentinel coexists (#377).
   */
  pauseState(project?: string): { source: string; reason?: string; since: number } | undefined;
  setPaused(value: boolean, project?: string): void;
  /** Host-global recovery fence for host-wide install surfaces. */
  setGlobalRecoveryPaused(value: boolean): void;
  restartDaemon(): Promise<void>;
  sleep(ms: number): Promise<void>;
  /**
   * The environment every surface of this transaction resolves from — the
   * herdr session included, through {@link resolveHerdrSession}. One source,
   * because the `herdr plugin list` read and the pane probe MUST agree on
   * which session this fleet is: when they did not, `upgrade` installed
   * against one session and failed verification against another, then rolled
   * a good release back (#976).
   */
  env: NodeJS.ProcessEnv;
  log(message: string): void;
  /**
   * Probe the running daemon's `/healthz`. The post-reload session
   * verification reads the live orchestrator attestation off the answer
   * (#832): the restarted daemon says which conductor version its orchestrator
   * session process actually loaded.
   */
  health(port: number): Promise<{ ok: boolean; body?: string }>;
  /**
   * Every omp process the fleet pane currently claims by herdr, resolved to
   * its start time — the evidence an external (pane-owned) orchestrator
   * reloaded after the install began (#832). `{ problem }` when the pane
   * surface answers unreadably: absence of evidence is never a reload.
   */
  probePaneOmp(session: string): Promise<PaneOmpProbe>;
  /**
   * The host runtime this version's package renders, compared against what is
   * actually installed: the daemon and herdr units, the recovery unit and its
   * playbook at {@link RECOVER_SCRIPT_INSTALL_PATH}, the herdr pane-shell
   * config and the herdr-conductor `config.env`. `upgrade` prints
   * {@link HostRuntimePlan.drift} after a release that re-rendered those
   * files, so the operator is told that `setup host` is owed before a later
   * `doctor` has to discover it. Read-only: the plan renders and compares,
   * it never writes a host file.
   */
  hostRuntime(): HostRuntimePlan;
  /**
   * The durable journal sink for the detached fleet installer (#486). Wired
   * to the state-dir journal by `upgrade-install`/`upgrade-rollback`; absent
   * on the human `upgrade` path, whose own process is its witness. The engine
   * journals each surface boundary so the post-restart tick can verify against
   * a record it did not write.
   */
  journal?(entry: UpgradeJournalEntry): void;
}

/** One command run to completion with captured output. Exported so the
 *  post-restart verifier and the rollback unit run processes the same way. */
export const DEFAULT_DEPS: UpgradeDeps = {
  run: runCommand,
  runIn: (cwd, command, args) => runCommand(command, args, cwd),
  readFile: (path) => readFileSync(path, "utf8"),
  snapshot: statusSnapshot,
  layers: fleetLayers,
  brief: (projectName) => {
    const project = findProject(loadConfig(), projectName);
    const layout = inspectBriefLayout(project.workspaceRoot, renderBriefForProject(project));
    if (layout.kind !== "overlay") return { kind: layout.kind, current: false };
    const policy = readFileSync(layout.policyPath, "utf8");
    const live = readFileSync(layout.orchestratorPath, "utf8");
    return { kind: layout.kind, current: live === renderBriefForProject(project, policy) };
  },
  projectNames: () => loadConfig().projects.map((p) => p.name),
  daemonIdentity: () => {
    const daemon = livingDaemon();
    if (daemon === undefined) return { running: false };
    return {
      running: true,
      project: daemon.project,
      generation: `${daemon.pid}@${daemon.startedAt}`,
    };
  },
  pauseState: (project) => pauseInstance(project),
  setPaused: (v, project) =>
    setPaused(v, { source: "upgrade", reason: "upgrade, draining" }, project),
  setGlobalRecoveryPaused: (v) => {
    const current = pauseInstance();
    if (v && current === undefined) {
      setPaused(true, { source: "upgrade-recovery", reason: "upgrade recovery verification required" });
    } else if (!v && current?.source === "upgrade-recovery") {
      setPaused(false);
    }
  },
  restartDaemon: async () => {
    await restartDaemon({});
  },
  health: async (port) => healthCheck(port),
  probePaneOmp: (session) => herdrPaneOmpStarts(runCommand, session, processStartTimeMs),
  // The bare host-global plan — the same render the advisory's `setup host`
  // command would install: no per-project tail, the recovery unit encoding no
  // one project's name, and FLEET_CWDS derived from every configured project.
  hostRuntime: () => {
    const cfg = loadConfig();
    return planHostRuntime(undefined, cfg.defaults, telegramStateDir(), undefined, totalConfiguredWorkers(cfg));
  },
  sleep: Bun.sleep,
  env: process.env,
  log: (message) => process.stdout.write(`${message}\n`),
};

function commandLine(command: string, args: readonly string[]): string {
  return [command, ...args].join(" ");
}

async function mustRun(
  deps: SurfaceProbeDeps,
  command: string,
  args: readonly string[],
): Promise<UpgradeCommandResult> {
  const ran = await deps.run(command, args);
  if (ran.code === 0) return ran;
  const detail = ran.stderr.trim() || ran.stdout.trim() || `exit ${ran.code}`;
  throw new Error(`${commandLine(command, args)} failed: ${detail}`);
}

function parseRegistry(raw: string): { version: string; gitHead: string } {
  let value: unknown;
  try {
    value = JSON.parse(raw);
  } catch {
    throw new Error("npm returned invalid release metadata");
  }
  if (value === null || typeof value !== "object" || Array.isArray(value)) {
    throw new Error("npm returned invalid release metadata");
  }
  const version = Reflect.get(value, "version");
  const gitHead = Reflect.get(value, "gitHead");
  if (typeof version !== "string" || version.length === 0) {
    throw new Error("npm release metadata has no version");
  }
  if (typeof gitHead !== "string" || !/^[0-9a-f]{40}$/i.test(gitHead)) {
    throw new Error(`npm release ${version} has no full gitHead`);
  }
  return { version, gitHead };
}

/**
 * One release identity plus how to install it (#908).
 *
 * Two kinds, one shape downstream: everything after this — the currentness
 * comparison, the journal, the install, the attestation and the rollback —
 * reads a `{ version, gitHead }` pair and an install spec, so a bootstrap is a
 * second accepted identity rather than a parallel installer.
 */
interface ReleaseIdentityPlan {
  version: string;
  gitHead: string;
  /** What the discovered CLI root / `omp plugin install` are given for this identity. */
  packageSpec: string;
  /** Named checks already run to earn this identity, journalled as evidence. */
  checks: UpgradeCheck[];
  /** True for a source/build identity: npm does not have this version. */
  bootstrap: boolean;
}

/** The version a source tree declares — the identity's other half, read from
 *  the tree itself because npm has never seen it. */
function sourceVersion(deps: UpgradeDeps, source: string): string {
  const manifest = join(source, "omp", "package.json");
  let raw: string;
  try {
    raw = deps.readFile(manifest);
  } catch (err) {
    throw new Error(
      `bootstrap refused: cannot read ${manifest} (${err instanceof Error ? err.message : String(err)}) — ` +
        "--source must be a checkout of this repository",
    );
  }
  let parsed: unknown;
  try {
    parsed = JSON.parse(raw);
  } catch {
    throw new Error(`bootstrap refused: ${manifest} is not valid JSON`);
  }
  const version = parsed !== null && typeof parsed === "object" ? Reflect.get(parsed, "version") : undefined;
  if (typeof version !== "string" || version.length === 0) {
    throw new Error(`bootstrap refused: ${manifest} declares no version`);
  }
  return version;
}

/**
 * Resolve which release this run installs, and prove a bootstrap identity
 * before it can touch anything (#908).
 *
 * The three refusals are the deliverable, not the install: a sha that is not a
 * sha, a source tree that is not at that sha, and a source tree whose own
 * checks fail. Each names the identity and what failed, and each happens before
 * the first surface is replaced — an unverified tree must never become a live
 * install, which is the difference between this and the `bun add github:...`
 * POLICY forbids.
 */
async function resolveIdentity(deps: UpgradeDeps, options: UpgradeOptions): Promise<ReleaseIdentityPlan> {
  const bootstrap = options.bootstrap;
  if (bootstrap === undefined) {
    const requested = options.version === undefined ? `${PACKAGE}@latest` : `${PACKAGE}@${options.version}`;
    const release = parseRegistry(
      (await mustRun(deps, "npm", ["view", requested, "version", "gitHead", "--json"])).stdout,
    );
    return { ...release, packageSpec: `${PACKAGE}@${release.version}`, checks: [], bootstrap: false };
  }
  if (options.version !== undefined) {
    throw new Error("bootstrap refused: pass either --to VERSION or --bootstrap SHA, never both");
  }
  const sha = bootstrap.sha.toLowerCase();
  if (!COMMIT_SHA.test(sha)) {
    throw new Error(`bootstrap refused: "${bootstrap.sha}" is not a 40-character commit sha`);
  }
  const checks: UpgradeCheck[] = [];
  // The tree must BE that commit. Without this the identity is a label the
  // operator typed, and the checks below would attest a different build than
  // the one installed.
  const head = await deps.run("git", ["-C", bootstrap.source, "rev-parse", "HEAD"]);
  const at = head.stdout.trim().toLowerCase();
  if (head.code !== 0 || at !== sha) {
    checks.push({ name: "source-at-sha", ok: false, detail: at.length === 0 ? head.stderr.trim() : at });
    throw new Error(
      `bootstrap refused: ${bootstrap.source} is at ${at.length === 0 ? "an unreadable HEAD" : at}, not ${sha} — ` +
        "the checks would attest a different build than the one installed",
    );
  }
  checks.push({ name: "source-at-sha", ok: true, detail: sha });
  const version = sourceVersion(deps, bootstrap.source);
  // The package's own declared gate, against the source, before anything is
  // installed. A failure refuses: a partially verified tree must not go live.
  deps.log(`bootstrap: running the package checks against ${bootstrap.source} (${sha.slice(0, 12)})`);
  const gate = await deps.runIn(join(bootstrap.source, "omp"), "bun", ["run", "check"]);
  if (gate.code !== 0) {
    const detail = (gate.stderr.trim() || gate.stdout.trim()).split("\n").slice(-3).join("; ");
    checks.push({ name: "package-checks", ok: false, detail });
    throw new Error(
      `bootstrap refused: \`bun run check\` failed against ${bootstrap.source} at ${sha} — ${detail}`,
    );
  }
  checks.push({ name: "package-checks", ok: true });
  return { version, gitHead: sha, packageSpec: `github:${PACKAGE_SOURCE}#${sha}`, checks, bootstrap: true };
}

interface OmpPluginEntry {
  version: string;
  path?: string;
}

function ompPluginInventory(raw: string): unknown[] {
  let parsed: unknown;
  try {
    parsed = JSON.parse(raw);
  } catch {
    throw new Error("omp plugin list returned invalid JSON");
  }
  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
    throw new Error("omp plugin list returned an invalid inventory");
  }
  const npm = Reflect.get(parsed, "npm");
  if (!Array.isArray(npm)) throw new Error("omp plugin list returned no npm inventory");
  return npm;
}

function ompPluginEntry(raw: string): OmpPluginEntry | undefined {
  const plugin = ompPluginInventory(raw).find(
    (entry) => entry !== null && typeof entry === "object" && Reflect.get(entry, "name") === PACKAGE,
  );
  if (plugin === undefined || plugin === null || typeof plugin !== "object") return undefined;
  const version = Reflect.get(plugin, "version");
  if (typeof version !== "string" || version.length === 0) {
    throw new Error("installed omp-conductor plugin has no version");
  }
  const path = Reflect.get(plugin, "path");
  return {
    version,
    ...(typeof path === "string" && path.length > 0 ? { path } : {}),
  };
}

function ompPluginVersion(raw: string): string | undefined {
  return ompPluginEntry(raw)?.version;
}

function nodeModulesRoot(path: string): string | undefined {
  const marker = "/node_modules/";
  const at = path.lastIndexOf(marker);
  return at <= 0 ? undefined : path.slice(0, at);
}

function defaultOmpPluginRoot(env: NodeJS.ProcessEnv): string {
  const configured = env["OMP_CODING_AGENT_DIR"];
  if (configured !== undefined && configured.length > 0) return join(configured, "plugins");
  return join(env["HOME"] ?? homedir(), ".omp", "plugins");
}

function ompPluginInstallRoot(raw: string, env: NodeJS.ProcessEnv): string {
  const inventory = ompPluginInventory(raw);
  const ownPath = ompPluginEntry(raw)?.path;
  const fallbackPath = inventory
    .map((entry) =>
      entry !== null && typeof entry === "object" ? Reflect.get(entry, "path") : undefined,
    )
    .find((path): path is string => typeof path === "string" && path.length > 0);
  const root = nodeModulesRoot(ownPath ?? fallbackPath ?? "");
  return root ?? defaultOmpPluginRoot(env);
}

function herdrPluginSource(raw: string): string | undefined {
  const line = raw.split("\n").find((candidate) => /^- herdr-conductor\b/.test(candidate.trim()));
  if (line === undefined) return undefined;
  const source = line.match(/\[([^\]]+)\]\s*$/)?.[1];
  if (source === undefined) {
    throw new Error("cannot determine installed Herdr plugin source");
  }
  return source;
}

function defaultHerdrInstallRoot(env: NodeJS.ProcessEnv): string {
  const config =
    env["HERDR_CONFIG_DIR"] ??
    join(env["XDG_CONFIG_HOME"] ?? join(env["HOME"] ?? homedir(), ".config"), "herdr");
  return join(config, "plugins", "github");
}

function herdrPluginManagement(
  raw: string,
  env: NodeJS.ProcessEnv,
): InstallSnapshotManagement["herdr"] {
  let parsed: unknown;
  try {
    parsed = JSON.parse(raw);
  } catch {
    throw new Error("herdr plugin list returned invalid JSON");
  }
  const result =
    parsed !== null && typeof parsed === "object" && !Array.isArray(parsed)
      ? Reflect.get(parsed, "result")
      : undefined;
  const plugins =
    result !== null && typeof result === "object" && !Array.isArray(result)
      ? Reflect.get(result, "plugins")
      : undefined;
  if (!Array.isArray(plugins)) throw new Error("herdr plugin list returned no plugin inventory");
  const plugin = plugins.find(
    (entry) => entry !== null && typeof entry === "object" && Reflect.get(entry, "plugin_id") === HERDR_PLUGIN,
  );
  const root =
    plugin !== undefined && typeof plugin === "object"
      ? Reflect.get(plugin, "plugin_root")
      : undefined;
  const source =
    plugin !== undefined && typeof plugin === "object"
      ? Reflect.get(plugin, "source")
      : undefined;
  const managedPath =
    source !== null && typeof source === "object" && !Array.isArray(source)
      ? Reflect.get(source, "managed_path")
      : undefined;
  const marker = "/plugins/github/";
  const markerAt = typeof root === "string" ? root.indexOf(marker) : -1;
  const installRoot =
    typeof managedPath === "string" && managedPath.length > 0
      ? dirname(managedPath)
      : markerAt >= 0 && typeof root === "string"
        ? root.slice(0, markerAt + marker.length - 1)
        : defaultHerdrInstallRoot(env);
  return {
    ...(typeof root === "string" && root.length > 0 ? { root } : {}),
    installRoot,
  };
}

function cliInstallRoot(target: string): string {
  const marker = `/node_modules/${PACKAGE}/`;
  const at = target.lastIndexOf(marker);
  if (at <= 0) {
    throw new Error(
      `installed omp-conductor CLI target ${target} is not inside a discoverable node_modules/${PACKAGE} root`,
    );
  }
  return target.slice(0, at);
}

/** Discover the exact managed roots the transaction will mutate (#1019). */
export async function discoverInstallManagement(
  deps: SurfaceProbeDeps,
): Promise<InstallSnapshotManagement> {
  const session = resolveHerdrSession(deps.env);
  const executable = (await mustRun(deps, "which", ["omp-conductor"])).stdout.trim();
  if (executable.length === 0) throw new Error("which omp-conductor returned no executable");
  const [resolved, omp, herdr] = await Promise.all([
    mustRun(deps, "readlink", ["-f", executable]),
    mustRun(deps, "omp", ["plugin", "list", "--json"]),
    mustRun(deps, "herdr", ["--session", session, "plugin", "list", "--json"]),
  ]);
  const target = resolved.stdout.trim();
  if (target.length === 0) throw new Error(`cannot resolve installed CLI ${executable}`);
  const ompRoot = ompPluginInstallRoot(omp.stdout, deps.env);
  return {
    cli: { executable, target, root: cliInstallRoot(target) },
    omp: { root: ompRoot, lock: join(ompRoot, "omp-plugins.lock.json") },
    herdr: herdrPluginManagement(herdr.stdout, deps.env),
  };
}

async function preflightInstall(
  deps: UpgradeDeps,
  management: InstallSnapshotManagement,
  packageSpec: string,
): Promise<void> {
  const writable: { label: string; path: string }[] = [
    { label: "CLI package root", path: management.cli.root },
    { label: "CLI package manifest", path: join(management.cli.root, "package.json") },
    { label: "CLI node_modules", path: join(management.cli.root, "node_modules") },
    { label: "OMP plugin root", path: management.omp.root },
    { label: "Herdr managed-plugin root", path: management.herdr.installRoot },
  ];
  if (existsSync(management.omp.lock)) {
    writable.push({ label: "OMP plugin lock", path: management.omp.lock });
  }
  for (const target of writable) {
    const checked = await deps.run("test", ["-w", target.path]);
    if (checked.code !== 0) {
      const detail = checked.stderr.trim() || checked.stdout.trim() || "missing or not writable";
      throw new Error(
        `upgrade preflight refused: ${target.label} ${target.path} is unusable (${detail}); ` +
          "nothing has been installed or paused",
      );
    }
  }
  try {
    await mustRun(deps, "bun", [
      "add",
      "--cwd",
      management.cli.root,
      "--exact",
      "--dry-run",
      packageSpec,
    ]);
  } catch (err) {
    throw new Error(
      `upgrade preflight refused: the CLI install command is unusable at ${management.cli.root} ` +
        `(${err instanceof Error ? err.message : String(err)}); nothing has been installed or paused`,
    );
  }
}

/** The three installed identities one host carries: the discovered CLI package
 * tree, the omp plugin, and the herdr recovery plugin's pin. */
export interface InstalledSurfaces {
  cliVersion: string;
  ompVersion?: string;
  herdrSource?: string;
  management?: InstallSnapshotManagement;
}

/** What reading the three surfaces needs — deliberately narrower than
 *  {@link UpgradeDeps} so a read-only consumer (`doctor`'s surface-parity
 *  finding, #904) can supply three fields instead of a whole transaction. */
export type SurfaceProbeDeps = Pick<UpgradeDeps, "run" | "log" | "env">;

/**
 * Read the CLI, omp-plugin and herdr-plugin identities actually installed on
 * this host. Exported because `upgrade` is not the only thing that needs to
 * know whether the three agree: a host whose installs are manual diverges
 * silently between transactions, and `doctor` reads them through this one
 * seam rather than growing a second implementation (#904).
 */
export async function inspectSurfaces(
  deps: SurfaceProbeDeps,
  /** `readHerdr: false` skips the herdr session probe and plugin read
   *  entirely, so a host with no herdr on PATH can still be asked what its

   *  CLI and omp plugin are (#904). The upgrade transaction always reads all
   *  three — a release pins every surface. */
  opts: { readHerdr?: boolean } = {},
): Promise<InstalledSurfaces> {
  if (opts.readHerdr === false) {
    const [cliOnly, ompOnly] = await Promise.all([
      mustRun(deps, "omp-conductor", ["--version"]),
      mustRun(deps, "omp", ["plugin", "list", "--json"]),
    ]);
    const version = cliOnly.stdout.trim();
    if (version.length === 0) throw new Error("installed omp-conductor CLI has no version");
    return { cliVersion: version, ompVersion: ompPluginVersion(ompOnly.stdout) };
  }
  const session = resolveHerdrSession(deps.env);
  const [cli, omp, herdr] = await Promise.all([
    mustRun(deps, "omp-conductor", ["--version"]),
    mustRun(deps, "omp", ["plugin", "list", "--json"]),
    mustRun(deps, "herdr", ["--session", session, "plugin", "list"]),
  ]);
  const cliVersion = cli.stdout.trim();
  if (cliVersion.length === 0) throw new Error("installed omp-conductor CLI has no version");
  return {
    cliVersion,
    ompVersion: ompPluginVersion(omp.stdout),
    herdrSource: herdrPluginSource(herdr.stdout),
  };
}

function packageVersionAt(deps: UpgradeDeps, manifest: string): string {
  let parsed: unknown;
  try {
    parsed = JSON.parse(deps.readFile(manifest));
  } catch (err) {
    throw new Error(
      `cannot read restored package manifest ${manifest}: ${err instanceof Error ? err.message : String(err)}`,
    );
  }
  const version =
    parsed !== null && typeof parsed === "object" && !Array.isArray(parsed)
      ? Reflect.get(parsed, "version")
      : undefined;
  if (typeof version !== "string" || version.length === 0) {
    throw new Error(`restored package manifest ${manifest} has no version`);
  }
  return version;
}

/** Verify the snapshot's recorded destinations directly, before PATH-based probes. */
function verifyRestoredManagement(deps: UpgradeDeps, previous: InstalledSurfaces): void {
  const management = previous.management;
  if (management === undefined) return;
  const cliManifest = join(management.cli.root, "node_modules", PACKAGE, "package.json");
  const cliVersion = packageVersionAt(deps, cliManifest);
  if (cliVersion !== previous.cliVersion) {
    throw new Error(
      `restored CLI root ${management.cli.root} has ${cliVersion}, expected ${previous.cliVersion}`,
    );
  }
  if (previous.ompVersion !== undefined) {
    const ompManifest = join(management.omp.root, "node_modules", PACKAGE, "package.json");
    const ompVersion = packageVersionAt(deps, ompManifest);
    if (ompVersion !== previous.ompVersion) {
      throw new Error(
        `restored OMP plugin root ${management.omp.root} has ${ompVersion}, expected ${previous.ompVersion}`,
      );
    }
  }
  if (previous.herdrSource !== undefined) {
    if (management.herdr.root === undefined) {
      throw new Error("restored Herdr plugin had no recorded root to verify");
    }
    const manifest = join(management.herdr.root, "herdr-plugin.toml");
    if (deps.readFile(manifest).trim().length === 0) {
      throw new Error(`restored Herdr plugin manifest ${manifest} is empty`);
    }
  }
}

export function expectedHerdrSource(gitHead: string): string {
  return `github:${HERDR_SOURCE}@${gitHead}`;
}

/**
 * The published identity of one release: its version and the exact commit it
 * was cut from, straight from the registry metadata `upgrade` already trusts.
 * Exported so a read-only consumer can answer "which commit should this
 * host's herdr plugin be pinned to?" without re-deriving the mapping (#904).
 */
export async function releaseIdentity(
  deps: SurfaceProbeDeps,
  version: string,
): Promise<{ version: string; gitHead: string }> {
  return parseRegistry((await mustRun(deps, "npm", ["view", `${PACKAGE}@${version}`, "version", "gitHead", "--json"])).stdout);
}

function surfacesCurrent(
  surfaces: InstalledSurfaces,
  version: string,
  gitHead: string,
): boolean {
  return (
    surfaces.cliVersion === version &&
    surfaces.ompVersion === version &&
    surfaces.herdrSource === expectedHerdrSource(gitHead)
  );
}

function previousHerdrInstall(source: string): readonly [string, readonly string[]] | undefined {
  if (source.startsWith("local:")) {
    return ["herdr", ["plugin", "link", source.slice("local:".length), "--enabled"]];
  }
  if (!source.startsWith("github:")) return undefined;
  const at = source.lastIndexOf("@");
  if (at < "github:".length) return undefined;
  return [
    "herdr",
    ["plugin", "install", source.slice("github:".length, at), "--ref", source.slice(at + 1), "--yes"],
  ];
}

/**
 * The projects one lifecycle transaction covers, and the single pause sentinel
 * that represents them.
 *
 * Everything an upgrade or a draining restart replaces is host-wide: the
 * discovered CLI package tree, the omp plugin, the Herdr plugin, and the one
 * daemon. That daemon serves *every* configured project — the generated unit's
 * ExecStart carries no `--project` — so its restart lands on all of them at
 * once. The scope therefore names every project whose workers must drain and
 * whose brief must be refreshed, instead of the single project a caller
 * happened to name (#389).
 *
 * `selectors` are what the rest of the module passes back to the project-aware
 * deps. A selector of `undefined` means "the one configured project", which is
 * how a single-project host has always addressed itself: keeping it verbatim
 * means the single-project path still emits exactly the same `brief-upgrade`
 * command and the same one `snapshot` call it did before.
 *
 * `pauseKey` is the sentinel this transaction owns. It is `undefined` for a
 * host-wide scope, which selects the legacy *global* sentinel — and that is
 * precisely the right instrument: `isPaused(name)` consults the global file
 * for every project, so one write pauses the whole host, and removing it lifts
 * exactly this transaction's pause while leaving any per-project operator hold
 * standing.
 */
export interface UpgradeScope {
  /** Project selectors to address the project-aware deps with, never empty. */
  selectors: readonly (string | undefined)[];
  /** The sentinel this transaction pauses and resumes. */
  pauseKey: string | undefined;
}

/**
 * The subset of {@link UpgradeDeps} the pause/drain fence needs. Factored out
 * so a caller that only wants the fence — `setup host` — can provide just
 * these members instead of a full install-and-verify deps object. Every
 * {@link UpgradeDeps} is a `DrainDeps`.
 */
/**
 * The daemon a transaction targets: running or not, its project when it
 * recorded one, and a `generation` identifying the exact instance (#377).
 */
export interface DaemonIdentity {
  running: boolean;
  project?: string;
  generation?: string;
}

/**
 * The subset of {@link UpgradeDeps} the pause/drain fence needs. Factored out
 * so a caller that only wants the fence — `setup host` — can provide just
 * these members instead of a full install-and-verify deps object. Every
 * {@link UpgradeDeps} is a `DrainDeps`.
 */
export interface DrainDeps {
  /** Live worker count per project selector; `undefined` = the one configured project. */
  snapshot(project?: string): { liveWorkers: number };
  /** The fleet's current dispatch/daemon layers for one project selector. */
  layers(project?: string): FleetLayers;
  /** Every configured project name, in config order (#389). */
  projectNames(): readonly string[];
  /**
   * The daemon the transaction targets: running or not, its project when it
   * recorded one, and a `generation` identifying the exact instance (#377).
   */
  daemonIdentity(): DaemonIdentity;
  /** The active pause sentinel as an instance, when one is readable (#377). */
  pauseState(project?: string): { source: string; reason?: string; since: number } | undefined;
  /** Write or clear the durable pause sentinel for the scope's project. */
  setPaused(value: boolean, project?: string): void;
  /**
   * The daemon's durable admission acknowledgement for the current fence, when
   * one is readable: the daemon attests it has observed the pause sentinel at
   * an admission boundary and will claim nothing while it stands (#651 review
   * #3). Absent when no daemon is running, when the file cannot be read, or
   * when the daemon has not (yet) observed the fence. The setup barrier
   * requires this before any mutation when a daemon is running.
   */
  admissionAck?(project?: string): AdmissionAckRecord | undefined;
  sleep(ms: number): Promise<void>;
  log(message: string): void;
}

/**
 * Decides which projects a transaction covers, and refuses the requests that
 * cannot be honoured truthfully.
 *
 * Two refusals, both about a caller naming a project the daemon does not agree
 * with:
 *
 *  - the daemon records a *different* project — it serves that one, so a
 *    request aimed at another is simply about a different fleet;
 *  - the daemon records *no* project while several are configured. That is not
 *    "unknown", it is the multi-project daemon: it serves them all. Narrowing
 *    is refused rather than obeyed, because draining one project and then
 *    restarting the shared daemon would kill the other projects' live workers
 *    without ever counting them (#389, and the shared-daemon rule in #378).
 *
 * A single-configured-project host keeps the historical bare selector, so its
 * behaviour is unchanged whether or not the daemon recorded its name.
 */
export function resolveScope(
  deps: DrainDeps,
  verb: DrainVerb,
  project?: string,
): UpgradeScope {
  let configured: readonly string[] = [];
  try {
    configured = deps.projectNames();
  } catch {
    // No readable config is not this function's error to raise: the caller
    // reaches a project-aware dep moments later and fails with the real
    // problem. Treat it as the historical bare case.
  }
  const daemon = deps.daemonIdentity();

  if (project !== undefined) {
    if (daemon.running && daemon.project !== undefined && daemon.project !== project) {
      throw new Error(
        `active daemon serves ${daemon.project}; refusing to ${verb} --project ${project} while it is running`,
      );
    }
    if (daemon.running && daemon.project === undefined && configured.length > 1) {
      throw new Error(
        `the running daemon serves all ${String(configured.length)} configured projects ` +
          `(${configured.join(", ")}); ${verb} replaces host-wide packages and restarts that one daemon, ` +
          `so it cannot be narrowed to --project ${project} — re-run \`omp-conductor ${verb}\` without --project`,
      );
    }
    return { selectors: [project], pauseKey: project };
  }

  // A daemon that recorded its project serves that project alone; the rest of
  // the config is somebody else's fleet as far as this restart is concerned.
  if (daemon.running && daemon.project !== undefined) {
    return { selectors: [daemon.project], pauseKey: daemon.project };
  }
  if (configured.length > 1) return { selectors: configured, pauseKey: undefined };
  return { selectors: [undefined], pauseKey: undefined };
}

export async function waitForDrain(
  deps: DrainDeps,
  scope: UpgradeScope,
  deadlineAt?: number,
  stale?: () => string | undefined,
): Promise<void> {
  let last = -1;
  while (true) {
    // A transaction that is no longer the fleet's current intent exits before
    // it waits on anyone else's workers: a resume lifted its pause or the
    // daemon it began with was replaced, so there is nothing left to drain
    // and nothing it may stop (#377).
    const cancelled = stale?.();
    if (cancelled !== undefined) throw new Error(cancelled);
    // Every project the scope covers: a host-wide drain that counted only one
    // would report zero while another project's worker was still writing to a
    // worktree the restart is about to orphan (#389).
    const workers = scope.selectors.reduce((n, s) => n + deps.snapshot(s).liveWorkers, 0);
    if (workers === 0) return;
    if (deadlineAt !== undefined && Date.now() >= deadlineAt) {
      throw new Error(
        `drain timed out with ${workers} live worker(s) still running — nothing was restarted; dispatch remains paused (omp-conductor resume to lift it, or re-run restart to keep waiting)`,
      );
    }
    if (workers !== last) deps.log(`waiting for ${workers} live worker(s) to finish`);
    last = workers;
    await deps.sleep(DRAIN_POLL_MS);
  }
}

/**
 * What a draining restart captured when it began: the restart-owned pause and
 * the daemon instance it is entitled to stop. Every later fence check compares
 * live state against this snapshot, so the destructive restart call only ever
 * acts on the world the request began in (#377).
 */
export interface RestartBegun {
  /** The durable pause sentinel this request relies on, proved readable as an instance. */
  pauseToken: { source: string; reason?: string; since: number };
  /** The daemon instance the restart targets, with its generation identity. */
  daemon: DaemonIdentity;
}

/**
 * Why a pause-and-drain transaction can no longer act, or `undefined` when it
 * still owns the pause and the daemon generation it began with. The pause
 * check is the *transaction-owned sentinel instance* — the caller proves
 * ownership up front, so a token that cannot be re-read (sentinel removed,
 * unreadable, or re-created — even under the same source and reason) means
 * this request's own pause no longer exists and it must cancel. The
 * generation check is the later-daemon fence: a daemon created after the
 * transaction began — by `restart --now`, a crash-restart, or an operator
 * start — must never be stopped by it.
 *
 * Exported so the setup apply's acknowledged quiescence barrier runs the same
 * fence instead of inventing a second counter-only one (#618): it freezes
 * admission host-wide through {@link pauseAndDrain}'s machinery, and this
 * check is what proves the freeze is still the one it acknowledged.
 */
export function restartFenceProblem(
  deps: DrainDeps,
  scope: UpgradeScope,
  begun: RestartBegun,
  actor: string,
): string | undefined {
  if (!deps.layers(scope.pauseKey).paused) {
    return `${actor} cancelled: dispatch was resumed while the drain was in progress — nothing was restarted`;
  }
  const owned = deps.pauseState(scope.pauseKey);
  if (
    owned === undefined ||
    owned.since !== begun.pauseToken.since ||
    owned.source !== begun.pauseToken.source ||
    owned.reason !== begun.pauseToken.reason
  ) {
    return `${actor} cancelled: the ${actor}-owned pause was lifted and replaced while the drain was in progress — nothing was restarted`;
  }
  const now = deps.daemonIdentity();
  if (now.running !== begun.daemon.running || (begun.daemon.running && now.generation !== begun.daemon.generation)) {
    return `${actor} cancelled: the daemon instance was replaced while the drain was in progress — the newer daemon was left running`;
  }
  return undefined;
}

/**
 * Every verb {@link pauseAndDrain} accepts, declared once and used for both the
 * `verb` parameter's type and the sentinel round-trip test — so a fourth verb
 * that cannot be encoded as a single `source=` token (see
 * {@link pauseSourceToken}; a verb with a space) fails the suite instead of
 * shipping an unprovable pause (#552).
 */
export const DRAIN_VERBS = ["upgrade", "restart", "setup host", "setup"] as const;
export type DrainVerb = (typeof DRAIN_VERBS)[number];

/**
 * Pause claims and drain live workers to idle — the destructive half of the
 * trusted restart transaction, minus the restart. Exported so `setup host`
 * runs the same fence `upgrade` and `restart` do, then performs its own unit
 * installs instead of {@link drainAndRestart}'s daemon restart.
 *
 * Generation-scoped (#377) exactly as the full transaction is: the caller
 * captures the restart-owned pause and the daemon generation, and the drain
 * aborts if a `resume` lifts the pause, replaces the sentinel, or a newer
 * daemon generation appears. The fence failure path (the pause it just
 * created cannot be read back as an instance) fails closed WITHOUT leaving
 * that pause behind: a pause the transaction cannot prove it is still using
 * is released, restoring the entry state, so the refusal never stops dispatch
 * on its own (#552). A drain that outlives `timeoutMs` stays paused (no
 * resume in a catch) — a drain cannot silently resume dispatch over a wedged
 * daemon. `timeoutMs` absent waits indefinitely, which is the `upgrade`
 * posture.
 */
export async function pauseAndDrain(
  deps: DrainDeps,
  verb: DrainVerb,
  o: { project?: string; timeoutMs?: number },
): Promise<{ scope: UpgradeScope; initialPaused: boolean }> {
  // Host-wide by default (#389): the daemon this restarts serves every
  // configured project, so the drain counts every project's workers and the
  // pause covers all of them.
  const scope = resolveScope(deps, verb, o.project);
  const initial = deps.layers(scope.pauseKey);
  if (!initial.paused) deps.setPaused(true, scope.pauseKey);
  // The lock must be provable NOW. A pause that cannot be read as an instance
  // — sentinel malformed or unreadable — would make the whole fence fail open
  // (nothing to compare against), so the transaction refuses before waiting
  // on anything it cannot act upon either way.
  const pauseToken = deps.pauseState(scope.pauseKey);
  if (pauseToken === undefined) {
    // Fail closed — but do not abandon our own pause. It is only correct to
    // hold a pause the transaction is still using, and a sentinel we cannot
    // prove is unusable: restore the entry state exactly, so a verb that
    // paused the fleet releases it (a fleet already paused stays paused)
    // before the refusal propagates (#552).
    if (!initial.paused) deps.setPaused(false, scope.pauseKey);
    throw new Error(
      `${verb} cancelled: cannot prove the ${verb}-owned pause — the active pause sentinel is unreadable or malformed; nothing was restarted`,
    );
  }
  const begun: RestartBegun = {
    pauseToken,
    daemon: deps.daemonIdentity(),
  };
  const stale = () => restartFenceProblem(deps, scope, begun, verb);
  await waitForDrain(
    deps,
    scope,
    o.timeoutMs === undefined ? undefined : Date.now() + o.timeoutMs,
    stale,
  );
  // The drain completed; the world may have moved on while it did. Re-prove
  // the pause and the generation an instant before the destructive call.
  const cancelled = stale();
  if (cancelled !== undefined) throw new Error(cancelled);
  return { scope, initialPaused: initial.paused };
}

/**
 * Pause, drain, restart, restore — the trusted restart transaction, minus the
 * install/verify steps of {@link upgradeConductor}.
 *
 * Generation-scoped (#377): the transaction captures the restart-owned pause
 * and the daemon generation when it begins and revalidates both on every drain
 * poll and once more immediately before the destructive restart call. A
 * `resume`, a `restart --now`, or any other action that lifts the pause or
 * replaces the daemon while the drain waits makes the transaction stale: it
 * exits nonzero without stopping anything, so a timed-out or disconnected
 * restart caller cannot leave a delayed stop aimed at a later daemon
 * generation. The stale transaction restores nothing and leaves whatever
 * pause the fleet now has exactly as it found it. Ownership is proven up
 * front: a sentinel that cannot be read as an instance at transaction start
 * aborts before any drain or restart, because a lock the fence cannot compare
 * against is a lock the fence cannot protect.
 *
 * Mirrors the upgrade's fail-closed posture: on ANY throw after the pause the
 * fleet stays paused (no resume in a catch) and the error rethrows, so a
 * systemd-owned restart that fails or a drain that outlives `timeoutMs` cannot
 * silently resume dispatch over a wedged daemon. The caller surfaces the message
 * and exits nonzero.
 */
export async function drainAndRestart(
  deps: UpgradeDeps,
  o: { project?: string; timeoutMs: number },
): Promise<void> {
  const { scope, initialPaused } = await pauseAndDrain(deps, "restart", o);
  await deps.restartDaemon();
  if (!initialPaused) deps.setPaused(false, scope.pauseKey);
}

function recoveryProblem(
  layers: FleetLayers,
  initial: FleetLayers,
  liveWorkers: number,
): string | undefined {
  if (layers.ticks !== initial.ticks) return `ticks changed from ${initial.ticks} to ${layers.ticks}`;
  if (layers.recovery !== initial.recovery) {
    return `recovery changed from ${initial.recovery} to ${layers.recovery}`;
  }
  if (initial.herdr === "active" && layers.herdr !== "active") return `Herdr is ${layers.herdr}`;
  if (initial.herdr === "active" && initial.recovery === "clear" && layers.pane !== "live") {
    return `orchestrator pane is ${layers.pane}`;
  }
  if (initial.dispatch !== "stopped" && !layers.daemon.running) return "dispatch daemon is not running";
  if (initial.dispatch === "stopped" && layers.daemon.running) return "stopped dispatch daemon was started";
  if (initial.dispatch !== "stopped" && !layers.paused) return "dispatch resumed before verification";
  if (liveWorkers !== 0) return `${liveWorkers} worker(s) appeared during verification`;
  return undefined;
}

async function waitForRecovery(
  deps: UpgradeDeps,
  initial: FleetLayers,
  scope: UpgradeScope,
): Promise<void> {
  let problem = "recovery did not settle";
  for (let attempt = 0; attempt < RECOVERY_ATTEMPTS; attempt += 1) {
    const live = scope.selectors.reduce((n, s) => n + deps.snapshot(s).liveWorkers, 0);
    problem = recoveryProblem(deps.layers(scope.pauseKey), initial, live) ?? "";
    if (problem === "") return;
    await deps.sleep(RECOVERY_POLL_MS);
  }
  throw new Error(`upgrade verification failed: ${problem}`);
}

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

/** One pass of the post-reload session verification (#832): read the restarted
 * daemon's attestation (embedded orchestrator: the live session's loaded
 * version; external: the pane probe's process starts) and answer whether
 * every in-scope live session proves it reloaded the target release after
 * the install began. `undefined` is the pass — every project the daemon
 * serves must have attested, because the restart covered the whole daemon at
 * once.
 *
 * The embedded leg is gated on the dispatch restart: a daemon that was
 * already stopped when the transaction began never hosted a session the
 * transaction could have reloaded, so its absence is not a failure. The
 * external leg is gated on the herdr restart: a transaction that left the
 * pane alone has no reason to demand proof the pane did anything.
 */
async function sessionVerifyProblem(
  deps: UpgradeDeps,
  targetVersion: string,
  reloadAfterMs: number,
  initial: FleetLayers,
): Promise<string | undefined> {
  const dispatchRestarted = initial.dispatch !== "stopped";
  const herdrRestarted = initial.herdr === "active";
  if (!dispatchRestarted && !herdrRestarted) return undefined;
  if (!dispatchRestarted) {
    // No daemon to attest: only the pane could have reloaded anything, so the
    // pane probe decides alone, exactly as the external leg of the detached
    // verifier does.
    const pane = await deps.probePaneOmp(resolveHerdrSession(deps.env));
    if ("problem" in pane) return `the fleet pane could not be probed: ${pane.problem}`;
    return paneRestartProblem(pane.starts, reloadAfterMs) ?? undefined;
  }
  const port = deps.layers().daemon.port ?? DEFAULT_PORT;
  const health = await deps.health(port);
  if (!health.ok) return `the restarted daemon did not answer /healthz on :${port}`;
  let parsed: unknown;
  try {
    parsed = JSON.parse(health.body ?? "");
  } catch {
    return `the restarted daemon's /healthz on :${port} is not JSON — the live session cannot be attested`;
  }
  const projects = projectsFromHealth(parsed);
  if (projects === undefined) {
    return `the restarted daemon's /healthz on :${port} carried no projects — the live session cannot be attested`;
  }
  const names: string[] = [];
  for (const entry of projects) {
    if (entry !== null && typeof entry === "object" && !Array.isArray(entry) && "project" in entry) {
      if (typeof entry.project === "string") names.push(entry.project);
    }
  }
  if (names.length === 0) {
    return `the restarted daemon's /healthz on :${port} named no project — the live session cannot be attested`;
  }
  const pane = await deps.probePaneOmp(resolveHerdrSession(deps.env));
  for (const project of names) {
    const problem = sessionReloadProblem({
      facts: orchestratorFactsFromHealth(health.body, project),
      targetVersion,
      reloadAfterMs,
      herdrRestarted: herdrRestarted && modeFromHealth(parsed, project) === "external",
      pane,
    });
    if (problem !== undefined) return problem;
  }
  return undefined;
}

/**
 * The `projects` array of a daemon /healthz payload, narrowed without an
 * unchecked cast: `undefined` when the payload is not an object with a
 * `projects` array — the caller decides what an unreadable answer means.
 */
function projectsFromHealth(parsed: unknown): unknown[] | undefined {
  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return undefined;
  if (!("projects" in parsed)) return undefined;
  return Array.isArray(parsed.projects) ? parsed.projects : undefined;
}

/**
 * One project's attested orchestrator mode off a parsed /healthz payload.
 * `undefined` when the entry or its mode cannot be read.
 */
function modeFromHealth(parsed: unknown, project: string): string | undefined {
  const projects = projectsFromHealth(parsed);
  if (projects === undefined) return undefined;
  for (const entry of projects) {
    if (entry === null || typeof entry !== "object" || Array.isArray(entry)) continue;
    if (!("project" in entry) || entry.project !== project) continue;
    if (!("orchestrator" in entry)) return undefined;
    const orchestrator = entry.orchestrator;
    if (orchestrator === null || typeof orchestrator !== "object" || Array.isArray(orchestrator)) {
      return undefined;
    }
    if (!("mode" in orchestrator) || typeof orchestrator.mode !== "string") return undefined;
    return orchestrator.mode;
  }
  return undefined;
}

/**
 * The verify leg after the reloads (#832): the upgrade only reports success
 * once the live orchestrator process has proven it executes the newly
 * installed extension. Polls the attestation within the same recovery window
 * as the other post-reload checks — the restarted daemon needs seconds to
 * boot its orchestrator child and answer — then throws into the transaction's
 * rollback path, so a failed session reload is a durable, actionable failure
 * and never a reported success.
 */
async function waitForSessionVerified(
  deps: UpgradeDeps,
  targetVersion: string,
  reloadAfterMs: number,
  initial: FleetLayers,
): Promise<void> {
  let problem = "the live orchestrator session did not attest a reload";
  for (let attempt = 0; attempt < RECOVERY_ATTEMPTS; attempt += 1) {
    problem = (await sessionVerifyProblem(deps, targetVersion, reloadAfterMs, initial)) ?? "";
    if (problem === "") return;
    await deps.sleep(RECOVERY_POLL_MS);
  }
  throw new Error(`upgrade verification failed: ${problem}`);
}

/** One project's brief, as the transaction found it. */
interface ScopedBrief {
  /** The selector to address the project-aware deps and the CLI with. */
  selector: string | undefined;
  kind: BriefLayout["kind"];
  current: boolean;
}

/**
 * Brings every in-scope brief up with the freshly installed package.
 *
 * Each project owns its own ORCHESTRATOR.md, so a host-wide upgrade migrates
 * them one at a time rather than leaving every project but one on a floor the
 * installed package no longer emits (#389).
 *
 * `brief-upgrade --migrate --apply` is the cross-version ABI: this code runs
 * from the *old* CLI while the new one is already installed, so the verb has to
 * keep working across the boundary. When it does not — a flag renamed, the verb
 * gone — the tolerable outcome depends entirely on the layout:
 *
 *  - `overlay` fleets need nothing. The floor is recomposed from the installed
 *    package on every tick, so a failed migrate costs nothing and stopping the
 *    upgrade over it would be the worse trade.
 *  - a legacy layout needs it. Skipping the migration silently would leave the
 *    fleet on a single-file brief whose floor no longer tracks the package —
 *    the exact drift this verb exists to end — so it fails loudly and rolls back.
 */
async function upgradeBriefs(deps: UpgradeDeps, briefs: readonly ScopedBrief[]): Promise<void> {
  for (const brief of briefs) {
    const selected = brief.selector === undefined ? [] : ["--project", brief.selector];
    if (brief.kind === "legacy-handwritten") {
      // Ungated on purpose: a hand-written brief cannot be migrated without it.
      await mustRun(deps, "omp-conductor", ["brief-upgrade", "--retrofit", "--apply", ...selected]);
    }
    try {
      await mustRun(deps, "omp-conductor", ["brief-upgrade", "--migrate", "--apply", ...selected]);
    } catch (err) {
      if (brief.kind !== "overlay") throw err;
      const msg = err instanceof Error ? err.message : String(err);
      deps.log(`brief: brief-upgrade unavailable in the target CLI (${msg}) — overlay recomposes each tick, continuing`);
    }
  }
}

/** Restore every surface the upgrade touched, in reverse, from the identities
 *  captured before the first install. Exported so the detached rollback unit
 *  can restore what the detached installer recorded (#486). */
export async function rollbackUpgrade(
  deps: UpgradeDeps,
  previous: InstalledSurfaces,
  scope: UpgradeScope,
  installTouched: boolean,
  briefChanged: boolean,
  herdrReloadStarted: boolean,
  daemonReloadStarted: boolean,
  configBefore?: string,
  preUpgradeBackup?: string,
): Promise<void> {
  const failures: string[] = [];

  // Restored FIRST, before the packages that have to read it.
  //
  // Rolling back binaries alone is not a rollback. The new daemon starts during
  // the upgrade and may migrate config.json; the reinstated older release then
  // cannot parse its own config and will not boot — which is exactly how an
  // upgrade failure turned into an outage. The snapshot is taken before the
  // first install, and restoring it is unconditional: writing back a file that
  // never changed is harmless, and detecting "did it change" is one more thing
  // to get wrong while the fleet is down.
  if (configBefore !== undefined) {
    try {
      const path = configPath();
      if (readFileSync(path, "utf8") !== configBefore) {
        deps.log("rollback: conductor config.json");
        // Name the durable snapshot: the in-memory restore only runs while
        // this process is alive, and an upgrade that died mid-flight leaves
        // the operator needing exactly these bytes from disk.
        deps.log(
          preUpgradeBackup === undefined
            ? "rollback: config.json changed during the upgrade; no durable snapshot was persisted"
            : `rollback: config.json changed during the upgrade; pre-upgrade snapshot at ${preUpgradeBackup}`,
        );
        // Exact bytes. `writeConfigFile(JSON.parse(...))` returns canonically
        // formatted output, which restores the keys but not the file.
        writeConfigRaw(configBefore);
      }
    } catch (err) {
      failures.push(
        `could not restore config.json: ${err instanceof Error ? err.message : String(err)}` +
          (preUpgradeBackup === undefined ? "" : `; pre-upgrade snapshot at ${preUpgradeBackup}`),
      );
    }
  }
  const restore = async (label: string, command: string, args: readonly string[]): Promise<void> => {
    deps.log(`rollback: ${label}`);
    try {
      await mustRun(deps, command, args);
    } catch (err) {
      failures.push(err instanceof Error ? err.message : String(err));
    }
  };
  const removeIfPresent = async (
    label: string,
    command: string,
    args: readonly string[],
  ): Promise<void> => {
    deps.log(`rollback: ${label}`);
    try {
      await deps.run(command, args);
    } catch {
      // Exact post-rollback identity verification decides whether absence is safe.
    }
  };
  if (installTouched) {
    if (previous.management === undefined) {
      await restore("legacy Bun-global omp-conductor CLI", "bun", [
        "add",
        "-g",
        `${PACKAGE}@${previous.cliVersion}`,
      ]);
    } else {
      await restore(`omp-conductor CLI at ${previous.management.cli.root}`, "bun", [
        "add",
        "--cwd",
        previous.management.cli.root,
        "--exact",
        `${PACKAGE}@${previous.cliVersion}`,
      ]);
    }
    if (previous.ompVersion === undefined) {
      await removeIfPresent("remove newly installed omp plugin", "omp", [
        "plugin",
        "uninstall",
        PACKAGE,
      ]);
    } else {
      await restore("omp plugin omp-conductor", "omp", [
        "plugin",
        "install",
        `${PACKAGE}@${previous.ompVersion}`,
      ]);
    }

    if (previous.herdrSource === undefined) {
      await removeIfPresent("remove newly installed Herdr plugin", "herdr", [
        "plugin",
        "uninstall",
        HERDR_PLUGIN,
      ]);
    } else {
      const install = previousHerdrInstall(previous.herdrSource);
      if (install === undefined) {
        failures.push(`cannot restore unknown Herdr source ${previous.herdrSource}`);
      } else {
        if (previous.herdrSource.startsWith("local:")) {
          await removeIfPresent("remove managed Herdr plugin", "herdr", [
            "plugin",
            "uninstall",
            HERDR_PLUGIN,
          ]);
        }
        await restore("Herdr plugin herdr-conductor", install[0], install[1]);
      }
    }

  }
  if (briefChanged) {
    try {
      deps.log("rollback: ORCHESTRATOR.md package floor");
      await upgradeBriefs(
        deps,
        scope.selectors.map((selector) => ({ selector, kind: "overlay" as const, current: false })),
      );
    } catch (err) {
      failures.push(err instanceof Error ? err.message : String(err));
    }
  }

  // The units the forward transaction reconciled (#905). Restoring them is the
  // same operation in the other direction: the *restored* CLI renders the
  // previous version's templates, so running its own reconcile puts the
  // pre-upgrade bytes back at every destination — and leaves anything the
  // operator edited alone, because that drift is protected on both legs.
  //
  // Ordered before the restarts below, exactly as the forward leg is, and
  // best-effort: a rollback that restored three surfaces must not fail because
  // a unit file could not be rewritten. The failure is collected and reported
  // with the rest.
  if (installTouched) {
    const restoreUnits = await deps.run("omp-conductor", ["reconcile-units", "--yes"]);
    const summary = (restoreUnits.stdout.trim() || restoreUnits.stderr.trim()).split("\n").pop() ?? "";
    if (restoreUnits.code === 0) {
      deps.log(`rollback: ${summary.length === 0 ? "host units restored" : summary}`);
    } else {
      failures.push(`host units not restored: ${summary}`);
    }
  }

  if (herdrReloadStarted) {
    await restore("restart herdr-fleet.service", "systemctl", ["restart", HERDR_UNIT]);
  }
  if (daemonReloadStarted) {
    try {
      deps.log("rollback: restart omp-conductor dispatch daemon");
      await deps.restartDaemon();
    } catch (err) {
      failures.push(err instanceof Error ? err.message : String(err));
    }
  }

  if (installTouched) {
    try {
      verifyRestoredManagement(deps, previous);
      const restored = await inspectSurfaces(deps);
      if (
        restored.cliVersion !== previous.cliVersion ||
        restored.ompVersion !== previous.ompVersion ||
        restored.herdrSource !== previous.herdrSource
      ) {
        failures.push(
          `restored identities differ: cli=${restored.cliVersion}, omp=${restored.ompVersion ?? "missing"}, ` +
            `herdr=${restored.herdrSource ?? "missing"}`,
        );
      }
    } catch (err) {
      failures.push(err instanceof Error ? err.message : String(err));
    }
  }

  if (failures.length > 0) throw new Error(failures.join("; "));
}

/**
 * Name the host-runtime destinations this version's render no longer matches,
 * after a successful upgrade — the installed units, the recovery playbook at
 * {@link RECOVER_SCRIPT_INSTALL_PATH}, the herdr pane-shell config and the
 * herdr-conductor `config.env`. A release that re-renders any of those files
 * silently invalidates the installed copies; telling the operator here — with
 * the exact command — is what stops the drift from surviving until a later
 * `doctor` finds it (#598).
 *
 * The plan *is* the comparison: {@link UpgradeDeps.hostRuntime} renders every
 * destination against disk, so a current host prints nothing and a drifted one
 * names exactly the files only a privileged `setup host` can refresh. This
 * never mutates a host file and never fails the upgrade — an unreadable host
 * state is a warning line, not a reason to roll a successful install back.
 */
function logHostRuntimeDrift(deps: UpgradeDeps): void {
  let plan: HostRuntimePlan;
  try {
    plan = deps.hostRuntime();
  } catch (err) {
    deps.log(
      `host runtime: could not compare against this version's render — ${err instanceof Error ? err.message : String(err)}`,
    );
    return;
  }
  if (plan.drift.length === 0) return;
  deps.log("host runtime:");
  for (const path of plan.drift) deps.log(`  ${path} differs from this version's render`);
  // Units are this transaction's own to reconcile (#905), and the reconcile
  // verb is bounded to them. Anything else the render owns — the herdr
  // pane-shell config, the herdr-conductor `config.env`, the worker identity —
  // still needs `setup host`, so the advisory names the right tool for each
  // rather than routing every drift into the flow that restarts (and kills)
  // the caller's own pane (#834).
  deps.log(
    "fix: `omp-conductor reconcile-units` re-installs the host units; anything else above needs " +
      "`omp-conductor setup host` from the fleet account",
  );
}

export async function upgradeConductor(
  options: UpgradeOptions = {},
  overrides: Partial<UpgradeDeps> = {},
): Promise<UpgradeResult> {
  const deps: UpgradeDeps = { ...DEFAULT_DEPS, ...overrides };
  const detached = options.detached === true;
  const journal = (entry: Omit<UpgradeJournalEntry, "at">, at = new Date()): void => {
    deps.journal?.({ ...entry, at: at.toISOString() });
  };
  if (deps.env["HERDR_ENV"] !== undefined) {
    throw new Error("run omp-conductor upgrade from a shell outside the target Herdr session");
  }

  // Host-wide by default (#389): one daemon serves every configured project,
  // so an upgrade that restarts it drains them all and refreshes every brief.
  const scope = resolveScope(deps, "upgrade", options.project);

  // Published semver, or an exact source/build identity whose checks have
  // already passed (#908). Everything below reads the same pair either way.
  const identity = await resolveIdentity(deps, options);
  const release = { version: identity.version, gitHead: identity.gitHead };
  const requestedProjects = (deps.env["OMP_CONDUCTOR_INSTALL_PROJECTS"] ?? "")
    .split(",")
    .filter((name) => name.length > 0);
  const requestedHolder = deps.env["OMP_CONDUCTOR_INSTALL_HOLDER"];
  const installHolder =
    requestedHolder === "human" || requestedHolder === "orchestrator"
      ? requestedHolder
      : undefined;
  const installProjects: readonly (string | undefined)[] =
    requestedProjects.length === 0 ? scope.selectors : requestedProjects;
  const installAttribution =
    `Host-global install holder: ${installHolder ?? "unknown"}; affected projects: ` +
    `${installProjects.map((selector) => selector ?? "(default)").join(", ")}.`;
  if (identity.bootstrap) {
    // The bootstrap's evidence lands in the journal before the first surface
    // moves, so an interrupted bootstrap leaves a record of what was verified
    // rather than an unexplained half-install.
    journal({
      kind: "request",
      ok: true,
      version: release.version,
      gitHead: release.gitHead,
      phase: "bootstrap",
      detail: `source identity from ${options.bootstrap?.source ?? "(source)"}`,
      checks: identity.checks,
      unit: deps.env["OMP_CONDUCTOR_UNIT"],
    });
  }
  if (detached) {
    // The durable request: written before anything can pause, so a unit that
    // dies after this line leaves a journal the returning process can act on
    // (roll back, resume, or report) instead of a paused fleet with no record.
    journal({
      kind: "request",
      ok: true,
      version: release.version,
      gitHead: release.gitHead,
      unit: deps.env["OMP_CONDUCTOR_UNIT"],
      selectors: scope.selectors.map((selector) => selector ?? null),
      ...(requestedProjects.length === 0 ? {} : { installProjects: requestedProjects }),
      ...(installHolder === undefined ? {} : { installHolder }),
    });
  }
  const initial = deps.layers(scope.pauseKey);
  const observedSurfaces = await inspectSurfaces(deps);
  const briefs: ScopedBrief[] = scope.selectors.map((selector) => ({
    selector,
    ...deps.brief(selector),
  }));
  const installNeeded = !surfacesCurrent(observedSurfaces, release.version, release.gitHead);
  const management = installNeeded ? await discoverInstallManagement(deps) : undefined;
  if (management !== undefined) await preflightInstall(deps, management, identity.packageSpec);
  const surfaces: InstalledSurfaces = {
    ...observedSurfaces,
    ...(management === undefined ? {} : { management }),
  };
  // The host-unit baseline, read while the OLD package is still the one
  // rendering (#905). A destination already differing from the old render was
  // edited outside conductor, so the reconcile after the install must not
  // overwrite it; `undefined` means the baseline could not be established at
  // all, and the reconcile is skipped rather than guessing.
  const preInstallDrift = ((): string[] | undefined => {
    try {
      return [...deps.hostRuntime().drift];
    } catch {
      return undefined;
    }
  })();
  if (!installNeeded && briefs.every((b) => b.current)) {
    if (detached) {
      journal({
        kind: "outcome",
        phase: "already-current",
        ok: true,
        version: release.version,
        detail: "all three surfaces and every brief were already pinned to the target release",
      });
    }
    // Packages are current, but the host units they render may be behind: the
    // same drift this advisory names after an install applies when a release
    // re-rendered a template since the operator's last `setup host`. Silent on
    // a host whose runtime matches (#598).
    logHostRuntimeDrift(deps);
    return {
      previousVersion: surfaces.cliVersion,
      ...release,
      alreadyCurrent: true,
      dispatch: initial.dispatch,
    };
  }

  // Snapshotted before anything is installed, because the new daemon starts
  // during this transaction and may migrate the file — see the restore in
  // `rollbackUpgrade`. Read as bytes, not through the loader: the point is to
  // put back exactly what was there, dialect and all. The same bytes are also
  // persisted under state, so a process death between here and the rollback
  // cannot take the only pre-upgrade copy with it.
  let configBefore: string | undefined;
  let preUpgradeBackup: string | undefined;
  try {
    configBefore = readFileSync(configPath(), "utf8");
    preUpgradeBackup = copyToUniqueBackup(
      configPath(),
      configBackupDir(),
      `config.json.pre-upgrade-${backupTimestamp()}`,
    );
    deps.log(`config backed up to ${preUpgradeBackup}`);
  } catch {
    // No readable config is not a reason to refuse an upgrade; there is simply
    // nothing to put back. A snapshot that failed to persist is not fatal
    // either — the in-memory restore below still covers a clean rollback.
  }

  const missing = briefs.find((b) => b.kind === "missing");
  if (missing !== undefined) {
    throw new Error(
      missing.selector === undefined
        ? "no ORCHESTRATOR.md exists for the configured project"
        : `no ORCHESTRATOR.md exists for project ${missing.selector}`,
    );
  }
  if (initial.herdr === "unknown") throw new Error("cannot determine whether herdr-fleet.service is active");

  // The one verification input that can be read before anything moves (#976).
  // Verification probes the fleet pane in `resolveHerdrSession(deps.env)`; a
  // host whose session answers to another name only discovers that after three
  // installs and two restarts, and a good release is then rolled back. Read it
  // first: the same probe, at zero cost, as a refusal.
  if (!detached && (initial.dispatch !== "stopped" || initial.herdr === "active")) {
    const session = resolveHerdrSession(deps.env);
    const pane = await deps.probePaneOmp(session);
    if ("problem" in pane) {
      throw new Error(
        `the fleet pane in herdr session "${session}" cannot be probed, so this upgrade could not be ` +
          `verified: ${pane.problem}. Set HERDR_SESSION to the session that owns the fleet pane and ` +
          `re-run, or bring that session up first — nothing has been installed.`,
      );
    }
  }

  if (detached) {
    // The durable record the returning process rolls back or verifies against:
    // what was installed, what dispatch state the fleet began in, which pause
    // sentinel this transaction owns, and the config snapshot for the restore.
    journal({
      kind: "snapshot",
      ok: true,
      version: release.version,
      gitHead: release.gitHead,
      unit: deps.env["OMP_CONDUCTOR_UNIT"],
      previous: {
        cliVersion: surfaces.cliVersion,
        ompVersion: surfaces.ompVersion,
        herdrSource: surfaces.herdrSource,
        ...(surfaces.management === undefined ? {} : { management: surfaces.management }),
      },
      ...(preUpgradeBackup === undefined ? {} : { configBackup: preUpgradeBackup }),
      pauseKey: scope.pauseKey,
      initialPaused: initial.paused,
      initial: {
        dispatch: initial.dispatch,
        ticks: initial.ticks,
        pane: initial.pane,
        herdr: initial.herdr,
      },
      selectors: scope.selectors.map((s) => s ?? null),
    });
  }

  deps.log(`target release: omp-conductor@${release.version} (${release.gitHead})`);
  if (!initial.paused) {
    deps.log("safety: pausing new issue claims");
    deps.setPaused(true, scope.pauseKey);
  }
  journal({
    kind: "phase",
    phase: "paused",
    surface: "dispatch",
    ok: true,
    version: release.version,
    pauseSince: deps.pauseState(scope.pauseKey)?.since,
  });
  deps.log("drain: waiting for live omp worker sessions");
  try {
    await waitForDrain(deps, scope);
  } catch (err) {
    journal({ kind: "phase", phase: "drain", ok: false, version: release.version, detail: err instanceof Error ? err.message : String(err) });
    const failure = err instanceof Error ? err.message : String(err);
    throw new Error(
      `upgrade failed while draining workers: ${failure}; no installation started; ` +
        `dispatch remains paused — resume dispatch with \`omp-conductor resume --all\``,
    );
  }
  journal({ kind: "phase", phase: "drain", ok: true, version: release.version });
  deps.log("drain: complete");

  let installTouched = false;
  let briefChanged = false;
  let herdrReloadStarted = false;
  let daemonReloadStarted = false;
  // The instant the transaction began replacing surfaces: the deadline the
  // live orchestrator session must prove it restarted after (#832).
  const reloadStarted = Date.now();
  try {
    if (installNeeded) {
      if (management === undefined) {
        throw new Error("upgrade invariant failed: install needed without discovered surface management");
      }
      installTouched = true;
      deps.log(`install 1/3: omp-conductor CLI at ${management.cli.root} → ${release.version}`);
      await mustRun(deps, "bun", [
        "add",
        "--cwd",
        management.cli.root,
        "--exact",
        identity.packageSpec,
      ]);
      journal({ kind: "phase", phase: "install", surface: "cli", ok: true, version: release.version, gitHead: release.gitHead });
      deps.log(`install 2/3: omp plugin omp-conductor → ${release.version}`);
      await mustRun(deps, "omp", ["plugin", "install", identity.packageSpec]);
      journal({ kind: "phase", phase: "install", surface: "omp", ok: true, version: release.version, gitHead: release.gitHead });
      if (surfaces.herdrSource?.startsWith("local:")) {
        await mustRun(deps, "herdr", ["plugin", "unlink", HERDR_PLUGIN]);
      }
      deps.log(`install 3/3: Herdr plugin herdr-conductor → ${release.gitHead}`);
      await mustRun(deps, "herdr", [
        "plugin",
        "install",
        HERDR_SOURCE,
        "--ref",
        release.gitHead,
        "--yes",
      ]);
      journal({ kind: "phase", phase: "install", surface: "herdr", ok: true, version: release.version, gitHead: release.gitHead });
    } else {
      deps.log("install: CLI, omp plugin, and Herdr plugin already pinned to the target release");
    }

    deps.log("brief: refreshing managed ORCHESTRATOR.md floor while preserving POLICY.md");
    briefChanged = true;
    await upgradeBriefs(deps, briefs);
    journal({ kind: "phase", phase: "brief", surface: "brief", ok: true, version: release.version });

    // #905: the templates this release re-rendered. The render is CODE, so
    // this process — which loaded the previous version before replacing it —
    // cannot see the new one: the reconcile runs as a fresh child of the
    // just-installed CLI, which is also why it is spawned rather than called.
    // (Not an escalation of conductor itself: the child escalates only its own
    // bounded `install`/`daemon-reload` steps, which is exactly the boundary
    // `privileged.ts` draws.)
    //
    // It lands here, before the restarts below, so the daemon and the herdr
    // session come up reading the units this release actually ships — and
    // inside the same pause/drain window, so nothing is admitted against a
    // half-reconciled host.
    {
      // No baseline means no *refresh*: drift the operator made cannot be told
      // from drift the upgrade made, so nothing is overwritten. It does not
      // mean no *retirement* (#895) — a unit this release stopped shipping is
      // not a comparison against any render, and skipping the whole child
      // there left an affected host mounting an obsolete bind after every
      // upgrade, forever. So the child always runs; only its refresh half is
      // gated.
      const refreshable = preInstallDrift !== undefined;
      if (!refreshable) {
        deps.log(
          "host units: refresh skipped — this version's host state could not be read before the install, " +
            "so drift the operator made cannot be told from drift the upgrade made. Retirement still runs; " +
            "run `omp-conductor reconcile-units --dry-run` to see what differs.",
        );
      }
      const reconcile = await deps.run("omp-conductor", [
        "reconcile-units",
        "--yes",
        ...(refreshable ? [] : ["--no-refresh"]),
        // Destinations that already differed from the OLD render are the
        // operator's own edits, not this upgrade's doing: named here so the
        // child leaves them alone and reports them.
        ...(preInstallDrift ?? []).flatMap((path) => ["--protect", path]),
      ]);
      const summary = (reconcile.stdout.trim() || reconcile.stderr.trim()).split("\n").pop() ?? "";
      if (reconcile.code === 0) {
        deps.log(summary.length === 0 ? "host units: reconciled" : summary);
        journal({ kind: "phase", phase: "reconcile", surface: "units", ok: true, version: release.version, detail: summary });
      } else {
        // A failed reconcile is a drifted host, not a failed upgrade: the
        // packages are installed and verifiable, and the units are exactly as
        // they were. Say so precisely instead of rolling three surfaces back.
        deps.log(`host units: reconcile did not complete — ${summary}`);
        journal({ kind: "phase", phase: "reconcile", surface: "units", ok: false, version: release.version, detail: summary });
      }
    }

    if (initial.herdr === "active") {
      herdrReloadStarted = true;
      deps.log("reload: restarting herdr-fleet.service and recovering the orchestrator pane");
      await mustRun(deps, "systemctl", ["restart", HERDR_UNIT]);
      journal({ kind: "phase", phase: "restart", surface: "herdr", ok: true, version: release.version });
    }
    if (initial.dispatch !== "stopped") {
      daemonReloadStarted = true;
      deps.log("reload: restarting the omp-conductor dispatch daemon");
      await deps.restartDaemon();
      journal({ kind: "phase", phase: "restart", surface: "daemon", ok: true, version: release.version });
    }

    if (detached) {
      // The installing process does not declare its own success: it journals
      // that the transaction reached the reloads and stops, leaving dispatch
      // paused exactly where it was. The first tick of the post-restart daemon
      // verifies against this journal and is the only witness that closes the
      // transaction (#486).
      journal({
        kind: "phase",
        phase: "awaiting-verification",
        ok: true,
        version: release.version,
        detail: "installed; the first tick after the restart verifies and restores dispatch",
      });
      return {
        previousVersion: surfaces.cliVersion,
        ...release,
        alreadyCurrent: false,
        dispatch: initial.dispatch,
      };
    }

    deps.log("verify 1/2: package identities, services, pane, ticks, and paused dispatch");
    const installed = await inspectSurfaces(deps);
    if (!surfacesCurrent(installed, release.version, release.gitHead)) {
      throw new Error(
        `installed identities differ: cli=${installed.cliVersion}, omp=${installed.ompVersion ?? "missing"}, ` +
          `herdr=${installed.herdrSource ?? "missing"}`,
      );
    }
    if (initial.herdr === "active") {
      const active = await mustRun(deps, "systemctl", ["is-active", HERDR_UNIT]);
      if (active.stdout.trim() !== "active") throw new Error(`${HERDR_UNIT} is not active after restart`);
    }
    await waitForRecovery(deps, initial, scope);
    deps.log("verify 2/2: recovered fleet remains stable");
    await deps.sleep(1_000);
    await waitForRecovery(deps, initial, scope);

    // verify 3/3 (#832): package identities are on disk, but the orchestrator
    // session process executes the module it LOADED at start — a session that
    // survived the restarts keeps running the pre-upgrade extension no matter
    // what the packages say. The transaction only reports success once the
    // live session (daemon-attested for the embedded orchestrator, pane start
    // times for the external one) proves it reloaded the installed release. A
    // fleet that restarted nothing — dispatch stopped AND herdr inactive —
    // has no live orchestrator to prove, so there is nothing to attest.
    if (initial.dispatch !== "stopped" || initial.herdr === "active") {
      deps.log("verify 3/3: the live orchestrator session loaded the installed extension");
      await waitForSessionVerified(deps, release.version, reloadStarted, initial);
      journal({ kind: "phase", phase: "verify", surface: "session", ok: true, version: release.version });
    } else {
      deps.log("verify 3/3: no live orchestrator session to attest (dispatch stopped, herdr inactive)");
    }

    if (!initial.paused) deps.setPaused(false, scope.pauseKey);
    const restoredDispatch = deps.layers(scope.pauseKey).dispatch;
    if (restoredDispatch !== initial.dispatch) {
      throw new Error(
        `dispatch state is ${restoredDispatch}; expected to restore ${initial.dispatch}`,
      );
    }
  } catch (err) {
    try {
      if (!deps.layers(scope.pauseKey).paused) deps.setPaused(true, scope.pauseKey);
    } catch {
      deps.setPaused(true, scope.pauseKey);
    }
    deps.setGlobalRecoveryPaused(true);
    const failure = err instanceof Error ? err.message : String(err);
    journal({
      kind: "phase",
      phase: "failed",
      ok: false,
      version: release.version,
      detail: failure,
    });
    deps.log(`upgrade failed: ${failure}`);
    const journalLog = (message: string): void => {
      journal({ kind: "phase", phase: "rollback", ok: true, version: release.version, detail: message });
      deps.log(message);
    };
    try {
      await rollbackUpgrade(
        { ...deps, log: journalLog },
        surfaces,
        scope,
        installTouched,
        briefChanged,
        herdrReloadStarted,
        daemonReloadStarted,
        configBefore,
        preUpgradeBackup,
      );
      journal({ kind: "outcome", phase: "rolled-back", ok: false, version: release.version, detail: failure });
      enqueueUpgradeReport(
        "tier2",
        reportProject(installProjects),
        `fleet upgrade to omp-conductor@${release.version} failed and was rolled back`,
        [
          `Reason: ${failure}`,
          installAttribution,
          `Restored: cli=${surfaces.cliVersion}, omp=${surfaces.ompVersion ?? "missing"}, ` +
            `herdr=${surfaces.herdrSource ?? "missing"}`,
          ...(preUpgradeBackup === undefined ? [] : [`Pre-upgrade config snapshot: ${preUpgradeBackup}`]),
          "Dispatch stays paused until the first tick after the restart records the rollback verified; " +
            "if it is still paused, resume it with `omp-conductor resume --all`.",
        ],
        `upgrade:${release.version}:rolled-back`,
      );
    } catch (rollbackErr) {
      const rollback = rollbackErr instanceof Error ? rollbackErr.message : String(rollbackErr);
      journal({ kind: "outcome", phase: "rollback-failed", ok: false, version: release.version, detail: rollback });
      enqueueUpgradeReport(
        "tier2",
        reportProject(installProjects),
        `upgrade to omp-conductor@${release.version} failed and ITS ROLLBACK ALSO FAILED`,
        [
          `Failure: ${failure}`,
          installAttribution,
          `Rollback failure: ${rollback}`,
          "The fleet may be at mixed versions; do not resume dispatch until versions are verified.",
          "Once one version is confirmed everywhere, `omp-conductor resume --all` lifts the pause.",
          "Restore manually from the journal at " + upgradeJournalPath(stateDir()) + ".",
        ],
        `upgrade:${release.version}:rollback-failed`,
      );
      throw new Error(
        `upgrade failed: ${failure}; rollback incomplete: ${rollback}; dispatch remains paused — ` +
          `\`omp-conductor resume --all\` lifts the pause once mixed versions are ruled out`,
      );
    }
    throw new Error(
      `upgrade failed: ${failure}; previous installation restored; dispatch remains paused — ` +
        `resume dispatch with \`omp-conductor resume --all\``,
    );
  }

  // The install landed; name any host-runtime destination this version no
  // longer renders identically, so the operator knows `setup host` is owed
  // before a later `doctor` reports it as drift (#598).
  logHostRuntimeDrift(deps);

  return {
    previousVersion: surfaces.cliVersion,
    ...release,
    alreadyCurrent: false,
    dispatch: initial.dispatch,
  };
}

// ---------------------------------------------------------------------------
// The detached fleet installer (#486)
// ---------------------------------------------------------------------------

/**
 * Run the upgrade transaction as the detached fleet installer: `upgrade`'s
 * engine with a journal wired to the state-dir journal and the in-process
 * verification handed to the first tick after the restart. This is the
 * `upgrade-install` command the transient unit runs — the one process a
 * `systemctl restart herdr-fleet.service` and a daemon restart both pass by.
 */
export async function runDetachedUpgrade(options: UpgradeOptions = {}): Promise<UpgradeResult> {
  return upgradeConductor(
    { ...options, detached: true },
    {
      journal: (entry) => appendJournal(stateDir(), entry),
    },
  );
}

// ---------------------------------------------------------------------------
// The detached rollback unit (#486)
// ---------------------------------------------------------------------------

export interface RollbackFromJournalResult {
  /** The version the rollback restored the fleet to. */
  restoredVersion: string;
  /** The version the failed upgrade had targeted. */
  failedVersion: string;
}

/**
 * Restore every surface the failed upgrade touched, from the journal the
 * detached installer wrote. This is the `upgrade-rollback` command the
 * verifier's transient unit runs — the same trust boundary as the install:
 * the process that rolls back is not the process that failed, and it acts
 * only on identities the journal recorded before the first install touched
 * anything.
 *
 * Scoped to the newest request's progression: an older, already-closed
 * upgrade in the same append-only journal neither supplies the snapshot to
 * restore nor the phases that say which surfaces were touched, and its
 * terminal outcome cannot refuse this rollback.
 */
export async function rollbackFromJournal(
  overrides: Partial<UpgradeDeps> = {},
  root = stateDir(),
): Promise<RollbackFromJournalResult> {
  const deps: UpgradeDeps = { ...DEFAULT_DEPS, ...overrides };
  const entries = readUpgradeJournal(root);
  const progression = requestProgression(entries);
  const request = pendingUpgradeRequest(entries);
  if (progression === undefined || request === undefined) {
    throw new Error("no upgrade request is journaled; nothing to roll back");
  }
  const snapshot = progression.find((entry) => entry.kind === "snapshot");
  if (snapshot === undefined || snapshot.previous === undefined) {
    throw new Error(
      `upgrade ${request.version} has no durable pre-install snapshot; refusing to guess what to restore`,
    );
  }
  const newestOutcome = [...progression].reverse().find((entry) => entry.kind === "outcome");
  if (newestOutcome !== undefined && newestOutcome.phase !== "rollback-requested") {
    // Already terminal (verified, rolled back in-process, aborted). A second
    // rollback would fight the newer outcome.
    throw new Error(
      `upgrade ${request.version} already closed as "${newestOutcome.phase}"; nothing to roll back`,
    );
  }

  const phases = progression.filter((entry) => entry.kind === "phase");
  const scope: UpgradeScope = {
    selectors: request.selectors,
    pauseKey: request.pauseKey,
  };
  const previous: InstalledSurfaces = {
    cliVersion: snapshot.previous.cliVersion,
    ompVersion: snapshot.previous.ompVersion,
    herdrSource: snapshot.previous.herdrSource,
    ...(snapshot.previous.management === undefined
      ? {}
      : { management: snapshot.previous.management }),
  };
  let configBefore: string | undefined;
  if (request.configBackup !== undefined && existsSync(request.configBackup)) {
    try {
      configBefore = readFileSync(request.configBackup, "utf8");
    } catch {
      configBefore = undefined;
    }
  }
  const journalLog = (message: string): void => {
    appendJournal(root, {
      at: new Date().toISOString(),
      kind: "phase",
      phase: "rollback",
      ok: true,
      version: request.version,
      detail: message,
    });
    process.stdout.write(`${message}\n`);
  };
  await rollbackUpgrade(
    { ...deps, log: journalLog },
    previous,
    scope,
    phases.some((entry) => entry.phase === "install"),
    phases.some((entry) => entry.phase === "brief"),
    phases.some((entry) => entry.phase === "restart" && entry.surface === "herdr"),
    phases.some((entry) => entry.phase === "restart" && entry.surface === "daemon"),
    configBefore,
    request.configBackup,
  );
  // The restored bytes remain host-wide unverified state until the returning
  // daemon reads all three surfaces. A project-scoped request cannot leave its
  // siblings admitting in that interval.
  deps.setGlobalRecoveryPaused(true);

  const restoredVersion = previous.cliVersion;
  appendJournal(root, {
    at: new Date().toISOString(),
    kind: "outcome",
    phase: "rolled-back",
    ok: false,
    version: request.version,
    detail: `restored ${previous.cliVersion} from the pre-install snapshot`,
  });
  enqueueUpgradeReport(
    "tier2",
    reportProject(request.installProjects ?? request.selectors),
    `fleet restored to omp-conductor@${restoredVersion} after the failed upgrade to ${request.version}`,
    [
      "The transient rollback unit restored every surface from the durable snapshot;",
      `Host-global install holder: ${request.installHolder ?? "unknown"}; affected projects: ${
        (request.installProjects ?? request.selectors).length === 0
          ? "all configured projects"
          : (request.installProjects ?? request.selectors)
              .map((selector) => selector ?? "(default)")
              .join(", ")
      }.`,
      "dispatch stays paused until the first tick after these restarts verifies the old version.",
      `Pre-upgrade config snapshot: ${request.configBackup ?? "none recorded"}`,
    ],
    `upgrade:${request.version}:rolled-back`,
  );
  return { restoredVersion, failedVersion: request.version };
}