/**
 * Operator surface for the two-process fleet: dispatch daemon + orchestrator
 * pane. Issue #61 — one answer to "stop the conductor" without naming four
 * files and units.
 *
 * Verbs:
 * - hold          — pause claiming AND disarm ticks; processes stay up
 * - halt          — hold, then stop the dispatch daemon (systemctl-aware)
 * - halt --pane   — pin recovery first, stop the exact omp conductor agent,
 *                   never systemctl stop herdr-fleet
 * - arm / disarm  — first-class armed marker (arm is inbound Telegram proof)
 * - releaseHold   — clear pause only; never re-arms
 * - resume        — clear pause and the stop --pane recovery pin
 */

import { spawnSync } from "node:child_process";
import {
  existsSync,
  mkdirSync,
  readFileSync,
  rmSync,
  writeFileSync,
} from "node:fs";
import { homedir } from "node:os";
import { dirname, join, sep } from "node:path";
import { findProject, loadConfig, resolveArmProof, stateDir } from "./config.ts";
import {
  clearArmTransaction,
  FLEET_ARM_KEY,
  recordArmChallenge,
  resolveArmReply,
  type ArmReplyVerdict,
  type ArmTarget,
} from "./arm-challenge.ts";
import {
  claimedTelegramTopics,
  lockPidAlive,
  pidAlive,
  readTelegramDmOwner,
  readTelegramMisroutes,
  readTelegramPollState,
  resolveClaimedSessionFile,
  resolveProjectTopicId,
  sendTelegram,
  telegramPlumbingVerdict,
  TELEGRAM_LOCK_FRESH_MS,
  type TelegramPlumbingFailureReason,
  type TelegramPlumbingProbe,
  type TelegramPlumbingVerdict,
} from "./escalate.ts";
import { readPlanUsage, sharedUsageSource } from "./usage.ts";
import { readApprovalSurface, readDaemonProfile } from "./approval-surface.ts";
import { inspectBriefLayout } from "./brief-upgrade.ts";
import { dbPath, openStore } from "./store.ts";
import { renderBriefForProject } from "./setup.ts";
import { DEFAULT_ARM_PROOF, type ArmProof, type DaemonStop, type ProjectConfig, type Store } from "./types.ts";
import { DEFAULT_DEPS, probeCodeGraph, type CodeGraphHealth } from "./graph-health.ts";
import { isPaused, setPaused, statusSnapshot, type StatusSnapshot } from "./daemon.ts";
// The drain surface belongs beside hold/releaseHold on the operator surface:
// the record is daemon state (implemented next to the pause sentinel), but the
// later CLI and release #484 children import from here, exactly where they find
// every other fleet control.
export {
  cancelDrain,
  consumeDrain,
  createDrain,
  drainPath,
  markDrained,
  readDrain,
  type CreateDrainOptions,
  type DrainProblem,
  type DrainRecord,
  type DrainStatus,
  type DrainVerdict,
} from "./daemon.ts";
import {
  healthCheck,
  isAlive,
  livingDaemon,
  probeUnit,
  runSystemctl,
  stopDaemon,
  type HealthCheckResult,
  type StopResult,
} from "./lifecycle.ts";
import type { WorkerPausePhase } from "./worker.ts";
import {
  formatFleetStatus,
  formatGroomingStatus,
  type DaemonProjectHealth,
  type DispatchLayer,
  type FleetDaemonProbe,
  type FleetLayers,
  type HerdrLayer,
  type PaneLayer,
  type TelegramHealth,
  type TicksLayer,
} from "./status-render.ts";
import { fetchRateLimit } from "./tracker/github.ts";
import {
  LEGACY_ARM_MARKER_DETAIL,
  legacyArmedMarkerPath,
  readTickConfig,
  readTickRuntimeStatus,
  resolveArmState,
  TICK_CONFIG_FILE,
  tickConfigMatchesProject,
  writeArmedMarker,
  type ArmState,
  type TickConfig,
  type TickConfigResult,
} from "./orchestrator-tick.ts";

export const PANE_HALT_FILE = ".conductor-pane-halted";
export const DEFAULT_HERDR_UNIT = "herdr-fleet.service";

/**
 * The Herdr session conductor fleets run in. `HERDR_SESSION` wins when set;
 * this is the only other name any surface may resolve. #320 renamed it from
 * `fleet` and shipped a one-release probe that preferred a populated legacy
 * session; #976 removed that probe, because a name resolved two ways is a
 * name two surfaces can disagree about — and they did, in `upgrade`.
 */
export const DEFAULT_HERDR_SESSION = "conductor";

/**
 * Agent name assumed when no tick config names one. Mirrors
 * `herdr-conductor`'s own `AGENT_NAME=${AGENT_NAME:-fleet}` default — a
 * documented default, not a guess. An *invalid* config is a different thing
 * and makes {@link stopConductorPane} refuse.
 */
export const DEFAULT_FLEET_AGENT_NAME = "fleet";
export const ARM_CHALLENGE_TIMEOUT_MS = 300_000;

/**
 * Resolve the Herdr session name — the single resolution every surface uses.
 * `HERDR_SESSION` is authoritative; otherwise {@link DEFAULT_HERDR_SESSION}.
 * Never probe for an alternative name here: a session that answers is not
 * evidence of which session this fleet is, and the probe that assumed it was
 * is what made `upgrade` install against `fleet` and verify against
 * `conductor` (#976).
 */
export function resolveHerdrSession(env: NodeJS.ProcessEnv = process.env): string {
  const explicit = env["HERDR_SESSION"];
  if (explicit !== undefined && explicit.length > 0) return explicit;
  return DEFAULT_HERDR_SESSION;
}

export function telegramStateDir(): string {
  const override = process.env["OMP_TELEGRAM_STATE_DIR"];
  if (override !== undefined && override.length > 0) return override;
  return join(homedir(), ".omp", "agent", "telegram");
}

export function tickConfigSearchRoots(projectName?: string): string[] {
  const roots: string[] = [stateDir()];
  try {
    const p = findProject(loadConfig(), projectName);
    const parent = dirname(p.workspaceRoot);
    if (parent !== roots[0]) roots.push(parent);
    if (p.workspaceRoot !== roots[0] && p.workspaceRoot !== parent) roots.push(p.workspaceRoot);
  } catch {
    /* no config */
  }
  return roots;
}

export type ResolvedTick =
  | { kind: "absent" }
  | { kind: "invalid"; path: string; cwd: string; problem: string }
  | { kind: "ok"; path: string; cwd: string; config: TickConfig };

/**
 * The tick config that belongs to this project, skipping any that names another
 * one. The search roots overlap between projects — `stateDir()` and the shared
 * parent of two fleet cwds are read for every one of them — so a stamped config
 * sitting in a shared root would otherwise answer for whichever project asked
 * first. An unstamped (pre-multi-project) config still matches anything, which
 * is what keeps a single-project fleet on the file it already has.
 */
export function resolveTickConfig(projectName?: string): ResolvedTick {
  for (const cwd of tickConfigSearchRoots(projectName)) {
    const r: TickConfigResult = readTickConfig(cwd);
    if (r.kind === "ok") {
      if (!tickConfigMatchesProject(r.config, projectName)) continue;
      return { kind: "ok", path: r.path, cwd, config: r.config };
    }
    if (r.kind === "invalid") return { kind: "invalid", path: r.path, cwd, problem: r.problem };
  }
  return { kind: "absent" };
}

/**
 * Where this project's arm marker lives. The per-project fallback is the point:
 * one shared `<stateDir>/armed` meant arming any project armed all of them. The
 * bare name survives only for the un-named single-project call, which is the
 * pre-multi-project spelling.
 */
export function armedMarkerPath(projectName?: string): string {
  return armGate(projectName).path;
}

/**
 * This project's arm marker together with the project name that identifies it —
 * the pair every arm decision needs, from one tick-config read. `named` prefers
 * the stamped {@link TickConfig.project} over the caller's `--project`, because
 * the stamp is what the marker was generated from.
 */
function armGate(projectName?: string): { path: string; named?: string } {
  const tick = resolveTickConfig(projectName);
  const configured = tick.kind === "ok" ? tick.config : undefined;
  const named = configured?.project ?? projectName;
  const path =
    configured?.armedFile ??
    (projectName === undefined ? legacyArmedMarkerPath() : join(stateDir(), `armed-${projectName}`));
  return named === undefined ? { path } : { path, named };
}

/**
 * This project's current tick-arm verdict — the marker the heartbeat reads,
 * returned with the path that carries it so a verb can report what actually
 * remains. Read-only: the same decision `disarm` takes, so control and status
 * about one fleet cannot disagree (a verb telling the operator to re-arm a
 * fleet that is already armed is the defect #375 removes).
 */
export function armState(projectName?: string): { path: string; armed: boolean } {
  const { path, named } = armGate(projectName);
  return { path, armed: resolveArmState(path, named).armed };
}

export type WatchMonitoring =
  | { monitored: true }
  | { monitored: false; reason: string };

/** Whether a durable watch can be observed and wake an orchestrator tick (#1024). */
export function watchMonitoringState(
  layers: Pick<FleetLayers, "daemon" | "ticks">,
): WatchMonitoring {
  if (!layers.daemon.running) return { monitored: false, reason: "daemon not running" };
  if (layers.ticks === "armed" || layers.ticks === "ungated") return { monitored: true };
  if (layers.ticks === "disarmed") {
    return {
      monitored: false,
      reason: "ticks disarmed — conditions may still be checked, but no tick can be woken",
    };
  }
  return { monitored: false, reason: `ticks ${layers.ticks}` };
}

/** Cheap watch liveness read: no Herdr, Telegram, healthz, or systemd probes. */
export function watchMonitoringForProject(projectName: string): WatchMonitoring {
  const daemon = livingDaemon();
  if (daemon === undefined) return { monitored: false, reason: "daemon not running" };
  if (daemon.project !== undefined && daemon.project !== projectName) {
    return { monitored: false, reason: `daemon serves ${daemon.project}, not ${projectName}` };
  }
  const tick = resolveTickConfig(projectName);
  const ticks: TicksLayer =
    tick.kind === "absent"
      ? "no-heartbeat-config"
      : tick.kind === "invalid"
        ? "invalid-heartbeat-config"
        : tick.config.armedFile === undefined
          ? "ungated"
          : armState(projectName).armed
            ? "armed"
            : "disarmed";
  return watchMonitoringState({ daemon: { running: true }, ticks });
}

/**
 * Clears the arm gate for this project — and `wasArmed` is the gate the
 * heartbeat reads, not merely one file's presence.
 *
 * The second removal is the whole reason this is not a one-line `rmSync`. While
 * the shared pre-per-project marker is still honoured (single-project host, an
 * `armedFile` just restamped to `armed-<name>`), it holds the gate open as soon
 * as this project's own marker is gone — so a disarm that removed only the
 * per-project file would leave the fleet ticking, which is exactly what `hold`
 * exists to prevent. A *stranded* shared marker arms nothing, so it is not this
 * command's to remove.
 */
export function disarmTicks(projectName?: string): { path: string; wasArmed: boolean } {
  const { path, named } = armGate(projectName);
  const wasArmed = resolveArmState(path, named).armed;
  rmSync(path, { force: true });
  if (resolveArmState(path, named).legacy === "honoured") rmSync(legacyArmedMarkerPath(), { force: true });
  return { path, wasArmed };
}

/** One project whose arm marker was written. */
export interface ArmedProject {
  /**
   * The configured project name, absent only for a legacy unstamped
   * single-project fleet.
   */
  project?: string;
  path: string;
  alreadyArmed: boolean;
}

/** Markers written without any reply: the `claim-only` policy proof (#613). */
export interface ArmMarkersWritten {
  outcome: "armed";
  proof: ArmProof;
  owner: string;
  armed: ArmedProject[];
}

/**
 * A challenge filed and sent, with nothing armed yet.
 *
 * Arming used to block here for up to five minutes on an in-session
 * acknowledgement. Nothing waits now: the reply is consumed mechanically by
 * the session whose topic the challenge went to (the orchestrator pane,
 * #1061), or by `omp-conductor arm --reply` on a console host whose DM the
 * operator answered in. This is the send half's receipt.
 */
export interface ArmChallengeSent {
  outcome: "challenge-sent";
  /** Always `challenge`: `claim-only` never sends, so it never reaches here. */
  proof: "challenge";
  /** The chat the challenge went to, recorded with the transaction. */
  owner: string;
  /** The pending transaction the reply must prove. */
  challengeId: string;
  /** Unix ms after which the code stops being a proof. */
  expiresAt: number;
  /** mm:ss the code stays good for — the same clock the message quotes. */
  validFor: string;
  /** Exactly what a matching reply will arm, as recorded with the challenge. */
  targets: ArmTarget[];
  /** The verbatim command that completes the ceremony, copy-pasteable as-is. */
  followUp: string;
}

export type ArmResult = ArmMarkersWritten | ArmChallengeSent;

export interface ArmDeps {
  sendChallenge?: (token: string, owner: string, text: string, topicId?: number) => Promise<void>;
  /**
   * The orchestrator session file the live omp-telegram claim names for this
   * project, or undefined when there is no (readable) claim. The default
   * resolves the claim the same way the send does; tests inject a fixture.
   * Claim-only feeds it to the plumbing verdict's session-identity checks;
   * the challenge proof never reads a transcript, so it ignores the claim.
   */
  claimedSessionFile?: () => string | undefined;
  now?: () => number;
  /** How long a sent code stays a proof. */
  timeoutMs?: number;
  /**
   * Liveness seams for the claim-only verdict (#613), with omp-telegram's own
   * rules (#612): `pidAlive` judges a claim or dm-owner (EPERM is dead),
   * `lockPidAlive` judges a bot.lock owner (EPERM is live), and `lockFresh`
   * applies the heartbeat window. Tests inject deterministic answers without
   * owning another uid's process; the verdict itself and every state file it
   * reads run for real.
   */
  pidAlive?: (pid: number) => boolean;
  lockPidAlive?: (pid: number) => boolean;
  lockFresh?: (mtimeMs: number) => boolean;
  /**
   * Where the ceremony's one-line narration goes. Nothing waits any more, so
   * this is no longer a liveness heartbeat (#861) — it is the send receipt and
   * the follow-up instruction, for a surface that prints as it goes. Absent
   * means no reporting.
   */
  progress?: (line: string) => void;
}

export async function armTicks(projectName?: string, deps: ArmDeps = {}): Promise<ArmResult> {
  const tick = resolveTickConfig(projectName);
  if (tick.kind === "invalid") {
    throw new Error(`tick config invalid at ${tick.path}: ${tick.problem}`);
  }
  if (tick.kind === "absent") {
    throw new Error(
      `no ${TICK_CONFIG_FILE} under ${tickConfigSearchRoots(projectName).join(" or ")} — ` +
        `nothing would read an arm marker; drop a tick config first`,
    );
  }
  if (tick.config.armedFile === undefined) {
    throw new Error(`${tick.path} has no armedFile — this heartbeat is ungated; add armedFile before arming`);
  }
  if (tick.config.accessFile === undefined) {
    throw new Error(`${tick.path} has no accessFile — arm cannot prove an inbound channel without it`);
  }

  const channel = readPairedChannel(tick.config.accessFile);
  if (channel.kind === "down") {
    throw new Error(
      `escalation channel is not up (${tick.config.accessFile}): ${channel.reason} — ` +
        `pair the bot (/telegram pair) and enable the bridge (/telegram on) before arming`,
    );
  }

  const token = readBotToken();
  if (token === undefined) {
    throw new Error(
      `no TELEGRAM_BOT_TOKEN in ${join(telegramStateDir(), ".env")} — the arm challenge cannot send without it`,
    );
  }

  // One bot, one chat, and — once a host runs more than one fleet — more than
  // one pane that can ask. The challenge names which one, or the operator is
  // answering a question they cannot attribute.
  const named = tick.config.project ?? projectName;
  // The handshake state key must be exactly what the reply step recomputes
  // from this same tick config: TickConfig.project, undefined for a legacy
  // unstamped config. `named` may fall back to the CLI argument for the
  // challenge text and config lookups; the state key must not, or the reply
  // would look for the challenge under a key nothing recorded.
  const stateKey = tick.config.project;

  // The arming proof is a declared per-project policy (#613). A config that
  // cannot name the project fails safe to `challenge` — the authenticated
  // round-trip — so a missing or unreadable config never silently weakens the
  // gate.
  let proof: ArmProof = DEFAULT_ARM_PROOF;
  try {
    proof = resolveArmProof(findProject(loadConfig(), named));
  } catch {
    /* no project config — keep the challenge behaviour */
  }

  // The orchestrator's live session file per omp-telegram's claim (#600) —
  // input to the claim-only verdict's session-identity checks below. The
  // challenge proof never reads it: its proof is conductor state, so where (or
  // whether) a transcript lives is no longer part of arming (#614).
  const claimed =
    deps.claimedSessionFile !== undefined
      ? deps.claimedSessionFile()
      : claimedOrchestratorSessionFile(named, deps.pidAlive ?? pidAlive);

  // Prefer the project's live forum topic so arm challenges land where
  // escalations already do (#318), following the bridge's current claim when the
  // pinned id has gone stale (#407). Missing project config keeps flat-chat 0.13.
  let sendTopic: number | undefined;
  let sendProject: { name: string; workspaceRoot?: string } | undefined;
  if (named !== undefined) {
    try {
      const cfg = findProject(loadConfig(), named);
      sendTopic = resolveProjectTopicId(cfg, deps.pidAlive ?? pidAlive);
      sendProject = { name: cfg.name, workspaceRoot: cfg.workspaceRoot };
    } catch {
      /* no project config */
    }
  }

  const path = tick.config.armedFile;

  if (proof === "claim-only") {
    // The human-intent gate is declared satisfied by policy, so #612's shared
    // verdict is the whole proof: the same state reads and the same liveness
    // rules the doctor's "telegram-plumbing" finding applies, on the route a
    // challenge would have ridden. No Telegram send, no reply step, no
    // pending-challenge record. A failed fact refuses arming by name — never
    // a silent pass from file existence, and never a marker.
    const scan = armVerdictScanDirs(tick.cwd, claimed);
    const probe: TelegramPlumbingProbe = {
      // `channel` is up here — the paired-channel block above already threw on
      // down — but the verdict re-reads it from the same state the challenge
      // would send over, so the two proofs cannot disagree about the transport.
      channel,
      registry: claimedTelegramTopics(),
      poll: readTelegramPollState(),
      dmOwner: readTelegramDmOwner(),
      alive: deps.pidAlive ?? pidAlive,
      lockAlive: deps.lockPidAlive ?? lockPidAlive,
      fresh: deps.lockFresh ?? ((mtimeMs) => (deps.now ?? Date.now)() - mtimeMs < TELEGRAM_LOCK_FRESH_MS),
    };
    const verdict = telegramPlumbingVerdict(sendTopic, { dirs: scan }, probe);
    if (!verdict.ok) {
      throw new Error(
        `arm: claim-only proof refused — ${claimOnlyFailureText(verdict.reason)}. ` +
          `NOT armed; no marker was written`,
      );
    }
    // The gate as the heartbeat reads it, so "replaced previous marker" is not
    // a lie about a fleet the shared marker was arming, and so the write knows
    // whether it is superseding that marker.
    const arm = resolveArmState(path, named);
    writeArmedMarker(path, channel.owner, arm);
    return {
      outcome: "armed",
      proof,
      owner: channel.owner,
      armed: [{ ...(named === undefined ? {} : { project: named }), path, alreadyArmed: arm.armed }],
    };
  }

  const send = deps.sendChallenge ?? sendTelegramMessage;
  const timeoutMs = deps.timeoutMs ?? ARM_CHALLENGE_TIMEOUT_MS;
  const code = makeChallengeCode();
  const followUp = armReplyCommand(projectName);
  // Self-describing (#991): with two live challenges in one chat the operator
  // was working out which was which from message order, and nothing said how
  // long a code stayed good — so the safe move was to scroll for the newest,
  // which is the work conductor-owned verification was meant to end.
  const text =
    `Fleet arming check${named === undefined ? "" : ` — project ${named}`}. ` +
    `Reply to this chat with exactly:\n${code}\n` +
    `Valid for ${armClock(timeoutMs)}. ` +
    `Nothing is dispatched until that reply is verified.`;
  const sentAt = (deps.now ?? Date.now)();
  // Recorded BEFORE the send, so a reply that beats this process's own return
  // still finds an active challenge — and so a send that fails has a
  // transaction to settle rather than a code loose in a chat (#415). The record
  // carries what the reply will arm: the config could change before the
  // operator answers, and arming anything but what the challenge named would
  // arm a fleet nobody was asked about.
  const challengeId = recordArmChallenge(stateKey, code, sentAt, sentAt + timeoutMs, {
    targets: [{ ...(named === undefined ? {} : { project: named }), armedFile: path }],
    owner: channel.owner,
  });
  try {
    await send(token, channel.owner, text, sendTopic, sendProject);
  } catch (err) {
    // The challenge never went out, so its transaction must not linger as a
    // classifiable proof either.
    clearArmTransaction(stateKey, challengeId);
    throw new Error(
      `arm: outbound sendMessage failed — NOT armed: ${err instanceof Error ? err.message : String(err)}`,
    );
  }
  deps.progress?.(
    `arm: challenge sent to ${channel.owner}${sendTopic === undefined ? "" : ` (topic ${sendTopic})`} — ` +
      `valid for ${armClock(timeoutMs)}. Nothing is armed yet: run ${followUp}`,
  );
  return {
    outcome: "challenge-sent",
    proof: "challenge",
    owner: channel.owner,
    challengeId,
    expiresAt: sentAt + timeoutMs,
    validFor: armClock(timeoutMs),
    targets: [{ ...(named === undefined ? {} : { project: named }), armedFile: path }],
    followUp,
  };
}

