import { existsSync, readdirSync, rmSync, statSync } from "node:fs";
import { join } from "node:path";
import { TMUX_SESSION_OWNER_OPTION, tmuxCommand } from "agent-relay-sdk/tmux-utils";
import { isClaudeExitConfirmationDialog } from "agent-relay-sdk/claude-exit-dialog";
import { classifyPidLiveness, killProcessGroup, readProcessStartMs, type PidLiveness } from "agent-relay-sdk/process-utils";
import { STATE_FILE, TOMBSTONE_DIR, orchestratorStateOwnerId } from "./spawn/constants";
import {
  captureSessionPaneBinding,
  classifyOwnerRunnerLiveness,
  clearRunnerTombstone,
  isManagedSessionOrphaned,
  isOwnerConfirmedDead,
  loadStateOutcome,
  logFilePath,
  readOwnerRunnerInfo,
  readRunnerTombstone,
  runnerInfoPath,
  type OwnerRunnerInfo,
  type RunnerTombstone,
  type SessionPaneBinding,
} from "./spawn/runtime";
import type { SessionRecord } from "./spawn/types";
import { tmuxSocketDir } from "./tmux-socket-sweeper";

// Each managed provider session runs on its own `agent-relay-<hash>` tmux socket
// (runner/src/adapters/claude-tmux.ts:tmuxSocketName). The wedged-session reaper
// scans those sockets — the same prefix the tmux-socket-sweeper cleans up.
const SOCKET_PREFIX = "agent-relay-";

// #1514 — a headless Claude/Codex session wedged at the interactive
// "Exit anyway / Move to background / Stay" dialog (or otherwise hung) stays a
// live PID with a live tmux server, so neither the PID-liveness health check nor
// the dead-socket sweeper reaps it. It sits forever holding ~120MB until the host
// is memory-starved. This reaper closes that gap by reaping a live session ONLY
// when ALL of the following positively hold — every one is a fail-safe guard:
//   1. State was positively LOADED this cycle (a missing/corrupt state file skips
//      the whole cycle, so a wiped file can never make live sessions look
//      unmanaged and mass-reap healthy agents — review finding 1).
//   2. The session is OWNED by THIS orchestrator: its @agent-relay-owner tag equals
//      our state-home id. Untagged/mismatched sessions (legacy, or a co-tenant
//      orchestrator sharing the UID + prefix) are never reaped — review finding 4.
//   3. The session is UNMANAGED — absent from orchestrator-sessions.json.
//   4. The session is IDLE past a generous threshold, measured by window_activity
//      (which advances with real pane output, unlike session_activity — review
//      finding 2), so a session that is actively producing output is never reaped.
//   5. The session is a POSITIVE ORPHAN — reaping requires AFFIRMATIVE proof the owning
//      runner terminated (isManagedSessionOrphaned → the single isOwnerConfirmedDead gate):
//      either the pid recorded in its runner-info file is entirely absent from the process
//      table, or (r7) — with runner-info absent — a durable confirmed-dead tombstone that is
//      exactly-name-matched, whose recorded runner pid still probes dead, AND which is bound
//      to this exact session incarnation via the pane identity (`#{pane_pid}` + pane start
//      time) captured from the live session at the moment death was confirmed, matching the
//      pane observed by this very scan. This is the INVERTED default (r4): runner-info is
//      written non-atomically and best-effort, so a MISSING / unreadable / invalid /
//      ambiguous / reused-pid runner-info is NOT evidence the runner is gone — and (r7) a
//      tombstone ALONE is not either: every unbound / mismatched / unobservable case FAILS
//      CLOSED and SPARES the session. A session is therefore reaped only when its recorded
//      runner is positively confirmed dead; when in doubt, spare. Leaking a wedged session
//      is acceptable, killing a live one is not.
//   6. (r8/r9) The kill is IDENTITY-ATOMIC — it targets the PROCESS INCARNATION, never the
//      session NAME. Session names are REUSABLE by design: sessionName() is a deterministic
//      function of (prefix, provider, label, spawnRequestId-or-agentId) — an idempotent spawn
//      retry reproduces the same name — the socket is a pure hash of the name
//      (runner/src/adapters/claude-tmux.ts:tmuxSocketName), and createOrReuseTmuxSession +
//      spawn-agent.ts's supersede path explicitly recreate a killed name on its old socket.
//      So `tmux kill-session -t <name>` re-resolves the name at kill time to WHATEVER session
//      currently holds it — a concurrent orchestrator that replaces the orphan with a
//      same-name/same-socket HEALTHY session in the observe→kill window would be killed by a
//      name-addressed kill no matter how late the name is re-observed (the r7/r8 TOCTOU).
//      r9 therefore removes the name from the lethal action entirely:
//        a. Immediately before killing, the tracked state must still positively load and NOT
//           contain the name, and the name-resolved pane must still match the authorized
//           incarnation (cheap early aborts on the name axis — catch most races upfront).
//        b. The authorization itself must be BINDABLE: the scan captured the orphan's pane
//           pid AND that process's OS start time. Unreadable start time (off-Linux, or the
//           pane already dead) ⇒ identity unconfirmable ⇒ SPARE — a kill is never authorized
//           by a bare pid match (a pid alone does not name an incarnation; see
//           paneIdentityUnchanged).
//        c. The LETHAL ACTION is SIGKILL to the authorized pane pid's process group, issued
//           only after re-reading that pid's /proc start time in the same synchronous instant
//           and matching it exactly against the authorized start time. A same-name healthy
//           replacement is structurally untargetable: its pane is a DIFFERENT process — a
//           different pid is never signalled, and a recycled pid has a start time in a
//           LATER ~10ms clock tick (starttime is tick-granular at USER_HZ=100) and is
//           spared by the exact-match. The residual (r10, finding 3) is therefore two
//           layers: pid free + full-pid-space wrap + reuse of the SAME pid within the same
//           ~10ms start-time tick between the scan read and the T-0 reread (plausible only
//           under an unusually small/churned pid space), and the microsecond no-I/O window
//           between the /proc reread and the kill(2) syscall. Accepted as the r9/r10 bound
//           (there is no name-shaped window left).
//        d. Killing the pane process tears the session down (pane exit ⇒ window ⇒ session ⇒
//           per-session server). If a config (e.g. remain-on-exit) lets the session WRAPPER
//           outlive the dead pane, the corpse is reaped by an atomic COMPARE-AND-KILL executed
//           inside the single-threaded tmux server (`if-shell -F` on `pane_pid==P && pane_dead`
//           guarding the kill-session in one command-queue item): the predicate is true only
//           of a session whose pane IS the dead authorized process, and a healthy session's
//           pane is live (pane_dead=0), so it can never satisfy it — no JS-side window exists
//           between the corpse check and the corpse kill.
//      Any mismatch, disappearance, unobservable pane, unreadable start time, newly-tracked
//      name, or unreadable state at/after kill time ⇒ SPARE this cycle, fail closed (a genuine
//      orphan whose identity merely could not be re-confirmed is retried next cycle; a
//      dead-pane orphan that can never be identity-bound leaks only a tmux session struct —
//      the ~120MB wedge is the LIVE TUI process, which always has a readable start time on
//      the platforms the reaper kills on).
//   7. (r9, finding 2) Reaping and state reconstruction only trust PROVABLY COMPLETE
//      observations of live tmux reality. enumerateLiveManagedSessions distinguishes a
//      socket that is positively server-less (zero sessions, provably) from one whose
//      enumeration FAILED; any failed socket marks the whole observation incomplete, which
//      blocks state reconstruction from persisting (a partial snapshot must never clobber
//      live session records) and blocks the tombstone sweep (a death-proof must never be
//      cleared because its session's socket merely failed to answer).
//
// r3/r4 also RECONSTRUCT tracked state from reality at startup (reconstructOwnedLiveSessions
// → recoverExistingSessions): after a missing/corrupt/invalid state file, owned live
// sessions are adopted back into the state file BEFORE any reaping and before the next
// spawn writes a partial record, so "managed" reflects reality rather than a lossy file.
// Reconstruction routes through the SAME single owner-death gate as guard 5, so the two
// layers are INDEPENDENT defenses: reconstruction skips a session only on the same
// positive proof of death, and even if it skips one, guard 5 still spares it. Neither
// layer can turn the ABSENCE of best-effort metadata into a reap.

