import { join } from "node:path";
import { chmodSync, closeSync, fsyncSync, lstatSync, openSync, readFileSync, renameSync, unlinkSync, writeFileSync } from "node:fs";
import { randomUUID } from "node:crypto";
import { homedir } from "node:os";
import { shellQuote } from "agent-relay-sdk/shell-utils";

/**
 * #1509 — the self-upgrade restart guard's IDENTITY, SCRIPT, and REAPER.
 *
 * This module owns everything about the launchd restart guard that is orthogonal to planning/
 * dispatching an upgrade (which lives in ./self-upgrade.ts): the per-generation identity scheme
 * (§2), the guard shell script (§8), the one-shot plist (§13), and the running-state reaper (§6,
 * §7, §13). See docs/self-upgrade-guard-design.md for the full concurrency argument.
 *
 * r7 (§13): on launchd the guard is a ONE-SHOT `launchctl bootstrap` job (RunAtLoad, KeepAlive=false),
 * NOT a KeepAlive `launchctl submit` job. It carries NO status/completion marker — its presence-and-
 * running state in the launchd listing IS the in-progress signal, so a not-running guard is terminal
 * (done or dead, both safe) and a dead guard can never wedge dispatch. The r5 marker machinery below
 * survives only to classify PRE-r7 KeepAlive guards during the one-time upgrade transition.
 *
 * r12+r13 (§18): every "provably absent / provably foreign / provably empty" determination in this
 * module is TOTAL over its input domain and fails CLOSED on ambiguity, through exactly ONE audited
 * implementation per determination: marker presence = defaultMarkerPresent (lstat — only a clean
 * ENOENT is absence); launchd listing trust = readLaunchctlGuardSnapshot (exit 0 AND well-formed
 * shape AND our own live job present as the completeness anchor — r13); systemd listing trust =
 * parseSystemdGuardUnits (well-formed rows; an unknown ActiveState contends); ExecStart ownership =
 * the couldBeOurGuard / isProvablyOurGuard predicate PAIR with opposite fail-defaults (r13 — block
 * on any could-be, neutralize only on strict positive proof); numeric manager readbacks fully
 * anchored (parsePsEtime, parseMainPidValue); the crash-loop store counts DISTINCT pids (r13).
 */

// ERE matched against /api/health by the readiness probe inside the restart guard.
export const HEALTH_PATTERN = '"status"[[:space:]]*:[[:space:]]*"ok"';

// ─────────────────────────────────────────────────────────────────────────────
// #1509 r8 (Finding 3) — structural bounds so a hung external command can never leave the one-shot
// guard running/blocking. Every launchctl/systemctl invocation inside the guard is wrapped in a
// portable `bounded` timeout (no `timeout` binary on macOS), curl carries --connect-timeout/--max-time,
// and the verify loop is capped by a wall-clock DEADLINE — so the guard's TOTAL runtime is bounded by
// construction (see the ~worst-case sum in the design doc §13/§14), not merely by attempt count.
// ─────────────────────────────────────────────────────────────────────────────

/** Hard per-command timeout (seconds) for each launchctl/systemctl call INSIDE the guard script. */
export const GUARD_CMD_TIMEOUT_SECONDS = 10;
/** curl health-probe cap (seconds) — bounds every readiness probe in the verify loop. */
export const GUARD_CURL_MAX_SECONDS = 5;
/** curl connect timeout (seconds). */
export const GUARD_CURL_CONNECT_SECONDS = 3;
/**
 * Wall-clock DEADLINE (seconds) on the guard's restart+verify loop — the REAL bound (the pre-r8 loop
 * was ~30 × (5s curl + 2s sleep) ≈ 210s, NOT the ~60s the doc claimed). The loop now breaks once this
 * many seconds have elapsed regardless of how slow any single curl was.
 */
export const GUARD_VERIFY_DEADLINE_SECONDS = 60;
/**
 * Hard cap (seconds) on the WHOLE guard unit's wall-clock runtime. Passed to systemd as
 * `RuntimeMaxSec` so systemd itself SIGTERM/SIGKILLs a wedged guard transient unit even if every
 * in-script bound somehow failed. (No launchd equivalent exists, but the in-script per-command
 * `bounded` timeouts + loop deadline already bound the launchd guard by construction.)
 */
export const GUARD_MAX_RUNTIME_SECONDS = 120;
/** Timeout (ms) the ORCHESTRATOR applies to the guard-dispatch command (launchctl bootstrap / systemd-run). */
export const GUARD_DISPATCH_TIMEOUT_MS = 15_000;
/**
 * Timeout (ms) on every ORCHESTRATOR-side manager call (`launchctl list/remove`, `systemctl
 * list-units/show/stop/reset-failed`, `ps`) — #1509 r10 Finding 5. These calls are synchronous by
 * design (the precondition→swap sequence must have no await), so a hung manager binary would
 * otherwise wedge the command poller's event loop indefinitely. Bounded ⇒ worst case a few seconds
 * of stall, and the call reports failure (fail closed) instead of hanging.
 */
export const MANAGER_CMD_TIMEOUT_MS = 10_000;
/**
 * #1509 r10 (Finding 2) — the GENEROUS wall-clock bound (seconds) past which a RUNNING own guard is
 * judged hung and neutralized (launchd: `launchctl remove <exact label>`; systemd: `systemctl stop`
 * + `reset-failed <exact unit>`) instead of blocking dispatch forever. A legitimate guard finishes in
 * ≤ ~120s by construction (GUARD_MAX_RUNTIME_SECONDS; even the unbounded-ish r7/r8 generations were
 * nominally ≤ ~210s), so 15 minutes is ≥ 4× any legitimate restart+verify. Ambiguity NEVER reaps:
 * when the elapsed runtime cannot be read, the guard is treated as legitimately working (block).
 */
export const HUNG_GUARD_NEUTRALIZE_SECONDS = 900;

/**
 * Synchronous, HARD-BOUNDED external command (#1509 r10 Finding 5) — the orchestrator-side analogue
 * of the guard script's `bounded` helper. Every default manager adapter below routes through this,
 * so a hung `launchctl`/`systemctl`/`ps` can stall the event loop for at most `timeoutMs` (SIGKILL —
 * a manager stuck in the kernel may ignore SIGTERM) and then reports exit 1, which every caller
 * already treats fail-closed. Synchronous on purpose: the dispatch precondition must run with no
 * await between its final snapshot and the symlink swap.
 */
export function spawnSyncBounded(cmd: string[], timeoutMs = MANAGER_CMD_TIMEOUT_MS): { exitCode: number; stdout: string } {
  try {
    const r = Bun.spawnSync({ cmd, stdout: "pipe", stderr: "ignore", timeout: timeoutMs, killSignal: "SIGKILL" });
    return { exitCode: r.exitCode ?? 1, stdout: new TextDecoder().decode(r.stdout) };
  } catch {
    return { exitCode: 1, stdout: "" };
  }
}

/**
 * Parse `ps -o etime=` output (`[[dd-]hh:]mm:ss`) into elapsed seconds; null on anything else.
 * `etime` (not `etimes`) because macOS/BSD ps has no `etimes` keyword — this is the portable form.
 */
export function parsePsEtime(raw: string): number | null {
  const m = /^(?:(?:(\d+)-)?(\d+):)?(\d{1,2}):(\d{2})$/.exec(raw.trim());
  if (!m) return null;
  const days = m[1] ? Number(m[1]) : 0;
  const hours = m[2] ? Number(m[2]) : 0;
  return ((days * 24 + hours) * 60 + Number(m[3])) * 60 + Number(m[4]);
}

/**
 * STRICT `systemctl show -p MainPID --value` parse (#1509 r12 §18): the value must be EXACTLY one
 * positive integer. r11 used parseInt, which accepts a numeric PREFIX of garble ("123 456",
 * "123junk" ⇒ 123), so a corrupt exit-0 readback could yield a WRONG pid whose elapsed runtime then
 * fed the hung-guard bound — a potential false "hung" (and neutralize) on a young own guard.
 * Anything but a clean positive integer ⇒ null ⇒ elapsed unknown ⇒ block-without-neutralize.
 */
export function parseMainPidValue(raw: string): number | null {
  const trimmed = raw.trim();
  if (!/^\d+$/.test(trimmed)) return null;
  const pid = Number(trimmed);
  return Number.isSafeInteger(pid) && pid > 0 ? pid : null;
}

/** Elapsed wall-clock seconds of a live PID via bounded `ps -p <pid> -o etime=`; null when unknown. */
function processElapsedSecondsViaPs(pid: number): number | null {
  if (!Number.isInteger(pid) || pid <= 0) return null;
  const r = spawnSyncBounded(["ps", "-p", String(pid), "-o", "etime="]);
  if (r.exitCode !== 0) return null;
  return parsePsEtime(r.stdout);
}

/** Resolve the runtime symlink path from an optional detected prefix (shared default otherwise). */
export function resolveRuntimePath(runtimePrefix?: string): string {
  return runtimePrefix ?? join(homedir(), ".agent-relay", "runtime");
}

// ─────────────────────────────────────────────────────────────────────────────
// #1509 r5 — generation identity (§2): persisted monotonic sequence + collision-free uuid.
// ─────────────────────────────────────────────────────────────────────────────

/**
 * A self-upgrade guard generation (#1509 r5 §2). `genid` = `s${seq}.${uuidhex}` is the single
 * token stamped into EVERY per-generation path (script/status/result/label) and into the marker
 * content — so two guards can never collide on any path (uuid) and ordering is total (seq).
 */
export interface GuardGeneration {
  /** Monotonic persisted sequence — total ordering, immune to clock rollback + same-ms collision. */
  seq: number;
  /** crypto.randomUUID() with hyphens stripped (32 hex) — collision-free identity. */
  uuidhex: string;
  /** `s${seq}.${uuidhex}`. */
  genid: string;
}

/** The persisted monotonic-sequence file path (`<runtimePath>.upgrade-guard-seq`). */
export function upgradeGuardSeqPath(runtimePath: string): string {
  return `${runtimePath}.upgrade-guard-seq`;
}

/**
 * Allocate a fresh guard generation (#1509 r5 §2). MUST be called under the single-flight lock.
 *
 * ORDERING FROM A COUNTER, IDENTITY FROM A UUID — never a clock (§12.6). seq is read from the
 * persisted file (missing/corrupt ⇒ 0, so ordering simply restarts at 1 — §11.2, harmless because
 * the uuid still makes every generation identity distinct), incremented, and written back
 * atomically (temp + fsync + rename). uuidhex is a fresh random 32-hex string. Date.now() is NEVER
 * consulted for either the sequence or the identity.
 *
 * FAIL-CLOSED (§2.4): a failure to PERSIST the incremented counter (temp write / fsync / rename)
 * throws, so the caller aborts the dispatch rather than submitting a guard whose ordering was never
 * durably recorded. (A read failure is NOT fatal — it is the normal first-upgrade/reset case.)
 */
export function allocateGuardGeneration(runtimePath: string): GuardGeneration {
  const seqPath = upgradeGuardSeqPath(runtimePath);
  const uuidhex = randomUUID().replace(/-/g, "");
  let n = 0;
  try {
    const parsed = parseInt(readFileSync(seqPath, "utf8").trim(), 10);
    if (Number.isFinite(parsed) && parsed > 0) n = parsed;
  } catch {
    n = 0; // missing or corrupt seq file — bootstrap at 0 (§11.2); uuid keeps identity distinct.
  }
  const seq = n + 1;
  // Persist atomically; any throw here fails the whole allocation closed (do NOT dispatch).
  const tmpPath = `${seqPath}.tmp.${uuidhex}`;
  const fd = openSync(tmpPath, "w", 0o644);
  try {
    writeFileSync(fd, `${seq}\n`);
    fsyncSync(fd);
  } finally {
    closeSync(fd);
  }
  renameSync(tmpPath, seqPath);
  return { seq, uuidhex, genid: `s${seq}.${uuidhex}` };
}

/** Atomic file write: temp (fresh uuid) + fsync + rename over the target; chmod if a mode is given. */
export function atomicWriteFile(path: string, content: string, mode?: number): void {
  const tmpPath = `${path}.tmp.${randomUUID().replace(/-/g, "")}`;
  const fd = openSync(tmpPath, "w", mode ?? 0o644);
  try {
    writeFileSync(fd, content);
    fsyncSync(fd);
  } finally {
    closeSync(fd);
  }
  renameSync(tmpPath, path);
  if (mode !== undefined) chmodSync(path, mode);
}

// ─────────────────────────────────────────────────────────────────────────────
// #1509 r5 — per-generation path + label helpers (§4).
// ─────────────────────────────────────────────────────────────────────────────

/** The own-unit guard-label BASE (`${selfUnit}.upgrade-guard`) — the own-vs-foreign scoping key. */
export function ownUpgradeGuardBase(selfUnit: string): string {
  return `${selfUnit}.upgrade-guard`;
}

/** The exact per-generation launchd label a dispatch submits under (`<base>.<genid>`). */
export function guardLabelForGenid(base: string, genid: string): string {
  return `${base}.${genid}`;
}

/** The per-generation status-marker path (`<runtimePath>.upgrade-guard-status.<genid>`), or the
 * shared r4 marker path (`<runtimePath>.upgrade-guard-status`) when no genid is given. */