/**
 * The second half of the ceremony, verbatim. Printed by every surface that
 * issues a challenge, and quoted in the refusals, because an agent reading a
 * console pane has to be able to paste it without composing anything: the
 * operator's message is the only variable.
 */
function armReplyCommand(projectName?: string): string {
  return (
    `omp-conductor arm --reply "<the operator's reply, verbatim>"` +
    `${projectName === undefined ? "" : ` --project ${projectName}`}`
  );
}

/**
 * One ceremony for the whole fleet (#991).
 *
 * Arming was per project: a two-project fleet meant two sequential handshakes
 * with two codes in one chat, though nothing about the fleet's state differed
 * between them. This sends **one** challenge whose single matching reply arms
 * every configured project — one fleet-wide pending record, not a loop that
 * sends N challenges and collects N replies.
 *
 * Every project is validated before anything is sent, so a fleet whose second
 * project has no `armedFile` refuses the ceremony instead of arming the first
 * and then failing — a half-armed fleet is worse than an unarmed one, because
 * only one of its halves dispatches.
 *
 * A project whose declared proof is `claim-only` is armed by this too: an
 * authenticated round-trip is strictly stronger than the plumbing verdict that
 * policy would have accepted, so satisfying the weaker gate with the stronger
 * proof cannot weaken it.
 *
 * Like {@link armTicks}, this returns as soon as the challenge is filed and
 * sent: the reply settles it mechanically on the orchestrator pane's own
 * topic, or through `omp-conductor arm --reply` from the console.
 */
export async function armFleet(
  projectNames: readonly string[],
  deps: ArmDeps = {},
): Promise<ArmChallengeSent> {
  if (projectNames.length === 0) throw new Error("arm: no projects are configured");

  // Resolve and validate every project first. Nothing is sent and no marker is
  // written until the whole fleet is known to be armable.
  const targets: { project: string; armedFile: string; accessFile: string; cwd: string; stateKey?: string }[] = [];
  for (const name of projectNames) {
    const tick = resolveTickConfig(name);
    if (tick.kind === "invalid") throw new Error(`tick config invalid at ${tick.path}: ${tick.problem}`);
    if (tick.kind === "absent") {
      throw new Error(
        `no ${TICK_CONFIG_FILE} under ${tickConfigSearchRoots(name).join(" or ")} for ${name} — ` +
          `nothing would read an arm marker; drop a tick config first`,
      );
    }
    if (tick.config.armedFile === undefined) {
      throw new Error(`${tick.path} has no armedFile — ${name}'s heartbeat is ungated; add armedFile before arming`);
    }
    if (tick.config.accessFile === undefined) {
      throw new Error(`${tick.path} has no accessFile — arm cannot prove an inbound channel for ${name}`);
    }
    targets.push({
      project: name,
      armedFile: tick.config.armedFile,
      accessFile: tick.config.accessFile,
      cwd: tick.cwd,
      ...(tick.config.project === undefined ? {} : { stateKey: tick.config.project }),
    });
  }

  // One bot, one chat: the channel is a host fact, so the first target's
  // access file speaks for the fleet. A project pointing at a different file
  // would be a different bot, which this ceremony cannot span — and would
  // announce itself here as a down channel rather than silently arming.
  const first = targets[0]!;
  const channel = readPairedChannel(first.accessFile);
  if (channel.kind === "down") {
    throw new Error(
      `escalation channel is not up (${first.accessFile}): ${channel.reason} — ` +
        `pair the bot (/telegram pair) and enable the bridge (/telegram on) before arming`,
    );
  }
  const token = readBotToken();
  if (token === undefined) {
    throw new Error(
      `no TELEGRAM_BOT_TOKEN in ${join(telegramStateDir(), ".env")} — the arm challenge cannot send without it`,
    );
  }

  // The first project that resolves a live topic carries the ceremony, and the
  // message says so: a fleet-wide question still has to land somewhere an
  // operator is reading, and the reply step reaches the record from any project
  // because it is fleet-wide rather than topic-scoped.
  let sendTopic: number | undefined;
  let sendProject: { name: string; workspaceRoot?: string } | undefined;
  for (const target of targets) {
    try {
      const cfg = findProject(loadConfig(), target.project);
      sendTopic = resolveProjectTopicId(cfg);
      sendProject = { name: cfg.name, workspaceRoot: cfg.workspaceRoot };
    } catch {
      continue; /* no project config — try the next */
    }
    if (sendTopic !== undefined) break;
  }

  const send = deps.sendChallenge ?? sendTelegramMessage;
  const timeoutMs = deps.timeoutMs ?? ARM_CHALLENGE_TIMEOUT_MS;
  const code = makeChallengeCode();
  const names = targets.map((t) => t.project).join(", ");
  // No `--project`: the reply resolves the fleet record, which is the one this
  // ceremony wrote. Naming a project here would be a narrower command than the
  // challenge the operator answered.
  const followUp = armReplyCommand();
  const text =
    `Fleet arming check — ${String(targets.length)} project(s): ${names}. ` +
    `Reply to this chat with exactly:\n${code}\n` +
    `Valid for ${armClock(timeoutMs)}, and one reply arms all of them. ` +
    `Nothing is dispatched until that reply is verified.`;

  const sentAt = (deps.now ?? Date.now)();
  // The record carries every project the message named, so the reply arms
  // exactly the fleet the operator was asked about — not whatever the config
  // says minutes later.
  const armTargets: ArmTarget[] = targets.map((target) => ({
    project: target.project,
    armedFile: target.armedFile,
  }));
  const challengeId = recordArmChallenge(FLEET_ARM_KEY, code, sentAt, sentAt + timeoutMs, {
    targets: armTargets,
    owner: channel.owner,
  });
  try {
    await send(token, channel.owner, text, sendTopic, sendProject);
  } catch (err) {
    clearArmTransaction(FLEET_ARM_KEY, challengeId);
    throw new Error(
      `arm: outbound sendMessage failed — NOT armed: ${err instanceof Error ? err.message : String(err)}`,
    );
  }
  deps.progress?.(
    `arm: one challenge sent to ${channel.owner}${sendTopic === undefined ? "" : ` (topic ${sendTopic})`} for ${names} — ` +
      `valid for ${armClock(timeoutMs)}. Nothing is armed yet: run ${followUp}`,
  );
  return {
    outcome: "challenge-sent",
    proof: "challenge",
    owner: channel.owner,
    challengeId,
    expiresAt: sentAt + timeoutMs,
    validFor: armClock(timeoutMs),
    targets: armTargets,
    followUp,
  };
}

/** What a reply that armed nothing was, and what to do about it. */
export interface ArmReplyRefused {
  outcome: "refused";
  /** Never `matched`: a match is the armed outcome. */
  verdict: Exclude<ArmReplyVerdict, "matched">;
  /** One operator-facing sentence naming the verdict and the next move. */
  message: string;
}

export interface ArmReplyAccepted {
  outcome: "armed";
  /** The chat the proved challenge was sent to — the marker's `owner=`. */
  owner: string;
  /** The transaction the reply proved and this call settled. */
  challengeId: string;
  /** Every project the challenge recorded, now armed. */
  armed: ArmedProject[];
}

export type ArmReplyResult = ArmReplyAccepted | ArmReplyRefused;

/**
 * The verification half of the ceremony: classify the operator's verbatim
 * message and, on a match, arm exactly the projects the challenge recorded.
 *
 * This is the host CLI half, and it is one of two consumers of the same
 * durable record: the orchestrator session's inbound path settles a reply
 * sent to the project topic mechanically (#1061), while this command remains
 * the console's route for a reply that landed in the operator DM. No session
 * waits for anything: the challenge is durable state, so the halves are
 * ordinary commands that can run minutes apart in different processes.
 *
 * The security properties are the send half's, unchanged. Only the project's
 * own record and the fleet record are read (no other project's ceremony can be
 * revealed or consumed), an expired record is never a proof, and a marker is
 * written only for a `matched` verdict — every other verdict returns a refusal
 * having written nothing.
 */
export function armReply(
  replyText: string,
  projectName?: string,
  deps: Pick<ArmDeps, "now"> = {},
): ArmReplyResult {
  const tick = resolveTickConfig(projectName);
  if (tick.kind === "invalid") {
    // Every derivation below would come from this file. A config that does not
    // parse cannot be trusted to name a state key or a marker.
    throw new Error(`tick config invalid at ${tick.path}: ${tick.problem}`);
  }
  // An absent config is not fatal here, which is the difference between the two
  // halves: a fleet ceremony's record carries its own targets, and on a
  // multi-project host the search roots for an unnamed project resolve no
  // config at all (the fleet cwds each carry their own). Refusing here would
  // make `arm --reply` unusable for exactly the ceremony that needs it most.
  const config = tick.kind === "ok" ? tick.config : undefined;
  // The same derivation the send half used, from the same file: the state key
  // is TickConfig.project (undefined for a legacy unstamped config), never the
  // CLI argument, or the reply would look under a key nothing recorded.
  const stateKey = config?.project;
  const now = (deps.now ?? Date.now)();
  const resolved = resolveArmReply(stateKey, replyText, now);
  if (resolved.verdict !== "matched") {
    return {
      outcome: "refused",
      verdict: resolved.verdict,
      message: armRefusalText(resolved.verdict, projectName),
    };
  }
  const match = resolved.match;
  // Recorded targets are the contract. A record written before targets existed
  // (one release of overlap) can only be read as naming the project it is keyed
  // under — or, for the fleet record, the project this command was given, since
  // re-deriving the fleet from today's config could arm a project the operator
  // was never asked about.
  const targets: ArmTarget[] = match.targets ?? [legacyArmTarget(config, projectName)];
  // The owner the challenge was actually sent to. Only a pre-targets record
  // lacks it, and then the paired channel is the only other honest source.
  const owner = match.owner ?? pairedChannelOwner(config?.accessFile);
  const armed = targets.map((target) => {
    const state = resolveArmState(target.armedFile, target.project);
    writeArmedMarker(target.armedFile, owner, state);
    return {
      ...(target.project === undefined ? {} : { project: target.project }),
      path: target.armedFile,
      alreadyArmed: state.armed,
    };
  });
  // Settled by id against the key that matched, so a replayed reply cannot arm
  // a second time and a newer replacement's record is never removed.
  clearArmTransaction(match.key, match.id);
  return { outcome: "armed", owner, challengeId: match.id, armed };
}

/**
 * What a pre-targets pending record arms. The tick config that resolved the
 * state key also names this project's marker, which is exactly what the send
 * half would have recorded. Without such a config there is nothing to fall back
 * to, and guessing a marker path is not an option — a challenge from before the
 * upgrade is simply re-run.
 */
function legacyArmTarget(config: TickConfig | undefined, projectName?: string): ArmTarget {
  const named = config?.project ?? projectName;
  if (config?.armedFile === undefined) {
    throw new Error(
      `arm: the pending challenge names no targets and no readable ${TICK_CONFIG_FILE} names an armedFile — ` +
        `nothing to arm; re-run \`omp-conductor arm\` for a challenge that records its own targets`,
    );
  }
  return { ...(named === undefined ? {} : { project: named }), armedFile: config.armedFile };
}

/** The paired owner, for the one record shape that does not carry its own. */
function pairedChannelOwner(accessFile: string | undefined): string {
  if (accessFile === undefined) {
    throw new Error(`${TICK_CONFIG_FILE} has no accessFile — the challenge's owner cannot be established`);
  }
  const channel = readPairedChannel(accessFile);
  if (channel.kind === "down") {
    throw new Error(
      `escalation channel is not up (${accessFile}): ${channel.reason} — ` +
        `this challenge predates owner recording, so the marker cannot name whom it armed`,
    );
  }
  return channel.owner;
}

/**
 * Each non-matching verdict in the operator's words, with the next move. A
 * silent no-op was the old failure mode: the operator could not tell a wrong
 * code from an expired one from a command that never looked.
 */
function armRefusalText(verdict: Exclude<ArmReplyVerdict, "matched">, projectName?: string): string {
  const reissue = `omp-conductor arm${projectName === undefined ? "" : ` --project ${projectName}`}`;
  switch (verdict) {
    case "expired":
      return (
        `arm: that code belongs to a challenge whose window has closed — NOT armed, no marker written. ` +
        `An expired code is never a proof: send a fresh challenge with \`${reissue}\` and verify that one.`
      );
    case "unknown":
      return (
        `arm: that message carries an arming code, but no live challenge here matches it — NOT armed, ` +
        `no marker written. Only this project's own challenge and the fleet ceremony are ever consulted. ` +
        `Send a challenge with \`${reissue}\` and reply to that one.`
      );
    case "none":
      return (
        `arm: that message contains no arming code at all — NOT armed, no marker written. ` +
        `Pass the operator's reply verbatim, including the FLEET-… code; if none was sent, run \`${reissue}\` first.`
      );
  }
}

export interface HoldResult {
  wasPaused: boolean;
  /** Absent when the caller kept the heartbeat armed. */
  disarmed?: { path: string; wasArmed: boolean };
}

export interface HaltResult {
  /** `stop` always disarms, so callers can print the marker path unconditionally. */
  hold: HoldResult & { disarmed: NonNullable<HoldResult["disarmed"]> };
  stop: StopResult;
}

export interface PaneStopResult {
  pinPath: string;
  /** How the pane was confirmed gone. Failures throw — never silent success. */
  stopped: "herdr-agent" | "already-gone";
  detail: string;
  agentName: string;
}

export interface HaltWithPaneResult extends HaltResult {
  pane: PaneStopResult;
}

/**
 * Stop claiming, leave the daemon and pane up.
 *
 * `keepTicks` is the old `pause` shape, and it is not a nicety: disarming is the
 * expensive half of a hold. Re-arming sends a Telegram challenge and blocks on the
 * operator answering it in the chat, so an operator who only wanted to stop new
 * claims for ten minutes pays a manual round trip to get the heartbeat back — and
 * meanwhile nothing shepherds the workers already in flight, which a hold
 * deliberately leaves running. Keeping the marker means `resume` alone restores
 * the fleet. The tick never consults the pause sentinel, so a held fleet with
 * armed ticks keeps reporting while it claims nothing.
 */
export function hold(
  projectName?: string,
  source: string = "hold",
  opts: { keepTicks?: boolean; reason?: string } = {},
): HoldResult {
  const wasPaused = isPaused(projectName);
  // `source` is a space-free token by contract (`pauseSourceToken`) — the fence
  // matches on it, and whitespace there makes a pause the fence cannot prove
  // (#552). The human sentence belongs in `reason=`, which is why this is a
  // separate field rather than something a caller concatenates into `source`.
  setPaused(true, { source, ...(opts.reason === undefined ? {} : { reason: opts.reason }) }, projectName);
  if (opts.keepTicks === true) return { wasPaused };
  return { wasPaused, disarmed: disarmTicks(projectName) };
}

export function releaseHold(projectName?: string): void {
  setPaused(false, undefined, projectName);
}

export async function halt(projectName?: string): Promise<HaltResult> {
  const held = hold(projectName, "halt");
  // `keepTicks` is a `hold` affordance: taking the fleet down always disarms, and
  // callers print the marker path. Assert the invariant rather than defaulting a
  // path we would then print as fact.
  if (held.disarmed === undefined) throw new Error("stop must disarm ticks");
  const stop = await stopDaemon();
  return { hold: { ...held, disarmed: held.disarmed }, stop };
}

/**
 * Where `halt --pane` pins recovery off, or why we cannot say.
 *
 * `herdr-conductor`'s `recover.sh` reads exactly one path —
 * `$FLEET_CWD/.conductor-pane-halted`. The only directory this package *knows*
 * is `FLEET_CWD` is the one holding `.conductor-tick.json`: the heartbeat
 * extension activates on that file in the session cwd, so the pane's cwd and
 * the tick config's directory are the same by construction.
 *
 * An *invalid* tick config still names that directory, so it resolves. An
 * *absent* one does not: the state dir is a guess, and a pin written there is
 * one `recover.sh` never reads — the agent would be respawned seconds after a
 * "successful" halt.
 */
export type ResolvedPaneHalt =
  | { kind: "ok"; path: string }
  | { kind: "unresolved"; reason: string };

export function resolvePaneHaltPath(projectName?: string): ResolvedPaneHalt {
  const tick = resolveTickConfig(projectName);
  if (tick.kind === "ok" || tick.kind === "invalid") {
    return { kind: "ok", path: join(tick.cwd, PANE_HALT_FILE) };
  }
  return {
    kind: "unresolved",
    reason:
      `no ${TICK_CONFIG_FILE} under ${tickConfigSearchRoots(projectName).join(" or ")} — ` +
      `the pane's own directory (FLEET_CWD) is unknown, and ${PANE_HALT_FILE} anywhere else ` +
      `is a file herdr-conductor never reads`,
  };
}

/**
 * {@link resolvePaneHaltPath}, refusing rather than guessing. Used by every
 * path that writes, clears or promises a pin.
 */
export function paneHaltPath(projectName?: string): string {
  const resolved = resolvePaneHaltPath(projectName);
  if (resolved.kind === "unresolved") {
    // Verb-neutral: this is also `resume`'s pin-clearing path.
    throw new Error(
      `cannot locate the pane recovery pin — ${resolved.reason}. ` +
        `Drop a tick config beside the pane, or stop the agent by hand; ` +
        `\`stop\` without \`--pane\` still stops the dispatch daemon.`,
    );
  }
  return resolved.path;
}

export function pinPaneHalt(projectName?: string): { path: string } {
  const path = paneHaltPath(projectName);
  mkdirSync(dirname(path), { recursive: true });
  writeFileSync(
    path,
    [
      `# Written by omp-conductor stop --pane at ${new Date().toISOString()}`,
      `# herdr-conductor recover.sh must not resume the fleet agent while this file exists.`,
      `# Clear with: omp-conductor resume  (or rm this file)`,
      "",
    ].join("\n"),
    { mode: 0o600 },
  );
  return { path };
}

/**
 * Clears the pin when its location is known, and reports plainly when it is not.
 *
 * Distinct from {@link clearPaneHalt}, which refuses an unresolvable location
 * because pinning somewhere recovery never reads is worse than failing. `resume`
 * needs the softer answer: it folds in the pin clear, and an operator resuming a
 * fleet from a host with no tick config has no pin to clear rather than an error
 * to work around.
 */
export function clearPaneHaltIfResolvable(projectName?: string): { path?: string; wasHalted: boolean } {
  const resolved = resolvePaneHaltPath(projectName);
  if (resolved.kind !== "ok") return { wasHalted: false };
  const wasHalted = existsSync(resolved.path);
  rmSync(resolved.path, { force: true });
  return { path: resolved.path, wasHalted };
}

export function clearPaneHalt(projectName?: string): { path: string; wasHalted: boolean } {
  const path = paneHaltPath(projectName);
  const wasHalted = existsSync(path);
  rmSync(path, { force: true });
  return { path, wasHalted };
}

export type HerdrStartResult =
  | { kind: "active"; unit: string; recoveryReleased: boolean }
  | { kind: "unmanaged"; unit: string; reason: string };

export interface HerdrStartDeps {
  systemctl?: (args: string[]) => { ok: boolean; stdout: string; stderr: string; missing?: boolean };
}

/**
 * Starts the dedicated Herdr fleet unit when it is installed. Hosts without
 * systemd or without that optional unit keep the standalone daemon behaviour.
 */