export interface LiveManagedSession {
  /** tmux `-L` socket the session lives on. */
  socket: string;
  /** tmux session name, e.g. `ar-claude-session-...`. */
  name: string;
  /** Epoch ms of the session's most recent pane output (max window_activity). */
  lastActivityMs: number;
  /** The @agent-relay-owner tag, or undefined when the session is untagged. */
  owner?: string;
  /** #1514 (r4) — pid of the live process in the session's active pane, if tmux
   * reported it. A real, alive pid the reconstruction path can anchor an adopted
   * record to when the best-effort runner-info file is absent/unreadable. */
  panePid?: number;
  /** #1514 (r8) — OS start time (epoch ms) of `panePid`, read in the SAME scan that
   * observed it. Together with panePid this is the observable identity of the session
   * incarnation this scan saw; the identity-atomic kill (guard 6) must re-observe and
   * match it exactly before killing. Absent when unreadable (e.g. the pane process is
   * already dead, or no /proc). */
  paneStartMs?: number;
}

export interface WedgedSessionReapCandidate {
  socket: string;
  name: string;
  idleMs: number;
  /** #1514 (r7) — the candidate's observed live pane pid, threaded to the positive-orphan
   * gate: the tombstone path can conclude "owner dead" only against this observable signal. */
  panePid?: number;
  /** #1514 (r8) — OS start time of `panePid` from the same scan; the second half of the
   * observable pane identity the identity-atomic kill re-checks at kill time. */
  paneStartMs?: number;
}

export interface WedgedSessionReapResult {
  scanned: number;
  reaped: Array<{ socket: string; name: string; idleMs: number; atExitDialog: boolean }>;
  /** True when the managed-session state file was missing/unreadable, so the
   * cycle skipped reaping entirely rather than risk killing healthy sessions. */
  skippedNoState: boolean;
  /** #1514 (r3) — how many otherwise-eligible sessions were SPARED because their
   * owning runner was still alive (positive-orphan backstop). A non-zero count in the
   * cycles right after a state loss is the signal that reconstruction/backstop just
   * prevented a mass-reap of healthy pre-restart sessions. */
  sparedLiveOwner: number;
  /** #1514 (r8) — how many gate-authorized kills were ABORTED at the last instant because
   * the session's observable pane identity or its tracked state changed (or could not be
   * re-confirmed) between the gate decision and the kill — the TOCTOU guard that stops a
   * stale cycle from killing a same-name/same-socket healthy replacement. */
  sparedKillRecheck: number;
  /** #1514 (r9, finding 2) — true when one or more sockets failed to enumerate this cycle.
   * Candidates on successfully-enumerated sockets are still processed (per-session
   * conclusions are sound — sessions cannot move between sockets), but fleet-wide
   * conclusions are suppressed: the tombstone sweep is skipped so a death-proof is never
   * cleared just because its session's socket failed to answer. */
  enumerationIncomplete: boolean;
}

/**
 * Decide which live managed sessions to reap. Pure so the policy is testable
 * without tmux or fs. A session is reaped iff it is a relay-managed provider
 * session (matches the orchestrator tmux prefix, but is not a guest terminal —
 * guests have their own reaper), is POSITIVELY owned by this orchestrator (its
 * owner tag equals `ownerId`), is NOT tracked in orchestrator-sessions.json, and
 * has been idle at least `idleThresholdMs`.
 */