export function upgradeGuardStatusPathFor(runtimePath: string, genid?: string): string {
  return genid ? `${runtimePath}.upgrade-guard-status.${genid}` : `${runtimePath}.upgrade-guard-status`;
}

/** The per-generation guard-script path (`<runtimePath>.upgrade-guard.<genid>.sh`). */
export function upgradeGuardScriptPathFor(runtimePath: string, genid: string): string {
  return `${runtimePath}.upgrade-guard.${genid}.sh`;
}

/** The per-generation result-marker path (`<runtimePath>.upgrade-guard-result.<genid>`). */
export function upgradeGuardResultPathFor(runtimePath: string, genid: string): string {
  return `${runtimePath}.upgrade-guard-result.${genid}`;
}

/** The per-generation one-shot launchd plist path (`<runtimePath>.upgrade-guard.<genid>.plist`, #1509 r7 §13). */
export function upgradeGuardPlistPathFor(runtimePath: string, genid: string): string {
  return `${runtimePath}.upgrade-guard.${genid}.plist`;
}

/** Minimal XML text escape for the per-generation guard plist (label/path values, #1509 r7 §13). */
function xmlEscape(value: string): string {
  return value
    .replace(/&/g, "&amp;")
    .replace(/</g, "&lt;")
    .replace(/>/g, "&gt;")
    .replace(/"/g, "&quot;")
    .replace(/'/g, "&apos;");
}

/**
 * Build the per-generation ONE-SHOT launchd plist that runs the restart guard exactly once
 * (#1509 r7 §13). The critical properties versus the old `launchctl submit` job:
 *  - `RunAtLoad=true` starts the guard immediately at bootstrap.
 *  - `KeepAlive` is FALSE (explicit): once the guard process exits, launchd NEVER relaunches
 *    it. This is the whole point of the r7 redesign — a not-running one-shot job is TERMINAL
 *    (done or dead, indistinguishable and both safe to re-dispatch over), so there is no
 *    best-effort completion marker that can lie and no KeepAlive relaunch to wedge dispatch.
 * A fixed default PATH is set so the guard's bare-name tools (launchctl/curl/grep/mv/…) resolve
 * under launchd's otherwise-minimal environment.
 */
export function buildGuardPlist(label: string, scriptPath: string): string {
  return [
    '<?xml version="1.0" encoding="UTF-8"?>',
    '<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">',
    '<plist version="1.0">',
    "<dict><!-- agent-relay self-upgrade restart guard (#1509 r7) — generated, do not edit. -->",
    "  <key>Label</key>",
    `  <string>${xmlEscape(label)}</string>`,
    "  <key>ProgramArguments</key>",
    "  <array>",
    "    <string>/bin/sh</string>",
    `    <string>${xmlEscape(scriptPath)}</string>`,
    "  </array>",
    "  <key>RunAtLoad</key>",
    "  <true/>",
    // KeepAlive FALSE: a one-shot that never relaunches — the structural fix for #1509.
    "  <key>KeepAlive</key>",
    "  <false/>",
    "  <key>ProcessType</key>",
    "  <string>Background</string>",
    "  <key>EnvironmentVariables</key>",
    "  <dict>",
    "    <key>PATH</key>",
    "    <string>/usr/bin:/bin:/usr/sbin:/sbin</string>",
    "  </dict>",
    "</dict>",
    "</plist>",
    "",
  ].join("\n");
}

// ─────────────────────────────────────────────────────────────────────────────
// #1509 r5 — the restart guard shell script (§8).
// ─────────────────────────────────────────────────────────────────────────────

/**
 * Build the full POSIX-sh restart guard as a self-contained script. It is written to a
 * file and dispatched as `sh <file>`, which is what makes the shell VALID and survivable:
 * the prior version string-joined fragments with "; " and emitted `… do; if curl …`, a
 * hard parse error (sh AND bash, exit 2). `sh -c` parses the whole string before running,
 * so the guard died at PARSE and the leading restart never executed — the upgrade swapped
 * the runtime symlink and silently restarted NOTHING while reporting success (#1315).
 *
 * The guard now:
 *  - gates on the new runtime's ExecStart target being executable (test -x) and rolls the
 *    symlink back WITHOUT restarting if it is missing (the #1312 203/EXEC landmine);
 *  - proves the restart actually replaced the process — captures MainPID before, requires
 *    a different nonzero MainPID after, because the OLD process serves /api/health fine;
 *  - requires a stable-healthy /api/health in addition to the PID change;
 *  - on success drops the rollback pointer AND the parked-aside original runtime (.legacy);
 *  - on any failure restores the previous runtime (single atomic rename over the live
 *    symlink — #1287 Bug B), restarts onto it, and exits nonzero;
 *  - always records an `ok|rolled-back <ts>` marker (atomic temp+rename — #1509 r5 §5) so
 *    verification can observe the outcome after the orchestrator reconnects, instead of
 *    trusting the dispatch exit code.
 *  - on launchd, runs as a ONE-SHOT bootstrapped job (`RunAtLoad`, `KeepAlive=false` — #1509
 *    r7 §13), so once it exits launchd NEVER relaunches it. It still BOOTS ITSELF OUT before
 *    every exit path (`launchctl bootout`, with `remove` as a fallback, plus an EXIT trap) to
 *    keep the launchd listing clean — but this is now purely COSMETIC: a one-shot that fails to
 *    self-remove sits loaded-but-idle at PID `-` and can neither loop nor wedge a later dispatch
 *    (a not-running one-shot is terminal). This replaces the old KeepAlive `launchctl submit`
 *    job, whose failure-to-self-remove re-ran the guard forever (~every 10s), SIGTERM-restart-
 *    looping the orchestrator (#1509). No-op on systemd (self-collecting transient unit).
 *  - carries NO status/completion marker (#1509 r7 §13). The old `active|done <genid>` marker was
 *    a best-effort completion signal that could lie (a guard that died before writing `done` left
 *    a forever-`active` marker that wedged all future upgrades). Under the one-shot model the job's
 *    PRESENCE-AND-RUNNING state in the launchd listing IS the in-progress signal, so no marker is
 *    needed and none is written. The `ok|rolled-back` RESULT marker (below) is unrelated — it is
 *    read by the orchestrator to REPORT the outcome, never to gate reaping or dispatch.
 */
export function buildArtifactRestartGuard(
  platform: "systemd" | "launchd",
  unit: string,
  runtimePath: string,
  readinessUrl: string,
  service?: string,
  guardLabel?: string,
  guardDomainTarget?: string,
  resultPath?: string,
  plistPath?: string,
): string {
  const previous = `${runtimePath}.previous`;
  const legacy = `${runtimePath}.legacy`;
  // systemd keeps the shared result path; launchd passes a per-generation one (#1509 r5 §4).
  const result = resultPath ?? `${runtimePath}.upgrade-guard-result`;
  // The unit's ExecStart / ProgramArguments target, resolved through the NEW runtime
  // symlink. materializeArtifactBins links it and chmod +x's the source; its absence is
  // the #1312 203/EXEC crash-loop, so gate on it before we ever tear the service down.
  const execTarget = join(runtimePath, "node_modules", ".bin", "agent-relay-orchestrator");
  const healthUrl = `${readinessUrl}/api/health`;
  // GNU mv needs -T to replace the symlink itself; BSD mv needs -h. Either way rollback is
  // a single atomic rename over the live pointer — .previous is always a symlink, so a
  // prior real directory rolls back without ENOTDIR and never through a no-pointer hole.
  const mvFlag = platform === "systemd" ? "-fT" : "-fh";
  const T = GUARD_CMD_TIMEOUT_SECONDS;
  // #1509 r8 (Finding 3): every launchctl/systemctl call runs under `bounded` so a hung invocation
  // cannot leave the guard running (and thus blocking dispatch) forever.
  const pidFn = platform === "systemd"
    ? `unit_pid() { bounded ${T} systemctl --user show -p MainPID --value "$UNIT" 2>/dev/null || echo 0; }`
    : `unit_pid() { bounded ${T} launchctl list "$UNIT" 2>/dev/null | sed -n 's/.*"PID" = \\([0-9][0-9]*\\).*/\\1/p'; }`;
  const restartFn = platform === "systemd"
    ? `restart_unit() { bounded ${T} systemctl --user restart "$UNIT"; }`
    : `restart_unit() { bounded ${T} launchctl kickstart -kp "$SERVICE"; }`;
  // #1509 r7 §13: on launchd the guard runs as a ONE-SHOT bootstrapped job (KeepAlive=false), so
  // self-removal is now COSMETIC (list hygiene) rather than loop-critical — but we still do it.
  // `launchctl bootout` is the counterpart to `bootstrap`; `remove` is a legacy fallback; the
  // per-generation plist file is deleted too. No-op on systemd (self-collecting transient unit).
  // #1509 r8: bootout/remove are `bounded` too, so a hung self-remove cannot leave the guard alive.
  const selfRemoveFn = platform === "launchd"
    ? `self_remove() { bounded ${T} launchctl bootout ${shellQuote(guardDomainTarget ?? "")} 2>/dev/null || bounded ${T} launchctl remove ${shellQuote(guardLabel ?? "")} 2>/dev/null || true; rm -f ${shellQuote(plistPath ?? "")} 2>/dev/null || true; }`
    : "self_remove() { :; }";
  return [
    "#!/bin/sh",
    `# agent-relay self-upgrade restart guard (${platform}) — generated, do not edit.`,
    "set -u",
    `RUNTIME=${shellQuote(runtimePath)}`,
    `PREVIOUS=${shellQuote(previous)}`,
    `LEGACY=${shellQuote(legacy)}`,
    `UNIT=${shellQuote(unit)}`,
    `SERVICE=${shellQuote(service ?? "")}`,
    `EXEC_TARGET=${shellQuote(execTarget)}`,
    `HEALTH_URL=${shellQuote(healthUrl)}`,
    `HEALTH_PATTERN=${shellQuote(HEALTH_PATTERN)}`,
    `RESULT=${shellQuote(result)}`,
    // #1509 r8 (Finding 3): portable per-command hard timeout — no `timeout` binary exists on macOS,
    // so run the command in the background and a detached watchdog TERM→KILLs it if it overruns. The
    // command's stdout is preserved (it inherits our fds) so `$(bounded … launchctl list …)` still
    // captures output; the watchdog is fully redirected so it never pollutes or holds the pipe open.
    //
    // #1509 r9 (Finding 4): the bounded command runs in its OWN PROCESS GROUP so a hung CHILD (and its
    // descendants) is signalled as a group. The r8 `bounded` TERM/KILLed only the immediate PID, so a
    // surviving child kept a command-substitution pipe open past the bound (repro: `bounded 1 sh -c
    // "printf hello; sleep 5"` took ~5s, not ~1s), and manager-side jobs a killed systemctl/launchctl
    // had already forked lived on. We prefer `setsid` (always present on Linux/systemd) to start a new
    // session+group; on macOS there is no `setsid` binary, but /bin/sh is bash, so we fall back to
    // monitor mode (`set -m`), which puts each backgrounded job in its own group. Either way `$!` is
    // the group leader (PGID == PID), so `kill -TERM -$pid` signals the WHOLE tree; the leader PID is
    // also signalled directly as a belt-and-suspenders if no separate group was created. stdout/exit
    // code are preserved (the command inherits our fds).
    "if command -v setsid >/dev/null 2>&1; then _AR_SETSID=1; else _AR_SETSID=0; fi",
    "bounded() {",
    "  _bt=$1; shift",
    '  if [ "$_AR_SETSID" = 1 ]; then',
    '    setsid "$@" &',
    "  else",
    '    set -m 2>/dev/null; "$@" &',
    "    set +m 2>/dev/null",
    "  fi",
    "  _bp=$!",
    '  ( sleep "$_bt"; kill -TERM "-$_bp" 2>/dev/null; kill -TERM "$_bp" 2>/dev/null; sleep 1; kill -KILL "-$_bp" 2>/dev/null; kill -KILL "$_bp" 2>/dev/null ) >/dev/null 2>&1 &',
    "  _bw=$!",
    '  wait "$_bp" 2>/dev/null',
    "  _brc=$?",
    '  kill "$_bw" 2>/dev/null',
    '  wait "$_bw" 2>/dev/null',
    '  return "$_brc"',
    "}",
    // Monotonic-ish wall clock for the verify-loop deadline; `echo 0` disables the deadline if date is absent.
    "now_s() { date +%s 2>/dev/null || echo 0; }",
    // Result marker written atomically (temp + rename — §5) so a reader never sees a torn line.
    "mark() { printf '%s %s\\n' \"$1\" \"$(date -u +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || echo unknown)\" > \"$RESULT.tmp.$$\" 2>/dev/null && mv -f \"$RESULT.tmp.$$\" \"$RESULT\" 2>/dev/null || true; }",
    pidFn,
    restartFn,
    selfRemoveFn,
    // #1509 r7 §13: boot our own one-shot job out on ANY exit (belt to the explicit self_remove
    // calls below) so the launchd listing stays clean. Cosmetic under KeepAlive=false (a one-shot
    // that skips this sits idle, never loops). Harmless on systemd. There is deliberately NO
    // status/completion marker: the job's presence-and-running state IS the in-progress signal.
    "trap 'self_remove' EXIT",
    // Pre-restart exec-ability gate: missing target → roll back and exit WITHOUT restarting.
    'if [ ! -x "$EXEC_TARGET" ]; then',
    `  mv ${mvFlag} "$PREVIOUS" "$RUNTIME"`,
    "  mark rolled-back",
    "  self_remove",
    "  exit 1",
    "fi",
    "old_pid=$(unit_pid)",
    '[ -n "$old_pid" ] || old_pid=0',
    "restart_unit",
    "ready=0",
    "new_pid=0",
    "attempt=0",
    // #1509 r8 (Finding 3): bound the verify loop by WALL CLOCK, not just attempt count. Even if
    // every curl takes its full --max-time, the loop exits at the deadline (~60s), so the guard's
    // total runtime is bounded by construction and a slow probe can never keep it running/blocking.
    "loop_start=$(now_s)",
    `loop_deadline=$((loop_start + ${GUARD_VERIFY_DEADLINE_SECONDS}))`,
    'while [ "$attempt" -lt 30 ]; do',
    "  attempt=$((attempt + 1))",
    '  if [ "$loop_start" != "0" ]; then now=$(now_s); [ "$now" != "0" ] && [ "$now" -ge "$loop_deadline" ] && break; fi',
    "  new_pid=$(unit_pid)",
    '  [ -n "$new_pid" ] || new_pid=0',
    `  if [ "$new_pid" != "0" ] && [ "$new_pid" != "$old_pid" ] && curl -fsS --connect-timeout ${GUARD_CURL_CONNECT_SECONDS} --max-time ${GUARD_CURL_MAX_SECONDS} "$HEALTH_URL" | grep -Eq "$HEALTH_PATTERN"; then`,
    "    ready=$((ready + 1))",
    "  else",
    "    ready=0",
    "  fi",
    '  [ "$ready" -ge 2 ] && break',
    "  sleep 2",
    "done",
    'if [ "$ready" -ge 2 ] && [ "$new_pid" != "0" ] && [ "$new_pid" != "$old_pid" ]; then',
    '  rm -rf "$PREVIOUS" "$LEGACY"',
    "  mark ok",
    "  self_remove",
    "  exit 0",
    "fi",
    // Restart never took or the process was not replaced: restore the previous runtime,
    // restart onto it, record the rollback, and fail loud.
    `mv ${mvFlag} "$PREVIOUS" "$RUNTIME"`,
    "restart_unit",
    "mark rolled-back",
    "self_remove",
    "exit 1",
  ].join("\n");
}

// ─────────────────────────────────────────────────────────────────────────────
// #1509 r5 — reaper: job model (§6.1), marker (§6.2), decision (§6.4, §7).
// ─────────────────────────────────────────────────────────────────────────────

/** The generation-scheme a guard label belongs to (#1509 r5 §6.1). */
export type GuardScheme = "r5" | "r4" | "legacy";

/**
 * One self-upgrade restart-guard launchd job from a SINGLE `launchctl list` snapshot (#1509 r5 §6.1),
 * parsed into its durable identity. PID is captured ONLY as a boolean `running` (positive PID column)
 * from the SAME line — never as an identity/ordering key (the r1–r3 mistake). Identity comes from the
 * label's genid (r5) / epoch (r4); completion comes from the per-generation marker. `base` is the
 * `${unit}.upgrade-guard` prefix used for own-vs-foreign scoping.
 */
export interface UpgradeGuardJob {
  /** The exact launchd label. */
  label: string;
  /** The `${unit}.upgrade-guard` prefix (own-vs-foreign scoping key). */
  base: string;
  /** Which generation scheme the label matches. */
  scheme: GuardScheme;
  /** PID column is a positive integer (not "-") in the same snapshot line. */
  running: boolean;
  /**
   * The PID from that same snapshot line, ONLY when running (#1509 r10 Finding 2). NEVER an
   * identity/ordering key (the r1–r3 mistake stands) — used solely to measure the process's
   * elapsed wall-clock runtime for the hung-guard bound, and every action taken on the job still
   * targets its exact LABEL via the manager (never a `kill <pid>`), so PID reuse can at worst
   * misread a duration, never touch a foreign process.
   */
  pid?: number;
  /** r5 only: `s${seq}.${uuidhex}` — matches the label suffix AND the marker content token. */
  genid?: string;
  /** r5 only: the monotonic sequence. */
  seq?: number;
  /** r5 only: the 32-hex uuid. */
  uuidhex?: string;
  /** r4 only: the `.g<epoch>` generation digits (string) — matches the shared marker content token. */
  epoch?: string;
}

// Anchored per-scheme label classifiers (#1509 r5 §6.1). Mutually exclusive by construction.
const GUARD_R5_LABEL_RE = /^(.*\.upgrade-guard)\.s(\d+)\.([0-9a-f]{32})$/;
const GUARD_R4_LABEL_RE = /^(.*\.upgrade-guard)\.g(\d+)$/;
const GUARD_LEGACY_LABEL_RE = /^(.*\.upgrade-guard)$/;

/**
 * Parse `launchctl list` output (tab-separated `PID\tStatus\tLabel` per line) into a
 * `{label, base, scheme, running, …}` for EVERY self-upgrade guard job. Raw/unscoped: it returns
 * foreign guards too (another orchestrator's, an unrelated app's) — scoping to the own-unit
 * base is the caller's job (see reapStaleUpgradeGuards).
 */
export function parseUpgradeGuardJobs(launchctlListOutput: string): UpgradeGuardJob[] {
  const jobs: UpgradeGuardJob[] = [];
  for (const line of launchctlListOutput.split("\n")) {
    const cols = line.split("\t");
    if (cols.length < 3) continue;
    const label = cols[2];
    if (!label) continue;
    const pidRaw = (cols[0] ?? "").trim();
    const running = /^\d+$/.test(pidRaw) && Number(pidRaw) > 0;
    const pid = running ? Number(pidRaw) : undefined;
    let m = GUARD_R5_LABEL_RE.exec(label);
    if (m) {
      jobs.push({ label, base: m[1]!, scheme: "r5", running, pid, seq: Number(m[2]), uuidhex: m[3]!, genid: `s${m[2]}.${m[3]}` });
      continue;
    }
    m = GUARD_R4_LABEL_RE.exec(label);
    if (m) {
      jobs.push({ label, base: m[1]!, scheme: "r4", running, pid, epoch: m[2]! });
      continue;
    }
    m = GUARD_LEGACY_LABEL_RE.exec(label);
    if (m) {
      jobs.push({ label, base: m[1]!, scheme: "legacy", running, pid });
      continue;
    }
  }
  return jobs;
}

/**
 * #1509 r12 (Finding 3) — "provably absent from the launchd listing" requires a WELL-FORMED listing,
 * not merely exit 0. r11 fed exit-0 stdout straight into the row extractor, so garbled/truncated/
 * foreign-shaped output (a partial write, an error banner, binary junk) parsed to ZERO rows and read
 * as proof that no guard exists — in the dispatch precondition, in its confirm-absent recheck, and in
 * the post-dispatch fence probe — admitting a swap (or a fence-lowering rollback) beside a
 * live-but-invisible guard. A TRUSTED listing requires EVERY non-empty line to be either the
 * `PID\tStatus\tLabel` header or a `<pid|-> <status> <label>` row. Anything else ⇒ null ⇒ the caller
 * fails closed. A well-formed, anchored listing that simply contains none of OUR guard labels is the
 * legitimate "zero own guards" case and still parses to [].
 *
 * #1509 r13 (Finding 2) — well-formedness alone is NOT completeness. A truncated or wrong-domain
 * exit-0 snapshot whose surviving lines are all individually valid (a bare header, a few foreign
 * rows) passed the r12 shape check and still read as proof of absence — beside an OMITTED live
 * guard. A listing proves anything about OUR guards only when it ALSO contains this orchestrator's
 * OWN job (`selfUnit`) as a LIVE row (positive PID): every one of these code paths runs inside that
 * very process, so a truthful, complete view of our user domain MUST show it. Its absence — or a
 * PID of `-`/0 on it — means the snapshot is not a trustworthy complete view ⇒ null (fail closed).
 * The anchor also subsumes the r12 "never empty" rule (an anchored listing is necessarily
 * non-empty).
 */
export function parseLaunchctlGuardListing(output: string, selfUnit: string): UpgradeGuardJob[] | null {
  const lines = output.split("\n").filter((line) => line.trim() !== "");
  if (lines.length === 0) return null;
  let ownJobLive = false;
  for (const line of lines) {
    if (line === "PID\tStatus\tLabel") continue;
    const cols = line.split("\t");
    if (cols.length < 3 || !cols[2]) return null;
    const pid = cols[0]!.trim();
    if (!/^(-|\d+)$/.test(pid) || !/^(-|-?\d+)$/.test(cols[1]!.trim())) return null;
    if (cols[2] === selfUnit && /^\d+$/.test(pid) && Number(pid) > 0) ownJobLive = true;
  }
  if (!ownJobLive) return null; // r13 Finding 2: no live own-job anchor ⇒ not a complete view
  return parseUpgradeGuardJobs(output);
}

/**
 * THE single trusted launchd-snapshot reader (#1509 r12 §18; own-job-anchored r13 Finding 2):
 * every production consumer of `launchctl list` in the guard routes through this. Exit 0 AND a
 * well-formed shape AND our own live job present ⇒ jobs; anything else ⇒ null, which each caller
 * treats as an untrusted snapshot in its own fail-closed direction (dispatch: throw; fence probe:
 * keep the fence up; cosmetic reaper: no-op).
 */
export function readLaunchctlGuardSnapshot(launchctl: LaunchctlControl, selfUnit: string): UpgradeGuardJob[] | null {
  const list = launchctl.list();
  return list.exitCode === 0 ? parseLaunchctlGuardListing(list.stdout, selfUnit) : null;
}

/**
 * The self-upgrade guard's completion/liveness marker content (#1509 r5), parsed from
 * `<phase> <token>`. `phase` is `active` while the guard is doing its restart/verify/rollback
 * work and `done` once it has finished (recorded right before self-remove). `token` is the
 * genid (r5) or the epoch digits (r4) and MUST match the job's own token for the marker to
 * count (defensive — §4). This — NOT a PID, NOT a settle timer — tells the reaper a lingering
 * job apart from a live one.
 */
export interface GuardMarker {
  phase: "active" | "done";
  /** The generation token embedded in the marker content (genid for r5, epoch digits for r4). */
  token: string;
}

/**
 * Parse a `<active|done> <token>` status marker; null if absent/empty/malformed (#1509 r7).
 *
 * FULLY ANCHORED + SINGLE-LINE (fail-closed). The trimmed content must be EXACTLY
 * `<active|done> <token>` — the end anchor rejects trailing junk (`done <genid> <junk>`) and,
 * with no `m` flag, a second line (`done\nactive`) fails to match entirely. r6's unanchored
 * `^(active|done)\s+(\S+)` matched both of those as `done` (MEDIUM fail-open — a competing swap
 * slipped through). Anything that is not exactly one clean `<phase> <token>` line ⇒ null, which
 * every caller treats as "not a completion signal" (never counts as `done`).
 *
 * NOTE: after the r7 one-shot redesign (§13) r7 guards write NO status marker at all; this parser
 * survives only to classify PRE-r7 (KeepAlive) guards during the one-time upgrade transition, and
 * it must be fail-closed so a malformed leftover marker can never be read as a completion.
 */
export function parseGuardMarker(raw: string | null | undefined): GuardMarker | null {
  if (!raw) return null;
  const m = /^(active|done)[ \t]+(\S+)[ \t]*$/.exec(raw.trim());
  if (!m) return null;
  return { phase: m[1] as "active" | "done", token: m[2]! };
}

/** A path-keyed marker reader (injectable for tests); default reads the file, null on any error. */
export type MarkerReader = (path: string) => string | null;

export const defaultReadMarker: MarkerReader = (path) => {
  try {
    return readFileSync(path, "utf8");
  } catch {
    return null;
  }
};

/**
 * A path-keyed marker CLEARER (injectable for tests). #1509 r8 (Finding 2): a stale pre-r7/legacy
 * `active` marker must never PERMANENTLY block dispatch, so the dispatch precondition clears it
 * in-place (best-effort). Default unlinks the file; a missing file is a no-op.
 */
export type MarkerClearer = (path: string) => void;

export const defaultClearMarker: MarkerClearer = (path) => {
  try {
    unlinkSync(path);
  } catch {
    // best-effort self-heal — a missing/undeletable marker is harmless (it no longer blocks either).
  }
};

/**
 * A path-keyed marker PRESENCE probe (#1509 r11 Finding 1; TOTAL over the filesystem domain r12
 * Finding 2), injectable for tests. It distinguishes "marker provably absent" (no directory entry)
 * from "marker present or unprovable". The two must fail in OPPOSITE directions: provable absence is
 * the only POSITIVE proof that an r5-labelled job is an r7+ one-shot leftover (r7+ guards write no
 * marker), while anything else is AMBIGUITY and must classify the job as a possible pre-r7 KeepAlive
 * guard (boot it out).
 *
 * #1509 r12 (Finding 2): r11 probed with existsSync, which FOLLOWS symlinks and returns false on any
 * error — so a marker that was a DANGLING symlink, a symlink into an unreadable directory, an
 * EACCES/EPERM path, or an ENOTDIR path all read "provably absent ⇒ r7+ ⇒ leave", and an idle pre-r7
 * KeepAlive guard in any of those states could relaunch across the swap. The probe is now lstat over
 * the ENTRY itself: ANY existing entry — regular file, symlink (dangling included), directory, fifo
 * — is present, and of the failure modes ONLY the kernel's clean positive "no such entry" (ENOENT)
 * is absence. EACCES/EPERM (cannot look), ENOTDIR (a path component is not a directory), EIO/ELOOP,
 * and every other error are ambiguity ⇒ present (fail closed).
 */
export type MarkerPresence = (path: string) => boolean;

export const defaultMarkerPresent: MarkerPresence = (path) => {
  try {
    lstatSync(path); // lstat, NOT stat/existsSync: a dangling symlink is still an existing entry
    return true;
  } catch (error) {
    return (error as NodeJS.ErrnoException).code !== "ENOENT";
  }
};

/** Reconstruct a job's OWN status-marker path from its scheme (per-generation r5; shared r4; none for legacy). */
function ownGuardMarkerPath(job: UpgradeGuardJob, runtimePath: string): string | null {
  if (job.scheme === "r5") return job.genid ? upgradeGuardStatusPathFor(runtimePath, job.genid) : null;
  if (job.scheme === "r4") return upgradeGuardStatusPathFor(runtimePath);
  return null; // legacy: no marker
}

/**
 * Read + match a job's OWN marker (#1509 r5 §6.2). Reconstructs the marker path from the job's
 * scheme (per-generation for r5; the shared marker for r4; none for legacy) and returns the
 * marker ONLY when its embedded token matches the job's own token — so a reaper that reconstructs
 * a path from a label can never act on a marker that belongs to a different generation.
 */
function readOwnGuardMarker(job: UpgradeGuardJob, runtimePath: string, readMarker: MarkerReader): GuardMarker | null {
  if (job.scheme === "r5") {
    const marker = parseGuardMarker(readMarker(upgradeGuardStatusPathFor(runtimePath, job.genid)));
    return marker && marker.token === job.genid ? marker : null;
  }
  if (job.scheme === "r4") {
    const marker = parseGuardMarker(readMarker(upgradeGuardStatusPathFor(runtimePath)));
    return marker && marker.token === job.epoch ? marker : null;
  }
  return null; // legacy: no marker
}

/**
 * Decide whether one own-unit guard job is genuinely STALE (safe + necessary to reap). Under the
 * r7 one-shot model (§13) reaping is COSMETIC — a not-running one-shot job never loops and never
 * wedges dispatch, so removing it is only listing hygiene — but the same predicate also safely
 * cleans up PRE-r7 KeepAlive guards during the one-time transition. The discriminator is still
 * RUNNING-STATE (+ the pre-r7 marker where present), never a PID-as-identity, never a settle timer.
 *
 *  - r7 (`.s<seq>.<uuid>`, NO status marker): running-state only. not running → REAP (terminal
 *    one-shot); running → LEAVE (a bounded, genuinely-working guard). This is the `readOwnGuardMarker
 *    → null → return !job.running` path.
 *  - r5/r6 (`.s<seq>.<uuid>`, pre-r7 status marker present — transition): [`done <genid>`, either
 *    running] → REAP; [`active <genid>`, running] → LEAVE; [`active <genid>`, not running] → REAP
 *    (health-proven); [absent/mismatched marker] → running-state.
 *  - r4 (`.g<epoch>`, shared marker — transition, §7): [`done <epoch>` matching] → REAP;
 *    [`active <epoch>` matching] → LEAVE; [absent/mismatched] → running-state.
 *  - legacy bare (`^…upgrade-guard$`, no marker — §7): running-state only.
 *
 * INVARIANT across all schemes: a guard that is `active` AND running (or any running guard with no
 * completion signal) is NEVER reaped.
 */
function guardIsReapable(job: UpgradeGuardJob, runtimePath: string, readMarker: MarkerReader): boolean {
  const marker = readOwnGuardMarker(job, runtimePath, readMarker);
  if (marker?.phase === "done") return true;
  if (job.scheme === "r4" && marker?.phase === "active") return false; // §7: r4 active matching → LEAVE
  // r5 active, or any scheme with no (matching) marker: reap only when NOT running (health-proven).
  return !job.running;
}

/**
 * Marker EVIDENCE for the pre-r7 classification (#1509 r11 Finding 1) — a tri-state, never a bare
 * null. r10 collapsed "no valid matching marker" into one bucket, so a marker that was PRESENT but
 * corrupt/unreadable/token-mismatched read the same as a marker that provably never existed — and
 * the job was left in place as an "r7+ one-shot leftover" when it could equally be a pre-r7
 * KeepAlive guard whose marker rotted. The states:
 *  - `matching`: a clean `<phase> <token>` whose token equals the job's own — positive r5/r6 dating.
 *  - `absent`: the marker file provably does not exist — the ONLY positive proof of r7+ (which
 *    writes no marker at all).
 *  - `ambiguous`: present-but-unreadable (reader null yet the file exists), present-but-malformed
 *    (parse fails), or token-mismatched content at the per-generation path. Cannot prove r7+ ⇒
 *    must be treated as a possible pre-r7 guard (fail closed).
 */
type OwnMarkerEvidence = "matching" | "absent" | "ambiguous";

function ownGuardMarkerEvidence(
  job: UpgradeGuardJob,
  runtimePath: string,
  readMarker: MarkerReader,
  markerPresent: MarkerPresence,
): OwnMarkerEvidence {
  const path = ownGuardMarkerPath(job, runtimePath);
  if (path === null) return "absent"; // legacy scheme has no marker path at all
  const raw = readMarker(path);
  if (raw === null) return markerPresent(path) ? "ambiguous" : "absent";
  const marker = parseGuardMarker(raw);
  if (!marker) return "ambiguous"; // present but malformed/corrupt content
  const token = job.scheme === "r5" ? job.genid : job.epoch;
  return marker.token === token ? "matching" : "ambiguous";
}

/**
 * True iff an own-unit job is a PRE-r7 KeepAlive guard that could still (re)launch (#1509 r9
 * Finding 2; BROADENED r10 Finding 1; FAIL-CLOSED ON MARKER AMBIGUITY r11 Finding 1). Pre-r7 guards
 * were created by `launchctl submit`, which KeepAlives them: even listed at PID `-` they can
 * relaunch at any moment, so before a swap every one of them must be BOOTED OUT — never merely
 * marker-cleared, never left in place. Identity is EXACT, never a substring: the anchored label
 * classifiers plus the own-`base` equality the caller already applied (the launchd counterpart of
 * systemd's exact ExecStart match — r10 Finding 4).
 *
 *  - `legacy` bare `<base>` and `r4` `<base>.g<epoch>` labels are pre-r7 BY SCHEME — only pre-r4
 *    code ever submitted them, marker or no marker. The r9 predicate required an `active` marker,
 *    which legacy jobs never had, so a bare KeepAlive job slipped past the bootout and could
 *    relaunch between the final snapshot and the swap — the exact r10 Finding 1 double-swap
 *    window. Scheme alone suffices.
 *  - an `r5` label whose marker evidence is `matching` is an r5/r6 KeepAlive-era guard (r7+ writes
 *    none, so the marker's existence dates it), and one whose evidence is `ambiguous` (present but
 *    corrupt/unreadable/mismatched — r11 Finding 1) CANNOT be proven r7+ — both are booted out.
 *  - only an `r5` label whose marker is provably ABSENT is an r7+ one-shot leftover: terminal,
 *    never relaunches, safe to leave.
 */
function isRelaunchablePreR7Guard(
  job: UpgradeGuardJob,
  runtimePath: string,
  readMarker: MarkerReader,
  markerPresent: MarkerPresence,
): boolean {
  if (job.scheme === "legacy" || job.scheme === "r4") return true;
  return ownGuardMarkerEvidence(job, runtimePath, readMarker, markerPresent) !== "absent";
}

/**
 * #1509 r11 (Finding 2) — the crash-loop relaunch detector's persisted observation store.
 *
 * A pre-r7 KeepAlive guard whose invocations crash and relaunch in under
 * HUNG_GUARD_NEUTRALIZE_SECONDS is observed RUNNING with a fresh young PID on every snapshot, so
 * the elapsed-based hung bound never fires, the Finding-1 bootout (idle jobs only) never applies,
 * and pre-r11 it wedged dispatch forever. But relaunch itself is PROOF of pre-r7 KeepAlive: every
 * r7+ label is bootstrapped exactly once (uuid identity) with KeepAlive=false, so the SAME own
 * label observed running under TWO DISTINCT PIDs has provably been relaunched by launchd — only
 * `launchctl submit` KeepAlive jobs can do that — and booting it out by exact label is precisely
 * the action every pre-r7 KeepAlive job gets. The store persists distinct-PID observations per
 * label across snapshots AND across orchestrator restarts (the crash-looping guard SIGTERM-restarts
 * the orchestrator every ~10s, so in-memory state would die with each loop — which is why this is a
 * file). All I/O is best-effort and one-directional: a broken store degrades to "block without
 * neutralizing" (fail closed) — it can stretch the bound, never produce a false neutralize.
 */
export interface GuardPidObservationStore {
  /** Record `pid` observed RUNNING under `label`; returns how many DISTINCT pids that label has shown. */
  recordRunningPid(label: string, pid: number): number;
  /** Forget labels no longer present in the listing (an absent label is unloaded and cannot relaunch). */
  prune(presentLabels: readonly string[]): void;
}

/** Distinct running PIDs on one label that PROVE it relaunches (2 ⇒ at least one relaunch happened). */
export const RELAUNCH_PROOF_DISTINCT_PIDS = 2;

/** The persisted crash-loop observation-store path (`<runtimePath>.upgrade-guard-observed-pids`). */
export function upgradeGuardObservedPidsPath(runtimePath: string): string {
  return `${runtimePath}.upgrade-guard-observed-pids`;
}

/** The default file-backed observation store (JSON `{ [label]: pid[] }`, atomic best-effort writes). */
export function fileGuardPidObservationStore(runtimePath: string): GuardPidObservationStore {
  const path = upgradeGuardObservedPidsPath(runtimePath);
  const load = (): Record<string, number[]> => {
    try {
      const parsed: unknown = JSON.parse(readFileSync(path, "utf8"));
      if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return {};
      const state: Record<string, number[]> = {};
      for (const [label, pids] of Object.entries(parsed)) {
        // #1509 r13 (Finding 3): DEDUP + sanitize on load. The relaunch threshold counts DISTINCT
        // pids, so a duplicate-laden/corrupt store (`[4242,4242]`) must never fabricate a second
        // distinct observation — pre-r13 its raw length met the threshold and neutralized a
        // legitimate one-shot on its FIRST observation. Suspicious entries degrade to fewer
        // observations (block-not-neutralize), never more.
        if (Array.isArray(pids)) state[label] = [...new Set(pids.filter((p): p is number => Number.isInteger(p) && p > 0))];
      }
      return state;
    } catch {
      return {}; // missing/corrupt store: history restarts — stretches the bound, never mis-neutralizes
    }
  };
  const save = (state: Record<string, number[]>): void => {
    try {
      atomicWriteFile(path, `${JSON.stringify(state)}\n`);
    } catch {
      // best-effort persistence — an unwritable store degrades to block-without-neutralize.
    }
  };
  return {
    recordRunningPid(label, pid) {
      const state = load();
      const pids = state[label] ?? []; // already deduped+sanitized by load (r13 Finding 3)
      if (pids.includes(pid)) return pids.length; // distinct count — a re-observation adds nothing
      state[label] = [...pids, pid].slice(-8); // cap per-label history
      save(state);
      return state[label]!.length;
    },
    prune(presentLabels) {
      const state = load();
      const keep = new Set(presentLabels);
      let changed = false;
      for (const label of Object.keys(state)) {
        if (!keep.has(label)) {
          delete state[label];
          changed = true;
        }
      }
      if (changed) save(state);
    },
  };
}

/**
 * Cross-process dispatch precondition (#1509 r5 §3, §8.2; hardened #1509 r6 defect 2, r8 Finding 2,
 * r9 Finding 2, r10 Findings 1+2). Guarantees that NO own self-upgrade guard can contend on — or be
 * (re)launched onto — the shared runtime symlink between this check and the swap, AND that no hung or
 * leftover guard can block dispatch forever. FAIL-CLOSED throughout.
 *
 * The rule set:
 *  1. Any RUNNING own-unit guard blocks (throws). A genuinely-working guard (r7+ one-shot, verify loop
 *     bounded to ~60s then it exits) must never be raced; a running pre-r7 KeepAlive guard likewise.
 *     #1509 r10 Finding 2 — but "blocks" must never mean "blocks FOREVER": a prior-generation guard
 *     that HANGS (r7/r8 generations predate the airtight in-script bounds and cannot be retrofitted)
 *     would otherwise wedge dispatch permanently, because a one-shot's running state only ever clears
 *     when the process exits. So a running own guard whose process has provably been alive LONGER than
 *     the generous HUNG_GUARD_NEUTRALIZE_SECONDS bound (≥4× any legitimate restart+verify) is
 *     NEUTRALIZED — `launchctl remove <exact label>` (manager-mediated; never a `kill <pid>`) — and
 *     dispatch is STILL refused for THIS attempt: the SIGTERM'd process may take a moment to die and
 *     must not overlap our swap. The next dispatch attempt finds it gone and proceeds — the wedge is
 *     bounded, not permanent. When the elapsed runtime is unreadable (no PID / `ps` failed), the guard
 *     is treated as legitimately working and dispatch simply blocks: ambiguity NEVER reaps.
 *  2. A NOT-running r7+ one-shot guard (r5-scheme label, no marker) is TERMINAL — KeepAlive=false ⇒ it
 *     can never relaunch — so it never blocks (the r8 anti-wedge: a dead one-shot cannot wedge dispatch).
 *  3. A NOT-running PRE-r7 KeepAlive guard can RELAUNCH via KeepAlive. #1509 r9 Finding 2 — the r8 code
 *     merely CLEARED its marker and proceeded to the synchronous swap, but launchd is external to the
 *     JS event loop: the job could relaunch AFTER the final `launchctl list` snapshot and BEFORE the
 *     swap (a snapshot→swap TOCTOU). So we definitively BOOT IT OUT (`launchctl remove` — the
 *     counterpart of the pre-r7 `launchctl submit`), so KeepAlive can never relaunch it, then CONFIRM
 *     ABSENT with a fresh snapshot before returning. Only once confirmed gone do we clear its stale
 *     marker (self-heal). If the confirm snapshot still shows an own guard (running, or a still-present
 *     KeepAlive candidate the remove did not take), we throw — never proceed while a guard could still
 *     (re)launch. #1509 r10 Finding 1 — "pre-r7 KeepAlive" now spans EVERY such job, marker or no
 *     marker: legacy bare `<base>` and r4 `.g<epoch>` labels are pre-r7 by scheme (the r9 predicate
 *     needed an `active` marker, which a legacy bare job never had, so the LIVE fleet's legacy guard
 *     could relaunch across the swap — the exact remaining double-swap window), and a marked r5/r6 job
 *     is pre-r7 by its marker's existence. See isRelaunchablePreR7Guard.
 *
 * Because the dispatcher calls this a SECOND time immediately before the swap with NO await between the
 * call and swapRuntimeSymlink (defect 2d), and this call leaves the listing with no own guard that can
 * (re)launch, the snapshot→swap window is closed on single-threaded JS: nothing can launch a guard
 * between the confirm-absent snapshot and the swap.
 */
export function assertNoContendingOwnGuard(
  base: string,
  runtimePath: string,
  launchctl: LaunchctlControl,
  readMarker: MarkerReader,
  clearMarker: MarkerClearer = defaultClearMarker,
  markerPresent: MarkerPresence = defaultMarkerPresent,
  pidObservations?: GuardPidObservationStore,
): void {
  // The completeness anchor is this orchestrator's own job label, recovered exactly from the base
  // it was constructed from (`ownUpgradeGuardBase(selfUnit)` = `${selfUnit}.upgrade-guard`).
  const selfUnit = base.replace(/\.upgrade-guard$/, "");
  const snapshot = readLaunchctlGuardSnapshot(launchctl, selfUnit);
  if (snapshot === null) {
    // fail-closed on an unreadable, unparseable, or INCOMPLETE snapshot (#1509 r12 Finding 3 + r13
    // Finding 2): a blind swap could race a guard we simply can't see — exit-0 garble is NOT an
    // empty listing, and a listing that omits our own live job is NOT a complete view.
    throw new Error("a self-upgrade restart may be pending (launchctl list unreadable or unparseable, or our own live job is missing from it — refusing to dispatch blind)");
  }
  const jobs = snapshot.filter((job) => job.base === base);
  const observations = pidObservations ?? fileGuardPidObservationStore(runtimePath);
  // Housekeeping: a label absent from the listing is unloaded and can never relaunch — forget it, so
  // stale PID history can never contribute to a later (false) relaunch proof.
  observations.prune(jobs.map((job) => job.label));

  // (1) Any RUNNING own guard blocks — never race a guard that is actively swapping/verifying. A guard
  // provably running BEYOND the generous hung bound is first neutralized by exact label (r10 Finding 2)
  // so it cannot wedge dispatch forever; this attempt still fails closed (the dying process must not
  // overlap the swap) and the retry proceeds once it is gone. #1509 r11 Finding 2: a CRASH-LOOPING
  // pre-r7 KeepAlive guard is observed young on every snapshot, so the elapsed bound alone never fires
  // — but the SAME label running under a SECOND distinct PID (persisted across snapshots and across
  // the restarts the looper itself inflicts) PROVES relaunch, and only KeepAlive jobs relaunch: it is
  // neutralized by exact label too. Identity stays exact either way (anchored scheme + own base); a
  // pid is only ever a measurement, never a kill target.
  const running = jobs.filter((job) => job.running);
  if (running.length > 0) {
    let hungNeutralized = 0;
    let relaunchNeutralized = 0;
    for (const job of running) {
      const elapsed = job.pid !== undefined ? (launchctl.processElapsedSeconds?.(job.pid) ?? null) : null;
      if (elapsed !== null && elapsed >= HUNG_GUARD_NEUTRALIZE_SECONDS) {
        launchctl.remove(job.label); // exact-label bootout; never `kill <pid>`
        hungNeutralized++;
        continue;
      }
      if (job.pid !== undefined && observations.recordRunningPid(job.label, job.pid) >= RELAUNCH_PROOF_DISTINCT_PIDS) {
        launchctl.remove(job.label); // exact-label bootout of a proven relauncher (pre-r7 KeepAlive)
        relaunchNeutralized++;
      }
    }
    if (hungNeutralized > 0) {
      throw new Error("a self-upgrade restart is already pending (a hung prior-generation guard was neutralized; retry once it is gone)");
    }
    if (relaunchNeutralized > 0) {
      throw new Error("a self-upgrade restart is already pending (a crash-looping pre-r7 KeepAlive guard was neutralized; retry once it is gone)");
    }
    throw new Error("a self-upgrade restart is already pending (an own-unit guard is running)");
  }

  // (3) NOT-running PRE-r7 KeepAlive candidates (legacy/r4 by scheme; r5/r6 by marker EVIDENCE —
  // matching OR ambiguous, r10 Finding 1 + r11 Finding 1) can relaunch — bootout each so launchd
  // cannot relaunch it, then confirm absent. (2) r7+ one-shot leftovers (r5 label, marker provably
  // absent) are terminal and left untouched.
  const keepAlive = jobs.filter((job) => isRelaunchablePreR7Guard(job, runtimePath, readMarker, markerPresent));
  if (keepAlive.length === 0) return; // only terminal one-shot leftovers (or nothing) remain — safe.

  for (const job of keepAlive) launchctl.remove(job.label); // bootout: prevent KeepAlive relaunch

  // Confirm absent: a fresh snapshot must show NO own guard that could still (re)launch onto the swap.
  const recheck = readLaunchctlGuardSnapshot(launchctl, selfUnit);
  if (recheck === null) {
    throw new Error("a self-upgrade restart may be pending (launchctl list unreadable or unparseable, or our own live job is missing from it — refusing to dispatch blind)");
  }
  const survivors = recheck.filter((job) => job.base === base);
  if (survivors.some((job) => job.running)) {
    // A KeepAlive guard relaunched before our bootout landed — refuse; the next dispatch retries.
    throw new Error("a self-upgrade restart is already pending (a pre-r7 guard relaunched)");
  }
  if (survivors.some((job) => isRelaunchablePreR7Guard(job, runtimePath, readMarker, markerPresent))) {
    // The bootout did not take (job still present with a live `active` marker) — we cannot prove it
    // will not relaunch into the swap window, so fail closed rather than swap over a relaunchable guard.
    throw new Error("a self-upgrade restart is already pending (a pre-r7 guard could not be booted out)");
  }

  // #1509 r14 (Finding 4): the booted labels are KNOWN, so confirm each gone with the
  // truncation-immune per-label probe (control-first) where available — a full-list scan can be
  // truncated at a clean row boundary after the anchor and silently omit a survivor. A label the
  // probe still reports present (or a failed control) fails closed exactly like a list survivor.
  for (const job of keepAlive) {
    if (launchdLabelProvablyAbsent(launchctl, selfUnit, job.label) === false) {
      throw new Error("a self-upgrade restart is already pending (a pre-r7 guard could not be booted out)");
    }
  }

  // Removed and confirmed absent — now clear each stale `active` marker (best-effort self-heal).
  for (const job of keepAlive) {
    const markerPath = ownGuardMarkerPath(job, runtimePath);
    if (markerPath) clearMarker(markerPath);
  }
}

export interface LaunchctlControl {
  list(): { exitCode: number; stdout: string };
  remove(label: string): { exitCode: number };
  /**
   * Elapsed wall-clock seconds the given PID has been alive (#1509 r10 Finding 2), or null when
   * unknown (dead PID, unreadable `ps`, or a test stub that omits this). Null is treated as
   * "legitimately working" — ambiguity blocks dispatch but NEVER neutralizes. Optional so existing
   * lightweight test controls keep compiling; the default adapter always provides it.
   */
  processElapsedSeconds?(pid: number): number | null;
  /**
   * Bounded `launchctl list <label>` — a DIRECT per-label existence probe (#1509 r14 Finding 4),
   * immune to full-listing truncation: exit 0 ⇒ the label is loaded in the domain (present);
   * nonzero ⇒ not-found OR a launchctl failure — which is why per-label absence proof always pairs
   * a target probe with a CONTROL probe of the known-present selfUnit (launchdLabelProvablyAbsent).
   * Optional so lightweight test controls keep compiling; callers fall back to the anchored-list
   * path when absent. The default adapter always provides it.
   */
  probeLabel?(label: string): { exitCode: number };
}

/**
 * #1509 r14 (Finding 4) — per-label PROOF of absence, immune to list truncation. An exit-0 full
 * listing truncated at a clean row boundary AFTER the header + selfUnit anchor passes every shape
 * check yet can omit a live guard — completeness of a list scan is unprovable. When the label to
 * check is KNOWN, probe it DIRECTLY instead, with a positive control: first `probeLabel(selfUnit)`
 * must report PRESENT (proving the probe mechanism works and we are looking at the right domain —
 * this orchestrator runs under that very label), and only then does a nonzero `probeLabel(label)`
 * prove the target absent. A hung/broken launchctl fails the control probe ⇒ no proof. Returns
 * null when the control is unavailable (caller falls back to the anchored-list determination).
 */
export function launchdLabelProvablyAbsent(launchctl: LaunchctlControl, selfUnit: string, label: string): boolean | null {
  if (!launchctl.probeLabel) return null;
  if (launchctl.probeLabel(selfUnit).exitCode !== 0) return false; // control failed ⇒ nothing proven
  return launchctl.probeLabel(label).exitCode !== 0;
}

// #1509 r10 (Finding 5): every default adapter call is routed through spawnSyncBounded, so a hung
// `launchctl`/`ps` binary can stall the event loop for at most MANAGER_CMD_TIMEOUT_MS and then
// reports exit 1 — which every caller treats fail-closed. The bootout/confirm path can never wedge.
export const defaultLaunchctl: LaunchctlControl = {
  list() {
    return spawnSyncBounded(["launchctl", "list"]);
  },
  remove(label) {
    return { exitCode: spawnSyncBounded(["launchctl", "remove", label]).exitCode };
  },
  processElapsedSeconds(pid) {
    return processElapsedSecondsViaPs(pid);
  },
  probeLabel(label) {
    // Per-label existence probe (#1509 r14 Finding 4): exit 0 ⇔ the label is loaded in our domain.
    return { exitCode: spawnSyncBounded(["launchctl", "list", label]).exitCode };
  },
};

// ─────────────────────────────────────────────────────────────────────────────
// #1509 r8 (Finding 1) — the systemd one-shot transient-unit guard: the Linux analogue of the
// launchd one-shot bootstrap job (§13). The whole invariant carried over:
//
//   The guard runs as a per-generation transient unit created with
//     `systemd-run --user --collect --unit=<per-gen> --property=Type=oneshot --property=RuntimeMaxSec=…`
//   (NO Restart=, NO KeepAlive-equivalent). So:
//     - unit ACTIVE/ACTIVATING  ⇒ in progress (still swapping/verifying)  ⇒ BLOCK dispatch.
//     - unit INACTIVE/FAILED/ABSENT ⇒ TERMINAL (done or dead, both safe)  ⇒ never blocks.
//   `--collect` garbage-collects the unit on inactive-OR-failed, so a finished guard's unit simply
//   disappears — there is no leftover to reap and, crucially, no completion marker (systemd guards
//   never wrote one), so the Finding-2 stale-marker wedge cannot exist on systemd at all.
//
// Per-generation unit names (dot-free, own-scoped) mean two guards never collide, exactly like the
// launchd per-generation labels. Cross-process contention is enforced by `systemctl list-units`,
// mirroring assertNoContendingOwnGuard.
// ─────────────────────────────────────────────────────────────────────────────

/**
 * The own-scoped, systemd-safe guard-unit BASE. Derived from the orchestrator's own service unit so a
 * foreign orchestrator's guards are never mistaken for ours, and sanitized to a dot-free token so the
 * per-generation `<base>-<genid>.service` units glob cleanly in `systemctl list-units`.
 */
export function systemdGuardUnitBase(selfUnit: string): string {
  const stem = selfUnit.replace(/\.(service|scope)$/, "");
  const safe = stem.replace(/[^A-Za-z0-9_-]+/g, "-").replace(/^-+|-+$/g, "");
  return `${safe || "agent-relay"}-upgrade-guard`;
}

/** The per-generation systemd transient guard unit name (genid dots → dashes, systemd-unit-safe). */
export function systemdGuardUnitFor(base: string, genid: string): string {
  return `${base}-${genid.replace(/\./g, "-")}`;
}

/** One own-unit systemd guard transient unit parsed from a `systemctl list-units` snapshot. */
export interface SystemdGuardUnit {
  /** The transient unit name (e.g. `agent-relay-orchestrator-upgrade-guard-s7-<uuid>.service`). */
  name: string;
  /** False ONLY on a POSITIVE terminal ActiveState (`inactive`/`failed`); true for the running
   * family (active/activating/reloading/deactivating) AND any unknown state (fail closed — r12). */
  running: boolean;
}

/**
 * Parse `systemctl --user list-units --plain --no-legend` output (UNIT LOAD ACTIVE SUB DESCRIPTION…)
 * into guard units + running-state. `--plain` strips the leading status bullet, so column 0 is the
 * unit name and column 2 is ActiveState.
 *
 * #1509 r12 (Finding 3) — TOTAL and fail-closed:
 *  - A TRUSTED listing requires EVERY non-empty line to be a well-formed row: ≥4 columns, a dotted
 *    unit name, and lowercase state tokens for LOAD/ACTIVE/SUB. ANY malformed line makes the WHOLE
 *    listing untrusted ⇒ null. (The pre-r12 parser silently skipped malformed lines, so garbled
 *    exit-0 output parsed to [] and read as "provably zero units" — in the contention check AND the
 *    post-dispatch fence probe. Proof of absence must never come from output we could not parse.)
 *  - EMPTY output stays a well-formed ZERO-UNIT listing: unlike launchd (whose domain always
 *    contains at least our own job), the `<base>-*.service run-*.service` glob legitimately matches
 *    nothing. Distinguishing these two is exactly the "well-formed listing that legitimately
 *    contains zero of our units" vs "output we could not parse" split.
 *  - Running-state is fail-closed: only the POSITIVE terminal ActiveStates (`inactive`, `failed`)
 *    read as not-running; the running family AND any unknown/new state token still contend (r11
 *    recognized only a fixed running set, so an unrecognized state read as "terminal ⇒ safe").
 *    A contending-but-unknown unit is then classified by ExecStart like any other candidate.
 */
export function parseSystemdGuardUnits(listOutput: string): SystemdGuardUnit[] | null {
  const units: SystemdGuardUnit[] = [];
  for (const raw of listOutput.split("\n")) {
    const line = raw.trim();
    if (!line) continue;
    const cols = line.split(/[ \t]+/);
    if (cols.length < 4 || !/^\S+\.[a-z]+$/.test(cols[0]!) || !/^[a-z-]+$/.test(cols[1]!) || !/^[a-z-]+$/.test(cols[2]!) || !/^[a-z-]+$/.test(cols[3]!)) {
      return null; // one malformed line ⇒ the whole snapshot is untrusted (fail closed)
    }
    if (!cols[0]!.endsWith(".service")) continue; // well-formed non-service rows are simply not candidates
    units.push({ name: cols[0]!, running: cols[2] !== "inactive" && cols[2] !== "failed" });
  }
  return units;
}

export interface SystemctlControl {
  /**
   * List CANDIDATE own-unit guard transient units, with running-state. This must enumerate BOTH:
   *  - the r8+ per-generation units `<base>-*.service` (own-scoped by name), AND
   *  - legacy auto-named `run-*.service` transient units — the shape a PRIOR generation's guard took
   *    when it was dispatched WITHOUT `--unit` (#1509 r9 Finding 1: the r7/pre-r8 systemd guard ran a
   *    shared `.upgrade-guard.sh` via `systemd-run` with no `--unit`, so systemd auto-named it
   *    `run-*.service`). A `run-*.service` unit is only CONFIRMED ours via its ExecStart (see below).
   */
  listGuardUnits(base: string): { exitCode: number; stdout: string };
  /**
   * Query one unit's ExecStart line (#1509 r9 Finding 1) so a legacy auto-named `run-*.service`
   * candidate can be confirmed to genuinely be OUR guard (its ExecStart runs `<runtimePath>.upgrade-
   * guard*.sh`) before it blocks dispatch — a bare `run-*.service` could belong to any `systemd-run`
   * on the box. Optional so lightweight test controls that only exercise r8 per-gen units can omit it —
   * but #1509 r10 Finding 3: a RUNNING `run-*.service` that cannot be classified (this method absent,
   * nonzero exit, or unreadable output) now BLOCKS dispatch (fail closed) instead of being ignored.
   */
  showExecStart?(unit: string): { exitCode: number; stdout: string };
  /**
   * Elapsed wall-clock seconds the unit's main process has been alive (#1509 r10 Finding 2), or null
   * when unknown. Null ⇒ treated as legitimately working (block, never neutralize). Optional for
   * lightweight test controls; the default adapter provides it (MainPID → bounded `ps -o etime=`).
   */
  unitElapsedSeconds?(unit: string): number | null;
  /** Bounded `systemctl --user stop <unit>` (#1509 r10 Finding 2 — hung-guard neutralization). */
  stopUnit?(unit: string): { exitCode: number };
  /** Bounded `systemctl --user reset-failed <unit>` — clears the stopped unit's failed residue. */
  resetFailed?(unit: string): { exitCode: number };
  /**
   * Bounded `systemctl --user show <unit> --property=LoadState --value` (#1509 r14 Finding 4) — a
   * DIRECT per-unit probe whose `not-found` answer is systemd's POSITIVE statement of absence,
   * immune to list truncation and to `--collect` GC races. Optional; callers fall back to the
   * listing-based determination when absent. The default adapter always provides it.
   */
  probeUnitLoadState?(unit: string): { exitCode: number; stdout: string };
}

// #1509 r10 (Finding 5): every default adapter call is routed through spawnSyncBounded, so a hung
// `systemctl`/`ps` binary can stall the event loop for at most MANAGER_CMD_TIMEOUT_MS and then
// reports exit 1 — which every caller treats fail-closed (an unreadable listing/show BLOCKS dispatch).
export const defaultSystemctl: SystemctlControl = {
  listGuardUnits(base) {
    // Enumerate own per-generation guard units AND legacy auto-named transient units in one call.
    return spawnSyncBounded(["systemctl", "--user", "list-units", "--all", "--type=service", "--no-legend", "--plain", `${base}-*.service`, "run-*.service"]);
  },
  showExecStart(unit) {
    return spawnSyncBounded(["systemctl", "--user", "show", unit, "--property=ExecStart"]);
  },
  unitElapsedSeconds(unit) {
    const shown = spawnSyncBounded(["systemctl", "--user", "show", unit, "--property=MainPID", "--value"]);
    if (shown.exitCode !== 0) return null;
    // #1509 r12 §18: STRICT value parse — exit-0 garble must read as "elapsed unknown" (block
    // without neutralizing), never as a lenient numeric prefix that measures the WRONG process.
    const pid = parseMainPidValue(shown.stdout);
    return pid === null ? null : processElapsedSecondsViaPs(pid);
  },
  stopUnit(unit) {
    return { exitCode: spawnSyncBounded(["systemctl", "--user", "stop", unit]).exitCode };
  },
  resetFailed(unit) {
    return { exitCode: spawnSyncBounded(["systemctl", "--user", "reset-failed", unit]).exitCode };
  },
  probeUnitLoadState(unit) {
    return spawnSyncBounded(["systemctl", "--user", "show", unit, "--property=LoadState", "--value"]);
  },
};

function escapeRegExp(value: string): string {
  return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
}

/**
 * #1509 r12 (Finding 1) + r13 (Finding 1) — ExecStart ownership answers TWO different questions
 * whose safe defaults are OPPOSITE, so it is TWO predicates over ONE shared evidence parser —
 * never a single tri-state doing double duty (that is exactly why r10–r12 kept oscillating
 * between under- and over-match: one function cannot carry both defaults):
 *
 *  - couldBeOurGuard — Question A, "must MY OWN dispatch block?": true on ANY form that could
 *    plausibly execute our guard script AND on every unparseable/ambiguous input. Over-inclusive
 *    is SAFE here (worst case: dispatch waits). Only a CLEANLY-parsed ExecStart that provably
 *    does not execute the script answers false (mere argv mentions — `cat <script>`,
 *    `--output=<script>` — stay false).
 *  - isProvablyOurGuard — Question B, "may I STOP/reset-failed this unit (neutralize)?": true
 *    ONLY on a cleanly-parsed exec that positively runs our guard as the executed program through
 *    an EXACTLY-known interpreter path (never a basename), in a recognized form, with NO doubt
 *    anywhere in the value. Over-inclusive here stops unrelated services — catastrophic — so
 *    every doubt answers false. ONLY this predicate ever gates the hung-bound neutralize.
 *
 * #1509 r14 (Findings 1–3) — WHAT the predicates match against is no longer "anything that could
 * be interpreted as executing the script" but the CLOSED SET of command lines our own launcher has
 * ever emitted (see EMITTED_INTERPRETER_PATHS below for the git-audited enumeration). We launch
 * the guard; its command line is finite and known, not an adversarial surface to interpret.
 * Normalization covers only what systemd/launchd legitimately do to OUR OWN forms: the `@ - : + !`
 * prefix chars (any order, argv0 shift for `@`) and the /bin ↔ /usr/bin usr-merge interpreter
 * alias. Consequences, deterministic in both directions: a form we never emitted (`sh -n
 * <script>`, `sh -x <script>`, `bash`/`zsh`/env(1)/timeout(1) wrappers, `/opt/vendor/sh
 * <script>`, the script as the directly-executed program, option-bearing variants) is FOREIGN —
 * it cannot be a prior-generation guard, so it neither blocks nor is ever stopped; the emitted
 * form is matched with certainty — no "unsure" band exists for real-world inputs, only for show
 * output we cannot decode (and for a multi-command wrapper that embeds our exact form: it
 * executes the guard ⇒ blocks, but is not a unit we launched ⇒ never stopped).
 *
 * Evidence parsing is TOTAL and strict (r13): the show output must be EXACTLY ONE `ExecStart=`
 * line and nothing else non-empty — duplicate property lines or trailing garble on later lines
 * (which the r12 first-match regex silently ignored) ⇒ ambiguous. systemd prefix chars `@ - : + !`
 * are accepted in ANY combination/order on the simple form's program token; an `@` ANYWHERE in the
 * prefix run shifts real arguments one right (r12 only detected a leading `@`). Structured
 * `{ path=… ; argv[]=… }` values must be exactly well-formed blocks; ANY unparseable block or
 * residue poisons the WHOLE value (r12 still scanned other blocks for "ours") ⇒ ambiguous.
 *
 * The script grammar stays exact and own-scoped (r10 F4 / r11 F3): a FULL-token anchored match on
 * `<runtimePath>.upgrade-guard[.s<seq>.<32hex>].sh`.
 */
export type ExecStartOwnership = "ours" | "foreign" | "ambiguous";

/** Anchored full-token grammar of every guard-script path any generation ever dispatched. */
function ownGuardScriptRe(runtimePath: string): RegExp {
  return new RegExp(`^${escapeRegExp(runtimePath)}\\.upgrade-guard(?:\\.s\\d+\\.[0-9a-f]{32})?\\.sh$`);
}

/**
 * #1509 r14 (Findings 1–3) — ownership MATCHES the CLOSED SET of command lines our launcher has
 * ever emitted; it does not parse/interpret arbitrary shell (an unbounded surface that produced
 * the r12/r13 under- and over-matches: `sh -n <script>` — syntax-check only, never executes — sat
 * in r13's "argument-free option" set and was stoppable; `-o pipefail`/env option forms wedged as
 * "unsure"; interpreter zoos guessed at forms we never launch). WE launch the guard, so the set
 * is finite and known. Git-audited enumeration of every dispatch across ALL generations:
 *   - r4–r6 launchd:  `launchctl submit -l <label> -- /bin/sh <script>`      → program args `/bin/sh <script>`
 *   - r7+  launchd:   bootstrap plist ProgramArguments `["/bin/sh", <script>]`
 *   - pre-r8 systemd: `systemd-run … /bin/sh <rt>.upgrade-guard.sh`           → ExecStart `/bin/sh <shared script>`
 *   - r8+  systemd:   `systemd-run … --unit <per-gen> … /bin/sh <script>`     → ExecStart `/bin/sh <per-gen script>`
 *   - fallback (no unit, never an ExecStart): `setsid /bin/sh <script>`
 *   - EXTINCT (pre-r5, #1315-era): inline `/bin/sh -c "<inline guard text> &"` — carried NO script
 *     path token; those transient units were `--collect`-GC'd generations ago and are consciously
 *     EXCLUDED from the allowlist (documented residual: such a fossil would classify foreign).
 * NO generation ever used bash/dash/zsh, env(1), timeout(1), or any option flag (`-x`, `-n`,
 * `-o pipefail` — the r13-review conjecture that `-o pipefail` was a real form is disproven by
 * this audit). The ONE emitted ExecStart template is therefore:
 *     `<sh-interpreter> <own-script-path>`   — exactly two argv slots, nothing else,
 * where <sh-interpreter> is `/bin/sh` (or `/usr/bin/sh`, its usr-merge display alias) and
 * <own-script-path> matches the exact own-scoped grammar. A non-emitted command line is
 * DETERMINISTICALLY foreign — no guessing, no "unsure" band for real-world forms.
 */
const EMITTED_INTERPRETER_PATHS = new Set(["/bin/sh", "/usr/bin/sh"]);

/**
 * True iff ONE parsed exec command IS the emitted guard form: `<sh> <own-script>`, nothing else.
 *
 * #1509 r15 — the script path is matched against the RAW REMAINDER of the command line, never a
 * whitespace-tokenized field: `systemctl show` FLATTENS argv elements without preserving their
 * boundaries (a single argv `/opt/my agent relay/….sh` arrives as `argv[]=/bin/sh /opt/my agent
 * relay/….sh`), so tokenizing tore a whitespace-bearing emitted script path apart — the exact
 * ACTIVE own guard then classified FOREIGN and a second dispatch was admitted beside it
 * (fail-open under a supported whitespace runtimePrefix). We KNOW the exact emitted line for this
 * runtimePath, so the whole remainder after the interpreter must equal it (anchored scriptRe,
 * embedded whitespace and all). Extra tokens (options, wrappers) make the remainder non-matching
 * ⇒ every r14 foreign classification is preserved.
 */
function commandIsEmittedGuardForm(program: string, argvTail: string, scriptRe: RegExp): boolean {
  return EMITTED_INTERPRETER_PATHS.has(program) && scriptRe.test(argvTail);
}

/** Split ONE leading whitespace-delimited token off a raw argv string; returns [token, rawRest]
 * with the rest's interior whitespace preserved verbatim (r15 — never re-tokenize the remainder). */
function splitFirstToken(raw: string): [string, string] {
  const m = /^\s*(\S+)(?:[ \t]+([\s\S]*))?$/.exec(raw);
  return m ? [m[1]!, m[2] ?? ""] : ["", ""];
}

/**
 * Parse a `systemctl show -p ExecStart` output into its exec commands (program + args, argv[0]
 * display slot removed), or null when the output is not EXACTLY one well-formed `ExecStart=`
 * property line (missing/duplicate/extra non-empty lines, empty value, residue, any unparseable
 * block, quoting/escaping/`;` in the simple form, unresolvable prefix run) — null = ambiguous.
 */
function parseExecStartCommands(showOutput: string): Array<{ program: string; argvTail: string }> | null {
  if (!execStartHasReadableValue(showOutput)) return null;
  const nonEmpty = showOutput.split("\n").filter((line) => line.trim() !== "");
  if (nonEmpty.length !== 1 || !nonEmpty[0]!.startsWith("ExecStart=")) return null; // duplicates/garble lines
  const value = nonEmpty[0]!.slice("ExecStart=".length).trim();
  if (value.startsWith("{")) {
    if (value.replace(/\{[^{}]*\}/g, "").replace(/[;\s]/g, "") !== "") return null; // residue outside blocks
    const commands: Array<{ program: string; argvTail: string }> = [];
    for (const [, inner] of value.matchAll(/\{([^{}]*)\}/g)) {
      const block = /^\s*path=(\S+)\s+;\s+argv\[\]=(.*?)\s+;\s+ignore_errors=/.exec(inner!);
      if (!block) return null; // ANY unparseable block poisons the whole value (r13)
      // r15: drop only the argv0/display slot; keep the remainder RAW — systemd flattened the argv
      // boundaries, so re-tokenizing it would tear a whitespace-bearing script path apart.
      commands.push({ program: block[1]!, argvTail: splitFirstToken(block[2]!)[1] });
    }
    return commands.length > 0 ? commands : null;
  }
  if (/["'\\;]/.test(value)) return null; // quoting/escaping/multi-command defeats rigorous tokenization
  const [first, afterFirst] = splitFirstToken(value);
  const prefixed = /^([@\-:+!]*)(.*)$/.exec(first)!;
  const program = prefixed[2]!;
  if (program === "") return null;
  // `@` ANYWHERE in the prefix run overrides argv[0]: the next token is the display name and the
  // real arguments start one later. The tail stays RAW either way (r15).
  return [{ program, argvTail: prefixed[1]!.includes("@") ? splitFirstToken(afterFirst)[1] : afterFirst }];
}

/**
 * Question A — dispatch blocking (#1509 r13 Finding 1; allowlist-matched r14). TRUE when the show
 * output is unparseable/ambiguous (cannot rule anything out), or when ANY parsed command IS the
 * emitted guard form (a multi-command unit wrapping our exact form still executes the guard and
 * contends on the runtime symlink, even though we never emit multi-command units). A cleanly
 * parsed value containing NO emitted form is deterministically foreign — we know every form we
 * launch, so a non-emitted form cannot be a prior-generation guard ⇒ no block.
 */
export function couldBeOurGuard(showOutput: string | null | undefined, runtimePath: string): boolean {
  if (typeof showOutput !== "string") return true;
  const commands = parseExecStartCommands(showOutput);
  if (commands === null) return true;
  const scriptRe = ownGuardScriptRe(runtimePath);
  return commands.some((cmd) => commandIsEmittedGuardForm(cmd.program, cmd.argvTail, scriptRe));
}

/**
 * Question B — neutralize eligibility (#1509 r13 Finding 1; allowlist-matched r14). TRUE only when
 * the ExecStart parsed cleanly and the WHOLE unit is byte-shape-equal (post-normalization) to an
 * emitted dispatch: exactly ONE command, and it is `<sh> <own-script>`. Anything else — including
 * a multi-command wrapper around our form — is not a unit we launched and is never stopped.
 */
export function isProvablyOurGuard(showOutput: string | null | undefined, runtimePath: string): boolean {
  if (typeof showOutput !== "string") return false;
  const commands = parseExecStartCommands(showOutput);
  if (commands === null || commands.length !== 1) return false;
  return commandIsEmittedGuardForm(commands[0]!.program, commands[0]!.argvTail, ownGuardScriptRe(runtimePath));
}

/** The derived tri-state view (`ours` ⊂ could-be) kept for reporting and matrix tests; decision
 * points consult the two predicates directly. */
export function classifyExecStartOwnership(showOutput: string | null | undefined, runtimePath: string): ExecStartOwnership {
  if (isProvablyOurGuard(showOutput, runtimePath)) return "ours";
  return couldBeOurGuard(showOutput, runtimePath) ? "ambiguous" : "foreign";
}

/** Historical boolean face — POSITIVE proof only (Question B). */
export function execStartRunsOwnGuardScript(execStartOutput: string, runtimePath: string): boolean {
  return isProvablyOurGuard(execStartOutput, runtimePath);
}

/**
 * True iff a `systemctl show -p ExecStart` output carries a READABLE, NON-EMPTY ExecStart value
 * (#1509 r11 Finding 4). The empty-property form `ExecStart=\n` — what systemd emits when the unit
 * has no ExecStart, or when show races the unit's teardown — passed r10's `includes("ExecStart=")`
 * readability check and then, not matching the script regex, was treated as PROOF-of-foreign
 * (dispatch allowed). An empty value proves nothing about what the unit runs: it is AMBIGUITY, and
 * ambiguity must block dispatch, exactly like a failed `show`.
 */
export function execStartHasReadableValue(showOutput: string): boolean {
  const m = /^ExecStart=(.*)$/m.exec(showOutput);
  return m !== null && m[1]!.trim() !== "";
}

/**
 * True iff `name` is EXACTLY one of our per-generation guard transient units (#1509 r11 Finding 3):
 * `<base>-s<seq>-<32 hex>.service` — the only name shape any generation of our code has ever passed
 * to `systemd-run --unit` (r8+; pre-r8 guards were auto-named `run-*.service`, matched by ExecStart
 * instead). A bare `${base}-` PREFIX is not identity: an adjacent foreign unit
 * (`<base>-maintenance.service`) shares the prefix without being ours — r10's prefix test blocked
 * upgrades on such a unit and then STOPPED it at the hung bound, neutralizing a unit we never
 * launched. The grammar is anchored, regex-escaped, and collision-free (the 32-hex uuid).
 */
export function isOwnSystemdGuardUnitName(name: string, base: string): boolean {
  return new RegExp(`^${escapeRegExp(base)}-s\\d+-[0-9a-f]{32}\\.service$`).test(name);
}

/**
 * systemd cross-process dispatch precondition (#1509 r8 Finding 1; hardened r9 Finding 1) — the systemd
 * analogue of assertNoContendingOwnGuard. Throws (fail closed) if any own self-upgrade guard transient
 * unit is still ACTIVE/ACTIVATING, i.e. still swapping/verifying the shared runtime symlink. A guard
 * that has exited is inactive/failed (and `--collect`-GC'd), so it never blocks — "not running ⇒
 * terminal". FAIL-CLOSED on an unreadable listing (cannot enumerate ⇒ must not swap the shared symlink
 * blind). Cheap to call twice — the dispatcher re-evaluates it immediately before the swap (no await).
 *
 * #1509 r9 Finding 1 — the invariant "any running own guard blocks dispatch" MUST hold ACROSS the
 * naming-scheme change. The r7/pre-r8 guard that installs r8/r9 was dispatched WITHOUT `--unit`, so
 * systemd auto-named it `run-*.service`; after it restarts the orchestrator, the fresh process resets
 * `selfUpgradeDispatched` and — under the r8 detector, which only globbed `<base>-*.service` — would
 * admit a SECOND dispatch while that prior guard is still verifying. We now block on:
 *   - any RUNNING `<base>-*.service` unit (own per-generation, authoritative by own-scoped name), OR
 *   - any RUNNING `run-*.service` unit whose ExecStart runs OUR guard script (the legacy contender).
 * Going forward every guard is enumerable by BOTH its own-scoped name (r8+) and its guard-script
 * ExecStart (all generations), so a future naming change cannot silently re-open this hole.
 *
 * NOTE: systemd guards never wrote a completion marker, so there is no stale-marker clause here — a
 * dead systemd guard is simply an absent/inactive unit and can never wedge dispatch.
 *
 * #1509 r10 hardening:
 *  - Finding 3 (fail CLOSED on unclassifiable): a RUNNING `run-*.service` blocks unless it is
 *    DEFINITIVELY not ours. The r9 form blocked only when `systemctl show` succeeded AND matched, so
 *    a transient D-Bus/read failure on the show of a genuinely-ours prior guard silently ALLOWED a
 *    concurrent swap. Now: show absent/failed/unreadable ⇒ ambiguous ⇒ BLOCK. A readable non-match
 *    (provably a foreign unit) is the only thing that lets a running `run-*` unit pass.
 *  - Finding 2 (hung prior guard cannot wedge forever): a running unit POSITIVELY identified as ours
 *    whose main process has provably been alive beyond the generous HUNG_GUARD_NEUTRALIZE_SECONDS
 *    bound is neutralized — bounded `systemctl stop` + `reset-failed` on its EXACT unit — and this
 *    dispatch still fails closed; the retry proceeds once the unit is gone. Unknown elapsed ⇒
 *    treated as legitimately working ⇒ block without neutralizing (ambiguity never reaps). r9+
 *    units carry TimeoutStartSec/RuntimeMaxSec so systemd kills them itself; this covers r7/r8-era
 *    units that predate those caps.
 *
 * #1509 r11 hardening (Findings 3+4 — exact grammar, no prefix trust, non-empty ExecStart):
 *  - "Ours" now requires BOTH the EXACT per-generation unit-name grammar
 *    (`<base>-s<seq>-<32hex>.service` — isOwnSystemdGuardUnitName) AND an exact single-argv-token
 *    ExecStart match. r10 trusted any `${base}-` PREFIX, so an adjacent foreign
 *    `<base>-maintenance.service` was falsely ours: it blocked upgrades and was stop/reset-failed at
 *    the hung bound — a neutralize on a unit we never launched. A unit failing the exact grammar is
 *    now classified purely by its ExecStart, exactly like an auto-named `run-*` candidate.
 *  - An exact-grammar-named unit whose ExecStart cannot be read is AMBIGUOUS: it blocks (almost
 *    certainly ours — the name embeds our base and a 32-hex uuid) but is NEVER neutralized (identity
 *    not fully proven — we never stop what we cannot exactly identify).
 *  - An exact-grammar-named unit whose ExecStart is readable and provably NOT our script cannot
 *    contend on the runtime symlink (it does not run the guard) — it neither blocks nor is touched.
 *  - A readable ExecStart with an EMPTY value (`ExecStart=\n` — r11 Finding 4) proves nothing and is
 *    ambiguity ⇒ BLOCK, never "foreign ⇒ allow".
 *
 * #1509 r12 hardening (Findings 1+3 — totality of "foreign" and "empty"):
 *  - The listing itself is trusted only when exit 0 AND well-formed (parseSystemdGuardUnits ⇒ null
 *    on any malformed line) — exit-0 garble parsed to [] pre-r12 and read as "provably zero units".
 *  - A unit that merely MENTIONS our script path in a non-executed argv position
 *    (`/usr/bin/cat <script>`) is foreign — never blocked, never stopped.
 *
 * #1509 r13 (Finding 1): every RUNNING candidate is judged by the TWO ownership predicates with
 * their opposite fail-defaults, never one tri-state doing both jobs:
 *  - !couldBeOurGuard  ⇒ provably cannot be executing our guard ⇒ ignore (no block, no stop);
 *  - isProvablyOurGuard ⇒ block, and — as the ONLY fully-proven state — hung-bound neutralize
 *    eligible (exact interpreter paths, cleanly parsed, no doubt anywhere);
 *  - could-be but not proven (basename-matched interpreter like `/opt/vendor/sh <script>`, an
 *    `sh`/`env` form with unresolvable options, unreadable/garbled/duplicated show output) ⇒
 *    block WITHOUT neutralizing.
 */
export function assertNoContendingSystemdGuard(base: string, runtimePath: string, systemctl: SystemctlControl): void {
  const list = systemctl.listGuardUnits(base);
  const units = list.exitCode === 0 ? parseSystemdGuardUnits(list.stdout) : null;
  if (units === null) {
    throw new Error("a self-upgrade restart may be pending (systemctl list-units unreadable or unparseable — refusing to dispatch blind)");
  }
  for (const unit of units) {
    if (!unit.running) continue; // POSITIVELY terminal (inactive/failed) ⇒ done or dead, both safe
    const shown = systemctl.showExecStart?.(unit.name);
    const showOutput = shown !== undefined && shown.exitCode === 0 ? shown.stdout : null; // null ⇒ ambiguous
    if (!couldBeOurGuard(showOutput, runtimePath)) continue; // provably NOT our guard ⇒ cannot contend
    if (isProvablyOurGuard(showOutput, runtimePath)) {
      // Positively ours — blocks, and as the only fully-proven state it alone may be neutralized
      // at the hung bound (r11 Finding 3: never stop what we cannot exactly identify).
      blockOnRunningOwnSystemdGuard(systemctl, unit.name, isOwnSystemdGuardUnitName(unit.name, base)
        ? "an own-unit guard unit is running"
        : "a prior-generation self-upgrade guard is running");
    }
    // Could be ours but NOT proven ⇒ block WITHOUT neutralizing (fail closed both ways). The
    // exact-grammar name only picks the message: such a unit is almost certainly ours (the name
    // embeds our base + a 32-hex uuid).
    if (isOwnSystemdGuardUnitName(unit.name, base)) {
      throw new Error("a self-upgrade restart is already pending (an own-unit guard unit is running; its ExecStart is unreadable or unclassifiable, so identity is not fully proven — blocking without neutralizing)");
    }
    throw new Error(`a self-upgrade restart may be pending (running ${unit.name} could not be classified — refusing to dispatch blind)`);
  }
}

/**
 * A RUNNING, positively-OURS guard unit always fails the precondition — but a provably-HUNG one
 * (elapsed ≥ HUNG_GUARD_NEUTRALIZE_SECONDS) is neutralized first (#1509 r10 Finding 2) so it cannot
 * wedge dispatch forever: bounded `systemctl stop` + `reset-failed` on the exact unit, then throw —
 * the stopping unit must not overlap this dispatch's swap; the retry proceeds once it is gone.
 */
function blockOnRunningOwnSystemdGuard(systemctl: SystemctlControl, unitName: string, blockDetail: string): never {
  const elapsed = systemctl.unitElapsedSeconds?.(unitName) ?? null;
  if (elapsed !== null && elapsed >= HUNG_GUARD_NEUTRALIZE_SECONDS && systemctl.stopUnit) {
    systemctl.stopUnit(unitName);
    systemctl.resetFailed?.(unitName);
    throw new Error("a self-upgrade restart is already pending (a hung prior-generation guard was neutralized; retry once it is gone)");
  }
  throw new Error(`a self-upgrade restart is already pending (${blockDetail})`);
}

/** Best-effort delete of a reaped r5 generation's per-generation script/status/result files (§6.4). */
function cleanupReapedGuardFiles(job: UpgradeGuardJob, runtimePath: string): void {
  if (job.scheme !== "r5" || !job.genid) return; // only r5 owns per-generation files; shared r4/legacy files are left
  for (const path of [
    upgradeGuardScriptPathFor(runtimePath, job.genid),
    upgradeGuardPlistPathFor(runtimePath, job.genid), // #1509 r7 §13: one-shot plist
    upgradeGuardStatusPathFor(runtimePath, job.genid), // pre-r7 leftover marker, if any
    upgradeGuardResultPathFor(runtimePath, job.genid),
  ]) {
    try {
      unlinkSync(path);
    } catch {
      // best-effort disk hygiene only (§11.4)
    }
  }
}

/**
 * Reap every genuinely-stale OWN-unit self-upgrade guard generation (#1509 r5). Replaces the
 * r1–r4 PID/epoch-inference approaches wholesale.
 *
 * ATOMICITY — reap targets each job's EXACT unique label (`<base>.s<seq>.<uuid>` for r5). Because
 * every dispatch mints a fresh uuid, a concurrently-dispatched NEW upgrade ALWAYS lives under a
 * DIFFERENT label than any generation we decide to reap — so `launchctl remove <staleLabel>` can
 * never remove the new guard no matter how the list/remove interleaves. Supersession/ordering is
 * NEVER a reap input (§12.4): there is no `seq <` comparison gating a removal.
 *
 * COMPLETION over timing (§12.3, §12.5). The ONLY branch that reaps a RUNNING job requires its own
 * marker to read `done`; an `active` + running guard is left for every scheme. No sleep/settle
 * constant anywhere in the decision.
 *
 * Own-unit exact scoping preserved (r2): only jobs whose base is `${selfUnit}.upgrade-guard` are
 * ever candidates; foreign `*.upgrade-guard` labels are untouched. No-op off launchd/darwin or
 * without a selfUnit. Returns the labels removed (old generations can accumulate).
 */
export function reapStaleUpgradeGuards(
  selfUnit: string | undefined,
  runtimePath: string,
  platform: NodeJS.Platform = process.platform,
  launchctl: LaunchctlControl = defaultLaunchctl,
  readMarker: MarkerReader = defaultReadMarker,
): string[] {
  if (platform !== "darwin" || !selfUnit) return [];
  const base = ownUpgradeGuardBase(selfUnit);
  // #1509 r12 (Finding 3) + r13 (Finding 2): an unreadable, unparseable, or own-job-less listing
  // reaps NOTHING — for this cosmetic reaper the fail-closed action on an untrusted snapshot is a
  // no-op, never a guess.
  const jobs = readLaunchctlGuardSnapshot(launchctl, selfUnit);
  if (jobs === null) return [];
  const removed: string[] = [];
  for (const job of jobs) {
    if (job.base !== base) continue; // own-unit only — never a foreign guard
    if (!guardIsReapable(job, runtimePath, readMarker)) continue;
    if (launchctl.remove(job.label).exitCode === 0) {
      removed.push(job.label);
      cleanupReapedGuardFiles(job, runtimePath);
    }
  }
  return removed;
}

/** A Relay-INDEPENDENT readiness probe: resolves true when this process's own local
 * /api/health reports `status: "ok"`. Used to prove health without a relay round-trip. */
export type LocalHealthProbe = () => Promise<boolean>;

/** Default local health probe: GET `${readinessUrl}/api/health` and check `status === "ok"`. */
export function defaultLocalHealthProbe(readinessUrl: string, timeoutMs = 2_000): LocalHealthProbe {
  const healthUrl = `${readinessUrl}/api/health`;
  return async () => {
    try {
      const res = await fetch(healthUrl, { signal: AbortSignal.timeout(timeoutMs) });
      if (!res.ok) return false;
      const body = (await res.json().catch(() => null)) as { status?: unknown } | null;
      return body?.status === "ok";
    } catch {
      return false;
    }
  };
}

export interface SelfHealUpgradeGuardOptions {
  platform?: NodeJS.Platform;
  launchctl?: LaunchctlControl;
  /** Relay-independent local readiness probe; omit to skip the health gate (test-only). */
  probe?: LocalHealthProbe;
  /** Consecutive healthy probes required before reaping (mirrors the guard's 2-sample gate). */
  stableSamples?: number;
  /** Delay between health probes. */
  probeIntervalMs?: number;
  /** Max probe attempts before giving up the health wait (bounds the whole self-heal). */
  maxProbeAttempts?: number;
  /** Injectable sleep (tests). */
  sleep?: (ms: number) => Promise<void>;
  /** Runtime path; the reaper reconstructs per-generation marker paths from it. */
  runtimePath?: string;
  /** Injectable marker reader (tests) — overrides reading marker files from disk. */
  readMarker?: MarkerReader;
}

/**
 * Startup self-heal against a lingering OWN-unit self-upgrade restart guard (#1509 r7 §13).
 *
 * COSMETIC under the r7 one-shot model. An r7 guard is a one-shot job (RunAtLoad, KeepAlive=false),
 * so a guard that fails to boot itself out sits loaded-but-idle at PID `-`: it can neither loop nor
 * wedge dispatch (the precondition blocks only on a RUNNING guard). Reaping it is therefore listing
 * hygiene, not a safety requirement — which is why this once-at-startup, unawaited call is safe and
 * can no longer leave a wedge. It ALSO cleans up any PRE-r7 KeepAlive guard left across the upgrade
 * transition: such a job WOULD loop (SIGTERM-restarting this orchestrator every ~10s), so for those
 * the health-gated reap still matters. Either way we reap only genuinely-stale generations
 * (reapStaleUpgradeGuards): a not-running one-shot, a completed pre-r7 generation, or a not-running
 * pre-r7 one once health is proven. A guard that is RUNNING is NEVER reaped here — a RUNNING guard
 * that has provably HUNG is instead neutralized on the dispatch path (assertNoContendingOwnGuard,
 * #1509 r10 Finding 2), which re-evaluates on every dispatch attempt and is therefore the reliable
 * place to break a wedge; this once-at-startup pass stays deliberately conservative.
 *
 * ATOMIC: removal targets each stale generation's exact unique label, which a concurrently-dispatched
 * new upgrade can never share, so no list/remove interleaving can reap the wrong job. RELAY-INDEPENDENT:
 * gated on a LOCAL /api/health signal, never on Relay registration — registerUntilConnected() retries
 * forever, so a serving-but-unregistered orchestrator can still break an old (pre-r7) guard's loop. We
 * reap only AFTER proven healthy (`stableSamples` consecutive healthy probes); a still-unhealthy
 * process leaves a running guard free to roll it back.
 *
 * Scoped to this orchestrator's EXACT `${selfUnit}.upgrade-guard`. No-op off darwin / without a
 * launchd selfUnit. Returns removed labels.
 */
export async function selfHealLingeringUpgradeGuard(
  selfUnit: string | undefined,
  opts: SelfHealUpgradeGuardOptions = {},
): Promise<string[]> {
  const platform = opts.platform ?? process.platform;
  if (platform !== "darwin" || !selfUnit) return [];
  const launchctl = opts.launchctl ?? defaultLaunchctl;
  const readMarker = opts.readMarker ?? defaultReadMarker;
  const runtimePath = opts.runtimePath ?? resolveRuntimePath();
  const base = ownUpgradeGuardBase(selfUnit);

  // Cheap pre-check: nothing to heal unless an own-unit guard job is actually present right now.
  // This also avoids the health wait entirely when there is clearly nothing lingering. An untrusted
  // (unreadable/unparseable/own-job-less) snapshot heals nothing — fail-closed no-op (#1509 r12
  // Finding 3 + r13 Finding 2).
  const initial = readLaunchctlGuardSnapshot(launchctl, selfUnit);
  if (initial === null || !initial.some((job) => job.base === base)) return [];

  // Prove THIS process locally healthy before touching any launchd job (defense in depth,
  // Relay-independent). A still-unhealthy process must not start reaping.
  if (opts.probe) {
    const stableSamples = Math.max(1, opts.stableSamples ?? 2);
    const probeIntervalMs = opts.probeIntervalMs ?? 2_000;
    const maxProbeAttempts = Math.max(stableSamples, opts.maxProbeAttempts ?? 30);
    const sleep = opts.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms)));
    let consecutive = 0;
    for (let attempt = 0; attempt < maxProbeAttempts && consecutive < stableSamples; attempt++) {
      const healthy = await opts.probe().catch(() => false);
      consecutive = healthy ? consecutive + 1 : 0;
      if (consecutive >= stableSamples) break;
      await sleep(probeIntervalMs);
    }
    if (consecutive < stableSamples) return [];
  }

  // Reap only genuinely-stale generations. reapStaleUpgradeGuards re-lists internally (one fresh
  // snapshot), reads each job's OWN marker, and removes by exact per-generation label.
  return reapStaleUpgradeGuards(selfUnit, runtimePath, platform, launchctl, readMarker);
}