export function startHerdrFleet(projectName?: string, deps: HerdrStartDeps = {}): HerdrStartResult {
  const run = deps.systemctl ?? runSystemctl;

  const shown = run(["show", DEFAULT_HERDR_UNIT, "--property=LoadState", "--value"]);
  if (shown.missing) return { kind: "unmanaged", unit: DEFAULT_HERDR_UNIT, reason: "no systemctl" };
  if (!shown.ok) {
    const detail = (shown.stderr.trim() || shown.stdout.trim() || "systemctl show failed").split("\n")[0]!;
    throw new Error(`cannot inspect ${DEFAULT_HERDR_UNIT}: ${detail}`);
  }
  const loadState = shown.stdout.trim();
  if (loadState === "not-found" || loadState === "") {
    return { kind: "unmanaged", unit: DEFAULT_HERDR_UNIT, reason: "unit not installed" };
  }
  if (loadState !== "loaded") {
    throw new Error(`${DEFAULT_HERDR_UNIT} is ${loadState}, not startable`);
  }

  const halt = resolvePaneHaltPath(projectName);
  const recoveryReleased = halt.kind === "ok" && existsSync(halt.path);
  if (halt.kind === "ok") rmSync(halt.path, { force: true });

  const started = run(["start", DEFAULT_HERDR_UNIT]);
  if (!started.ok) {
    const detail = (started.stderr.trim() || started.stdout.trim() || "systemctl start failed").split("\n")[0]!;
    throw new Error(`systemctl start ${DEFAULT_HERDR_UNIT} failed: ${detail}`);
  }
  const active = run(["is-active", DEFAULT_HERDR_UNIT]);
  if (!active.ok || active.stdout.trim() !== "active") {
    const detail = (active.stderr.trim() || active.stdout.trim() || "not active").split("\n")[0]!;
    throw new Error(`${DEFAULT_HERDR_UNIT} did not become active: ${detail}`);
  }
  return { kind: "active", unit: DEFAULT_HERDR_UNIT, recoveryReleased };
}

export interface HerdrAgent {
  name?: string;
  paneId: string;
  agent?: string;
  sessionPath?: string;
}

/**
 * The Herdr representation of one live conductor worker (#840).
 *
 * ## Why this shape, and not `herdr agent start --kind omp`
 *
 * The authoritative worker is the `session-host` child conductor spawns itself:
 * it holds the typed control socket, the mediated verb socket, the worktree, the
 * transcript, the accounting and the output-schema settlement. Starting a second
 * OMP process in a pane would look right in the workspace and be authoritative
 * for nothing — the named silent fake of #840 — so nothing here starts an agent.
 *
 * Herdr already has the verb for exactly this case: an *external* supervisor
 * reports lifecycle onto a pane it owns (`pane report-agent`,
 * `pane report-agent-session`, `pane release-agent`). So conductor keeps the
 * child it already spawned, and reports that child's identity and state onto a
 * pane whose only job is display.
 *
 * ## What is displayed, and why that is not "a transcript-tail pane"
 *
 * The pane runs `omp-conductor tail <issue>`, which is the repository's existing
 * read-only follower. The distinction #840 draws is about *authority*, not about
 * pixels: nothing here parses that output, and no run state, settlement, turn,
 * spend or model fact is derived from it. Identity comes from the exact child
 * pid handed over by the spawn path, and state comes from the typed events the
 * proxy already emits.
 *
 * ## Observation-only is structural, not a gate
 *
 * The authoritative child is spawned with `stdin: "ignore"` — there is no file
 * descriptor for a keystroke to travel down, from a pane or anywhere else. The
 * pane's own process is a follower whose stdin reaches only itself. So "input
 * cannot reach OMP, change its prompt, invoke a verb, or settle the run" is a
 * property of the process tree rather than a check that could be forgotten.
 */
export interface WorkerPaneIdentity {
  project: string;
  issue: number;
  attempt: number;
  /** The run row's id — the durable identity the pane is reported under. */
  runId: string;
  /** The exact `session-host` pid, from the spawn path's own `onSpawn`. */
  pid: number;
  /** The transcript the child opened, when it has already reported one. */
  sessionFile?: string;
}

/** A tracked pane, or the explicit reason there is none. Never a silent claim.
 *  The tracked variant names the marked workspace the representation lives in,
 *  so a caller that just created one can protect it from the same pass's
 *  empty-workspace cleanup. `phase` names the step that refused — `"split"`
 *  marks the one failure the re-establishment budget spends whole (#998): a
 *  terminal that refuses to split will refuse every retry within the pass, so
 *  burning the budget once beats leaking an attempt per reconciliation pass
 *  against a wall. */
export type WorkerPaneOutcome =
  | { kind: "tracked"; paneId: string; label: string; pid: number; workspaceId?: string }
  | { kind: "unavailable"; reason: string; phase?: "split" };

/** One `herdr` invocation, injected so every path is testable with no terminal. */
export type HerdrRun = (args: readonly string[]) => { ok: boolean; stdout: string; stderr: string };

export interface WorkerPaneDeps {
  run?: HerdrRun;
  session?: string;
  /** The follower command the pane displays; the CLI by default. */
  viewer?: (identity: PaneCeremonyIdentity) => readonly string[];
  /** The durable ownership half of workspace discovery (#1035 review);
   *  production wires the conductor store, tests a recorder. */
  ownership?: WorkspaceOwnership;
}

/**
 * The pane's unique, human-readable name (#840: "uniquely named").
 *
 * Keyed by run id rather than issue: an issue can have several attempts, and two
 * attempts must never resolve to one pane. The issue and attempt ride along
 * because the operator reads this label in a workspace, not a database.
 */
/**
 * The external-supervisor source every conductor pane report carries.
 *
 * Written onto every pane conductor reports against. It is no longer what
 * proves a pane is conductor's: production repeatedly handed back panes whose
 * `agent_session` record Herdr had dropped (#1035), so filtering by reported
 * source quietly exempted exactly the panes that needed cleaning. Ownership is
 * structural now — conductor creates representations only inside a workspace
 * it marked itself (see {@link WORKER_WORKSPACE_TOKEN}) — and this source
 * remains on the reports as the display identity of who speaks for a pane.
 */
export const WORKER_PANE_SOURCE = "omp-conductor";

/**
 * How many times this daemon re-establishes one run's pane before it stops
 * trying (#998).
 *
 * The pathology being bounded is unboundedness, not frequency: a pane that
 * cannot be created is not going to start working on the next pass, and the
 * retries actively make it worse — `pane_split_failed: ghostty error -2` is the
 * terminal the loop eventually produces for itself. A run with no pane keeps
 * working; only its representation is missing.
 */
export const PANE_REATTEMPT_MAX = 3;

/**
 * The workspace metadata token that marks a workspace as conductor's
 * per-project worker surface (#1035).
 *
 * Written once at creation (`workspace report-metadata --token
 * conductor_workers=<project>`) and read back from `workspace list`, so
 * rediscovery survives daemon and Herdr restarts without trusting the label,
 * the sidebar position, or whichever workspace happens to be focused. Only
 * panes inside a marked workspace are ever reconciled here, which is what makes
 * the operator's own panes invisible to this surface by construction — and
 * what makes every pane inside a marked workspace conductor's, however Herdr
 * feels about reporting `agent_session` back.
 */
export const WORKER_WORKSPACE_TOKEN = "conductor_workers";

/** The dedicated sibling workspace's display label. Never identity. */
export function workerWorkspaceLabel(project: string): string {
  return `${project}-workers`;
}

/** The identity fields a representation's ceremony actually reports (#1110).
 *  The spawn pid rides on {@link WorkerPaneIdentity} for the record; naming,
 *  attribution and the viewer command never read it, so the retile's
 *  transplant can re-establish a representation without pretending to know a
 *  pid it cannot prove. */
type PaneCeremonyIdentity = Pick<WorkerPaneIdentity, "project" | "issue" | "attempt" | "runId"> & {
  /** The spawn pid, when the caller holds one; the ceremony never reads it. */
  pid?: number;
  sessionFile?: string;
};
export function workerPaneLabel(identity: PaneCeremonyIdentity): string {
  return `worker-${identity.project}-${identity.issue}-a${identity.attempt}-${identity.runId.slice(0, 8)}`;
}

const realHerdrRun: HerdrRun = (args) => {
  const res = spawnSync("herdr", [...args], { encoding: "utf8", timeout: 8_000, env: process.env });
  return {
    ok: !res.error && res.status === 0,
    stdout: res.stdout ?? "",
    stderr: res.stderr ?? (res.error === undefined ? "" : String(res.error.message)),
  };
};

/** `omp-conductor tail` — read-only by construction, and never parsed here. */
function defaultViewer(identity: PaneCeremonyIdentity): readonly string[] {
  return ["omp-conductor", "tail", String(identity.issue), "--project", identity.project];
}

/**
 * The pane id inside `herdr pane split`'s answer.
 *
 * herdr answers every CLI call with a one-line JSON envelope
 * (`{"id":"cli:pane:split","result":{"pane":{"pane_id":"w2:pT",…}}}`), so raw
 * stdout is never a pane id. Passing it through verbatim made every worker
 * launch fail `rename` with `pane_not_found` while leaking the pane it had just
 * created (#992) — and the test fake that answered a bare `%7` is why the suite
 * agreed. Parse the field; a shape this does not recognise is no pane id at
 * all, never a best guess scraped out of the payload.
 */
function parsePaneId(stdout: string): string | undefined {
  const line = firstLine(stdout);
  if (line === "") return undefined;
  let payload: unknown;
  try {
    payload = JSON.parse(line);
  } catch {
    return undefined;
  }
  if (payload === null || typeof payload !== "object") return undefined;
  const result = (payload as Record<string, unknown>)["result"];
  if (result === null || typeof result !== "object") return undefined;
  const pane = (result as Record<string, unknown>)["pane"];
  if (pane === null || typeof pane !== "object") return undefined;
  const paneId = (pane as Record<string, unknown>)["pane_id"];
  return typeof paneId === "string" && paneId.trim() !== "" ? paneId.trim() : undefined;
}

/**
 * The workspace and root-pane ids inside `herdr workspace create`'s answer
 * (`{"id":"cli:workspace:create","result":{"workspace":{…},"root_pane":{…}}}`).
 * The root pane is the anchor every later split targets, so a create answer
 * without one is only half a workspace — the caller cleans it up rather than
 * guessing at a pane to split.
 */
function parseCreatedWorkspace(stdout: string): { workspaceId: string; rootPaneId?: string } | undefined {
  const result = envelopeOf(stdout);
  if (result === undefined) return undefined;
  const workspace = result["workspace"];
  if (workspace === null || typeof workspace !== "object") return undefined;
  const workspaceId = (workspace as Record<string, unknown>)["workspace_id"];
  if (typeof workspaceId !== "string" || workspaceId.trim() === "") return undefined;
  const root = (result as Record<string, unknown>)["root_pane"];
  const rootPaneId =
    root !== null && typeof root === "object"
      ? (root as Record<string, unknown>)["pane_id"]
      : undefined;
  return {
    workspaceId: workspaceId.trim(),
    ...(typeof rootPaneId === "string" && rootPaneId.trim() !== "" ? { rootPaneId: rootPaneId.trim() } : {}),
  };
}

/** The `{result: …}` object of a herdr CLI envelope, or nothing. */

function envelopeOf(stdout: string): Record<string, unknown> | undefined {
  const line = firstLine(stdout);
  if (line === "") return undefined;
  let payload: unknown;
  try {
    payload = JSON.parse(line);
  } catch {
    return undefined;
  }
  if (payload === null || typeof payload !== "object") return undefined;
  const result = (payload as Record<string, unknown>)["result"];
  return result === null || typeof result !== "object" ? undefined : (result as Record<string, unknown>);
}

/** The root-pane id inside a `tab create` answer (`result.root_pane.pane_id`):
 *  a tab answer carries no workspace object, so {@link parseCreatedWorkspace}
 *  cannot read it. */
function parseRootPane(stdout: string): string | undefined {
  const result = envelopeOf(stdout);
  const root = result?.["root_pane"];
  const paneId =
    root !== null && typeof root === "object" ? (root as Record<string, unknown>)["pane_id"] : undefined;
  return typeof paneId === "string" && paneId.trim() !== "" ? paneId.trim() : undefined;
}

/** Every workspace in a `workspace list` answer, with the tokens ownership is read from. */
function parseWorkspaceList(stdout: string): { workspaces: { workspaceId: string; label?: string; tokens: Record<string, string> }[] } | undefined {
  const result = envelopeOf(stdout);
  const raw = result?.["workspaces"];
  if (!Array.isArray(raw)) return undefined;
  const workspaces: { workspaceId: string; label?: string; tokens: Record<string, string> }[] = [];
  for (const entry of raw) {
    if (entry === null || typeof entry !== "object") continue;
    const w = entry as Record<string, unknown>;
    if (typeof w["workspace_id"] !== "string" || w["workspace_id"].trim() === "") continue;
    const tokens: Record<string, string> = {};
    if (w["tokens"] !== null && typeof w["tokens"] === "object") {
      for (const [name, value] of Object.entries(w["tokens"] as Record<string, unknown>)) {
        if (typeof value === "string") tokens[name] = value;
      }
    }
    workspaces.push({
      workspaceId: w["workspace_id"],
      ...(typeof w["label"] === "string" ? { label: w["label"] } : {}),
      tokens,
    });
  }
  return { workspaces };
}

/** One pane record out of any `pane list` answer — global or `--workspace`-scoped. */
export interface ListedPane {
  paneId: string;
  /** The run id Herdr still reports for this pane, or "" when it reports none. */
  runId: string;
  agent?: string;
  workspaceId?: string;
  /** The tab holding this pane, when Herdr reports one (#1110). */
  tabId?: string;
}

/** The panes of a `pane list` answer, carrying whichever fields reconciliation reads. */
function parsePaneRecords(stdout: string): { panes: ListedPane[] } | undefined {
  const result = envelopeOf(stdout);
  const raw = result?.["panes"];
  if (!Array.isArray(raw)) return undefined;
  const panes: ListedPane[] = [];
  for (const entry of raw) {
    if (entry === null || typeof entry !== "object") continue;
    const p = entry as Record<string, unknown>;
    if (typeof p["pane_id"] !== "string" || p["pane_id"].trim() === "") continue;
    const session = p["agent_session"];
    const source = session !== null && typeof session === "object" ? (session as Record<string, unknown>)["source"] : undefined;
    const value = session !== null && typeof session === "object" ? (session as Record<string, unknown>)["value"] : undefined;
    panes.push({
      paneId: p["pane_id"].trim(),
      // Ours by construction inside a marked workspace; the reported session id
      // is a secondary signal only (#1035), because Herdr has been observed to
      // drop it while the pane lives on.
      runId: typeof value === "string" && source === WORKER_PANE_SOURCE ? value : "",
      ...(typeof p["agent"] === "string" ? { agent: p["agent"] } : {}),
      ...(typeof p["workspace_id"] === "string" ? { workspaceId: p["workspace_id"] } : {}),
      ...(typeof p["tab_id"] === "string" ? { tabId: p["tab_id"] } : {}),
    });
  }
  return { panes };
}

/** The new tab inside a `tab create` answer (`result.tab`, `result.root_pane`). */
function parseCreatedTab(stdout: string): { tabId?: string; rootPaneId?: string } | undefined {
  const result = envelopeOf(stdout);
  if (result === undefined) return undefined;
  const tab = result["tab"];
  const root = result["root_pane"];
  const tabId =
    tab !== null && typeof tab === "object" ? (tab as Record<string, unknown>)["tab_id"] : undefined;
  const rootPaneId =
    root !== null && typeof root === "object" ? (root as Record<string, unknown>)["pane_id"] : undefined;
  const id = typeof tabId === "string" && tabId.trim() !== "" ? tabId.trim() : undefined;
  const rootId =
    typeof rootPaneId === "string" && rootPaneId.trim() !== "" ? rootPaneId.trim() : undefined;
  if (id === undefined && rootId === undefined) return undefined;
  return {
    ...(id === undefined ? {} : { tabId: id }),
    ...(rootId === undefined ? {} : { rootPaneId: rootId }),
  };
}

/** One pane's rectangle out of a `pane layout` answer. */
interface PaneRect {
  paneId: string;
  /** Top row inside the tab; ascending top to bottom. */
  y: number;
  height: number;
}

/**
 * Every pane rect of the tab a `pane layout --pane` answer describes (#1110) —
 * one call measures the whole tab. An answer carrying no usable rects is
 * unreadable, never an excuse to mutate blind.
 */
function parsePaneLayout(stdout: string): { tabId?: string; panes: PaneRect[] } | undefined {
  const result = envelopeOf(stdout);
  const layout = result?.["layout"];
  if (layout === null || typeof layout !== "object") return undefined;
  const record = layout as Record<string, unknown>;
  const raw = record["panes"];
  if (!Array.isArray(raw)) return undefined;
  const panes: PaneRect[] = [];
  for (const entry of raw) {
    if (entry === null || typeof entry !== "object") continue;
    const pane = entry as Record<string, unknown>;
    const rect = pane["rect"];
    if (typeof pane["pane_id"] !== "string" || pane["pane_id"].trim() === "") continue;
    if (rect === null || typeof rect !== "object") continue;
    const height = (rect as Record<string, unknown>)["height"];
    const y = (rect as Record<string, unknown>)["y"];
    if (typeof height !== "number" || !Number.isFinite(height)) continue;
    if (typeof y !== "number" || !Number.isFinite(y)) continue;
    panes.push({ paneId: pane["pane_id"].trim(), height, y });
  }
  if (panes.length === 0) return undefined;
  const tabId = record["tab_id"];
  return {
    ...(typeof tabId === "string" && tabId.trim() !== "" ? { tabId: tabId.trim() } : {}),
    panes,
  };
}

/** One conductor-marked worker workspace, as discovery returns it. */
export interface WorkerWorkspace {
  workspaceId: string;
  /** The project the marking token binds it to — never derived from the label. */
  project: string;
  label?: string;
}

/**
 * The durable half of worker-workspace discovery (#1035 review).
 *
 * Herdr restores a session's workspaces and panes across a server restart but
 * drops `report-metadata` tokens — probed live: the restored worker workspace
 * comes back carrying no tokens at all. Token-only discovery would then ignore
 * the surface conductor created, build a duplicate beside it, and never be
 * able to reconcile or remove the restored one's stale panes — exactly the
 * accumulation this issue exists to end. Ownership is therefore two-legged:
 * the live token AND the conductor-store record of workspace ids the project
 * created. Either leg proves ownership; the store is the leg that survives.
 */
export interface WorkspaceOwnership {
  /** Store-recorded workspace ids for a project, oldest first. */
  recordedWorkspaces(project: string): readonly string[];
  rememberWorkspace(project: string, workspaceId: string): void;
  forgetWorkspace(project: string, workspaceId: string): void;
}

/** Every workspace in a `workspace list` answer, marked or not. */
function listWorkspacesRaw(
  deps: { run?: HerdrRun; session?: string } = {},
): {
  ok: true;
  workspaces: { workspaceId: string; label?: string; tokens: Record<string, string> }[];
} | { ok: false; reason: string } {
  const run = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  const res = run(["--session", session, "workspace", "list"]);
  if (!res.ok) {
    return { ok: false, reason: `herdr workspace list failed: ${firstLine(res.stderr) || "no output"}` };
  }
  const parsed = parseWorkspaceList(res.stdout);
  if (parsed === undefined) return { ok: false, reason: "herdr workspace list was unreadable" };
  return { ok: true, workspaces: parsed.workspaces };
}

/**
 * Every conductor-marked worker workspace Herdr currently has, keyed by nothing:
 * the caller filters and scopes. Discovery reads the token, not the label — a
 * renamed workspace is still ours, and a lookalike label without the token is
 * not (#1035).
 */
export function listWorkerWorkspaces(deps: { run?: HerdrRun; session?: string } = {}): {
  ok: true;
  workspaces: WorkerWorkspace[];
} | { ok: false; reason: string } {
  const listed = listWorkspacesRaw(deps);
  if (!listed.ok) return listed;
  const workspaces: WorkerWorkspace[] = [];
  for (const w of listed.workspaces) {
    const project = w.tokens[WORKER_WORKSPACE_TOKEN];
    if (project === undefined || project.trim() === "") continue;
    workspaces.push({
      workspaceId: w.workspaceId,
      project: project.trim(),
      ...(w.label === undefined ? {} : { label: w.label }),
    });
  }
  return { ok: true, workspaces };
}

/**
 * Every worker workspace these projects own on the live Herdr (#1035 review):
 * token-marked ones plus store-recorded ids that still exist. A recorded id
 * Herdr no longer lists is dead weight rather than evidence — its workspace is
 * gone, and reconciliation's removal path is what forgets it.
 */