export function selectWedgedSessions(input: {
  live: LiveManagedSession[];
  trackedNames: Set<string>;
  now: number;
  idleThresholdMs: number;
  tmuxPrefix: string;
  ownerId: string;
}): WedgedSessionReapCandidate[] {
  // No owner id means we can prove ownership of nothing — reap nothing (fail-safe).
  if (!input.ownerId) return [];
  const agentPrefix = `${input.tmuxPrefix}-`;
  const guestPrefix = `${input.tmuxPrefix}-guest-`;
  const candidates: WedgedSessionReapCandidate[] = [];
  for (const session of input.live) {
    if (!session.name.startsWith(agentPrefix)) continue; // not an orchestrator-spawned session
    if (session.name.startsWith(guestPrefix)) continue; // guest terminals have their own TTL reaper
    if (session.owner !== input.ownerId) continue; // not positively owned by THIS orchestrator — never reap
    if (input.trackedNames.has(session.name)) continue; // still managed — leave it alone
    const idleMs = input.now - session.lastActivityMs;
    if (idleMs < input.idleThresholdMs) continue; // too fresh — active output or a spawn/registration race
    candidates.push({ socket: session.socket, name: session.name, idleMs, panePid: session.panePid, paneStartMs: session.paneStartMs });
  }
  return candidates;
}

/**
 * Read the managed-session names from orchestrator-sessions.json.
 * Returns `null` — meaning "do not reap this cycle" — when the file is missing,
 * unparseable, OR structurally invalid (any record failing isValidSessionRecordShape,
 * e.g. `[{}]` / `[{"name":17}]`), so a wiped/corrupt/logically-corrupt state file can
 * never make every live session look unmanaged and trigger a mass reap of healthy
 * agents. An empty but valid `[]` file (the exact #1514 incident state) returns an
 * empty set, which correctly makes an orphaned session reap-eligible. Shares
 * loadStateOutcome with the orchestrator so the "loaded vs missing/corrupt/invalid"
 * decision is single-sourced.
 */
export function loadTrackedSessionNames(stateFile = STATE_FILE): Set<string> | null {
  const outcome = loadStateOutcome(stateFile);
  if (outcome.status !== "loaded") return null;
  const names = new Set<string>();
  for (const record of outcome.records) {
    if (record && typeof record === "object" && typeof (record as { name?: unknown }).name === "string") {
      names.add((record as { name: string }).name);
    }
  }
  return names;
}

/**
 * #1514 (r9, finding 2) — the result of enumerating live managed sessions across the
 * `agent-relay-*` sockets, WITH an explicit completeness verdict. `complete` is true only
 * when every managed socket was POSITIVELY observed: it either enumerated successfully or
 * was provably server-less (tmux reported no server on the socket — a dead socket file has
 * zero sessions by definition; the socket sweeper removes it). Any other per-socket failure
 * used to be silently swallowed, letting an empty/partial array masquerade as a genuine
 * completed observation — which state reconstruction would then PERSIST, wiping live session
 * records on a one-socket hiccup. Consumers that draw fleet-wide conclusions (persisting a
 * reconstructed state set, sweeping tombstones by "no longer live") MUST require `complete`;
 * per-session conclusions (a candidate on a successfully-enumerated socket) remain sound on
 * a partial result because sessions cannot move between sockets (the socket is a pure hash
 * of the session name).
 */
export interface LiveManagedSessionEnumeration {
  sessions: LiveManagedSession[];
  complete: boolean;
  /** Socket entries whose enumeration failed for a reason OTHER than "no server". */
  failedSockets: string[];
}

// tmux stderr signatures that POSITIVELY mean "no server is listening on this socket" —
// the socket file is a dead leftover (crash/exit), which provably hosts zero sessions.
// Anything else nonzero (unexpected error, ENOTSOCK anomaly, kill mid-enumeration, future
// tmux message change) is treated fail-closed as a FAILED observation of that socket.
function tmuxStderrMeansNoServer(stderr: string): boolean {
  return stderr.includes("no server running")
    || (stderr.includes("error connecting to")
      && (stderr.includes("No such file or directory") || stderr.includes("Connection refused")));
}

/**
 * Enumerate every live managed tmux session across the `agent-relay-*` sockets.
 * Reads window_activity (NOT session_activity — the latter is ~session creation
 * time and does not advance with pane output, #1514 finding 2), taking the max
 * across a session's windows, plus the @agent-relay-owner tag, in one tmux call
 * per socket. Per-socket failures are never swallowed silently: they are reported
 * in `failedSockets` and flip `complete` to false (#1514 r9 finding 2).
 */
export function enumerateLiveManagedSessions(socketDir = tmuxSocketDir(), now = Date.now(), env?: NodeJS.ProcessEnv): LiveManagedSessionEnumeration {
  // #1514 (r10, finding 2) — "no socket dir" must be PROVEN absent, not merely unobserved.
  // existsSync returns false BOTH for a truly-absent dir (ENOENT) and for one that cannot
  // be statted at all (EACCES/EPERM/ENOTDIR via an inaccessible ancestor), and only the
  // former is a complete observation of zero managed sessions. Any other errno means the
  // dir — and every socket in it — was NOT observed: fail closed (complete:false) so
  // reconstruction throws and the no-persist fail-safe protects the prior state.
  try {
    statSync(socketDir);
  } catch (error) {
    if ((error as NodeJS.ErrnoException).code === "ENOENT") {
      return { sessions: [], complete: true, failedSockets: [] }; // provably no socket dir ⇒ provably zero managed sessions
    }
    return { sessions: [], complete: false, failedSockets: [socketDir] }; // unstatable ⇒ a FAILED observation, not an empty one
  }
  const sessions: LiveManagedSession[] = [];
  const failedSockets: string[] = [];
  let entries: string[];
  try {
    entries = readdirSync(socketDir);
  } catch {
    // The socket dir exists but cannot be read — nothing was observed; the whole
    // enumeration is a failed observation, not an empty one.
    return { sessions: [], complete: false, failedSockets: [socketDir] };
  }
  for (const entry of entries) {
    if (!entry.startsWith(SOCKET_PREFIX)) continue;
    const result = Bun.spawnSync(
      tmuxCommand(entry, "list-windows", "-a", "-F", `#{session_name}\t#{window_activity}\t#{${TMUX_SESSION_OWNER_OPTION}}\t#{pane_pid}`),
      // `env` (tests only): pin TMUX_TMPDIR so `-L <entry>` resolves inside a private dir,
      // the same seam shape as tmux-socket-sweeper. Production inherits the process env.
      { stdin: "ignore", stdout: "pipe", stderr: "pipe", ...(env ? { env } : {}) },
    );
    if (result.exitCode !== 0) {
      // Distinguish "provably no server on this socket" (a complete observation of zero
      // sessions — the sweeper removes the file) from ANY other failure, which marks this
      // socket — and therefore the whole enumeration — incomplete (fail closed).
      if (!tmuxStderrMeansNoServer(result.stderr.toString())) failedSockets.push(entry);
      continue;
    }
    const byName = new Map<string, { lastActivityMs: number; owner?: string; panePid?: number }>();
    for (const line of result.stdout.toString().split("\n")) {
      const parts = line.split("\t");
      if (parts.length < 2) continue;
      const name = parts[0]!.trim();
      if (!name) continue;
      const activitySec = Number(parts[1]!.trim());
      const owner = (parts[2] ?? "").trim() || undefined;
      const panePidRaw = Number((parts[3] ?? "").trim());
      const panePid = Number.isFinite(panePidRaw) && panePidRaw > 0 ? panePidRaw : undefined;
      // tmux reports window_activity as epoch seconds. If a tmux build ever omits
      // it, fail SAFE: treat the window as just-active (never reaped) rather than
      // maximally idle, so an unreadable clock can't cause a false reap.
      const lastActivityMs = Number.isFinite(activitySec) && activitySec > 0 ? activitySec * 1000 : now;
      const prev = byName.get(name);
      if (!prev) {
        byName.set(name, { lastActivityMs, owner, panePid });
      } else {
        // Anchor panePid to the most-recently-active window's pane (freshest liveness).
        if (lastActivityMs >= prev.lastActivityMs && panePid) prev.panePid = panePid;
        prev.lastActivityMs = Math.max(prev.lastActivityMs, lastActivityMs);
        if (!prev.owner && owner) prev.owner = owner;
      }
    }
    for (const [name, info] of byName) {
      // #1514 (r8) — complete the observable pane identity in the same scan: the pane
      // process's OS start time. Unreadable (dead pane / no /proc) ⇒ left absent; the
      // identity-atomic kill then only matches an equally start-time-less re-observation.
      const paneStartMs = info.panePid !== undefined ? readProcessStartMs(info.panePid) ?? undefined : undefined;
      sessions.push({ socket: entry, name, lastActivityMs: info.lastActivityMs, owner: info.owner, panePid: info.panePid, ...(paneStartMs !== undefined ? { paneStartMs } : {}) });
    }
  }
  return { sessions, complete: failedSockets.length === 0, failedSockets };
}

/**
 * Sessions-only view of enumerateLiveManagedSessions, for callers whose per-session
 * conclusions are sound on a partial observation. Callers that persist or clear
 * fleet-wide state from the result MUST use the enumeration and require `complete`.
 */
export function listLiveManagedSessions(socketDir = tmuxSocketDir(), now = Date.now()): LiveManagedSession[] {
  return enumerateLiveManagedSessions(socketDir, now).sessions;
}

/**
 * #1514 (r3/r4) — RECONSTRUCT the tracked managed-session set from live tmux reality.
 * Called at startup when the state file is missing/corrupt/invalid, so "managed"
 * reflects the sessions that actually exist rather than a lossy file. Enumerates the
 * live sessions THIS orchestrator owns (owner tag matches, non-guest managed prefix)
 * and adopts every one whose owning runner is NOT positively dead — routing the decision
 * through the exact single owner-death gate (isOwnerConfirmedDead, with this session's
 * observed pane pid) the reaper's positive-orphan backstop uses. Only a session whose
 * owner is positively confirmed dead (runner-info pid absent, or an identity-bound
 * pane-matched tombstone — r7) is left unadopted (a genuine orphan for the reaper); a
 * MISSING / unreadable / invalid / reused-pid runner-info FAILS CLOSED and is adopted
 * (r4), so a lost runner-info file can never make a live session look orphaned in
 * EITHER layer. When runner-info is present-and-alive the
 * record is enriched from it (real pid/agentId/provider); otherwise the adopted record
 * is anchored to the session's live pane pid (a real, alive process) so it is not
 * spuriously reaped as "dead" on a later recovery. Each record is flagged `adopted` so
 * its best-effort fields are identifiable. Persisting these BEFORE the next spawn's
 * `addSessionRecord` means that spawn appends to a complete set instead of an empty
 * one, durably closing the delayed mass-reap of healthy pre-restart sessions. Pure
 * w.r.t. injectable deps so the policy is testable without tmux/fs.
 */