export function ownedWorkerWorkspaces(
  projects: readonly string[],
  deps: { run?: HerdrRun; session?: string; ownership?: WorkspaceOwnership } = {},
): { ok: true; workspaces: WorkerWorkspace[] } | { ok: false; reason: string } {
  const raw = listWorkspacesRaw(deps);
  if (!raw.ok) return raw;
  const scope = new Set(projects);
  const owned = new Map<string, WorkerWorkspace>();
  for (const w of raw.workspaces) {
    const project = w.tokens[WORKER_WORKSPACE_TOKEN]?.trim();
    if (project === undefined || project === "" || !scope.has(project)) continue;
    owned.set(w.workspaceId, {
      workspaceId: w.workspaceId,
      project,
      ...(w.label === undefined ? {} : { label: w.label }),
    });
    // Adopt token-owned surfaces into the durable leg even when no pane needs
    // opening this pass. Otherwise an upgrade followed by a Herdr restart can
    // drop the token before ensureWorkerWorkspace ever records the workspace.
    deps.ownership?.rememberWorkspace(project, w.workspaceId);
  }
  for (const project of projects) {
    for (const id of deps.ownership?.recordedWorkspaces(project) ?? []) {
      if (owned.has(id)) continue;
      const hit = raw.workspaces.find((w) => w.workspaceId === id);
      if (hit === undefined) {
        // The workspace disappeared outside conductor. Forget the stale id now
        // so a future Herdr session cannot reuse it as false ownership.
        deps.ownership?.forgetWorkspace(project, id);
        continue;
      }
      owned.set(id, { workspaceId: id, project, ...(hit.label === undefined ? {} : { label: hit.label }) });
    }
  }
  return { ok: true, workspaces: [...owned.values()] };
}

export type WorkerWorkspaceResolution =
  | { kind: "ready"; workspaceId: string; anchorPaneId: string; created?: true }
  | { kind: "unavailable"; reason: string };

/**
 * Find this project's worker workspace or create it — the single door every
 * split goes through (#1035).
 *
 * Identity is two-legged now (#1035 review): the marking token OR the
 * conductor-store record, so rediscovery after any restart finds the same
 * workspace whatever Herdr preserved. Duplicates are deliberately NOT
 * resolved here: closing one blind — without listing its panes or checking
 * the result — could destroy the only live representation and strand the
 * durable row pointing at a dead pane. Convergence belongs to
 * {@link reconcileWorkerPanes}, which attributes every pane first and removes
 * a duplicate only once nothing living holds it. A discovered workspace with
 * no panes gets a fresh tab rather than failing forever; a workspace CREATED
 * by this call is closed again on every later failure, because newly-made
 * residue is exactly what the acceptance criterion forbids.
 */
export function ensureWorkerWorkspace(
  project: string,
  deps: { run?: HerdrRun; session?: string; ownership?: WorkspaceOwnership } = {},
): WorkerWorkspaceResolution {
  const run = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  const base = ["--session", session];

  let listed = ownedWorkerWorkspaces([project], deps);
  if (!listed.ok) return { kind: "unavailable", reason: listed.reason };

  for (const existing of [...listed.workspaces].sort((a, b) => (a.workspaceId < b.workspaceId ? -1 : 1))) {
    const res = run([...base, "pane", "list", "--workspace", existing.workspaceId]);
    if (!res.ok) {
      return { kind: "unavailable", reason: `herdr pane list failed: ${firstLine(res.stderr) || "no output"}` };
    }
    const parsed = parsePaneRecords(res.stdout);
    if (parsed === undefined) return { kind: "unavailable", reason: "herdr pane list was unreadable" };
    let anchor = parsed.panes[0]?.paneId;
    if (anchor === undefined) {
      // No pane to split: give the workspace a fresh tab and use its root pane.
      const tabbed = run([...base, "tab", "create", "--workspace", existing.workspaceId, "--no-focus"]);
      if (!tabbed.ok) {
        return { kind: "unavailable", reason: `herdr tab create failed: ${firstLine(tabbed.stderr) || "no output"}` };
      }
      anchor = parseRootPane(tabbed.stdout);
      if (anchor === undefined) {
        return { kind: "unavailable", reason: "herdr tab create reported no root pane" };
      }
    }
    // Adopted into the store: if this workspace's token is what Herdr drops
    // next restart, the record below is what still finds it.
    deps.ownership?.rememberWorkspace(project, existing.workspaceId);
    return { kind: "ready", workspaceId: existing.workspaceId, anchorPaneId: anchor };
  }

  const made = run([
    ...base,
    "workspace",
    "create",
    "--label",
    workerWorkspaceLabel(project),
    "--no-focus",
  ]);
  if (!made.ok) {
    return { kind: "unavailable", reason: `herdr workspace create failed: ${firstLine(made.stderr) || "no output"}` };
  }
  const parsed = parseCreatedWorkspace(made.stdout);
  if (parsed === undefined) {
    return { kind: "unavailable", reason: "herdr workspace create reported no workspace id" };
  }
  /** This attempt made the workspace, so every exit from here takes it back. */
  const discardMade = (): void => {
    run([...base, "workspace", "close", parsed.workspaceId]);
    deps.ownership?.forgetWorkspace(project, parsed.workspaceId);
  };
  // Mark before use: an unmarked workspace would be invisible to token-side
  // discovery on the next pass — an empty shell accumulating precisely as
  // #1035 describes. A failure here takes the half-made workspace with it.
  const marked = run([
    ...base,
    "workspace",
    "report-metadata",
    parsed.workspaceId,
    "--source",
    WORKER_PANE_SOURCE,
    "--token",
    `${WORKER_WORKSPACE_TOKEN}=${project}`,
  ]);
  if (!marked.ok) {
    discardMade();
    return {
      kind: "unavailable",
      reason: `herdr workspace report-metadata failed: ${firstLine(marked.stderr) || "no output"}`,
    };
  }
  if (parsed.rootPaneId === undefined) {
    // No anchor means no split can ever be targeted; a fresh-made workspace
    // without one is residue this very attempt must not leave behind
    // (#1035 review) — so it goes now, not on some later pass.
    discardMade();
    return { kind: "unavailable", reason: "herdr workspace create reported no root pane" };
  }
  deps.ownership?.rememberWorkspace(project, parsed.workspaceId);
  return { kind: "ready", workspaceId: parsed.workspaceId, anchorPaneId: parsed.rootPaneId, created: true };
}

/**
 * Name a pane for its run, report the run's identity onto it, start the
 * read-only follower, and report the working state — the ceremony every fresh
 * representation goes through, shared verbatim by {@link openWorkerPane} and
 * the #1110 retile's identity transplant. Reasons come back exactly as the
 * inline steps worded them, because callers surface them unchanged.
 */
function establishWorkerPane(
  paneId: string,
  identity: PaneCeremonyIdentity,
  deps: WorkerPaneDeps & { session: string },
): { ok: true; label: string } | { ok: false; reason: string } {
  const run = deps.run ?? realHerdrRun;
  const base = ["--session", deps.session, "pane"];
  const label = workerPaneLabel(identity);

  const named = run([...base, "rename", paneId, label]);
  if (!named.ok) {
    return { ok: false, reason: `herdr pane rename failed: ${firstLine(named.stderr) || "no output"}` };
  }

  // Identity before display: the pane must be attributable to this exact run
  // before it shows anything, so a pane that appears is never a pane nobody can
  // trace back to a worker.
  const identified = run([
    ...base,
    "report-agent-session",
    paneId,
    "--source",
    WORKER_PANE_SOURCE,
    "--agent",
    label,
    "--agent-session-id",
    identity.runId,
    ...(identity.sessionFile === undefined ? [] : ["--agent-session-path", identity.sessionFile]),
    "--session-start-source",
    WORKER_PANE_SOURCE,
  ]);
  if (!identified.ok) {
    return {
      ok: false,
      reason: `herdr pane report-agent-session failed: ${firstLine(identified.stderr) || "no output"}`,
    };
  }

  const started = run([...base, "run", paneId, ...(deps.viewer?.(identity) ?? defaultViewer(identity))]);
  if (!started.ok) {
    return { ok: false, reason: `herdr pane run failed: ${firstLine(started.stderr) || "no output"}` };
  }

  // A worker that has just spawned is working by definition. The ongoing
  // projection of turn/pause/blocked transitions is #842's, from the same typed
  // events — never from the pane's output.
  const reported = reportWorkerPaneState(paneId, label, "working", { run, session: deps.session });
  if (!reported.ok) return { ok: false, reason: reported.reason };
  return { ok: true, label };
}

/**
 * Create the pane, run the follower in it, and report the child's identity and
 * initial state — or say exactly why it could not.
 *
 * The representation lives in the project's worker workspace (#1035), found or
 * created first, and the split targets an anchor pane inside it explicitly, so
 * the operator's own layout is never touched. Every step is checked, and the
 * first failure returns `unavailable` with the reason and leaves ZERO residue
 * of this attempt (#1035 review): the pane the split had already created is
 * closed, and a workspace this call CREATED is closed with it — only surfaces
 * that already existed are preserved. A partially established representation
 * is worse than none: it is a pane the operator would read as a tracked
 * worker, and before #992 it was left behind on every single launch. What
 * this deliberately does NOT do is decide what a failure means for the launch
 * — failing closed versus running degraded is #841's policy, and inventing it
 * here would pre-empt it.
 */
export function openWorkerPane(identity: WorkerPaneIdentity, deps: WorkerPaneDeps = {}): WorkerPaneOutcome {
  const run = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  const label = workerPaneLabel(identity);
  const ensured = ensureWorkerWorkspace(identity.project, deps);
  if (ensured.kind === "unavailable") return ensured;
  const base = ["--session", session, "pane"];

  /** Undo everything this attempt built so far — the split's pane once there
   *  is one, and always a workspace this call itself created (#1035 review). */
  const abandon = (reason: string, phase?: "split", paneId?: string): WorkerPaneOutcome => {
    if (paneId !== undefined) run([...base, "close", paneId]);
    if (ensured.created === true) {
      run(["--session", session, "workspace", "close", ensured.workspaceId]);
      deps.ownership?.forgetWorkspace(identity.project, ensured.workspaceId);
    }
    return { kind: "unavailable", reason, ...(phase === undefined ? {} : { phase }) };
  };

  // The explicit target is the whole point (#1035): the split names a pane
  // inside the worker workspace, so whichever pane the operator happens to be
  // looking at is never touched.
  const split = run([
    ...base,
    "split",
    "--pane",
    ensured.anchorPaneId,
    "--direction",
    "down",
    "--ratio",
    "0.3",
    "--no-focus",
  ]);
  if (!split.ok) {
    return abandon(
      `herdr pane split failed: ${firstLine(split.stderr) || "no output"}`,
      "split",
    );
  }
  const paneId = parsePaneId(split.stdout);
  if (paneId === undefined) {
    return abandon("herdr pane split reported no pane id");
  }

  const established = establishWorkerPane(paneId, identity, { ...deps, session });
  if (!established.ok) {
    return abandon(established.reason, undefined, paneId);
  }

  // Layout hygiene runs last, so a refused retile can never fail the launch
  // it follows (#1110). A workspace this call itself created holds one worker
  // beside its bare skeleton — nothing to rebalance yet; the junk sweep takes
  // the skeleton and later passes keep the tab even. When a rebuilt tab's
  // skeleton carries this run's identity, the tracked outcome names the pane
  // that now represents it.
  const retiled: WorkerTabRebalance =
    ensured.created === true
      ? { ok: true, rebuilt: false }
      : rebalanceWorkerTabs(ensured.workspaceId, {
          run,
          session,
          ...(deps.viewer === undefined ? {} : { viewer: deps.viewer }),
          reidentify: (candidate) => (candidate === paneId ? identity : undefined),
        });
  const displayed =
    retiled.ok && retiled.transplanted !== undefined && retiled.transplanted.fromPaneId === paneId
      ? retiled.transplanted.toPaneId
      : paneId;

  return {
    kind: "tracked",
    paneId: displayed,
    label: established.label,
    pid: identity.pid,
    workspaceId: ensured.workspaceId,
  };
}

/** Report one lifecycle state for a tracked pane. Monotonic `seq` is the caller's. */
export function reportWorkerPaneState(
  paneId: string,
  label: string,
  state: "working" | "idle" | "blocked" | "unknown",
  deps: { run?: HerdrRun; session?: string; seq?: number; message?: string } = {},
): { ok: true } | { ok: false; reason: string } {
  const run = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  const res = run([
    "--session",
    session,
    "pane",
    "report-agent",
    paneId,
    "--source",
    WORKER_PANE_SOURCE,
    "--agent",
    label,
    "--state",
    state,
    ...(deps.seq === undefined ? [] : ["--seq", String(deps.seq)]),
    ...(deps.message === undefined ? [] : ["--message", deps.message]),
  ]);
  return res.ok
    ? { ok: true }
    : { ok: false, reason: `herdr pane report-agent failed: ${firstLine(res.stderr) || "no output"}` };
}

/**
 * Hand lifecycle authority back when conductor no longer speaks for a pane's
 * agent. Half of {@link retireWorkerPane}: it says "this agent record is no
 * longer ours" without removing anything, which is what reconciliation wants
 * for a pane it is about to close anyway but also stands alone for callers
 * that only re-report identity (#842's adopt path).
 */
export function releaseWorkerPane(
  paneId: string,
  label: string,
  deps: { run?: HerdrRun; session?: string; seq?: number } = {},
): { ok: true } | { ok: false; reason: string } {
  const run = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  const res = run([
    "--session",
    session,
    "pane",
    "release-agent",
    paneId,
    "--source",
    WORKER_PANE_SOURCE,
    "--agent",
    label,
    ...(deps.seq === undefined ? [] : ["--seq", String(deps.seq)]),
  ]);
  return res.ok
    ? { ok: true }
    : { ok: false, reason: `herdr pane release-agent failed: ${firstLine(res.stderr) || "no output"}` };
}

/** Close one pane outright. The pane holds only ever a follower, so closing
 *  it cannot reach the authoritative worker — that child belongs to the
 *  daemon's process tree, not to any terminal (#1035). */
export function closeWorkerPane(
  paneId: string,
  deps: { run?: HerdrRun; session?: string } = {},
): { ok: true } | { ok: false; reason: string } {
  const run = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  const res = run(["--session", session, "pane", "close", paneId]);
  return res.ok
    ? { ok: true }
    : { ok: false, reason: `herdr pane close failed: ${firstLine(res.stderr) || "no output"}` };
}

/**
 * Retire one run's visual representation (#1035): hand lifecycle authority
 * back to Herdr, then close the pane.
 *
 * #841 originally stopped at release, and production showed what that leaves:
 * panes advertising `working` for runs that had settled hours earlier, their
 * follower long exited, a bare shell where a worker used to be. A settled run
 * keeps no representation. The close reaches only the follower — the
 * authoritative child was never in the pane, so nothing here can signal it.
 *
 * A close failure is reported, not swallowed: the next reconciliation pass
 * sees the pane as unclaimed junk and closes it again, so one refusal costs a
 * pass rather than the representation.
 */
export function retireWorkerPane(
  paneId: string,
  label: string,
  deps: { run?: HerdrRun; session?: string; seq?: number } = {},
): { ok: true } | { ok: false; reason: string } {
  const released = releaseWorkerPane(paneId, label, deps);
  if (!released.ok) return released;
  return closeWorkerPane(paneId, deps);
}

/**
 * How far the tallest pane may sit above the shortest before a tab counts as
 * crooked (#1110). Five workers attached by repeated fixed-ratio splits of one
 * anchor come out geometrically smaller — roughly 30%, 21%, 15%, 10%, 7% of
 * the workspace — so the fifth lands far past this line; a freshly tiled tab
 * sits at about 1.0.
 */
export const PANE_HEIGHT_TOLERANCE = 1.5;

/**
 * Split ratios that leave `slots` equal rows behind when every newcomer is
 * anchored on the same shrinking skeleton (#1110).
 *
 * Measured on a live Herdr: a split or move with `--ratio r` leaves the ANCHOR
 * an r share of itself and hands the newcomer the rest. Taking
 * `r_j = (slots - j) / (slots - j + 1)` for move j therefore hands each
 * newcomer exactly one slot of the whole — the arithmetic that turns the fifth
 * attach from a fourteenth of the height into a fifth.
 */
export function evenTileRatios(slots: number): readonly number[] {
  if (!Number.isInteger(slots) || slots < 2) return [];
  const ratios: number[] = [];
  for (let j = 1; j <= slots - 1; j++) ratios.push((slots - j) / (slots - j + 1));
  return ratios;
}

export type WorkerTabRebalance =
  | {
      ok: true;
      /** True when at least one crooked tab was rebuilt by this call. */
      rebuilt: boolean;
      /** The representation that rode the rebuilt tab's skeleton, when the old
       *  tab could only be disposed of by re-establishing one run's visual. */
      transplanted?: { fromPaneId: string; toPaneId: string; identity: PaneCeremonyIdentity };
    }
    | { ok: false; reason: string };

/**
 * Retile every crooked tab of one worker workspace into equal rows (#1110).
 * Repeated fixed-ratio splits of a single anchor make each new worker
 * geometrically smaller than the last, and a settling pane hands its space to
 * a split sibling instead of sharing it out — attaches and detaches both
 * drift. `pane resize` was probed live and rejected: it moves borders through
 * the split tree proportionally, so nudges cascade across neighbours, invert
 * under sign, and wedge into states where they no-op forever. The honest
 * rebuild moves the standing panes — ids, followers and scrollback surviving
 * the move — into a fresh tab whose sequential anchor ratios land them even,
 * re-establishes one run's representation on the skeleton root so no bare
 * shell remains (a closed skeleton dumps its whole share into one split
 * sibling, which was measured skewing a six-pane tab back to 11:6), and
 * closes the emptied tab.
 *
 * Only callers that already own the workspace invoke this: reconciliation
 * scopes it to token-marked or store-recorded surfaces (#1035), and
 * {@link openWorkerPane} calls it for the workspace it just found or created
 * through the same discovery. No pane is created, written to, or given input:
 * a move re-parents the terminal whole, and the one follower that restarts
 * (the transplanted run's) restarts the same read-only `omp-conductor tail`.
 *
 * Best effort by design. Any refusal stops the tab's rebuild where it stands
 * and is reported; partial rebuilds converge because already-moved panes
 * measure even in their new tab, so each retry moves strictly fewer panes.
 * Callers treat a failure as next pass's identical request, never as a launch
 * or settlement problem: a crooked layout never outweighs the representation
 * it hosts.
 */