export function reconstructOwnedLiveSessions(input: {
  tmuxPrefix: string;
  ownerId: string;
  now?: number;
  socketDir?: string;
  /** Test seam: a plain session list is treated as a PROVABLY COMPLETE observation.
   * Production resolution goes through `enumerate` (completeness-aware). */
  listLive?: (socketDir?: string) => LiveManagedSession[];
  /** #1514 (r9, finding 2) — completeness-aware enumeration seam; the production default. */
  enumerate?: (socketDir?: string) => LiveManagedSessionEnumeration;
  ownerRunner?: (name: string) => OwnerRunnerInfo | null;
  classify?: (pid: number) => PidLiveness;
  startMs?: (pid: number) => number | null;
  readTombstone?: (name: string) => RunnerTombstone | null;
}): SessionRecord[] {
  // No owner id means we can prove ownership of nothing — adopt nothing (fail-safe).
  if (!input.ownerId) return [];
  const now = input.now ?? Date.now();
  const ownerRunner = input.ownerRunner ?? readOwnerRunnerInfo;
  const readTombstone = input.readTombstone ?? readRunnerTombstone;
  const enumeration = input.enumerate
    ? input.enumerate(input.socketDir)
    : input.listLive
      ? { sessions: input.listLive(input.socketDir), complete: true, failedSockets: [] }
      : enumerateLiveManagedSessions(input.socketDir);
  // #1514 (r9, finding 2) — reconstruction may only CONCLUDE from a provably complete
  // observation. A per-socket tmux failure used to be silently swallowed, making an
  // empty/partial list look like completed reality; persisting that snapshot wiped live
  // session records on a one-socket hiccup. Throwing routes the caller
  // (recoverExistingSessions) into its fail-safe branch: keep the prior state, skip the
  // persist, retry on a later recovery — never clobber live state on ignorance.
  if (!enumeration.complete) {
    throw new Error(
      `live-session enumeration incomplete (failed sockets: ${enumeration.failedSockets.join(", ") || "unknown"}) — refusing to reconstruct from a partial observation`,
    );
  }
  const live = enumeration.sessions;
  const agentPrefix = `${input.tmuxPrefix}-`;
  const guestPrefix = `${input.tmuxPrefix}-guest-`;
  const adopted: SessionRecord[] = [];
  for (const session of live) {
    if (!session.name.startsWith(agentPrefix)) continue; // not an orchestrator-spawned session
    if (session.name.startsWith(guestPrefix)) continue; // guest terminals have their own reaper
    if (session.owner !== input.ownerId) continue; // only sessions THIS orchestrator positively owns
    // #1514 (r6/r7, finding 3) — the skip-adoption ("genuine orphan") decision routes through
    // the ONE owner-death gate, with this session's observed pane pid as the observable
    // binding — the same decision, on the same evidence, the reaper's positive-orphan test
    // makes. Adopting a genuine orphan would mark it "managed" and re-leak it; skipping a
    // live-owned one would expose it to the reaper. One gate, one answer, both layers.
    if (isOwnerConfirmedDead(
      { name: session.name, panePid: session.panePid },
      { classify: input.classify, startMs: input.startMs, readInfo: ownerRunner, readTombstone },
    )) continue; // positive proof the runner is gone — a genuine orphan; leave it for the reaper
    // NOT confirmed dead — fail closed and adopt. The classifier below is used ONLY to pick
    // the pid anchor (it can never conclude death here: had it been "dead", the gate above
    // would already have skipped this session). Anchor the record's pid to the runner-info
    // pid only when it is the identity-consistent, live runner; otherwise use the live pane
    // pid so the adopted record is not later mistaken for dead.
    const info = ownerRunner(session.name);
    const liveness = classifyOwnerRunnerLiveness(info, { classify: input.classify, startMs: input.startMs });
    const pid = liveness === "alive" && info ? info.pid : session.panePid;
    if (pid === undefined) continue; // no live pid to anchor to — nothing safe to persist; the reaper spares it anyway
    adopted.push({
      name: session.name,
      pid,
      provider: info?.provider ?? "unknown",
      cwd: "",
      logFile: logFilePath(session.name),
      runnerInfoFile: runnerInfoPath(session.name),
      agentId: info?.agentId ?? "",
      approvalMode: "",
      startedAt: info?.startedAt ?? now,
      adopted: true,
    });
  }
  return adopted;
}

// #1514 (r8/r9, guard 6) — does the pane identity RE-OBSERVED at kill time still match exactly
// the pane identity of the incarnation the gate authorized (observed at scan time)? The kill
// proceeds only on an exact FULL match; every other shape fails closed and SPARES this cycle:
//   - the scan never observed a pane pid ⇒ there is no incarnation identity to bind the kill
//     to ⇒ never kill on an unbindable authorization;
//   - (r9, finding 3) the scan could not read the pane process's OS start time ⇒ the pid alone
//     does NOT name an incarnation (pids are recycled; readProcessStartMs is unavailable
//     off-Linux and can transiently fail on Linux) ⇒ identity UNCONFIRMED ⇒ spare. The old
//     "both-unreadable identifies the lingering dead-pane orphan" special case was pid-only
//     identity and is gone: a dead-pane corpse holds only a tmux session struct (the ~120MB
//     wedge is a LIVE TUI, whose start time is readable where the reaper kills), so sparing
//     it costs a struct while killing on pid-only could hit a recycled pid's session;
//   - the pane is not observable now, or its pid differs ⇒ the session disappeared or was
//     replaced (a new session's pane is a new process) ⇒ spare;
//   - same pid but a different or unreadable OS start time now ⇒ the pid was reused by a
//     different process, or identity cannot be re-confirmed ⇒ spare.
function paneIdentityUnchanged(
  authorized: { panePid?: number; paneStartMs?: number },
  observed: SessionPaneBinding | null,
): boolean {
  if (authorized.panePid === undefined || authorized.paneStartMs === undefined) return false;
  if (!observed || observed.panePid !== authorized.panePid) return false;
  return observed.paneStartMs !== undefined && observed.paneStartMs === authorized.paneStartMs;
}

function captureExitDialogFlag(socket: string, name: string): boolean {
  try {
    const result = Bun.spawnSync(tmuxCommand(socket, "capture-pane", "-p", "-t", name, "-S", "-80"), {
      stdin: "ignore", stdout: "pipe", stderr: "ignore",
    });
    if (result.exitCode !== 0) return false;
    return isClaudeExitConfirmationDialog(result.stdout.toString());
  } catch {
    return false;
  }
}

// Exact-match session existence probe. The `=` prefix disables tmux's prefix/fnmatch target
// matching, so a partially-matching DIFFERENT session name can never answer for this one.
function hasSessionExact(socket: string, name: string): boolean {
  const result = Bun.spawnSync(tmuxCommand(socket, "has-session", "-t", `=${name}`), {
    stdin: "ignore", stdout: "ignore", stderr: "ignore",
  });
  return result.exitCode === 0;
}

// #1514 (r9) — atomic COMPARE-AND-KILL of a corpse session, executed INSIDE the tmux server.
// `if-shell -F` evaluates the format predicate and runs the guarded command in one
// command-queue item of the single-threaded tmux server, so no other client action (e.g. a
// replacement spawn claiming the name) can interleave between the check and the kill. The
// predicate is true ONLY of a session whose active pane is pane_pid==<panePid> AND dead
// (pane_dead=1, the remain-on-exit corpse of the incarnation this reaper already
// identity-verified and SIGKILLed); a healthy session's pane is a live process (pane_dead=0),
// so a same-name replacement can never satisfy it. On any failure (old tmux, races, target
// gone) nothing is killed — callers re-observe and retry/give up fail-closed.
function killCorpseSessionExact(socket: string, name: string, panePid: number): void {
  // Target `=name:` — exact-match session, its active pane — because a bare `=name` pane
  // target does not resolve pane formats (observed on tmux 3.4: #{pane_pid}/#{pane_dead}
  // evaluate empty, making the predicate constant-false and the corpse unreapable).
  Bun.spawnSync(
    tmuxCommand(
      socket, "if-shell", "-F", "-t", `=${name}:`,
      `#{&&:#{==:#{pane_pid},${panePid}},#{==:#{pane_dead},1}}`,
      `kill-session -t "=${name}"`,
    ),
    { stdin: "ignore", stdout: "ignore", stderr: "ignore" },
  );
}

/** Outcome of an identity-atomic kill attempt (#1514 r9). */
export type IdentityKillOutcome =
  // The authorized incarnation was re-verified at T-0 and destroyed, and its session is gone.
  | "killed"
  // Identity could not be re-confirmed at/after T-0, or the name is now held by a different
  // incarnation — nothing was (further) harmed; fail closed, retry next cycle if still real.
  | "spared"
  // The pane process was signalled but the session could not be confirmed gone in time (or
  // the corpse kill failed). Death-proof must be retained; the next cycle re-detects.
  | "failed";

/**
 * #1514 (r9, finding 1) — the IDENTITY-ATOMIC KILL. Destroys exactly ONE authorized session
 * incarnation, identified by its pane process (pid + OS start time), never by name resolution:
 *   1. T-0 recheck: re-read the authorized pid's /proc start time in the same synchronous
 *      instant as the kill; unreadable or different ⇒ SPARE (the incarnation is gone or the
 *      pid was recycled). Residual (r10, finding 3): /proc starttime is clock-tick granular
 *      (~10ms at USER_HZ=100), so the equality bound assumes no pid wrap+reuse onto the
 *      SAME pid within the same tick between the scan read and this reread (plausible only
 *      under an unusually small/churned pid space), layered on the microsecond no-I/O
 *      window between this reread and the kill(2) syscall (the review-accepted bound;
 *      there is no name-shaped window at all).
 *   2. SIGKILL the pane process GROUP (the pane child is a setsid leader, so the group is its
 *      foreground tree; SIGTERM is pointless here — a wedged TUI's whole failure mode is
 *      prompting instead of exiting). A same-name replacement's pane is a different process:
 *      different pid ⇒ never signalled; recycled pid ⇒ a later start-time tick ⇒ spared in
 *      step 1 (up to the same-tick wrap+reuse residual above).
 *   3. Verify teardown: pane exit normally cascades pane ⇒ window ⇒ session ⇒ per-session
 *      server. Poll (bounded) until the session is positively gone.
 *   4. Corpse fallback: if the session WRAPPER outlives its dead pane (e.g. user tmux config
 *      sets remain-on-exit; tmuxCommand loads user config), reap it via the in-server atomic
 *      compare-and-kill (killCorpseSessionExact) — the predicate (pane_pid==P && pane_dead)
 *      can only be true of the corpse of the verified-and-killed incarnation, never of a
 *      healthy (live-pane) session.
 * A re-observation that shows the name held by ANY other incarnation (different pane pid, or
 * same pid but live with a different start time) returns "spared" immediately — that session
 * is someone else's, structurally never targeted. Timeout ⇒ "failed" (retain death-proof).
 */
export function killWedgedSessionByIdentity(input: {
  socket: string;
  name: string;
  panePid: number;
  paneStartMs: number;
  startMs?: (pid: number) => number | null;
  classify?: (pid: number) => PidLiveness;
  killPaneProcess?: (pid: number) => void;
  observePane?: (socket: string, name: string) => SessionPaneBinding | null;
  hasSession?: (socket: string, name: string) => boolean;
  killCorpse?: (socket: string, name: string, panePid: number) => void;
  sleep?: (ms: number) => void;
  deadlineMs?: number;
  pollMs?: number;
}): IdentityKillOutcome {
  // #1514 (r10, finding 1) — defense in depth: an "authorized" pane pid below 2 must never
  // reach a group kill, no matter what the scan recorded. killProcessGroup re-checks this,
  // but an injected killPaneProcess seam (or a future default) must never see 0/1/-1 either
  // (kill(-1) is "signal everything the caller may signal", not "process group 1").
  if (!Number.isInteger(input.panePid) || input.panePid < 2) return "spared";
  const startMs = input.startMs ?? readProcessStartMs;
  const classify = input.classify ?? classifyPidLiveness;
  const observePane = input.observePane ?? ((socket: string, name: string) => captureSessionPaneBinding(name, socket));
  const hasSession = input.hasSession ?? hasSessionExact;
  const killPaneProcess = input.killPaneProcess ?? ((pid: number) => killProcessGroup(pid, "SIGKILL"));
  const killCorpse = input.killCorpse ?? killCorpseSessionExact;
  const sleep = input.sleep ?? ((ms: number) => Bun.sleepSync(ms));
  const deadlineMs = input.deadlineMs ?? 2_000;
  const pollMs = Math.max(1, input.pollMs ?? 50);

  // T-0: the lethal action is bound to the process INCARNATION (pid + start time), re-read
  // synchronously here — never to the session name (finding 1) and never to a bare pid
  // (finding 3). Unreadable or changed ⇒ the authorized incarnation no longer exists ⇒ spare.
  const t0StartMs = startMs(input.panePid);
  if (t0StartMs === null || t0StartMs !== input.paneStartMs) return "spared";
  killPaneProcess(input.panePid);

  for (let waited = 0; ; waited += pollMs) {
    // Positive teardown: the pane's death cascades to the session (and its per-session
    // server); exact-name probe so a similarly-named session can never answer for it.
    if (!hasSession(input.socket, input.name)) return "killed";
    const observed = observePane(input.socket, input.name);
    if (observed !== null) {
      if (observed.panePid !== input.panePid) return "spared"; // the name is now a DIFFERENT incarnation (replacement / user split) — hands off
      const liveness = classify(input.panePid);
      if (liveness === "alive") {
        const nowStartMs = startMs(input.panePid);
        if (nowStartMs !== null && nowStartMs !== input.paneStartMs) return "spared"; // pid recycled by a NEW live pane — not ours
        // Still our incarnation with the SIGKILL pending (or momentarily unreadable) — wait.
      } else if (liveness === "dead") {
        // The session wrapper outlived its dead pane (remain-on-exit corpse). Reap it with
        // the in-server atomic compare-and-kill; re-check disappearance on the next loop.
        killCorpse(input.socket, input.name, input.panePid);
        if (!hasSession(input.socket, input.name)) return "killed";
      }
      // liveness === "unknown": unprobeable — keep waiting, fail closed on timeout.
    }
    if (waited >= deadlineMs) return "failed";
    sleep(pollMs);
  }
}