export function rebalanceWorkerTabs(
  workspaceId: string,
  deps: {
    run?: HerdrRun;
    session?: string;
    viewer?: (identity: PaneCeremonyIdentity) => readonly string[];
    /** Whose representation a pane carries, as far as the caller knows. A pane
     *  nobody can identify is never chosen to ride the skeleton. */
    reidentify?: (paneId: string) => PaneCeremonyIdentity | undefined;
  } = {},
): WorkerTabRebalance {
  const run = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  const base = ["--session", session];

  const res = run([...base, "pane", "list", "--workspace", workspaceId]);
  if (!res.ok) {
    return { ok: false, reason: `herdr pane list failed: ${firstLine(res.stderr) || "no output"}` };
  }
  const parsed = parsePaneRecords(res.stdout);
  if (parsed === undefined) return { ok: false, reason: "herdr pane list was unreadable" };

  /** Tab id → member pane ids, in listing order. */
  const tabs = new Map<string, string[]>();
  for (const pane of parsed.panes) {
    const members = tabs.get(pane.tabId ?? "");
    if (members === undefined) tabs.set(pane.tabId ?? "", [pane.paneId]);
    else members.push(pane.paneId);
  }

  let rebuilt = false;
  for (const [tabId, paneIds] of tabs) {
    // A tab Herdr did not name cannot be torn down safely after the rebuild,
    // so measuring it is where its story ends.
    if (tabId === "" || paneIds.length < 2) continue;
    const measured = run([...base, "pane", "layout", "--pane", paneIds[0]!]);
    if (!measured.ok) {
      return { ok: false, reason: `herdr pane layout failed: ${firstLine(measured.stderr) || "no output"}` };
    }
    const layout = parsePaneLayout(measured.stdout);
    if (layout === undefined) return { ok: false, reason: "herdr pane layout was unreadable" };
    const rectOf = new Map(layout.panes.map((rect) => [rect.paneId, rect]));
    const rects = paneIds.map((id) => rectOf.get(id)).filter((rect): rect is PaneRect => rect !== undefined);
    // A listed pane the layout does not describe is a tab changing under us:
    // mutate nothing on half evidence.
    if (rects.length !== paneIds.length) continue;
    const heights = rects.map((rect) => rect.height);
    const shortest = Math.min(...heights);
    const tallest = Math.max(...heights);
    // A zero-height pane is crooked beside anything alive; the tolerance is
    // inclusive at exactly 1.5×, since the criterion reads "no more than".
    // Below three average cells a pane is illegible whatever the ratio —
    // tiling cannot help a terminal that small.
    if (tallest <= shortest * PANE_HEIGHT_TOLERANCE) continue;
    if (heights.reduce((sum, h) => sum + h, 0) / heights.length < 3) continue;

    // The rider is the topmost pane the caller can identify; it re-founds its
    // representation on the skeleton, which lands as the tab's top row, so the
    // operator's reading order survives. The rest move bottom-to-top,
    // reversing into their original order underneath it.
    const ascending = [...rects].sort((a, b) => a.y - b.y);
    const rider = ascending.find((rect) => deps.reidentify?.(rect.paneId) !== undefined);
    if (rider === undefined) continue;
    const identity = deps.reidentify!(rider.paneId)!;
    const movers = ascending.filter((rect) => rect.paneId !== rider.paneId).reverse();

    const made = run([...base, "tab", "create", "--workspace", workspaceId, "--no-focus"]);
    if (!made.ok) {
      return { ok: false, reason: `herdr tab create failed: ${firstLine(made.stderr) || "no output"}` };
    }
    const created = parseCreatedTab(made.stdout);
    if (created?.tabId === undefined || created.rootPaneId === undefined) {
      // Half an answer: the tab may exist and must not linger as an empty
      // shell this attempt cannot account for.
      if (created?.tabId !== undefined) run([...base, "tab", "close", created.tabId]);
      return { ok: false, reason: "herdr tab create reported no root pane" };
    }

    const slots = movers.length + 1;
    const ratios = evenTileRatios(slots);
    let placed = 0;
    for (; placed < movers.length; placed++) {
      const moved = run([
        ...base,
        "pane",
        "move",
        movers[placed]!.paneId,
        "--tab",
        created.tabId,
        "--split",
        "down",
        "--target-pane",
        created.rootPaneId,
        "--ratio",
        String(ratios[placed]),
      ]);
      if (!moved.ok) {
        return { ok: false, reason: `herdr pane move failed: ${firstLine(moved.stderr) || "no output"}` };
      }
    }

    // The skeleton kept the top slot only by inheriting the rider's identity:
    // the same name, the same session report, the same read-only tail. A bare
    // shell up there would be junk, and junk closure is exactly what skews.
    const founded = establishWorkerPane(created.rootPaneId, identity, {
      run,
      session,
      ...(deps.viewer === undefined ? {} : { viewer: deps.viewer }),
    });
    if (!founded.ok) {
      return { ok: false, reason: founded.reason };
    }
    const retiredRider = run([...base, "pane", "close", rider.paneId]);
    if (!retiredRider.ok) {
      return { ok: false, reason: `herdr pane close failed: ${firstLine(retiredRider.stderr) || "no output"}` };
    }
    // Herdr tears an emptied tab down by itself (probed live), so a refusal
    // here is only fatal when panes still stand in the old tab — verified,
    // never assumed.
    const closed = run([...base, "tab", "close", tabId]);
    if (!closed.ok) {
      const verify = run([...base, "pane", "list", "--workspace", workspaceId]);
      const leftover = verify.ok
        ? (parsePaneRecords(verify.stdout)?.panes ?? []).filter((pane) => pane.tabId === tabId)
        : null;
      if (leftover === null || leftover.length > 0) {
        return { ok: false, reason: `herdr tab close failed: ${firstLine(closed.stderr) || "no output"}` };
      }
    }
    rebuilt = true;
    return {
      ok: true,
      rebuilt: true,
      transplanted: { fromPaneId: rider.paneId, toPaneId: created.rootPaneId, identity },
    };
  }
  return { ok: true, rebuilt };
}

/**
 * Hand back the representation a dead worker left behind (#842), retired for
 * good by closing it (#1035) — but only once the recorded pane is PROVEN still
 * conductor's (#1035 review).
 *
 * Called for every run a restart reaps. The recorded pid is deliberately NOT
 * consulted for liveness: a `session-host` child dies with the daemon that owned
 * its verb socket, so an orphaned row's worker is gone whatever pid it carries —
 * and pids are reused, so checking one is how a stranger's process comes to read
 * as a live worker. The same reuse argument applies to pane ids: after a Herdr
 * restart the stored `w2:pT` may belong to a pane in somebody else's workspace,
 * so a blind release+close could destroy a stranger's terminal. Ownership is
 * therefore proven before anything is closed:
 *
 * - the pane must be listed inside a workspace conductor OWNS — a
 *   {@link WORKER_WORKSPACE_TOKEN}-marked one, or one the store recorded for
 *   the run's project (#1035 review: Herdr restarts restore workspaces without
 *   their tokens, and a token-only check would leave the restored pane
 *   unretired forever); and
 */
export function releaseOrphanedWorkerPane(
  run: { paneId?: string; paneLabel?: string; runId?: string; project?: string },
  deps: { run?: HerdrRun; session?: string; seq?: number; ownership?: WorkspaceOwnership } = {},
):
  | { kind: "none" }
  | { kind: "released"; paneId: string }
  | { kind: "failed"; paneId: string; reason: string }
  | { kind: "unowned"; paneId: string; reason: string } {
  if (run.paneId === undefined || run.paneLabel === undefined) return { kind: "none" };
  const discovered = ownedWorkerWorkspaces(run.project === undefined ? [] : [run.project], deps);
  if (!discovered.ok) {
    return { kind: "failed", paneId: run.paneId, reason: discovered.reason };
  }
  const runFn = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  let home: string | undefined;
  let reportedAgent: string | undefined;
  let reportedSession: string | undefined;
  for (const ws of discovered.workspaces) {
    const res = runFn(["--session", session, "pane", "list", "--workspace", ws.workspaceId]);
    if (!res.ok) {
      return {
        kind: "failed",
        paneId: run.paneId,
        reason: `herdr pane list failed: ${firstLine(res.stderr) || "no output"}`,
      };
    }
    const parsed = parsePaneRecords(res.stdout);
    if (parsed === undefined) {
      return { kind: "failed", paneId: run.paneId, reason: "herdr pane list was unreadable" };
    }
    const found = parsed.panes.find((pane) => pane.paneId === run.paneId);
    if (found !== undefined && home === undefined) {
      home = ws.workspaceId;
      if (found.agent !== undefined) reportedAgent = found.agent;
      if (found.runId !== "") reportedSession = found.runId;
    }
  }
  if (home === undefined) {
    // Not in any marked workspace: either already retired or never ours to
    // touch. Both answers close nothing.
    return { kind: "none" };
  }
  if (
    (reportedAgent !== undefined && reportedAgent !== run.paneLabel) ||
    (reportedSession !== undefined && run.runId !== undefined && reportedSession !== run.runId)
  ) {
    return {
      kind: "unowned",
      paneId: run.paneId,
      reason: `pane ${run.paneId} in marked workspace ${home} reports another identity (agent ${reportedAgent ?? "none"}, session ${reportedSession ?? "none"})`,
    };
  }
  const retired = retireWorkerPane(run.paneId, run.paneLabel, deps);
  return retired.ok
    ? { kind: "released", paneId: run.paneId }
    : { kind: "failed", paneId: run.paneId, reason: retired.reason };
}

/**
 * Every pane inside a conductor-marked worker workspace (#1035).
 *
 * Ownership here is structural: conductor created the workspace, marked it,
 * and splits into it exclusively, so every pane inside it is conductor's
 * however Herdr feels about reporting `agent_session` back — production has
 * repeatedly handed back worker panes with that record dropped to null, which
 * is why the previous source-filtered listing could not see the very panes
 * that needed cleaning. The reported session id still rides along as a
 * secondary attribution signal; an empty run id means "unattributable", and
 * unattributable panes in a marked workspace are exactly the stale shells
 * this surface exists to converge away.
 */
export function listWorkerPanes(
  deps: { run?: HerdrRun; session?: string } = {},
): { ok: true; panes: (ListedPane & { project: string })[] } | { ok: false; reason: string } {
  const workspaces = listWorkerWorkspaces(deps);
  if (!workspaces.ok) return workspaces;
  const run = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  const owned: (ListedPane & { project: string })[] = [];
  for (const ws of workspaces.workspaces) {
    const res = run(["--session", session, "pane", "list", "--workspace", ws.workspaceId]);
    if (!res.ok) {
      return { ok: false, reason: `herdr pane list failed: ${firstLine(res.stderr) || "no output"}` };
    }
    const parsed = parsePaneRecords(res.stdout);
    if (parsed === undefined) return { ok: false, reason: "herdr pane list carried no pane array" };
    for (const pane of parsed.panes) owned.push({ ...pane, project: ws.project });
  }
  return { ok: true, panes: owned };
}

/** One live worker, as the reconciler needs to see it. */
export interface LiveWorkerPane {
  runId: string;
  issue: number;
  attempt: number;
  project: string;
  /** The recorded session-host pid; absent when nothing ever reported one. */
  pid?: number;
  paneId?: string;
  paneLabel?: string;
  sessionFile?: string;
}

export type StaleCause = "settled" | "duplicate" | "unidentified";

export type WorkerPaneReconciliation =
  /** The recorded pane is still there and its follower is alive — nothing done. */
  | { kind: "intact"; runId: string; paneId: string }
  /** The pane stands but its follower table could not decide life or death:
   *  no evidence, no destruction (#1035 review). The visual may be stale; the
   *  caller says so once instead of gambling a live worker's representation. */
  | { kind: "visual-unreadable"; runId: string; paneId: string; reason: string }
  /** The run had no usable representation; a new one now represents the same
   *  child — a Herdr restart, a missing pane, or a follower that had exited. */
  | { kind: "reassociated"; runId: string; paneId: string; label: string }
  /** No representation, and the reason. The run keeps working regardless.
   *  `attempted` marks the ones that spent a re-establishment attempt (#998),
   *  so the caller can bound them; the no-pid case costs nothing. `phase`
   *  carries {@link WorkerPaneOutcome}'s split marker through. */
  | { kind: "untracked"; runId: string; reason: string; attempted?: true; phase?: "split" }
  /** The attempt budget for this run is spent, so nothing was tried this pass
   *  (#998). Before this existed, a run whose pane could not be created was
   *  retried on every reconciliation pass forever — and each attempt leaked a
   *  pane until #992, ending in `ghostty error -2` once enough had piled up. */
  | { kind: "attempts-exhausted"; runId: string; attempts: number }
  /** A conductor-workspace pane with no live run behind it: authority handed
   *  back and the pane closed. `settled` — the run finished and its retirement
   *  never landed; `duplicate` — a second representation of one live run, or
   *  the one a fresh replacement displaced; `unidentified` — a representation
   *  whose run id nobody can resolve any more, the bare shells production
   *  accumulated (#1035). */
  | { kind: "stale-released"; paneId: string; runId: string; cause: StaleCause }
  | { kind: "stale-release-failed"; paneId: string; runId: string; reason: string; cause: StaleCause }
  /** A scoped marked workspace nobody's live run holds any more, emptied by
   *  the cleanup above, is closed (#1035). Every marked workspace converges,
   *  not just one per project (#1035 review). */
  | { kind: "workspace-removed"; project: string; workspaceId: string }
  | { kind: "workspace-remove-failed"; project: string; workspaceId: string; reason: string };

/** The verdict the foreground process table supports about one pane's
 *  read-only follower. Terminal output is never read: this says whether the
 *  *visual* is alive, nothing more, and settlement stays the store's business
 *  (#1035). `unknown` is a real verdict, not an error — recognition of
 *  `omp-conductor` in the table is positive evidence of life, but its absence
 *  is only evidence of death when every entry carries enough identity (name or
 *  argv) to rule it out. Real Herdr answers name + argv; a table with blanked
 *  or missing fields, no shell pid, or no entries at all cannot decide, and a
 *  pane destroyed on a guess we did not have is a live worker gone blind. */
export type FollowerPulse =
  | { pulse: "alive" }
  | { pulse: "dead" }
  | { pulse: "unknown"; reason: string };

export function followerPulseFromProcessInfo(info: ProcessInfo): FollowerPulse {
  const procs = Array.isArray(info.foreground_processes) ? info.foreground_processes : undefined;
  if (procs === undefined || procs.length === 0) {
    return { pulse: "unknown", reason: "process-info carried no foreground process table" };
  }
  const shell = typeof info.shell_pid === "number" ? info.shell_pid : undefined;
  let nonShell = false;
  for (const proc of procs) {
    const parts = [proc.name ?? "", proc.argv0 ?? "", ...(proc.argv ?? [])];
    const isShell = shell !== undefined && proc.pid === shell;
    if (!isShell && !parts.some((a) => a !== "")) {
      return { pulse: "unknown", reason: `foreground pid ${proc.pid ?? "unknown"} carries no name or argv` };
    }
    if (!isShell) {
      nonShell = true;
      if (parts.some((a) => a.toLowerCase() === "omp-conductor" || a.toLowerCase().endsWith("/omp-conductor"))) {
        return { pulse: "alive" };
      }
    }
  }
  if (!nonShell && shell === undefined) {
    return { pulse: "unknown", reason: "process-info named neither a shell nor any identifiable process" };
  }
  return { pulse: "dead" };
}

/**
 * Probe one pane's follower liveness. A probe that cannot be read, or a table
 * that cannot decide, answers `unknown` rather than dead: destroying a
 * representation on evidence we do not have is how a healthy worker goes blind
 * mid-run.
 */
export function paneFollowerAlive(
  paneId: string,
  deps: { run?: HerdrRun; session?: string } = {},
): FollowerPulse {
  const run = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  const res = run(["--session", session, "pane", "process-info", "--pane", paneId]);
  if (!res.ok) {
    return {
      pulse: "unknown",
      reason: `herdr pane process-info failed: ${firstLine(res.stderr) || "no output"}`,
    };
  }
  try {
    return followerPulseFromProcessInfo(parseHerdrProcessInfo(res.stdout, paneId));
  } catch (err) {
    return { pulse: "unknown", reason: err instanceof Error ? err.message : String(err) };
  }
}

function firstLine(text: string): string {
  return text.split("\n", 1)[0]?.trim() ?? "";
}
/** The kill syscall, injectable so error mapping is testable without one. */
export type KillFn = (pid: number, sig: NodeJS.Signals | 0) => void;

const realKill: KillFn = (pid, sig) => {
  process.kill(pid, sig);
};

/**
 * Make the conductor-owned worker workspaces agree with the live run set
 * (#841, reworked by #1035).
 *
 * Idempotent by construction: a second pass over an already-reconciled fleet
 * returns `intact` for every live run and finds nothing left to close, so
 * repeated launch/settlement/restart cycles converge instead of accumulating.
 *
 * Four rules, and each exists to refuse a specific way this goes wrong:
 *
 * - **Only owned workspaces are touched.** Scope is the projects of the live
 *   set plus the caller's managed list; ownership is the marking token OR the
 *   conductor-store record (#1035 review), so a Herdr restart that drops
 *   tokens still finds the restored surface. A pane in another project's
 *   worker workspace is invisible to this pass, which is what keeps conductor
 *   and veltrosecurity from discovering, reusing, or cleaning each other.
 * - **Attribution is exact, structural first.** A pane belongs to the live run
 *   the store recorded it under; the session id Herdr reports is the fallback.
 *   Everything else in an owned workspace — a settled run's leftover, a
 *   duplicate of a live run, a representation whose identity Herdr dropped —
 *   is closed, because inside an owned workspace "ours" and "junk" are the
 *   only options and leaving junk is what filled the operator's layout.
 * - **A live run keeps exactly one living representation.** Duplicates fold
 *   into the recorded pane; a pane whose follower has exited is replaced while
 *   the run is live; a missing pane is recreated — all bounded by the same
 *   re-establishment budget (#998).
 * - **Nothing here can stop a worker.** Panes hold followers only; the
 *   authoritative child lives in the daemon's process tree and is never
 *   signalled, and worker liveness comes from the caller's live set alone.
 */
export function reconcileWorkerPanes(
  live: readonly LiveWorkerPane[],
  deps: WorkerPaneDeps = {},
  /** Attempts already spent per run id, owned by the caller so this stays pure
   *  and a daemon restart legitimately gets a fresh budget (#998). */
  attempts: ReadonlyMap<string, number> = new Map(),
  maxAttempts = PANE_REATTEMPT_MAX,
  /** Projects this pass manages even when they have no live runs: their stale
   *  panes are cleaned and a workspace left empty by that cleanup is removed. */
  managedProjects: readonly string[] = [],
): { ok: true; outcomes: WorkerPaneReconciliation[] } | { ok: false; reason: string } {
  const run = deps.run ?? realHerdrRun;
  const session = deps.session ?? resolveHerdrSession();
  const base = ["--session", session];

  const scope = new Set([...managedProjects, ...live.map((worker) => worker.project)]);
  const discovered = ownedWorkerWorkspaces([...scope], deps);
  if (!discovered.ok) return { ok: false, reason: discovered.reason };
  // Deterministic order so outcomes and closes are reproducible pass to pass.
  const scoped = [...discovered.workspaces].sort((a, b) => (a.workspaceId < b.workspaceId ? -1 : 1));

  // Every marked workspace of every scoped project converges (#1035 review):
  // duplicate token-marked workspaces are cleaned like any other, so a settled
  // project cannot keep stale shells in a second marked surface nobody tracks.
  const entries: { ws: WorkerWorkspace; panes: ListedPane[] }[] = [];
  for (const ws of scoped) {
    const res = run([...base, "pane", "list", "--workspace", ws.workspaceId]);
    if (!res.ok) {
      // An unreadable workspace is not evidence that anything is stale, so the
      // pass mutates nothing rather than closing on a guess.
      return { ok: false, reason: `herdr pane list failed: ${firstLine(res.stderr) || "no output"}` };
    }
    const parsed = parsePaneRecords(res.stdout);
    if (parsed === undefined) return { ok: false, reason: "herdr pane list carried no pane array" };
    entries.push({ ws, panes: parsed.panes });
  }

  const liveByRun = new Map(live.map((worker) => [worker.runId, worker]));
  const outcomes: WorkerPaneReconciliation[] = [];
  /** Panes this pass will close, with the identity the outcome names. */
  const junk = new Map<string, { runId: string; cause: StaleCause; agent?: string }>();
  /** Panes still standing when cleanup ends, keyed to their workspace — an
   *  occupied workspace survives, whichever marked surface it sits in. */
  const standingWs = new Map<string, string>();
  /** Workspaces that gained a representation this pass, read from the tracked
   *  outcomes: their listing predates the new pane, so removal never judges
   *  them on it. */
  const openedWs = new Set<string>();

  /** One representation per live run: the pane plus the marked workspace that
   *  holds it. Attribution runs across ALL of a project's marked workspaces,
   *  because the recorded pane may sit in any of them. */
  const held = new Map<string, { pane: ListedPane; wsId: string }>();
  for (const entry of entries) {
    const projectLive = live.filter((worker) => worker.project === entry.ws.project);
    for (const pane of entry.panes) {
      standingWs.set(pane.paneId, entry.ws.workspaceId);
      const owner =
        projectLive.find((worker) => worker.paneId === pane.paneId) ??
        (pane.runId === "" ? undefined : liveByRun.get(pane.runId));
      if (owner === undefined || owner.project !== entry.ws.project) {
        // Not attributable to any live run of this project: a settled run's
        // leftover (still carrying its old id) or an unidentified shell.
        junk.set(pane.paneId, {
          runId: pane.runId,
          cause: pane.runId === "" ? "unidentified" : "settled",
          ...(pane.agent === undefined ? {} : { agent: pane.agent }),
        });
        continue;
      }
      const bucket = held.get(owner.runId);
      if (bucket === undefined) {
        held.set(owner.runId, { pane, wsId: entry.ws.workspaceId });
        continue;
      }
      // Exactly one focusable representation per live run: the recorded pane
      // wins so the durable row stays true without a write; the other is a
      // duplicate (#1035).
      const keep =
        bucket.pane.paneId === owner.paneId
          ? bucket
          : pane.paneId === owner.paneId
            ? { pane, wsId: entry.ws.workspaceId }
            : bucket;
      const drop = keep.pane.paneId === bucket.pane.paneId ? pane : bucket.pane;
      held.set(owner.runId, keep);
      junk.set(drop.paneId, {
        runId: owner.runId,
        cause: "duplicate",
        ...(drop.agent === undefined ? {} : { agent: drop.agent }),
      });
    }
  }

  for (const worker of live) {
    const hold = held.get(worker.runId);
    if (hold === undefined) {
      // No representation anywhere in the marked workspaces. Only the pid makes
      // a replacement honest: the pane represents an exact child, so without
      // one there is nothing to represent and inventing a pane would be the
      // silent claim this whole surface exists to avoid.
      if (worker.pid === undefined) {
        outcomes.push({
          kind: "untracked",
          runId: worker.runId,
          reason: "no session-host pid was ever recorded for this run",
        });
        continue;
      }
      // The budget, checked before the attempt: an exhausted run is skipped
      // silently here and reported once by the caller (#998).
      const spent = attempts.get(worker.runId) ?? 0;
      if (spent >= maxAttempts) {
        outcomes.push({ kind: "attempts-exhausted", runId: worker.runId, attempts: spent });
        continue;
      }
      const opened = openWorkerPane(
        {
          project: worker.project,
          issue: worker.issue,
          attempt: worker.attempt,
          runId: worker.runId,
          pid: worker.pid,
          ...(worker.sessionFile === undefined ? {} : { sessionFile: worker.sessionFile }),
        },
        deps,
      );
      if (opened.kind === "tracked") {
        if (opened.workspaceId !== undefined) openedWs.add(opened.workspaceId);
        outcomes.push({
          kind: "reassociated",
          runId: worker.runId,
          paneId: opened.paneId,
          label: opened.label,
        });
      } else {
        outcomes.push({
          kind: "untracked",
          runId: worker.runId,
          reason: opened.reason,
          attempted: true,
          ...(opened.phase === undefined ? {} : { phase: opened.phase }),
        });
      }
      continue;
    }
    // The pane stands. Whether it still shows anything is the follower's
    // question, asked only to decide the *visual* — never settlement (#1035).
    // A probe that cannot be read, or a table that cannot decide, is no
    // evidence the visual died: it is reported and the pane keeps standing,
    // because a live worker's representation is never destroyed on a guess.
    const pulse = paneFollowerAlive(hold.pane.paneId, deps);
    if (pulse.pulse !== "dead") {
      outcomes.push(
        pulse.pulse === "alive"
          ? { kind: "intact", runId: worker.runId, paneId: hold.pane.paneId }
          : {
              kind: "visual-unreadable",
              runId: worker.runId,
              paneId: hold.pane.paneId,
              reason: pulse.reason,
            },
      );
      continue;
    }
    // The follower has provably exited. A replacement represents the exact
    // child, so a run with no recorded pid keeps its dead pane rather than
    // gaining a representation of nobody — reported so the record drops and
    // the next pass reads the orphan as junk.
    if (worker.pid === undefined) {
      outcomes.push({
        kind: "untracked",
        runId: worker.runId,
        reason: "follower exited but no session-host pid was ever recorded; not replacing",
      });
      continue;
    }
    const spent = attempts.get(worker.runId) ?? 0;
    if (spent >= maxAttempts) {
      // Out of budget: the dead pane keeps its last state until a restart
      // refreshes the budget or the run settles and retires it outright.
      outcomes.push({ kind: "attempts-exhausted", runId: worker.runId, attempts: spent });
      continue;
    }
    const replaced = openWorkerPane(
      {
        project: worker.project,
        issue: worker.issue,
        attempt: worker.attempt,
        runId: worker.runId,
        pid: worker.pid,
        ...(worker.sessionFile === undefined ? {} : { sessionFile: worker.sessionFile }),
      },
      deps,
    );
    if (replaced.kind === "tracked") {
      if (replaced.workspaceId !== undefined) openedWs.add(replaced.workspaceId);
      outcomes.push({
        kind: "reassociated",
        runId: worker.runId,
        paneId: replaced.paneId,
        label: replaced.label,
      });
      junk.set(hold.pane.paneId, {
        runId: worker.runId,
        cause: "duplicate",
        ...(hold.pane.agent === undefined ? {} : { agent: hold.pane.agent }),
      });
    } else {
      outcomes.push({
        kind: "untracked",
        runId: worker.runId,
        reason: replaced.reason,
        attempted: true,
        ...(replaced.phase === undefined ? {} : { phase: replaced.phase }),
      });
      // The replacement failed; the dead pane has nothing left to show, so it
      // goes now rather than advertising a worker for another pass. Its record
      // is dropped either way, so a survivor is junk next pass.
      closeWorkerPane(hold.pane.paneId, deps);
      standingWs.delete(hold.pane.paneId);
    }
  }

  for (const [paneId, info] of junk) {
    const released =
      info.agent === undefined ? ({ ok: true } as const) : releaseWorkerPane(paneId, info.agent, deps);
    if (released.ok) {
      const closed = closeWorkerPane(paneId, deps);
      if (closed.ok) {
        standingWs.delete(paneId);
        outcomes.push({ kind: "stale-released", paneId, runId: info.runId, cause: info.cause });
        continue;
      }
      outcomes.push({
        kind: "stale-release-failed",
        paneId,
        runId: info.runId,
        reason: closed.reason,
        cause: info.cause,
      });
      continue;
    }
    outcomes.push({
      kind: "stale-release-failed",
      paneId,
      runId: info.runId,
      reason: released.reason,
      cause: info.cause,
    });
  }

  // Attach and detach both leave geometry behind: repeated fixed-ratio splits
  // shrink every new pane geometrically (#1110), and a settle hands its space
  // to a split sibling instead of sharing it out. Every scoped workspace that
  // still holds two or more panes gets measured now; a tab past the tolerance
  // is rebuilt evenly — same panes, same ids, followers untouched. The one
  // representation riding the skeleton keeps its reported identity, so the
  // next pass attributes it through the session id exactly as #1035 already
  // requires; this pass deliberately says nothing about layout, because a
  // crooked tile is cosmetics and an outcome row would claim otherwise.
  for (const entry of entries) {
    let standingInWorkspace = 0;
    for (const wsId of standingWs.values()) if (wsId === entry.ws.workspaceId) standingInWorkspace += 1;
    if (standingInWorkspace < 2) continue;
    rebalanceWorkerTabs(entry.ws.workspaceId, {
      run,
      session,
      ...(deps.viewer === undefined ? {} : { viewer: deps.viewer }),
      reidentify: (paneId) => {
        const worker = live.find((candidate) => candidate.paneId === paneId);
        if (worker === undefined) return undefined;
        return {
          project: worker.project,
          issue: worker.issue,
          attempt: worker.attempt,
          runId: worker.runId,
          ...(worker.sessionFile === undefined ? {} : { sessionFile: worker.sessionFile }),
        };
      },
    });
  }

  // Every scoped marked workspace whose panes the cleanup above emptied, and
  // that gained no representation this pass, is closed — however many marked
  // surfaces a project accumulated (#1035 review). A live project's primary
  // workspace survives because its representation keeps it standing; extras
  // converge away; repeated cycles hold at zero shells.
  for (const entry of entries) {
    if (openedWs.has(entry.ws.workspaceId)) continue;
    if ([...standingWs.values()].some((wsId) => wsId === entry.ws.workspaceId)) continue;
    const res = run([...base, "workspace", "close", entry.ws.workspaceId]);
    if (res.ok) deps.ownership?.forgetWorkspace(entry.ws.project, entry.ws.workspaceId);
    // A failed close keeps the record: the workspace may still exist, and
    // forgetting ownership of a live surface is how orphans are manufactured.
    outcomes.push(
      res.ok
        ? { kind: "workspace-removed", project: entry.ws.project, workspaceId: entry.ws.workspaceId }
        : {
            kind: "workspace-remove-failed",
            project: entry.ws.project,
            workspaceId: entry.ws.workspaceId,
            reason: `herdr workspace close failed: ${firstLine(res.stderr) || "no output"}`,
          },
    );
  }
  return { ok: true, outcomes };
}

/**
 * What a failed `kill` proves.
 *
 * Only `ESRCH` — "no such process" — proves the process is gone. `EPERM` means
 * it *exists* and merely is not ours to signal; every other errno is an answer
 * we cannot read. Collapsing either into "gone" is how a live conductor gets
 * reported as stopped.
 */
export function classifyKillError(err: unknown): "gone" | "unknown" {
  return (err as NodeJS.ErrnoException | undefined)?.code === "ESRCH" ? "gone" : "unknown";
}

/** Whether `pid` is still running. Throws when the answer cannot be read. */
export function pidLiveness(pid: number, kill: KillFn = realKill): boolean {
  try {
    kill(pid, 0);
    return true;
  } catch (err) {
    if (classifyKillError(err) === "gone") return false;
    const code = (err as NodeJS.ErrnoException).code;
    throw new Error(`cannot probe pid ${pid} (${code ?? String(err)}) — liveness unknown, not dead`);
  }
}

/**
 * Deliver `sig` to `pid`. An already-gone process is success — that is the
 * outcome we wanted. Every other failure throws: an undelivered signal must
 * never read as a kill.
 */
export function deliverSignal(pid: number, sig: NodeJS.Signals, kill: KillFn = realKill): boolean {
  try {
    kill(pid, sig);
    return true;
  } catch (err) {
    if (classifyKillError(err) === "gone") return true;
    const code = (err as NodeJS.ErrnoException).code;
    throw new Error(`cannot send ${sig} to pid ${pid} (${code ?? String(err)})`);
  }
}

export interface PaneStopDeps {
  herdrBin?: string;
  herdrSession?: string;
  /** Deliver a signal. Return `false` or throw when delivery is unproven. */
  signalPid?: (pid: number, sig: NodeJS.Signals) => boolean;
  /** Liveness probe. MUST throw when it cannot tell — never answer `false`. */
  isAlive?: (pid: number) => boolean;
  /** Clock for deterministic deadline tests. */
  now?: () => number;
  sleep?: (ms: number) => Promise<void>;
  listAgents?: () => Promise<HerdrAgent[]>;
  panePids?: (paneId: string) => Promise<number[]>;
}

/**
 * hold + pin recovery first + stop exact omp conductor agent.
 * Never systemctl stop herdr-fleet.
 *
 * Pin is written even when the pane stop throws — recovery stays off so a
 * failed kill cannot be undone by herdr-conductor respawning the agent. The
 * command still fails (throws) so operators never see a green "halted" while
 * the pane is still alive.
 */
export async function haltWithPane(
  projectName?: string,
  deps: PaneStopDeps = {},
): Promise<HaltWithPaneResult> {
  const pin = pinPaneHalt(projectName);
  const result = await halt(projectName);
  // Throws on uncertainty or if the agent survives SIGTERM+SIGKILL.
  const stopped = await stopConductorPane(projectName, deps);
  return { ...result, pane: { pinPath: pin.path, ...stopped } };
}

/**
 * Stop the configured conductor agent only. Recovery must already be pinned.
 *
 * Success returns only when the pane is confirmed gone (`herdr-agent` or
 * `already-gone`). Any uncertainty **throws**: an unparseable tick config (the
 * agent name is then a guess), herdr unreachable, output we cannot read as an
 * explicit agent list, ambiguous identity, a live omp claim with no recognized
 * PID, a liveness probe or signal delivery that fails (`EPERM` is "exists but
 * not ours", never "dead"), or a process that survives SIGKILL. Callers must not treat a thrown
 * error as "maybe stopped".
 */
export async function stopConductorPane(
  projectName?: string,
  deps: PaneStopDeps = {},
): Promise<Omit<PaneStopResult, "pinPath">> {
  const tick = resolveTickConfig(projectName);
  if (tick.kind === "invalid") {
    // The file that names the agent does not parse, so the identity we would
    // stop is a guess. Stopping the wrong pane is worse than refusing.
    throw new Error(
      `halt --pane: tick config invalid at ${tick.path} (${tick.problem}) — ` +
        `the conductor agent name cannot be read, so refusing to stop a guessed identity. ` +
        `Recovery pin was written; fix the config and re-run.`,
    );
  }
  const agentName =
    tick.kind === "ok" && tick.config.agentName !== undefined
      ? tick.config.agentName
      : DEFAULT_FLEET_AGENT_NAME;
  const sleep = deps.sleep ?? ((ms: number) => new Promise<void>((r) => setTimeout(r, ms)));
  const now = deps.now ?? Date.now;
  const signalPid = deps.signalPid ?? deliverSignal;
  const alive = deps.isAlive ?? pidLiveness;
  const pinned = "Recovery pin was written";

  /** Any-alive across `pids`; an unreadable probe refuses instead of "dead". */
  const anyAlive = (pids: number[], stage: string): boolean => {
    try {
      return pids.some((pid) => alive(pid));
    } catch (err) {
      throw new Error(
        `halt --pane: cannot tell whether the conductor agent is still running ${stage} ` +
          `(${err instanceof Error ? err.message : String(err)}) — refusing to report success. ` +
          `${pinned}; check the pane by hand.`,
      );
    }
  };

  /** Send `sig` to every pid, refusing on any delivery we cannot prove. */
  const deliver = (pids: number[], sig: NodeJS.Signals): void => {
    for (const pid of pids) {
      let delivered: boolean;
      try {
        delivered = signalPid(pid, sig);
      } catch (err) {
        throw new Error(
          `halt --pane: ${sig} to pid ${pid} failed ` +
            `(${err instanceof Error ? err.message : String(err)}) — refusing to report success. ${pinned}.`,
        );
      }
      if (!delivered) {
        throw new Error(
          `halt --pane: ${sig} to pid ${pid} was not delivered — refusing to report success. ${pinned}.`,
        );
      }
    }
  };

  let agents: HerdrAgent[];
  try {
    agents = deps.listAgents ? await deps.listAgents() : await herdrAgentList(deps);
  } catch (err) {
    throw new Error(
      `halt --pane: herdr agent list failed — refusing to guess a pane ` +
        `(${err instanceof Error ? err.message : String(err)}). ` +
        `Recovery pin was written; fix herdr and re-run halt --pane or kill the agent by hand.`,
    );
  }

  const matches = agents.filter((a) => a.name === agentName);
  if (matches.length === 0) {
    return { stopped: "already-gone", detail: `no herdr agent named ${agentName}`, agentName };
  }
  if (matches.length > 1) {
    throw new Error(
      `halt --pane: refusing to stop — ${matches.length} herdr agents named ${agentName} (not unique). ` +
        `Recovery pin was written; resolve the identity and re-run or kill by hand.`,
    );
  }
  const claim = matches[0]!;
  if (claim.agent !== "omp") {
    if (claim.agent === undefined || claim.agent === "") {
      // Sticky name after exit — herdr still lists the claim but no live agent.
      return {
        stopped: "already-gone",
        detail: `agent ${agentName} pane ${claim.paneId} has no live agent label (sticky name only)`,
        agentName,
      };
    }
    throw new Error(
      `halt --pane: agent ${agentName} pane ${claim.paneId} is ${claim.agent}, not omp — refusing. ` +
        `Recovery pin was written.`,
    );
  }

  let pids: number[];
  try {
    pids = deps.panePids
      ? await deps.panePids(claim.paneId)
      : await herdrOmpForegroundPids(claim.paneId, deps);
  } catch (err) {
    throw new Error(
      `halt --pane: pane process-info failed for ${claim.paneId} ` +
        `(${err instanceof Error ? err.message : String(err)}). ` +
        `Live omp claim exists but PIDs are unknown — refusing to report success. Recovery pin was written.`,
    );
  }

  const live = pids.filter((pid) => anyAlive([pid], `for pane ${claim.paneId}`));
  if (live.length === 0) {
    // Live agent=omp claim but no recognizable omp PID: we cannot prove gone.
    // Treating this as already-gone would exit 0 while the pane may still run.
    throw new Error(
      `halt --pane: agent ${agentName} pane ${claim.paneId} is live omp but no omp foreground PID ` +
        `was recognized — refusing to report success. Recovery pin was written; inspect the pane and kill by hand if needed.`,
    );
  }

  deliver(live, "SIGTERM");
  const softDeadline = now() + 10_000;
  while (now() < softDeadline) {
    if (!anyAlive(live, "after SIGTERM")) {
      return {
        stopped: "herdr-agent",
        detail: `SIGTERM omp in agent ${agentName} pane ${claim.paneId} pids ${live.join(",")}`,
        agentName,
      };
    }
    await sleep(100);
  }

  deliver(live, "SIGKILL");
  const hardDeadline = now() + 2_000;
  while (now() < hardDeadline) {
    if (!anyAlive(live, "after SIGKILL")) {
      return {
        stopped: "herdr-agent",
        detail: `SIGKILL omp in agent ${agentName} pane ${claim.paneId} pids ${live.join(",")} after SIGTERM grace`,
        agentName,
      };
    }
    await sleep(50);
  }

  const survivors = live.filter((pid) => anyAlive([pid], "after SIGKILL"));
  throw new Error(
    `halt --pane: agent ${agentName} pane ${claim.paneId} still alive after SIGTERM+SIGKILL ` +
      `(pids ${survivors.join(",")}). Recovery pin was written; kill by hand before trusting a quiet fleet.`,
  );
}

async function herdrAgentList(deps: PaneStopDeps): Promise<HerdrAgent[]> {
  const bin = deps.herdrBin ?? "herdr";
  const session = deps.herdrSession ?? resolveHerdrSession(process.env);
  const res = spawnSync(bin, ["--session", session, "agent", "list"], {
    encoding: "utf8",
    timeout: 8_000,
    env: process.env,
  });
  if (res.error) throw res.error;
  if (res.status !== 0) {
    throw new Error((res.stderr ?? res.stdout ?? `herdr exit ${String(res.status)}`).trim());
  }
  return parseHerdrAgentList(res.stdout ?? "");
}

export function parseHerdrAgentList(rawOutput: string): HerdrAgent[] {
  // Only an explicit `agents: []` means no agents. Every malformed answer is
  // uncertainty, never permission to report the conductor pane missing.
  const raw = rawOutput.trim();
  if (raw.length === 0) {
    throw new Error("herdr agent list printed nothing — cannot tell whether the pane is running");
  }
  let parsed: { result?: { agents?: unknown }; agents?: unknown };
  try {
    parsed = JSON.parse(raw) as { result?: { agents?: unknown }; agents?: unknown };
  } catch (err) {
    throw new Error(
      `herdr agent list output is not JSON (${err instanceof Error ? err.message : String(err)})`,
    );
  }
  const list = parsed.result?.agents ?? parsed.agents;
  if (!Array.isArray(list)) {
    throw new Error("herdr agent list output has no `agents` array — unrecognized schema");
  }
  const out: HerdrAgent[] = [];
  for (const row of list) {
    if (row === null || typeof row !== "object") {
      throw new Error("herdr agent list contains a non-object agent row — unrecognized schema");
    }
    const agent = row as { readonly [key: string]: unknown };
    const name = agent["name"];
    const paneId = agent["pane_id"];
    if (typeof paneId !== "string") {
      throw new Error(
        "herdr agent list row is missing a string `pane_id` — cannot identify the pane",
      );
    }
    if (name !== undefined && name !== null && typeof name !== "string") {
      throw new Error(
        `herdr agent list row for pane ${paneId} has a non-string \`name\` (${typeof name}) — ` +
          "cannot tell whether it is the conductor pane",
      );
    }
    const parsedName = typeof name === "string" ? name : undefined;
    const rawAgent = agent["agent"];
    if (rawAgent !== undefined && rawAgent !== null && typeof rawAgent !== "string") {
      throw new Error(
        `herdr agent list row for ${parsedName ?? `pane ${paneId}`} has a non-string \`agent\` (${typeof rawAgent}) — ` +
          `cannot tell whether an agent is live`,
      );
    }
    const liveAgent = typeof rawAgent === "string" ? rawAgent : undefined;
    let sessionPath: string | undefined;
    const session = agent["agent_session"];
    if (session !== null && typeof session === "object") {
      const value = session as { readonly [key: string]: unknown };
      if (value["source"] === "herdr:omp" && typeof value["value"] === "string") {
        sessionPath = value["value"];
      }
    }
    out.push({
      ...(parsedName === undefined ? {} : { name: parsedName }),
      paneId,
      ...(liveAgent === undefined ? {} : { agent: liveAgent }),
      ...(sessionPath === undefined ? {} : { sessionPath }),
    });
  }
  return out;
}

/**
 * The `process_info` document out of `herdr pane process-info` output
 * (#832): every caller that maps the fleet pane's live omp processes starts
 * from this schema, whether the output carries herdr's CLI `result` envelope
 * or the bare document. Throws when the output cannot be read; the caller
 * decides what an unreadable answer means.
 */
export function parseHerdrProcessInfo(stdout: string, paneId: string): ProcessInfo {
  const raw = stdout.trim();
  if (raw.length === 0) {
    throw new Error(`herdr pane process-info printed nothing for ${paneId}`);
  }
  let parsed: { result?: { process_info?: ProcessInfo }; process_info?: ProcessInfo };
  try {
    parsed = JSON.parse(raw) as { result?: { process_info?: ProcessInfo }; process_info?: ProcessInfo };
  } catch (err) {
    throw new Error(
      `herdr pane process-info output is not JSON (${err instanceof Error ? err.message : String(err)})`,
    );
  }
  const info = parsed.result?.process_info ?? parsed.process_info;
  if (info === undefined) {
    throw new Error(`herdr pane process-info has no process_info for ${paneId} — unrecognized schema`);
  }
  return info;
}