// #1514 (Defect 3/4) — tombstone hygiene sweep. Deletes any tombstone that is INVALID
// (unreadable / unparseable / marker-less / invalid pid / internal-name≠filename — all of which
// readRunnerTombstone rejects to null, Defect 4) by removing the ACTUAL file read, and clears
// any VALID tombstone whose session is no longer live (its owner's death-proof is spent). A
// name-mismatched tombstone must be deleted by its real path, NOT clearRunnerTombstone(internalName)
// which would clear the WRONG path and leak this file forever. Best-effort: failures are swallowed
// (a lingering tombstone is harmless — it is re-validated and only ever authorizes reaping a
// session the single fail-closed gate already proves dead).
// #1514 (r7, r6 finding 4) — how old a crash-left `<name>.json.tmp` must be before the sweep
// deletes it. writeRunnerTombstone's tmp→rename window is microseconds; anything older is a
// crash leftover. The generous margin guarantees the sweep can never race an in-flight write
// and delete its tmp between the write and the rename (which would silently drop a genuine
// death-proof → orphan leak).
const TOMBSTONE_TMP_SWEEP_MIN_AGE_MS = 60_000;

function sweepStaleTombstones(liveNames: Set<string>): void {
  try {
    if (!existsSync(TOMBSTONE_DIR)) return;
    for (const entry of readdirSync(TOMBSTONE_DIR)) {
      // r6 finding 4 — a crash between writeRunnerTombstone's tmp write and its rename leaks
      // a `.json.tmp` no reader ever visits; sweep it once it is unambiguously not in-flight.
      if (entry.endsWith(".json.tmp")) {
        try {
          const path = join(TOMBSTONE_DIR, entry);
          if (Date.now() - statSync(path).mtimeMs >= TOMBSTONE_TMP_SWEEP_MIN_AGE_MS) rmSync(path, { force: true });
        } catch {}
        continue;
      }
      if (!entry.endsWith(".json")) continue;
      const tombstone = readRunnerTombstone(entry.slice(0, -".json".length));
      if (!tombstone) {
        rmSync(join(TOMBSTONE_DIR, entry), { force: true }); // invalid/malformed/misplaced → delete the real file
        continue;
      }
      if (!liveNames.has(tombstone.name)) clearRunnerTombstone(tombstone.name); // valid but stale → clear
    }
  } catch {
    // best-effort hygiene only
  }
}

/**
 * Reap orphaned, idle managed tmux sessions this orchestrator positively owns (the
 * #1514 wedge safety net). Kills live sessions that carry THIS orchestrator's owner
 * tag, are unmanaged (absent from orchestrator-sessions.json), and idle past
 * `idleThresholdMs`, AND for which there is positive evidence of orphan-hood — the
 * owning runner is gone (`isOrphaned`) — classifying each by whether it was parked at
 * the exit-confirmation dialog (for the reap log). Runs on a periodic timer.
 */