async function herdrOmpForegroundPids(paneId: string, deps: PaneStopDeps): Promise<number[]> {
  const bin = deps.herdrBin ?? "herdr";
  const session = deps.herdrSession ?? resolveHerdrSession(process.env);
  const res = spawnSync(bin, ["--session", session, "pane", "process-info", "--pane", paneId], {
    encoding: "utf8",
    timeout: 8_000,
    env: process.env,
  });
  if (res.error) throw res.error;
  if (res.status !== 0) {
    throw new Error((res.stderr ?? res.stdout ?? `process-info exit ${String(res.status)}`).trim());
  }
  return ompPidsFromProcessInfo(parseHerdrProcessInfo(res.stdout ?? "", paneId));
}

export interface ProcessInfo {
  shell_pid?: number;
  foreground_processes?: ForegroundProc[];
}

interface ForegroundProc {
  pid?: number;
  name?: string;
  argv0?: string;
  argv?: string[];
}

export function ompPidsFromProcessInfo(info: ProcessInfo): number[] {
  const shell = typeof info.shell_pid === "number" ? info.shell_pid : undefined;
  const out: number[] = [];
  for (const proc of info.foreground_processes ?? []) {
    if (typeof proc.pid !== "number" || !Number.isInteger(proc.pid) || proc.pid <= 1) continue;
    if (shell !== undefined && proc.pid === shell) continue;
    if (!isOmpProcess(proc)) continue;
    out.push(proc.pid);
  }
  return out;
}

function isOmpProcess(proc: ForegroundProc): boolean {
  const name = (proc.name ?? "").toLowerCase();
  const argv0 = (proc.argv0 ?? "").toLowerCase();
  const argv = (proc.argv ?? []).map((a) => a.toLowerCase());
  if (name === "omp" || argv0 === "omp" || argv0.endsWith("/omp")) return true;
  if (name === "bun" || argv0 === "bun" || argv0.endsWith("/bun")) {
    if (argv.some((a) => a === "omp" || a.endsWith("/omp") || /(^|\/)omp$/.test(a))) return true;
  }
  return false;
}

/**
 * The omp processes the fleet pane claims, resolved to their start times
 * (#832), or why they could not be read. Structurally identical to
 * upgrade-verify's `PaneOmpProbe` — the upgrade engine and the post-restart
 * verifier pass this straight into the pure verdicts, and the type is
 * repeated here because the verifier leaf must not import this module (fleet
 * reaches the daemon, which reaches the verifier).
 */
export type PaneProbeResult = { starts: readonly number[] } | { problem: string };

/**
 * Every omp process the fleet pane currently claims by herdr, resolved to
 * its start time (#832): the pane is where an *external* orchestrator
 * session lives, and the only fact that proves it reloaded is the live
 * process's own start — a pane process still running from before the install
 * began loaded the pre-upgrade extension. Every unreadable answer is a
 * `problem`: absence of evidence is never a reload.
 *
 * `run` is the injected command runner (the upgrade's scripted seam, the
 * daemon's `runCommand`), and `startTime` resolves one pid to its start —
 * the /proc read lives with the verifier's other live-process facts.
 */
export async function herdrPaneOmpStarts(
  run: (
    command: string,
    args: readonly string[],
  ) => Promise<{ code: number; stdout: string; stderr: string }>,
  session: string,
  startTime: (pid: number) => number | undefined,
): Promise<PaneProbeResult> {
  const agents = await run("herdr", ["--session", session, "agent", "list"]);
  if (agents.code !== 0) {
    return {
      problem:
        `herdr agent list failed: ${agents.stderr.trim() || agents.stdout.trim() || `exit ${agents.code}`}`,
    };
  }
  let parsed: HerdrAgent[];
  try {
    parsed = parseHerdrAgentList(agents.stdout);
  } catch (err) {
    return { problem: err instanceof Error ? err.message : String(err) };
  }
  const starts: number[] = [];
  for (const agent of parsed) {
    // A name without a live omp claim is a leftover label, not a pane process.
    if (agent.agent === undefined) continue;
    const info = await run("herdr", [
      "--session",
      session,
      "pane",
      "process-info",
      "--pane",
      agent.paneId,
    ]);
    if (info.code !== 0) {
      return {
        problem:
          `herdr pane process-info for ${agent.paneId} failed: ` +
          `${info.stderr.trim() || info.stdout.trim() || `exit ${info.code}`}`,
      };
    }
    let pids: number[];
    try {
      pids = ompPidsFromProcessInfo(parseHerdrProcessInfo(info.stdout, agent.paneId));
    } catch (err) {
      return { problem: err instanceof Error ? err.message : String(err) };
    }
    for (const pid of pids) {
      const startedAt = startTime(pid);
      if (startedAt === undefined) {
        return { problem: `cannot read the start time of pane process ${pid} — reload unproven` };
      }
      starts.push(startedAt);
    }
  }
  return { starts };
}

/**
 * The project a bare read means, when the config leaves no doubt.
 *
 * The lifecycle verbs resolve their target before writing, so `hold` on a
 * single-project host writes `paused-<name>`. A status read that passed
 * `undefined` straight through saw only the legacy global sentinel and reported
 * `dispatch running` for a fleet the operator had just held — control and status
 * disagreeing about the same fleet, which is the one thing this surface exists to
 * settle. A host with several projects and no name keeps the legacy-only read:
 * guessing which project a bare `status` meant is what {@link findProject}
 * refuses to do.
 */
function bareReadProject(): string | undefined {
  try {
    return findProject(loadConfig()).name;
  } catch {
    return undefined;
  }
}

export function fleetLayers(projectName?: string): FleetLayers {
  const rec = livingDaemon();
  // systemd is the authoritative liveness witness for a unit-owned daemon: a
  // pidfile that is missing, stale, or skewed against the unit's MainPID must
  // not declare the daemon dead (#716). Consulted only when the pidfile says
  // dead — a healthy record needs no shell-out, and a host without systemctl
  // answers `inactive`, so those hosts behave exactly as before. `failed`,
  // `unknown`, and a mid-restart unit (MainPID 0, process gone) all answer
  // undefined: a genuinely stopped daemon never reports running.
  let unitPid: number | undefined;
  if (rec === undefined) {
    const ownership = probeUnit();
    if (ownership.kind === "active" && isAlive(ownership.pid)) unitPid = ownership.pid;
  }
  const daemon: FleetLayers["daemon"] =
    rec !== undefined
      ? { running: true, pid: rec.pid, port: rec.port }
      : unitPid !== undefined
        ? { running: true, pid: unitPid }
        : { running: false };
  const paused = isPaused(projectName ?? bareReadProject());
  const dispatch: DispatchLayer = daemon.running ? (paused ? "paused" : "running") : "stopped";

  const tick = resolveTickConfig(projectName);
  let ticks: TicksLayer;
  let ticksDetail: string | undefined;
  let nextTickAt: string | undefined;
  let armedPath: string | undefined;
  let tickConfigPath: string | undefined;
  if (tick.kind === "absent") {
    ticks = "no-heartbeat-config";
  } else if (tick.kind === "invalid") {
    ticks = "invalid-heartbeat-config";
    ticksDetail = `${tick.path}: ${tick.problem}`;
    tickConfigPath = tick.path;
  } else {
    tickConfigPath = tick.path;
    armedPath = tick.config.armedFile;
    if (tick.config.armedFile === undefined) {
      ticks = "ungated";
      ticksDetail = "no armedFile in tick config — heartbeat sends without an arm marker";
    } else {
      // The same decision the heartbeat itself makes, so status can never claim
      // armed while the tick skips (or the reverse) — including the two shared
      // marker cases: honoured through an upgrade on a single-project host,
      // stranded once a second project exists.
      const arm = resolveArmState(tick.config.armedFile, tick.config.project ?? projectName);
      ticks = arm.armed ? "armed" : "disarmed";
      ticksDetail =
        arm.legacy === undefined
          ? tick.config.armedFile
          : `${tick.config.armedFile} — ${legacyArmedMarkerPath()}: ${
              arm.legacy === "honoured"
                ? "legacy global arm marker still honoured; re-arm this project to replace it"
                : LEGACY_ARM_MARKER_DETAIL
            }`;
    }
    const runtime = readTickRuntimeStatus(tick.cwd);
    if (runtime !== undefined && isAlive(runtime.pid)) nextTickAt = runtime.nextTickAt;
  }
  if (armedPath === undefined) armedPath = armedMarkerPath(projectName);

  // Status reports; it never refuses. An unresolvable pin location is its own
  // answer — "clear" would claim recovery is armed and ready to be pinned.
  const resolvedHalt = resolvePaneHaltPath(projectName);
  const haltPath = resolvedHalt.kind === "ok" ? resolvedHalt.path : undefined;
  const recoveryPinned = haltPath !== undefined && existsSync(haltPath);
  const omp = probeOmpPane(projectName);
  let pane: PaneLayer;
  let paneDetail: string | undefined;
  if (omp.kind === "live") {
    pane = "live";
    paneDetail = omp.summary;
  } else if (omp.kind === "missing") {
    pane = "missing";
  } else {
    pane = "unknown";
    paneDetail = omp.reason;
  }

  const herdr = probeHerdrUnit();

  return {
    dispatch,
    ticks,
    ...(ticksDetail === undefined ? {} : { ticksDetail }),
    ...(nextTickAt === undefined ? {} : { nextTickAt }),
    pane,
    ...(paneDetail === undefined ? {} : { paneDetail }),
    recovery: haltPath === undefined ? "unpinnable" : recoveryPinned ? "pinned" : "clear",
    ...(haltPath === undefined
      ? { recoveryDetail: resolvedHalt.kind === "unresolved" ? resolvedHalt.reason : undefined }
      : recoveryPinned
        ? { recoveryDetail: haltPath }
        : {}),
    herdr: herdr.kind,
    ...(herdr.detail === undefined ? {} : { herdrDetail: herdr.detail }),
    armedPath,
    ...(tickConfigPath === undefined ? {} : { tickConfigPath }),
    ...(haltPath === undefined ? {} : { paneHaltPath: haltPath }),
    paused,
    daemon,
  };
}

export function codeGraphFromHealthz(body: string | undefined, project: string): CodeGraphHealth | undefined {
  if (body === undefined) return undefined;
  try {
    const parsed = JSON.parse(body) as Record<string, unknown>;
    let block: Record<string, unknown> | undefined;
    if (parsed["project"] === project) {
      block = parsed;
    } else {
      const projects = parsed["projects"];
      if (Array.isArray(projects)) {
        for (const entry of projects) {
          if (
            entry !== null &&
            typeof entry === "object" &&
            !Array.isArray(entry) &&
            (entry as Record<string, unknown>)["project"] === project
          ) {
            block = entry as Record<string, unknown>;
            break;
          }
        }
      }
    }
    if (block === undefined) return undefined;
    const graph = block["codeGraph"];
    if (graph === null || typeof graph !== "object" || Array.isArray(graph)) return undefined;
    const value = graph as Record<string, unknown>;
    if (value["configured"] === false) return { configured: false };
    if (
      value["configured"] !== true ||
      !["healthy", "degraded", "unknown"].includes(String(value["status"])) ||
      typeof value["checkedAt"] !== "string" ||
      value["prerequisites"] === null ||
      typeof value["prerequisites"] !== "object" ||
      !Array.isArray(value["repos"]) ||
      value["timer"] === null ||
      typeof value["timer"] !== "object" ||
      value["refresh"] === null ||
      typeof value["refresh"] !== "object" ||
      !Array.isArray(value["reasons"])
    ) {
      return undefined;
    }
    return graph as CodeGraphHealth;
  } catch {
    return undefined;
  }
}


/** One live pause as `/healthz` reports it: the phase, plus who asked and
 *  when where the daemon recorded provenance (#997). */
export interface WorkerPauseView {
  phase: WorkerPausePhase;
  source?: string;
  pausedAtMs?: number;
}

/** Live worker pause phases from a daemon `/healthz` body for one project. */
export function workerPhasesFromHealthz(
  body: string | undefined,
  project: string,
): ReadonlyMap<number, WorkerPauseView> {
  const phases = new Map<number, WorkerPauseView>();
  if (body === undefined) return phases;
  try {
    const payload = JSON.parse(body) as unknown;
    if (payload === null || typeof payload !== "object") return phases;
    let block: object | undefined;
    if (Reflect.get(payload, "project") === project) {
      block = payload;
    } else {
      const projects = Reflect.get(payload, "projects");
      if (Array.isArray(projects)) {
        for (const entry of projects) {
          if (
            entry !== null &&
            typeof entry === "object" &&
            Reflect.get(entry, "project") === project
          ) {
            block = entry;
            break;
          }
        }
      }
    }
    if (block === undefined) return phases;
    const workers = Reflect.get(block, "workers");
    if (!Array.isArray(workers)) return phases;
    for (const worker of workers) {
      if (worker === null || typeof worker !== "object") continue;
      const issue = Reflect.get(worker, "issue");
      const phase = Reflect.get(worker, "phase");
      if (
        Number.isSafeInteger(issue) &&
        (issue as number) > 0 &&
        (phase === "pausing" || phase === "paused")
      ) {
        const source = Reflect.get(worker, "source");
        const pausedAtMs = Reflect.get(worker, "pausedAtMs");
        phases.set(issue as number, {
          phase,
          ...(typeof source === "string" && source !== "" ? { source } : {}),
          ...(Number.isSafeInteger(pausedAtMs) && (pausedAtMs as number) > 0
            ? { pausedAtMs: pausedAtMs as number }
            : {}),
        });
      }
    }
  } catch {
    // An unreadable health body means no trustworthy pause phase.
  }
  return phases;
}


/**
 * Decide whether a living record + optional `/healthz` answer serve `project`.
 *
 * Mirrors the board gate: a record pinned to another project is never probed
 * further; a multi-project payload is ok only when `project` appears in
 * `projects[]` (or as the top-level single-project `project` field).
 */
export function classifyDaemonProjectHealth(
  record: { project?: string } | undefined,
  health: HealthCheckResult | undefined,
  project: string,
): DaemonProjectHealth {
  if (record === undefined) return { kind: "stopped" };
  // Record project is authoritative when set — a single-project leftover from
  // #376 must not be presented as this project's daemon just because its port
  // still answers.
  if (record.project !== undefined && record.project !== project) {
    return { kind: "other-project", serves: record.project };
  }
  // #685: a timed-out probe is a probe outcome, never a death verdict — the
  // pid is up but wedged, so it must not read as `not running`. Only a
  // refusal or other failure (nothing listening, a torn answer) stays
  // `unreachable`.
  if (health?.ok !== true) {
    return health?.failure === "timeout" ? { kind: "unresponsive" } : { kind: "unreachable" };
  }
  try {
    const payload = JSON.parse(health.body ?? "null") as unknown;
    if (payload === null || typeof payload !== "object") {
      return { kind: "unreachable" };
    }
    if (Reflect.get(payload, "project") === project) return { kind: "ok" };
    const projects = Reflect.get(payload, "projects");
    if (
      Array.isArray(projects) &&
      projects.some(
        (entry) =>
          entry !== null &&
          typeof entry === "object" &&
          Reflect.get(entry, "project") === project,
      )
    ) {
      return { kind: "ok" };
    }
    return { kind: "other-project", serves: servedProjectName(payload) };
  } catch {
    return { kind: "unreachable" };
  }
}

/** Best-effort name of who a foreign `/healthz` payload is actually serving. */
function servedProjectName(payload: object): string | undefined {
  const top = Reflect.get(payload, "project");
  if (typeof top === "string" && top.length > 0) return top;
  const projects = Reflect.get(payload, "projects");
  if (!Array.isArray(projects)) return undefined;
  for (const entry of projects) {
    if (entry === null || typeof entry !== "object") continue;
    const name = Reflect.get(entry, "project");
    if (typeof name === "string" && name.length > 0) return name;
  }
  return undefined;
}

export type FleetStatusReport = StatusSnapshot & {
  observedAt: number;
  layers: FleetLayers;
  daemon: FleetDaemonProbe | undefined;
  telegram: TelegramHealth;
  codeGraph: CodeGraphHealth;
  brief: string | undefined;
  decisions: string | undefined;
  failureClasses: string | undefined;
  workerPhases: { issue: number; view: WorkerPauseView }[];
  intake: string | undefined;
  /** The project-scoped durable to-spec grooming lifecycle as status lines
   *  (#809), or nothing when there is nothing to report. */
  grooming: string | undefined;
  /** The sends this project's stale pin already delivered to the flat chat
   *  (#1094), or undefined when the pin is healthy or freshly repaired. */
  topicMisroute: string | undefined;
  lastStop: DaemonStop | undefined;
  siblings: { project: string; live: number }[];
};

/** Collects the complete status payload once for both text and JSON renderers. */
export async function collectFleetStatus(projectName?: string): Promise<FleetStatusReport> {
  const snapshot = statusSnapshot(projectName);
  const layers = fleetLayers(projectName);
  const project = findProject(loadConfig(), projectName);
  const rec = livingDaemon();
  // Gate the port probe on record membership first — a leftover single-project
  // daemon from #376 must not be health-checked as if it served us (#379).
  const wrongRecord = rec?.project !== undefined && rec.project !== project.name;
  const [rawHealth, telegram, planUsage, github] = await Promise.all([
    rec === undefined || wrongRecord ? undefined : healthCheck(rec.port),
    probeTelegramHealth(projectName),
    // Read here rather than in `statusSnapshot`, which is synchronous and used
    // by callers that must not shell out. An unmetered project never spawns
    // the provider at all.
    readPlanUsage(snapshot.caps.planUsage, sharedUsageSource()),
    // Same reasoning as `planUsage`: the snapshot is synchronous, this read is
    // a shell-out, and undefined on any failure — one missing row, never a
    // broken report (#188).
    fetchRateLimit(),
  ]);
  const projectHealth = classifyDaemonProjectHealth(rec, rawHealth, project.name);
  // Ownership is probed, never assumed from the unit name constant (#379).
  const unit = rec === undefined ? undefined : probeUnit();
  const daemon: FleetDaemonProbe | undefined =
    rec === undefined
      ? undefined
      : {
          project: projectHealth,
          ...(projectHealth.kind === "ok" && rawHealth?.body !== undefined
            ? { body: rawHealth.body }
            : {}),
          ...(unit === undefined ? {} : { unit }),
        };
  const healthBody = projectHealth.kind === "ok" ? rawHealth?.body : undefined;
  const cached = codeGraphFromHealthz(healthBody, project.name);
  // The runtime half of the code-graph finding (#726): the probe reads the
  // store-backed per-run observations, so "observed" is always grounded in
  // dispatched runs, never in the daemon's own process or in mcp.json. The
  // store stays scoped to the same try/finally as the other reads so a
  // throwing probe cannot leak its handle.
  const store = openStore(dbPath());
  const workerPhases = [...workerPhasesFromHealthz(healthBody, project.name)].map(
    ([issue, view]) => ({ issue, view }),
  );
  // The newest host-wide stop/restart provenance (#378). Read here — not in
  // `statusSnapshot`, which is synchronous and belongs to the daemon module —
  // and rendered identically from either project: the daemon_stops table is
  // deliberately not partitioned by project, because the daemon serves every
  // project and the uninvolved one must see who stopped it too.
  let codeGraph: CodeGraphHealth;
  let lastStop: DaemonStop | undefined;
  let siblings: { project: string; live: number }[] = [];
  let grooming: string | undefined;
  try {
    codeGraph =
      cached ??
      (await probeCodeGraph(project, {
        ...DEFAULT_DEPS,
        graphToolsObservations: () => store.graphToolsObservationCounts(project.name),
      }));
    lastStop = store.latestDaemonStop();
    // Shared-daemon visibility (#545): every configured project other than the
    // one being viewed, with its live-run count.
    siblings = loadConfig()
      .projects.filter((p) => p.name !== project.name)
      .map((p) => ({ project: p.name, live: store.liveRuns(p.name).length }));
    // The durable to-spec grooming lifecycle (#809): project-scoped verdict
    // rows from the store, plus the dispatch snapshot's routed/parked counts —
    // no tracker read, no second cache, no evidence parsing. Rendered only
    // when it has something to say.
    grooming = formatGroomingStatus({
      records: store.groomingVerdicts(project.name),
      routed: snapshot.dispatch?.routed ?? 0,
      parked: snapshot.dispatch?.parked ?? 0,
    });
  } finally {
    store.close();
  }
  return {
    ...snapshot,
    planUsage,
    github,
    observedAt: Date.now(),
    layers,
    daemon,
    telegram,
    codeGraph,
    brief: briefStatusLine(project),
    decisions: decisionStatusLine(project.name, layers),
    failureClasses: failureClassBlock(project.name),
    workerPhases,
    intake: intakeStatusLine(project.name),
    grooming,
    topicMisroute: telegramMisrouteStatusLine(project),
    lastStop,
    siblings,
  };
}