export function reapWedgedSessions(input: {
  tmuxPrefix: string;
  idleThresholdMs: number;
  now?: number;
  socketDir?: string;
  stateFile?: string;
  ownerId?: string;
  /** Test seam: a plain session list is treated as a PROVABLY COMPLETE observation.
   * Production resolution goes through `enumerate` (completeness-aware, r9 finding 2). */
  listLive?: (socketDir?: string) => LiveManagedSession[];
  /** #1514 (r9, finding 2) — completeness-aware enumeration seam; the production default. */
  enumerate?: (socketDir?: string) => LiveManagedSessionEnumeration;
  captureExitDialog?: (socket: string, name: string) => boolean;
  /** #1514 (Defect 3 / r9) — destroy the authorized session incarnation. The PRODUCTION
   * default is the identity-atomic kill (killWedgedSessionByIdentity): SIGKILL to the
   * identity-rechecked pane process, session teardown verified, corpse compare-and-killed —
   * the session NAME is never what the lethal action resolves. Return contract:
   * "spared" ⇒ identity could not be re-confirmed at kill time (counted in
   * sparedKillRecheck); strict `false` or "failed" ⇒ the kill did not complete, so the
   * session's death-proof is retained and the reap is not reported; anything else
   * (void/number/true — legacy test doubles) reads as killed. */
  kill?: (socket: string, name: string, identity: { panePid: number; paneStartMs: number }) => unknown;
  /** #1514 (r3/r7) — positive-orphan test: true iff the session's owning runner is gone.
   * Receives the candidate's observed live pane pid so the tombstone path can bind to the
   * observable session. A candidate whose runner is still alive is spared, never reaped. */
  isOrphaned?: (name: string, panePid?: number) => boolean;
  /** #1514 (r8, guard 6) — re-observe a session's pane identity immediately before the kill.
   * Returns the pane pid (+ its OS start time when readable) currently in the session, or
   * null when the session/pane is not observable. The kill is aborted (SPARED) unless the
   * re-observation exactly matches the incarnation the gate authorized. */
  observePane?: (socket: string, name: string) => SessionPaneBinding | null;
  /** #1514 (Defect 3) — clear a reaped session's durable death-proof tombstone. */
  clearTombstone?: (name: string) => void;
  /** #1514 (Defect 3) — best-effort sweep of tombstones whose session is confirmed gone. */
  sweepTombstones?: (liveNames: Set<string>) => void;
}): WedgedSessionReapResult {
  const now = input.now ?? Date.now();
  const ownerId = input.ownerId ?? orchestratorStateOwnerId(input.stateFile);
  const trackedNames = loadTrackedSessionNames(input.stateFile);
  const enumeration = input.enumerate
    ? input.enumerate(input.socketDir)
    : input.listLive
      ? { sessions: input.listLive(input.socketDir), complete: true, failedSockets: [] }
      : enumerateLiveManagedSessions(input.socketDir);
  const live = enumeration.sessions;
  if (trackedNames === null) {
    return { scanned: live.length, reaped: [], skippedNoState: true, sparedLiveOwner: 0, sparedKillRecheck: 0, enumerationIncomplete: !enumeration.complete };
  }
  const candidates = selectWedgedSessions({
    live,
    trackedNames,
    now,
    idleThresholdMs: input.idleThresholdMs,
    tmuxPrefix: input.tmuxPrefix,
    ownerId,
  });
  const capture = input.captureExitDialog ?? captureExitDialogFlag;
  const kill = input.kill
    ?? ((socket: string, name: string, identity: { panePid: number; paneStartMs: number }) =>
      killWedgedSessionByIdentity({ socket, name, ...identity }));
  const isOrphaned = input.isOrphaned ?? ((name: string, panePid?: number) => isManagedSessionOrphaned({ name, panePid }));
  const observePane = input.observePane ?? ((socket: string, name: string) => captureSessionPaneBinding(name, socket));
  const clearTombstone = input.clearTombstone ?? clearRunnerTombstone;
  const reaped: WedgedSessionReapResult["reaped"] = [];
  let sparedLiveOwner = 0;
  let sparedKillRecheck = 0;
  for (const candidate of candidates) {
    // Positive-orphan backstop: never reap a session whose owning runner is still
    // alive — it is healthy and only "unmanaged" because state was recently lost.
    // The candidate's observed pane pid is the observable-session binding the
    // tombstone path must match before it may conclude "owner dead" (r7).
    if (!isOrphaned(candidate.name, candidate.panePid)) {
      sparedLiveOwner++;
      continue;
    }
    // #1514 (r9, finding 3) — a kill must be BINDABLE to a process incarnation: pane pid AND
    // its OS start time, both captured by this cycle's scan. A candidate whose start time was
    // unreadable (off-Linux, dead pane, transient /proc failure) has pid-only identity, which
    // does not name an incarnation ⇒ never killable this cycle. Checked before any kill-time
    // work so the TS-narrowed identity below is total.
    if (candidate.panePid === undefined || candidate.paneStartMs === undefined) {
      sparedKillRecheck++;
      continue;
    }
    const atExitDialog = capture(candidate.socket, candidate.name);
    // #1514 (r8, guard 6a) — cheap name-axis rechecks immediately before the kill: (a) the
    // tracked state must still positively load and still NOT contain this name — a name
    // tracked/adopted since the scan is managed again, and an unreadable state can no longer
    // confirm anything; (b) the name-resolved pane must still match the authorized
    // incarnation. These catch most races early, but they are NOT what makes the kill safe —
    // the lethal action below never resolves the name (r9): it is bound to the pane process
    // incarnation itself, so a replacement claiming the name after these checks is still
    // structurally untargetable.
    const trackedAtKill = loadTrackedSessionNames(input.stateFile);
    if (trackedAtKill === null || trackedAtKill.has(candidate.name)) {
      sparedKillRecheck++;
      continue;
    }
    if (!paneIdentityUnchanged(candidate, observePane(candidate.socket, candidate.name))) {
      sparedKillRecheck++;
      continue;
    }
    // #1514 (Defect 3 / r9) — the identity-atomic kill. Only a POSITIVE completed kill
    // ("killed") consumes the death-proof and counts as reaped. "spared" means the
    // incarnation identity could not be re-confirmed at the lethal instant (or the name is
    // now someone else's) — fail closed, count with the kill-time spares. `false`/"failed"
    // (kill did not complete: transient tmux/process error, teardown unverified) RETAINS the
    // tombstone and does NOT report the reap, so the next cycle re-detects the
    // still-proven-dead owner and retries — rather than clearing the proof and permanently
    // sparing a session that is still there.
    const outcome = kill(candidate.socket, candidate.name, { panePid: candidate.panePid, paneStartMs: candidate.paneStartMs });
    if (outcome === "spared") {
      sparedKillRecheck++;
      continue;
    }
    if (outcome === false || outcome === "failed") continue;
    // The session is gone, so its durable death-proof is spent: clear the tombstone (if any).
    clearTombstone(candidate.name);
    reaped.push({ socket: candidate.socket, name: candidate.name, idleMs: candidate.idleMs, atExitDialog });
  }
  // #1514 (Defect 3) — hygiene: drop tombstones for sessions that are no longer live at all
  // (reaped above, or died on their own). Skippable via a no-op dep in pure unit tests.
  // (r9, finding 2) — only on a PROVABLY COMPLETE enumeration: with any socket unobserved,
  // "not in liveNames" is ignorance, not death, and clearing a tombstone on it would destroy
  // the only durable proof that lets a genuine orphan ever be reaped.
  if (enumeration.complete) {
    (input.sweepTombstones ?? sweepStaleTombstones)(new Set(live.map((s) => s.name)));
  }
  return { scanned: live.length, reaped, skippedNoState: false, sparedLiveOwner, sparedKillRecheck, enumerationIncomplete: !enumeration.complete };
}