export function renderFleetStatusReport(report: FleetStatusReport): string {
  return formatFleetStatus(
    report,
    report.layers,
    report.daemon,
    report.telegram,
    report.observedAt,
    report.codeGraph,
    report.brief,
    report.decisions,
    report.failureClasses,
    new Map(report.workerPhases.map(({ issue, view }) => [issue, view])),
    report.intake,
    report.lastStop,
    report.siblings,
    report.grooming,
    report.topicMisroute,
  );
}

export async function renderStatus(projectName?: string): Promise<string> {
  return renderFleetStatusReport(await collectFleetStatus(projectName));
}

/**
 * One status line naming the sends this project's stale pin already delivered
 * to the flat chat (#1094) — the send-side evidence behind doctor's topic-pin
 * warning. Rows match the *current* pin id, so re-pinning retires the row
 * without anyone deleting history.
 */
function telegramMisrouteStatusLine(project: ProjectConfig): string | undefined {
  const pinned = project.escalation.telegramTopicId;
  if (pinned === undefined) return undefined;
  const rows = readTelegramMisroutes().filter(
    (row) => row.project === project.name && row.staleTopicId === pinned,
  );
  if (rows.length === 0) return undefined;
  const last = new Date(rows[rows.length - 1]!.at).toISOString();
  return (
    `misroute  ${rows.length} send(s) went to the flat chat because pinned topic ${pinned} was stale — ` +
    `last ${last}; re-pin escalation.telegramTopicId to clear`
  );
}

/**
 * One line naming the brief layout, or nothing when it cannot be read.
 *
 * `brief-upgrade` stays the verb that migrates a legacy layout — this is the
 * surface that tells an operator they still have one, in the place they already
 * look (#131). Silent on any throw: a status that cannot render because the
 * brief moved is worse than a status missing one row.
 */
function briefStatusLine(project: ProjectConfig): string | undefined {
  try {
    const layout = inspectBriefLayout(project.workspaceRoot, renderBriefForProject(project));
    return layout.kind === "overlay"
      ? "brief     overlay (package floor + POLICY.md)"
      : `brief     ${layout.kind} — run omp-conductor brief-upgrade`;
  } catch {
    return undefined;
  }
}

/**
 * Failure classes, in two blocks that mean different things (#132).
 *
 * Reported as classes rather than as row states because a row state is not an
 * issue state: the FAILED column counted four completed issues on this fleet
 * while the genuinely stuck ones were invisible (#109).
 *
 * That defect then reappeared one level up. A single "unrecovered" list keyed on
 * `recoveredAt IS NULL` swept in the two recorded-only actions, which never
 * stamp it — so this fleet's status read `returned-for-revision 44`, growing by
 * one on every review return, while every class an operator could actually act
 * on sat at zero and invisible beneath it. A count nobody can act on, printed
 * where the actionable ones go, is the same lie in a new column.
 *
 * So: `awaiting recovery` is what the sweep will pick up, and it is omitted when
 * empty because empty is the healthy answer. `classified, no action` is the
 * signal — review returns and holds — named as such, never as a backlog.
 */
/** How many surviving `unknown` rows a status render explains (#986). A board
 *  is read at a glance: enough to show a pattern, never a backlog dump. */
const UNKNOWN_EVIDENCE_SHOWN = 5;

function failureClassBlock(projectName: string): string | undefined {
  const path = dbPath();
  if (!existsSync(path)) return undefined;
  let store: Store | undefined;
  try {
    store = openStore(path);
    const awaiting = store.failureClassCounts(projectName);
    const recorded = store.recordedOnlyClassCounts(projectName);
    if (awaiting.length === 0 && recorded.length === 0) return undefined;
    const lines: string[] = [];
    if (awaiting.length > 0) {
      lines.push(
        "failure classes (awaiting recovery)",
        ...awaiting.map((c) => `  ${c.cls.padEnd(24)}${c.n}`),
      );
    }
    if (recorded.length > 0) {
      lines.push(
        "failure classes (classified, no action by design)",
        ...recorded.map((c) => `  ${c.cls.padEnd(24)}${c.n}`),
      );
    }
    // A surviving `unknown` is an unnamed cause, and the only way anyone closes
    // that gap is by seeing what the settle sweep actually knew about the row
    // (#986). Bounded to the newest few: this is a status board, not a log, and
    // a fleet with a long unknown history must not turn one render into a scan.
    const unknown = store.unknownRuns(projectName, UNKNOWN_EVIDENCE_SHOWN);
    const withEvidence = unknown.filter((run) => run.terminalEvidence !== undefined);
    if (withEvidence.length > 0) {
      lines.push(
        "unknown, with terminal evidence",
        ...withEvidence.map(
          (run) => `  #${run.issue} ${(run.terminalEvidence ?? "").slice(0, 90)}`,
        ),
      );
    }
    return lines.join("\n");
  } catch {
    return undefined;
  } finally {
    store?.close();
  }
}

/**
 * One line naming what the orchestrator still owes its operator (#136).
 *
 * Reported whenever the store exists, `none open` included: a row that appears
 * only when something is outstanding is a row whose absence means nothing, and
 * "did it forget to record the question, or is there genuinely none?" is exactly
 * the ambiguity this ledger exists to remove. Omitted on any throw — a status
 * that fails to render because the store is mid-migration is worse than one
 * missing a row.
 *
 * `decisions N open` counts only rows a human must answer (#459). Watches the
 * orchestrator set for itself are reported separately — a fleet with three
 * watches behind GitHub's checks is not a fleet waiting on its operator, and
 * lumping them in made `decisions 3 open` read as three unanswered questions.
 */
export function decisionStatusLine(
  projectName: string,
  layers: Pick<FleetLayers, "daemon" | "ticks">,
): string | undefined {
  const path = dbPath();
  if (!existsSync(path)) return undefined;
  let store: Store | undefined;
  try {
    store = openStore(path);
    const open = store.openDecisions(projectName);
    const questions = open.filter((d) => d.kind !== "watch");
    const watches = open.filter((d) => d.kind === "watch");
    if (questions.length === 0 && watches.length === 0) return "decisions none open";
    let line: string;
    if (questions.length === 0) {
      line = "decisions none open";
    } else {
      const oldest = questions[0]!;
      const hours = Math.max(0, Math.round((Date.now() - oldest.askedAt) / 3_600_000));
      line = `decisions ${questions.length} open (oldest ${hours}h)`;
    }
    if (watches.length > 0) {
      const monitoring = watchMonitoringState(layers);
      line += monitoring.monitored
        ? ` · watches ${watches.length} monitored`
        : ` · watches ${watches.length} durable, unmonitored (${monitoring.reason})`;
    }
    return line;
  } catch {
    return undefined;
  } finally {
    store?.close();
  }
}

/**
 * One line naming how many raw ideas are still pending (#299), or nothing when
 * there are none.
 *
 * Deliberately absent at zero, unlike the decision line: an empty intake is the
 * common state of a healthy fleet, and a row that prints "0 pending" every
 * status would be permanent noise. Only a nonzero backlog is news, exactly like
 * {@link failureClassBlock}. Omitted on any throw, like the other optional rows.
 * Exported so the conditional is testable against a real store.
 */
export function intakeStatusLine(projectName: string): string | undefined {
  const path = dbPath();
  if (!existsSync(path)) return undefined;
  let store: Store | undefined;
  try {
    store = openStore(path);
    const pending = store.pendingIntake(projectName);
    if (pending.length === 0) return undefined;
    return `intake    ${pending.length} pending idea(s) — omp-conductor intake list`;
  } catch {
    return undefined;
  } finally {
    store?.close();
  }
}

// ---------------------------------------------------------------------------
// arm proof helpers
// ---------------------------------------------------------------------------

type Channel = { kind: "up"; owner: string } | { kind: "down"; reason: string };

function readPairedChannel(path: string): Channel {
  let parsed: unknown;
  try {
    parsed = JSON.parse(readFileSync(path, "utf8"));
  } catch {
    return { kind: "down", reason: "unreadable or missing access.json" };
  }
  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
    return { kind: "down", reason: "access.json is not an object" };
  }
  const access = parsed as { readonly [key: string]: unknown };
  if (access["enabled"] !== true) return { kind: "down", reason: "bridge disabled (enabled !== true)" };
  const allowFrom = access["allowFrom"];
  if (!Array.isArray(allowFrom) || allowFrom.length !== 1) {
    return { kind: "down", reason: "allowFrom must hold exactly one paired owner id" };
  }
  const owner = allowFrom[0];
  if (typeof owner !== "string" && typeof owner !== "number") {
    return { kind: "down", reason: "paired owner id is not a string or number" };
  }
  return { kind: "up", owner: String(owner) };
}

/**
 * The approval half of the Telegram surface is a different question from
 * whether inbound works, and it is answered by {@link readApprovalSurface} —
 * shared with the orchestrator tick so this row and the tick's own warning can
 * never disagree about whether an amendment can be asked.
 */

function readBotToken(): string | undefined {
  const env = process.env["TELEGRAM_BOT_TOKEN"];
  if (env !== undefined && env.length > 0) return env;
  const file = join(telegramStateDir(), ".env");
  if (!existsSync(file)) return undefined;
  try {
    for (const line of readFileSync(file, "utf8").split("\n")) {
      const m = /^TELEGRAM_BOT_TOKEN=(.*)$/.exec(line.trim());
      if (m?.[1]) return m[1].replace(/^["']|["']$/g, "");
    }
  } catch {
    return undefined;
  }
  return undefined;
}

export async function probeTelegramHealth(
  projectName?: string,
  request: (input: string, init?: RequestInit) => Promise<Response> = fetch,
): Promise<TelegramHealth> {
  const tick = resolveTickConfig(projectName);
  const accessPath =
    tick.kind === "ok" && tick.config.accessFile !== undefined
      ? tick.config.accessFile
      : join(telegramStateDir(), "access.json");
  const channel = readPairedChannel(accessPath);
  const token = readBotToken();
  if (token === undefined) {
    return {
      kind: "unconfigured",
      detail: `no TELEGRAM_BOT_TOKEN; inbound ${channel.kind === "up" ? "configured" : channel.reason}`,
    };
  }

  let response: Response;
  let body: unknown;
  try {
    response = await request(`https://api.telegram.org/bot${token}/getMe`, {
      signal: AbortSignal.timeout(5_000),
    });
    body = await response.json();
  } catch {
    return {
      kind: "down",
      detail: `Telegram API request failed; inbound ${channel.kind === "up" ? "configured" : channel.reason}`,
    };
  }

  if (body === null || typeof body !== "object" || Array.isArray(body)) {
    return { kind: "down", detail: "Telegram API returned an invalid response" };
  }
  const result = body as Record<string, unknown>;
  if (!response.ok || result["ok"] !== true) {
    const description =
      typeof result["description"] === "string" ? result["description"] : `HTTP ${response.status}`;
    return { kind: "down", detail: `${description}; inbound ${channel.kind === "up" ? "configured" : channel.reason}` };
  }

  const user =
    result["result"] !== null && typeof result["result"] === "object" && !Array.isArray(result["result"])
      ? (result["result"] as Record<string, unknown>)
      : undefined;
  const username = typeof user?.["username"] === "string" ? `@${user["username"]}` : "authenticated";
  // Inbound first: a bridge that is down says nothing about the approval
  // surface, and stacking two remedies on one row buries the one to act on.
  if (channel.kind === "down") return { kind: "degraded", detail: `${username}; inbound ${channel.reason}` };
  // The token is already proven: `getMe` succeeded above. What is left is
  // whether an answer could come back, which is the access file's business.
  const approval = readApprovalSurface(accessPath);
  if (approval.kind === "missing") {
    return { kind: "degraded", detail: `${username}; inbound configured; ${approval.reason}` };
  }
  // Last, and only once the surface is answerable: whether it is *quiet*.
  //
  // The order is the order an operator should fix things in, and one row carries
  // one remedy. A missing token means nothing outbound works, so the approval
  // surface is not worth discussing; a missing approval surface means an
  // amendment cannot be asked, which outranks noise; an interactive profile is
  // real but strictly the least severe — the fleet works, it is just loud. And
  // the two lower checks are not independent of each other: setting `notifyMode`
  // to fix the approval surface is what *arms* the `agent_end` notify post this
  // one warns about, so a fleet that fixes them in the other order would see the
  // noise appear as the reward for fixing the silence.
  const profile = readDaemonProfile(accessPath);
  if (profile.kind === "interactive") {
    return {
      kind: "degraded",
      detail: `${username}; inbound configured; interactive profile relays assistant text — /telegram set profile daemon`,
    };
  }
  return { kind: "ok", detail: `${username}; inbound configured; telegram_ask expected per config (mounted per turn by omp-telegram — a tick that finds it missing falls back to telegram_send and says so)` };
}

export function sessionDirForCwd(cwd: string): string {
  const home = homedir();
  const slug = cwd.startsWith(home) ? cwd.slice(home.length) : cwd;
  return join(home, ".omp", "agent", "sessions", slug.replaceAll("/", "-"));
}

/** The omp session tree the claim-only verdict judges session identity against. */
export function sessionsRoot(): string {
  return join(homedir(), ".omp", "agent", "sessions");
}

/**
 * The orchestrator session file omp-telegram's live claim names for this
 * project — the transcript a resumed/restored pane actually writes, which can
 * live in a different session directory than the tick cwd implies (#600).
 * Never throws: no claim, or an unreadable config, means undefined. Only the
 * claim-only verdict consumes this — the challenge proof reads conductor
 * state, not transcripts (#614).
 */
function claimedOrchestratorSessionFile(
  named: string | undefined,
  alive?: (pid: number) => boolean,
): string | undefined {
  if (named === undefined) return undefined;
  try {
    return resolveClaimedSessionFile(findProject(loadConfig(), named), alive);
  } catch {
    return undefined;
  }
}

function makeChallengeCode(): string {
  const bytes = new Uint8Array(4);
  crypto.getRandomValues(bytes);
  const hex = [...bytes].map((b) => b.toString(16).padStart(2, "0")).join("").toUpperCase();
  return `FLEET-${hex}`;
}

/**
 * mm:ss for the window a challenge code stays good for. Deliberately not
 * `formatDownDuration`, which rounds to whole minutes because it reports
 * hours-to-days outages: rounded to the minute, a code with 30 seconds left
 * would read "5m", which is the exact ambiguity these lines exist to remove
 * (#861).
 */
function armClock(ms: number): string {
  const total = Math.max(0, Math.round(ms / 1000));
  const minutes = Math.floor(total / 60);
  const seconds = total % 60;
  return minutes === 0 ? `${seconds}s` : `${minutes}m${String(seconds).padStart(2, "0")}s`;
}

/**
 * The session surface the claim-only verdict judges: the tick-cwd-derived
 * session directory plus, when the live claim's file lives elsewhere in the
 * tree, that file's directory (#600). An outside-tree claim is not a reason
 * to stop here — the verdict names that failure itself
 * (`claim-session-outside` / `dm-owner-unrelated`), so the refusal carries
 * the fact.
 */
function armVerdictScanDirs(cwd: string, claimed: string | undefined): string[] {
  const cwdDir = sessionDirForCwd(cwd);
  if (claimed === undefined) return [cwdDir];
  const root = sessionsRoot();
  const claimDir = dirname(claimed);
  if (claimDir === cwdDir) return [cwdDir];
  const insideTree = claimDir === root || claimDir.startsWith(join(root, sep));
  return insideTree ? [cwdDir, claimDir] : [cwdDir];
}

/**
 * What each claim-only refusal names, in the operator's words (#613). The
 * verdict's reason enum is the fact; this is the sentence that says it — the
 * same facts the doctor's `telegram-plumbing` finding spells out, without the
 * finding's remediation (the refusal is one line an unattended recovery can
 * escalate verbatim).
 */
function claimOnlyFailureText(reason: TelegramPlumbingFailureReason): string {
  switch (reason) {
    case "channel-down":
      return "the paired inbound channel is not up";
    case "registry-unreadable":
      return "omp-telegram's threads.json is absent, unreadable or malformed — the topic's claim liveness cannot be verified";
    case "no-topic-claim":
      return "no live omp-telegram claim carries the topic arming would send into";
    case "claim-dead":
      return "the topic's omp-telegram claim records no live pid — the pane behind it is gone";
    case "claim-session-outside":
      return "the topic's claim names a session file outside the session directories arm scans — the reply could never be seen";
    case "no-dm-owner":
      return "no flat-chat reply recipient: no dm-owner.json, and bot.lock names no session inside the arm scan surface";
    case "dm-owner-dead":
      return "omp-telegram's dm-owner.json records no live pid — the flat-chat recipient is gone";
    case "dm-owner-unrelated":
      return "the dm-owner.json session sits outside the arm scan surface — a flat reply would land where no challenge can read it";
    case "no-poller-state":
      return "omp-telegram's bot.lock is absent, unreadable or malformed — nothing owns the poll, so nothing is receiving inbound";
    case "poller-dead":
      return "omp-telegram's bot.lock records no live owner and no fresh heartbeat — Telegram isn't being polled";
  }
}

async function sendTelegramMessage(
  token: string,
  owner: string,
  text: string,
  topicId?: number,
  project?: { name: string; workspaceRoot?: string },
): Promise<void> {
  // Shared transport: stale-topic retry + message_thread_id live in one place.
  // The project pair rides along (#1094), so a challenge whose pinned topic is
  // refused follows the project's live claim or lands labelled flat like every
  // other send on this seam.
  await sendTelegram(token, owner, text, { topicId, ...(project === undefined ? {} : { project }) });
}

// ---------------------------------------------------------------------------
// probes
// ---------------------------------------------------------------------------

function probeHerdrUnit(unit = DEFAULT_HERDR_UNIT): { kind: HerdrLayer; detail?: string } {
  const res = runSystemctl(["is-active", unit]);
  if (res.missing) return { kind: "unknown", detail: "no systemctl" };
  const out = res.stdout.trim();
  if (out === "active") return { kind: "active", detail: unit };
  if (out === "inactive" || out === "failed" || out === "dead") {
    return { kind: "inactive", detail: `${unit} ${out}` };
  }
  const error = res.stderr.trim();
  return {
    kind: "unknown",
    detail: `${unit} ${out || error || "systemctl is-active failed"}`,
  };
}

export function paneLayerFromAgents(
  agents: HerdrAgent[],
  agentName: string,
):
  | { kind: "live"; summary: string }
  | { kind: "missing" }
  | { kind: "unknown"; reason: string } {
  const matches = agents.filter((agent) => agent.name === agentName);
  if (matches.length === 0) return { kind: "missing" };
  if (matches.length > 1) {
    return { kind: "unknown", reason: `${matches.length} herdr agents named ${agentName}` };
  }
  const match = matches[0]!;
  if (match.agent === "omp") {
    return { kind: "live", summary: `agent ${agentName} pane ${match.paneId}` };
  }
  if (match.agent === undefined || match.agent === "") return { kind: "missing" };
  return {
    kind: "unknown",
    reason: `agent ${agentName} pane ${match.paneId} is ${match.agent}, not omp`,
  };
}

function probeOmpPane(
  projectName?: string,
):
  | { kind: "live"; summary: string }
  | { kind: "missing" }
  | { kind: "unknown"; reason: string } {
  const tick = resolveTickConfig(projectName);
  if (tick.kind === "invalid") {
    return {
      kind: "unknown",
      reason:
        `tick config invalid at ${tick.path} (${tick.problem}) — ` +
        "the conductor agent name cannot be read",
    };
  }
  const agentName =
    tick.kind === "ok" ? (tick.config.agentName ?? DEFAULT_FLEET_AGENT_NAME) : DEFAULT_FLEET_AGENT_NAME;
  const session = resolveHerdrSession(process.env);
  try {
    const res = spawnSync("herdr", ["--session", session, "agent", "list"], {
      encoding: "utf8",
      timeout: 8_000,
      env: process.env,
    });
    if (res.error) {
      const err = res.error as NodeJS.ErrnoException;
      if (err.code === "ENOENT") return { kind: "unknown", reason: "no herdr" };
      return { kind: "unknown", reason: err.message };
    }
    if (res.status !== 0) {
      const detail = (res.stderr ?? res.stdout ?? `herdr exit ${String(res.status)}`).trim();
      return { kind: "unknown", reason: detail };
    }
    return paneLayerFromAgents(parseHerdrAgentList(res.stdout ?? ""), agentName);
  } catch (err) {
    return { kind: "unknown", reason: err instanceof Error ? err.message : String(err) };
  }
}