/**
 * Lifecycle for the standalone daemon: start it in the background, prove it
 * came up, stop it, and answer "is it running?" honestly.
 *
 * The record under the runtime directory is a hint, never a fact. A pidfile
 * outlives the process that wrote it — a crash, a reboot, an OOM kill all
 * leave one behind — and pids are recycled, so the file's *existence* says
 * nothing. Every read probes the pid before believing the record; a stale
 * entry is cleared rather than honoured, because the two failures that cost an
 * operator real time are `start` refusing against a ghost and `status`
 * reporting a daemon that died hours ago.
 *
 * When the live pid is the MainPID of the `omp-conductor.service` unit,
 * `stop`/`restart` go through `systemctl` rather than a raw `SIGTERM`. A
 * raw signal against a unit with `Restart=on-failure` is read as a crash and
 * the unit comes straight back — the bug that made `omp-conductor stop` look
 * like a no-op on a systemd-managed host.
 *
 * Deliberately free of every other module in this package: nothing here opens
 * the store, loads the config or talks to `gh`, so `stop` and `status` keep
 * working when the config is the very thing that is broken.
 */

import { spawn, spawnSync } from "node:child_process";
import { chmodSync, closeSync, mkdirSync, openSync, readFileSync, readSync, renameSync, rmSync, statSync, writeFileSync } from "node:fs";
import { homedir } from "node:os";
import { join } from "node:path";
// Type-only: erased at runtime, so the "free of every other module" property
// below survives — this module still opens no store, loads no config and runs
// no `gh`, and `stop`/`status` keep working when the config is broken.
import type { DaemonStopDraft } from "./types.ts";

/**
 * The systemd unit name operators are expected to install for a supervised
 * daemon. Hardcoded rather than configured: one name on the host is the
 * whole point of a unit, and a wrong name would silently fall back to the
 * SIGTERM path and reintroduce the restart loop this module exists to avoid.
 */
export const SYSTEMD_UNIT = "omp-conductor.service";

/**
 * Mirrors `DEFAULT_PORT` in ./daemon.ts. Duplicated rather than imported so
 * this module stays free of the dispatcher's dependency tree; exported so the
 * CLI can name the port a foreground daemon will serve on without adding a
 * third literal.
 * // ponytail: two constants that must agree. If a third *definition* ever
 * // appears, move it into ./types.ts and import it everywhere.
 */
export const DEFAULT_PORT = 8787;

/** How long `startDaemon` waits for the first successful `/healthz`. */
const READY_TIMEOUT_MS = 15_000;
const READY_POLL_MS = 250;

/** `/healthz` is a local, in-memory answer; a slow one means something is wrong. */
export const HEALTH_TIMEOUT_MS = 1_500;

/** Default grace period between `SIGTERM` and `SIGKILL`. */
const STOP_TIMEOUT_MS = 10_000;

/** Bytes of `daemon.log` quoted back when the daemon dies during boot. */
const LOG_TAIL_BYTES = 4_096;

/**
 * Where the pidfile and the log live. Sits under `~/.omp/run/daemons/` with
 * the other omp daemons rather than beside the config, because this is
 * runtime state that is meaningless after a reboot — the config directory is
 * for things worth keeping.
 */
export function runtimeDir(): string {
  const override = process.env["OMP_CONDUCTOR_RUNTIME_DIR"];
  if (override !== undefined && override.length > 0) return override;
  if (process.env["NODE_ENV"] === "test") {
    // A test process must never read the host-global runtime dir: doing so
    // names the real installed daemon's pid, and a test `stop` that inherits
    // that record would signal — and stop — production's own unit (#399).
    // Fail closed: a lifecycle test that omitted OMP_CONDUCTOR_RUNTIME_DIR
    // (the runtime half of the test lifecycle seam) gets a loud error rather
    // than a silent read of the operator's pidfile.
    throw new Error(
      "OMP_CONDUCTOR_RUNTIME_DIR must be set under NODE_ENV=test — " +
        "refusing to read the host runtime directory (~/.omp/run/daemons/omp-conductor) from a test process",
    );
  }
  return join(homedir(), ".omp", "run", "daemons", "omp-conductor");
}

/** What a running daemon was started with, as far as the pidfile knows. */
export interface DaemonRecord {
  pid: number;
  port: number;
  project?: string;
  startedAt: number;
  logFile: string;
}

export function recordPath(): string {
  return join(runtimeDir(), "daemon.json");
}

/** Creates the runtime directory 0700; only chmods what this call created. */
function ensureRuntimeDir(): string {
  const dir = runtimeDir();
  const created = mkdirSync(dir, { recursive: true, mode: 0o700 });
  // mkdir's mode is masked by umask, so fix up what we made; an existing
  // directory keeps whatever the operator chose for it.
  if (created !== undefined) chmodSync(dir, 0o700);
  return dir;
}

/**
 * Reads the pidfile. Every failure — absent, unreadable, truncated mid-write,
 * hand-edited into nonsense — is the same answer: `undefined`. A lifecycle
 * command that throws because a runtime file is garbage is a lifecycle command
 * that cannot clean up after itself.
 */
export function readRecord(): DaemonRecord | undefined {
  let parsed: unknown;
  try {
    parsed = JSON.parse(readFileSync(recordPath(), "utf8"));
  } catch {
    return undefined;
  }
  if (parsed === null || typeof parsed !== "object") return undefined;
  const r = parsed as Record<string, unknown>;

  const pid = r["pid"];
  const port = r["port"];
  const startedAt = r["startedAt"];
  const logFile = r["logFile"];
  const project = r["project"];

  // pid 0 and 1 are never ours: 0 means "this process group" to `kill`, which
  // would make a corrupt file signal the whole group.
  if (typeof pid !== "number" || !Number.isInteger(pid) || pid <= 1) return undefined;
  if (typeof port !== "number" || !Number.isInteger(port) || port < 1 || port > 65535) return undefined;
  if (typeof startedAt !== "number" || !Number.isFinite(startedAt)) return undefined;
  if (typeof logFile !== "string" || logFile.length === 0) return undefined;
  if (project !== undefined && typeof project !== "string") return undefined;

  return { pid, port, startedAt, logFile, ...(project === undefined ? {} : { project }) };
}

/**
 * Writes the pidfile atomically — temp file, then rename — so a reader never
 * catches a half-written record and concludes the daemon is gone. Mode 0600:
 * it names a process another user has no business signalling.
 */
export function writeRecord(r: DaemonRecord): void {
  const dir = ensureRuntimeDir();
  const target = recordPath();
  const tmp = join(dir, `.daemon.json.${process.pid.toString(36)}.${Date.now().toString(36)}.tmp`);
  try {
    writeFileSync(tmp, `${JSON.stringify(r, null, 2)}\n`, { mode: 0o600 });
    renameSync(tmp, target);
  } catch (err) {
    rmSync(tmp, { force: true });
    throw err;
  }
  chmodSync(target, 0o600);
}

/** Removes the pidfile. Absent is success — this is how staleness is repaired. */
export function clearRecord(): void {
  rmSync(recordPath(), { force: true });
}

/**
 * Whether a pid names a live process. Signal 0 performs the permission and
 * existence checks without delivering anything: `ESRCH` is dead, `EPERM` is
 * alive but owned by somebody else — still alive, so still a reason not to
 * start a second daemon.
 */
export function isAlive(pid: number): boolean {
  if (!Number.isInteger(pid) || pid <= 1) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch (err) {
    return (err as NodeJS.ErrnoException).code === "EPERM";
  }
}

/**
 * The record, but only when the process it names is actually running. A dead
 * pid clears the file on the way out, so the next `start` is not blocked by a
 * daemon that stopped existing three reboots ago.
 */
export function livingDaemon(): DaemonRecord | undefined {
  const rec = readRecord();
  if (rec === undefined) return undefined;
  if (isAlive(rec.pid)) return rec;
  clearRecord();
  return undefined;
}

/**
 * `DaemonRecord.logFile` for a daemon nobody spawned — which is how systemd
 * runs it (`ExecStart … omp-conductor daemon`) and how a hand-run foreground
 * `daemon` works too. The field is required and `status` prints it, so it has
 * to say something true: a foreground daemon opened no log of its own —
 * whoever started it owns its stdout, be that the journal, a terminal, or a
 * pane.
 */
const FOREGROUND_LOG = "<inherited stdout — started in the foreground>";

/** What this process would publish as its own boot record, given its flags. */
function foregroundRecord(o: { port?: number; project?: string }): DaemonRecord {
  return {
    pid: process.pid,
    port: o.port ?? DEFAULT_PORT,
    startedAt: Date.now(),
    logFile: FOREGROUND_LOG,
    ...(o.project === undefined ? {} : { project: o.project }),
  };
}

export type ForegroundBoot =
  | { kind: "claimed"; record: DaemonRecord }
  | { kind: "foreign-live"; pid: number };

/**
 * The boot half of pidfile ownership (#1114): a daemon that starts without a
 * managed spawn — systemd's `ExecStart`, a reboot, a bare hand-run — publishes
 * its own record here, so the single-owner guard (`startDaemon`'s live-record
 * refusal), `wake`, `status` and orphan reconciliation are armed no matter who
 * started the process.
 *
 * A foreign live record is refused, never stolen; a stale one is cleared by
 * {@link livingDaemon} on the way past; and a record already naming this pid
 * was written by whoever spawned us and knows the real log file — replacing
 * it with a guess would be a downgrade.
 */
export function publishForegroundRecord(
  o: { port?: number; project?: string } = {},
): ForegroundBoot {
  const running = livingDaemon();
  if (running !== undefined && running.pid !== process.pid) {
    return { kind: "foreign-live", pid: running.pid };
  }
  if (running !== undefined) return { kind: "claimed", record: running };
  const record = foregroundRecord(o);
  writeRecord(record);
  return { kind: "claimed", record };
}

/**
 * The running half of pidfile ownership (#1114): the record can vanish under
 * a perfectly healthy daemon — an external sweep of the shared runtime
 * directory, a tmp cleaner — and nothing else re-arms the single-owner guard
 * until the next restart. The daemon itself repairs that: absent or stale
 * means it publishes again; a live foreign record is left alone (a `--once`
 * drill may legitimately own visibility mid-tick).
 */
export function republishSelfRecord(o: { port?: number; project?: string } = {}): void {
  const current = livingDaemon();
  // Ours is already published, or a live foreign process owns visibility:
  // either way there is nothing to repair.
  if (current !== undefined) return;
  writeRecord(foregroundRecord(o));
}

/** How often a running daemon re-asserts its record ({@link republishSelfRecord}). */
export const RECORD_REPAIR_INTERVAL_MS = 60_000;

/** Stops the periodic repair started by {@link startRecordRepair}. */
export interface RecordRepair {
  stop(): void;
}

/**
 * Re-assert the record on an interval while the daemon runs. A failed repair
 * never propagates: the pidfile is visibility, not correctness, and a failed
 * attempt retries on the next tick rather than killing the dispatcher over it.
 */
export function startRecordRepair(
  o: { port?: number; project?: string } = {},
  intervalMs = RECORD_REPAIR_INTERVAL_MS,
): RecordRepair {
  const timer = setInterval(() => {
    try {
      republishSelfRecord(o);
    } catch (err) {
      process.stderr.write(
        `omp-conductor: could not re-assert the daemon pidfile: ${(err as Error).message}\n`,
      );
    }
  }, intervalMs);
  // Bookkeeping must never hold the event loop open at shutdown.
  timer.unref?.();
  return { stop: () => clearInterval(timer) };
}

/**
 * The `daemon --once` lease file, one per process. Created `O_EXCL` so two
 * simultaneous `--once` runs cannot both win ownership, and it names the
 * holder's pid so a stale file (holder died) is removable rather than a
 * permanent lock.
 *
 * The pidfile (`daemon.json`) is what makes this process *visible* as the owner
 * for the duration of the tick (`status`, `worker stop`, a concurrent `daemon`
 * start, orphan reconciliation all probe it); the lease file is what makes
 * ownership *exclusive* — it is the thing that lets two simultaneous `--once`
 * processes serialise, which the pidfile alone cannot (each would just
 * overwrite it). They are deliberately two files: the lease is never the
 * daemon record, so a short-lived drill can never impersonate or clear the
 * managed daemon's record by virtue of holding a lock.
 */
function onceLeasePath(): string {
  return join(runtimeDir(), "once.lock");
}

interface OnceLeaseHolder {
  pid: number;
  startedAt: number;
}

/** Parses the lease file; garbage is the same answer as absence — "not held". */
function readOnceLeaseHolder(): OnceLeaseHolder | undefined {
  let parsed: unknown;
  try {
    parsed = JSON.parse(readFileSync(onceLeasePath(), "utf8"));
  } catch {
    return undefined;
  }
  if (parsed === null || typeof parsed !== "object") return undefined;
  const pid = Reflect.get(parsed, "pid");
  const startedAt = Reflect.get(parsed, "startedAt");
  if (typeof pid !== "number" || !Number.isInteger(pid) || pid <= 1) return undefined;
  if (typeof startedAt !== "number" || !Number.isFinite(startedAt)) return undefined;
  return { pid, startedAt };
}

/** Bounded attempts to clear a stale lease and re-acquire it. */
const ONCE_LEASE_RETRIES = 3;

/** A once tick has no log file of its own; the record's field is informational. */
const ONCE_RECORD_LOG = "<once tick — no daemon log>";

export interface OnceLease {
  /** The owned record that makes this process the visible daemon for the tick. */
  record: DaemonRecord;
  /**
   * Drops exclusivity — but only while it is still ours. A manager that
   * claimed the record mid-tick must be left alone: clearing it would clear
   * somebody else's daemon.
   */
  release(): void;
}

export type OnceLeaseAcquire = { ok: true; lease: OnceLease } | { ok: false; pid: number };

/**
 * Acquires the exclusive daemon/state ownership a `--once` tick must hold for
 * its whole lifecycle (#431). A drill is still a dispatcher — it settles rows,
 * projects labels, admits and launches workers — so it takes the same
 * ownership the long-running dispatcher holds via the pidfile, and refuses
 * (having done nothing) while a live daemon owns the record or another live
 * `--once` holds the lease.
 *
 * Returns `{ ok: false, pid }` — having performed no mutation — when another
 * live process owns the project/state. A stale lease and a stale record whose
 * holder died are cleared and retried. On success the process both holds the
 * exclusive lease file and has published itself as the daemon owner, so every
 * ownership probe (`livingDaemon`, a `daemon` start, a second `--once`) sees
 * this process for the whole tick.
 */
export function acquireOnceLease(
  o: { port?: number; project?: string; logFile?: string } = {},
): OnceLeaseAcquire {
  // A live dispatcher already owns the shared record. Never run beside it.
  const daemon = livingDaemon();
  if (daemon !== undefined && daemon.pid !== process.pid) {
    return { ok: false, pid: daemon.pid };
  }

  const dir = ensureRuntimeDir();
  const path = onceLeasePath();

  for (let attempt = 0; attempt < ONCE_LEASE_RETRIES; attempt++) {
    let created = false;
    try {
      const fd = openSync(path, "wx", 0o600);
      writeFileSync(fd, `${JSON.stringify({ pid: process.pid, startedAt: Date.now() })}\n`);
      closeSync(fd);
      created = true;
    } catch (err) {
      if ((err as NodeJS.ErrnoException).code !== "EEXIST") throw err;
      const holder = readOnceLeaseHolder();
      if (holder !== undefined && isAlive(holder.pid)) {
        // A live holder — another `--once` (or a process still finishing one).
        return { ok: false, pid: holder.pid };
      }
      // Stale or unreadable holder: clear it and try once more.
      rmSync(path, { force: true });
      continue;
    }
    if (!created) continue;

    const record: DaemonRecord = {
      pid: process.pid,
      port: o.port ?? DEFAULT_PORT,
      startedAt: Date.now(),
      logFile: o.logFile ?? ONCE_RECORD_LOG,
      ...(o.project === undefined ? {} : { project: o.project }),
    };
    try {
      writeRecord(record);
    } catch (err) {
      rmSync(path, { force: true });
      throw err;
    }

    // Confirm we still own the shared record: a manager may have claimed it in
    // the instant between our lease create and our record write.
    const current = readRecord();
    if (current !== undefined && current.pid === process.pid) {
      return {
        ok: true,
        lease: {
          record,
          release: () => {
            const owned = readRecord();
            if (owned !== undefined && owned.pid === process.pid) clearRecord();
            const holder = readOnceLeaseHolder();
            if (holder !== undefined && holder.pid === process.pid) {
              rmSync(path, { force: true });
            }
          },
        },
      };
    }
    rmSync(path, { force: true });
    return { ok: false, pid: current?.pid ?? process.pid };
  }

  throw new Error("could not acquire the daemon lease — a stale holder kept reappearing");
}

/**
 * Why a health probe came back negative. A timeout means the process is up
 * but not answering (wedged or overloaded); a refusal means nothing is
 * listening on the port at all. Everything else — a 500, a torn response —
 * is `other`, a health answer rather than a probe verdict.
 */
export type HealthCheckFailure = "timeout" | "refused" | "other";

export type HealthCheckResult =
  | { ok: true; body?: string }
  | { ok: false; failure: HealthCheckFailure; body?: string };

/**
 * Probes the daemon's own health endpoint. Never throws: a refused connection,
 * a DNS-less host, a hung socket and a 500 are all just "not healthy", and the
 * callers of this are the ones responsible for saying so nicely. The `failure`
 * kind lets a caller tell "up but slow" (timeout) from "nothing listening"
 * (refused) instead of collapsing both into `{ ok: false }` (#716).
 */
export async function healthCheck(port: number): Promise<HealthCheckResult> {
  try {
    const res = await fetch(`http://127.0.0.1:${port}/healthz`, {
      signal: AbortSignal.timeout(HEALTH_TIMEOUT_MS),
    });
    const body = (await res.text()).trim();
    if (!res.ok) {
      return { ok: false, failure: "other", ...(body.length > 0 ? { body } : {}) };
    }
    return { ok: true, ...(body.length > 0 ? { body } : {}) };
  } catch (err) {
    return { ok: false, failure: healthFailureKind(err) };
  }
}

/**
 * Classify a fetch rejection: the timeout signal raises `TimeoutError`; a
 * refused connection is `ConnectionRefused` under Bun and `ECONNREFUSED`
 * under undici. Anything else is an unclassified probe failure.
 */
function healthFailureKind(err: unknown): HealthCheckFailure {
  if (err instanceof Error && err.name === "TimeoutError") return "timeout";
  const code = (err as { code?: unknown } | null)?.code;
  if (code === "ConnectionRefused" || code === "ECONNREFUSED" || code === "ECONNRESET") {
    return "refused";
  }
  return "other";
}

/** Whether a `/healthz` body names `project` as served: the daemon's health
 *  endpoint lists every project it loaded, as bare strings or `{ project }`
 *  entries, so callers can prove a runtime generation serves a project —
 *  the setup rollback's prior-set proof and the restart's owned proof both
 *  build on it (#650). An unparseable body is a served-nothing answer. */
export function healthServesProject(body: string | undefined, project: string | undefined): boolean {
  if (project === undefined) return true;
  if (body === undefined) return false;
  try {
    const parsed = JSON.parse(body) as unknown;
    if (parsed === null || typeof parsed !== "object") return false;
    const projects = Reflect.get(parsed, "projects");
    if (!Array.isArray(projects)) return false;
    return projects.some((entry) => {
      if (typeof entry === "string") return entry === project;
      return entry !== null && typeof entry === "object" && Reflect.get(entry, "project") === project;
    });
  } catch {
    return false;
  }
}

/**
 * Starts the daemon and does not return until it answers `/healthz`.
 *
 * An installed unit owns the host-wide daemon lifecycle. `start --project X`
 * may select the project whose health is proved, but it must not turn the unit
 * into a detached single-project competitor (#400). Only a host proven not to
 * have the unit keeps the standalone spawn path.
 */
export async function startDaemon(
  o: { port?: number; project?: string; timeoutMs?: number } = {},
): Promise<DaemonRecord> {
  const running = livingDaemon();
  if (running !== undefined) {
    throw new Error(
      `daemon already running (pid ${running.pid}, port ${running.port}) — ` +
        `use "omp-conductor restart" to replace it, or "omp-conductor stop" first`,
    );
  }

  const installation = probeUnitInstallation();
  if (installation.kind === "unknown") {
    throw new Error(
      `cannot determine whether ${SYSTEMD_UNIT} is installed (${installation.reason}); ` +
        "refusing to launch a detached daemon that could compete with the service manager",
    );
  }
  if (installation.kind === "installed") {
    const ownership = probeUnit();
    if (ownership.kind === "unknown") {
      throw new Error(ownershipUnknown("start", ownership.reason));
    }
    if (ownership.kind === "failed") {
      return await restoreFailedUnit(o.timeoutMs, o.project);
    }
    const started = systemctl(["start", SYSTEMD_UNIT]);
    if (!started.ok) {
      throw new Error(systemctlFailure("start", started));
    }
    return await waitForOwnedDaemon("start", o.timeoutMs, o.project);
  }

  const port = o.port ?? DEFAULT_PORT;
  const logFile = join(ensureRuntimeDir(), "daemon.log");

  // Append, never truncate: the previous boot's failure is usually the reason
  // somebody is running `start` again.
  const logFd = openSync(logFile, "a", 0o600);
  let pid: number | undefined;
  try {
    const child = spawn(
      process.execPath,
      [join(import.meta.dir, "cli.ts"), "daemon", "--port", String(port), ...(o.project === undefined ? [] : ["--project", o.project])],
      {
        // The runtime directory is one we know exists and will not be deleted
        // out from under a long-running process; the daemon itself resolves
        // every path from the config, so cwd is only about not holding a
        // stale directory open.
        cwd: runtimeDir(),
        detached: true,
        stdio: ["ignore", logFd, logFd],
      },
    );
    child.once("error", (err) => {
      process.stderr.write(`omp-conductor: daemon process failed: ${err.message}\n`);
    });
    // Detach from the parent's event loop and process group so the daemon
    // outlives the shell that started it.
    child.unref();
    pid = child.pid;
  } finally {
    closeSync(logFd);
  }

  if (pid === undefined) throw new Error(`could not spawn the daemon; see ${logFile}`);

  const record: DaemonRecord = {
    pid,
    port,
    startedAt: Date.now(),
    logFile,
    ...(o.project === undefined ? {} : { project: o.project }),
  };
  // Written before the wait so a concurrent `status` sees a booting daemon
  // rather than nothing at all.
  writeRecord(record);

  const deadline = Date.now() + (o.timeoutMs ?? READY_TIMEOUT_MS);
  for (;;) {
    // Liveness first. If the child is gone, a healthy answer on that port came
    // from somebody else's server, and reporting it as ours would be worse
    // than reporting nothing.
    if (!isAlive(pid)) {
      clearRecord();
      throw new Error(`daemon exited during startup${tailLog(logFile)}`);
    }
    const health = await healthCheck(port);
    if (health.ok && healthServesProject(health.body, o.project)) return record;
    if (Date.now() >= deadline) break;
    await sleep(READY_POLL_MS);
  }

  // Alive but never healthy: a wedged process is not a running daemon, and
  // leaving it behind would block the next `start` for no benefit.
  await terminate(pid, STOP_TIMEOUT_MS);
  clearRecord();
  throw new Error(
    `daemon did not answer http://127.0.0.1:${port}/healthz for the requested project within ${Math.round((o.timeoutMs ?? READY_TIMEOUT_MS) / 1000)}s${tailLog(logFile)}`,
  );
}

/**
 * The delivery facts a mediated stop/restart records at the exact chokepoint:
 * the request's own provenance, how the stop is about to be delivered, and the
 * daemon pid it targets (#378). The recorder is injected by the CLI callers
 * (which own the store); this module stays free of it.
 */
export interface StopDelivery {
  /** The request facts built by the CLI (commands/stop.ts, commands/restart.ts). */
  provenance: DaemonStopDraft;
  /** Whether the stop is about to go through `systemctl` or a raw signal. */
  via: "systemctl" | "signal";
  /** The daemon pid being stopped, when one was known. */
  pid?: number;
}

/** The injected recorder invoked immediately before a stop/restart is signalled. */
export type StopDeliveryFn = (stop: StopDelivery) => void;

/**
 * The honest provenance for a daemon that received a stop with no mediated
 * request anywhere: unattributed by construction — Linux exposes no sender
 * identity for a signal, so the record must say "unattributed" rather than
 * guess one — plus what the receiving process actually knows: its own pid,
 * the moment (stamped by the store), the runtime directory, and the projects
 * it serves with their live-run counts.
 *
 * Storage belongs at the caller: the daemon's SIGINT/SIGTERM path writes the
 * returned draft through the store before/while draining. This builds the
 * facts, store-free, so a raw `kill`/out-of-band `systemctl stop` — no
 * conductor CLI in the delivery path at all — still leaves a durable row the
 * next `omp-conductor status` can show.
 */
export function externalStopProvenance(o: {
  /** The daemon pid that received the signal, when known. */
  daemonPid?: number;
  /** The daemon's runtime directory (host paths only — never secrets). */
  runtimeDir: string;
  /** Every served project with its live-run count at signal time. */
  affected: { project: string; live: number }[];
  reason?: string;
}): DaemonStopDraft {
  return {
    controlPath: "external signal",
    scope: "global",
    ...(o.daemonPid === undefined ? {} : { daemonPid: o.daemonPid }),
    runtimeDir: o.runtimeDir,
    affected: o.affected,
    reason: o.reason ?? "external signal — no mediated stop request was recorded",
    unattributed: true,
  };
}

/**
 * How the last stop actually landed. Callers print this so an operator can
 * tell a supervised stop from a bare SIGTERM without reading the journal.
 */
export type StopResult =
  | { kind: "stopped"; pid: number; via: "systemctl" | "signal" }
  | { kind: "not-running" };

/** What {@link restartDaemon} returns: the pre-restart record, the new one, and the path taken. */
export interface RestartResult {
  previous: DaemonRecord | undefined;
  record: DaemonRecord;
  via: "systemctl" | "cli";
}

/**
 * Stops the daemon.
 *
 * Prefer `systemctl stop omp-conductor.service` when that unit's MainPID is
 * the live daemon: systemd then owns the stop and will not schedule a restart
 * for the exit it just requested. A raw `SIGTERM` against a unit with
 * `Restart=on-failure` is what made `omp-conductor stop` look like a no-op on
 * the reference fleet — exit 143 was a "failure", and the unit came back five
 * seconds later.
 *
 * Falls back to `SIGTERM` then `SIGKILL` when there is no unit, the unit is
 * not running, or its MainPID is somebody else (a hand-started daemon next to
 * a stopped unit). The grace period is a deadline, not a clean drain: a tick
 * with a worker in flight can run for that worker's whole wall clock.
 */
export async function stopDaemon(
  o: {
    timeoutMs?: number;
    /**
     * The request facts for this stop, when a mediated caller carries them.
     * With `record`, the durable row is written immediately before the signal
     * is sent — the exact ordering "provenance precedes signalling" refers to
     * — and only when a stop actually lands: a request that finds no daemon
     * stops nothing and records nothing. Drain-style callers whose restart may
     * never execute additionally record the request at entry themselves.
     */
    provenance?: DaemonStopDraft;
    /** Invoked just before the daemon is signalled, with the delivery method. */
    record?: StopDeliveryFn;
  } = {},
): Promise<StopResult> {
  const recordStop = (via: "systemctl" | "signal", pid: number | undefined): void => {
    if (o.provenance === undefined || o.record === undefined) return;
    o.record({
      provenance: {
        ...o.provenance,
        controlPath: `${o.provenance.controlPath} via ${via}`,
        ...(pid === undefined ? {} : { daemonPid: pid }),
      },
      via,
      pid,
    });
  };

  const rec = livingDaemon();
  if (rec === undefined) {
    // `livingDaemon` already cleared a stale file; this covers the unparseable
    // one it refused to read. Still ask systemd: a unit can be running with
    // no pidfile (boot race, wiped runtime dir) and stop must still land.
    const decision = decideSystemdStop(undefined);
    if (decision.kind === "unknown") {
      throw new Error(ownershipUnknown("stop", decision.reason));
    }
    if (decision.kind === "stop") {
      recordStop("systemctl", decision.mainPid);
      await runSystemdStop(decision.mainPid, o.timeoutMs);
      clearRecord();
      return { kind: "stopped", pid: decision.mainPid, via: "systemctl" };
    }
    clearRecord();
    return { kind: "not-running" };
  }

  const decision = decideSystemdStop(rec.pid);
  if (decision.kind === "unknown") {
    // Pidfile stays: we could not prove the unit does *not* own this pid, and
    // signalling it under that uncertainty is the bounce under Restart=on-failure.
    throw new Error(ownershipUnknown("stop", decision.reason));
  }
  if (decision.kind === "stop") {
    recordStop("systemctl", rec.pid);
    await runSystemdStop(rec.pid, o.timeoutMs);
    clearRecord();
    return { kind: "stopped", pid: rec.pid, via: "systemctl" };
  }

  // decision.kind === "not-ours": confirmed no unit, inactive unit, no systemd
  // binary, or a unit whose MainPID is somebody else. Only a *confirmed*
  // negative is safe to signal.
  recordStop("signal", rec.pid);
  const gone = await terminate(rec.pid, o.timeoutMs ?? STOP_TIMEOUT_MS);
  if (!gone) {
    // The record stays: something is still holding that pid, and forgetting
    // about it would let the next `start` bind a port that is already taken.
    throw new Error(`daemon pid ${rec.pid} survived SIGTERM and SIGKILL — it may belong to another user`);
  }
  clearRecord();
  return { kind: "stopped", pid: rec.pid, via: "signal" };
}

/**
 * Restarts the daemon.
 *
 * Same ownership rule as {@link stopDaemon}: when the unit owns the live pid,
 * `systemctl restart` is the *only* path — a failed manager call is terminal,
 * never a fallthrough to raw signals. An *unanswered* ownership query is also
 * terminal: "dbus blipped" is not "no unit". A *failed* installed unit is the
 * third non-fallthrough: it must be restored through the manager
 * (`reset-failed` + `start`), because a detached CLI daemon beside a failed
 * unit is exactly the "restarted, but the fleet is still down" lie this
 * module exists to prevent (#376). Falling back to stop+start is reserved for
 * a confirmed hand-started daemon (no systemd, inactive unit, or a unit whose
 * MainPID is someone else).
 *
 * Success is reported only after the service manager is re-probed and proven
 * to own the daemon: the unit's MainPID must equal the daemon record pid AND
 * `/healthz` must answer. The manager call returning is not that proof — for
 * `Type=simple` the unit is active as soon as the process is forked, before
 * the daemon has bound its port.
 *
 * Returns the record of the process that is now answering `/healthz`.
 */
export async function restartDaemon(
  o: {
    port?: number;
    project?: string;
    timeoutMs?: number;
    /** Request facts and recorder, same semantics as {@link stopDaemon}: the
     *  durable row is written immediately before the restarting signal. */
    provenance?: DaemonStopDraft;
    record?: StopDeliveryFn;
  } = {},
): Promise<RestartResult> {
  const recordStop = (via: "systemctl" | "signal", pid: number | undefined): void => {
    if (o.provenance === undefined || o.record === undefined) return;
    o.record({
      provenance: {
        ...o.provenance,
        controlPath: `${o.provenance.controlPath} via ${via}`,
        ...(pid === undefined ? {} : { daemonPid: pid }),
      },
      via,
      pid,
    });
  };

  const previous = livingDaemon();
  const ownership = probeUnit();
  if (ownership.kind === "unknown") {
    throw new Error(ownershipUnknown("restart", ownership.reason));
  }

  if (ownership.kind === "failed") {
    // An installed unit in failed state: reset the failure record and let
    // systemd start the service. Never fall through to stopDaemon() +
    // startDaemon() — that leaves the unit failed while handing the operator
    // a healthy-looking but unmanaged pid (the 12:17Z incident in #376).
    // Any manager refusal is terminal; so is unproven ownership afterwards.
    recordStop("systemctl", previous?.pid);
    const record = await restoreFailedUnit(o.timeoutMs, o.project);
    return { previous, record, via: "systemctl" };
  }

  const unitOwns =
    ownership.kind === "active" && (previous === undefined || previous.pid === ownership.pid);

  if (unitOwns) {
    // Ownership is proven. A refused/timed-out restart must not fall through
    // to stopDaemon's signal path — that is the exact bounce this module exists
    // to prevent (SIGTERM → exit 143 → Restart=on-failure → new MainPID).
    recordStop("systemctl", ownership.pid);
    const ran = systemctl(["restart", SYSTEMD_UNIT]);
    if (!ran.ok) {
      throw new Error(systemctlFailure("restart", ran));
    }
    // systemctl restart returns once the new MainPID is up; the pidfile is
    // written by the daemon itself on boot, so wait for that rather than
    // inventing a record from the unit alone.
    const record = await waitForOwnedDaemon("restart", o.timeoutMs, o.project);
    return { previous, record, via: "systemctl" };
  }

  // Confirmed unmanaged: no unit, an inactive unit, or a unit whose MainPID
  // is somebody else. The detached CLI daemon is the only path left.
  recordStop("signal", previous?.pid);
  await stopDaemon({ timeoutMs: o.timeoutMs });
  const record = await startDaemon({ port: o.port ?? previous?.port, project: o.project ?? previous?.project });
  return { previous, record, via: "cli" };
}

/**
 * Restores a failed installed unit through the manager: `reset-failed` then
 * `start`, and proves the result the same way any managed restart is proven
 * (see {@link waitForOwnedDaemon}). Refusal or unproven ownership is terminal
 * — never a fallthrough to the detached CLI daemon.
 */
async function restoreFailedUnit(timeoutMs?: number, project?: string): Promise<DaemonRecord> {
  const reset = systemctl(["reset-failed", SYSTEMD_UNIT]);
  if (!reset.ok) {
    throw new Error(systemctlFailure("reset-failed", reset));
  }
  const started = systemctl(["start", SYSTEMD_UNIT]);
  if (!started.ok) {
    throw new Error(systemctlFailure("start", started));
  }
  return await waitForOwnedDaemon("start", timeoutMs, project);
}

/**
 * Waits for the daemon a manager call was just made to boot, and refuses to
 * report success until the service manager is proven to own it: the unit's
 * MainPID must equal the daemon record pid AND `/healthz` must answer. A unit
 * that went back to `failed` is a confirmed negative and fails fast — waiting
 * cannot un-fail it. Any other unproven state fails at the deadline with a
 * diagnostic that names the mismatch, never a bare "not ready".
 */
async function waitForOwnedDaemon(
  verb: "restart" | "start",
  timeoutMs?: number,
  project?: string,
): Promise<DaemonRecord> {
  const via = `systemctl ${verb} ${SYSTEMD_UNIT} returned`;
  const inspect = `systemctl status ${SYSTEMD_UNIT}`;
  const deadline = Date.now() + (timeoutMs ?? READY_TIMEOUT_MS);
  for (;;) {
    const rec = livingDaemon();
    const ownership = probeUnit();
    if (rec !== undefined && ownership.kind === "active" && ownership.pid === rec.pid) {
      const health = await healthCheck(rec.port);
      if (health.ok && healthServesProject(health.body, project)) return rec;
    } else if (ownership.kind === "failed") {
      // A confirmed negative: the start did not take and the unit is failed
      // again. Waiting longer cannot un-fail it.
      throw new Error(
        `${via}, but the unit went back to failed — the daemon is NOT running; check \`${inspect}\``,
      );
    }
    if (Date.now() >= deadline) break;
    await sleep(READY_POLL_MS);
  }

  // Failed to prove ownership. Name what is wrong rather than a bare timeout,
  // so the operator knows whether to fix the unit or hunt the foreign pid.
  const rec = livingDaemon();
  const ownership = probeUnit();
  if (rec === undefined) {
    throw new Error(
      `${via}, but no live daemon record appeared — the daemon did not come back; check \`${inspect}\``,
    );
  }
  if (ownership.kind === "active") {
    if (ownership.pid === rec.pid) {
      const projectDetail = project === undefined ? "" : ` for project ${project}`;
      throw new Error(
        `${via}, but the daemon never answered /healthz${projectDetail} on :${rec.port} — ` +
          `the service manager owns pid ${rec.pid}, but it is not serving the requested project; ` +
          `check \`${inspect}\` and ${rec.logFile}`,
      );
    }
    throw new Error(
      `${via}, but the unit's MainPID (${ownership.pid}) does not match the daemon record pid (${rec.pid}) — ` +
        `the reported pid is NOT owned by the service manager; check \`${inspect}\``,
    );
  }
  if (ownership.kind === "unknown") {
    throw new Error(
      `${via}, but ownership can no longer be determined (${ownership.reason}) — ` +
        `cannot confirm the daemon is managed; check \`${inspect}\``,
    );
  }
  throw new Error(`${via}, but the unit is not active; check \`${inspect}\``);
}

/**
 * What we know about {@link SYSTEMD_UNIT}.
 *
 * - `active` — unit has a live MainPID (> 1). Includes transitional manager
 *   states (`activating`, `deactivating`, `reloading`, `reactivating`) — the
 *   pid is still systemd-owned, so a raw SIGTERM would bounce under
 *   `Restart=on-failure`.
 * - `failed` — the unit is installed and its last activation failed
 *   (ActiveState=failed, no MainPID). systemd still owns this unit's
 *   lifecycle, so restoring the daemon must go through the manager
 *   (`reset-failed` + `start`); classifying a failed unit as "not supervised
 *   here" is how `restart` launched an unmanaged daemon next to a unit that
 *   stayed failed (#376).
 * - `inactive` — confirmed not running (MainPID absent/0, ActiveState not
 *   `failed`), unit absent, *or* no `systemctl` binary on this host. Safe to
 *   treat as "not supervised here".
 * - `unknown` — the manager exists (or we cannot tell it does not) but the
 *   query failed: dbus blip, permission, timeout. Must not be collapsed into
 *   `inactive` — that is how a unit-owned daemon gets a raw SIGTERM.
 */
export type UnitOwnership =
  | { kind: "active"; pid: number }
  | { kind: "failed" }
  | { kind: "inactive" }
  | { kind: "unknown"; reason: string };

/**
 * Probe {@link SYSTEMD_UNIT} ownership. Never throws — the caller decides
 * whether `unknown` is fatal (stop/restart: yes).
 */
export function probeUnit(unit = SYSTEMD_UNIT): UnitOwnership {
  const ran = systemctl(["show", unit, "--property=MainPID", "--property=ActiveState", "--value"]);
  if (!ran.ok) {
    // No binary at all → this host has no systemd manager, so nothing here can
    // be unit-owned. Any other failure (dbus, auth, timeout) is unknown.
    if (ran.missing) return { kind: "inactive" };
    const detail = (ran.stderr.trim() || ran.stdout.trim() || "systemctl show failed").split("\n")[0]!;
    return { kind: "unknown", reason: detail };
  }
  // `systemctl show --value` prints one property per line, MainPID then
  // ActiveState, in the order requested. Tolerate either order and blank
  // lines so a future systemctl rearrange does not silently disable the path.
  // Unknown units still exit 0 with `0` / `inactive`, which is the confirmed
  // negative we want — not an error.
  //
  // Ownership is the MainPID, not ActiveState. A successful probe that names a
  // live pid (> 1) means systemd still owns that process — including during
  // `activating` / `deactivating` / `reloading`. Filtering on ActiveState here
  // used to label those transitional states "inactive" and hand the pid to a
  // raw SIGTERM, which is exactly the Restart=on-failure bounce. The one
  // place ActiveState decides is the `failed` classification below: with no
  // MainPID there is no pid to hand to anything, but the unit has not gone
  // away either.
  const lines = ran.stdout
    .split("\n")
    .map((l) => l.trim())
    .filter((l) => l.length > 0);
  let pid: number | undefined;
  let activeState: string | undefined;
  for (const line of lines) {
    if (/^\d+$/.test(line)) {
      const n = Number(line);
      if (Number.isInteger(n) && n > 1) pid = n;
    } else {
      activeState = line;
    }
  }
  if (pid !== undefined) return { kind: "active", pid };
  // A failed unit is NOT the confirmed negative `inactive` claims to be: the
  // unit is installed and the manager still owns its lifecycle, so restoring
  // the daemon has to go through systemd (`reset-failed` + `start`). Reading
  // `failed` as "not supervised here" is precisely how #376 ended with a
  // healthy-looking but unmanaged pid next to a unit that stayed failed.
  if (activeState === "failed") return { kind: "failed" };
  return { kind: "inactive" };
}

type UnitInstallation =
  | { kind: "installed" }
  | { kind: "absent" }
  | { kind: "unknown"; reason: string };

/**
 * Installation is separate from ownership: an inactive installed unit and an
 * absent unit both have MainPID 0, but only the latter permits a detached
 * fallback.
 */
function probeUnitInstallation(unit = SYSTEMD_UNIT): UnitInstallation {
  const ran = systemctl(["show", unit, "--property=LoadState", "--value"]);
  if (!ran.ok) {
    if (ran.missing) return { kind: "absent" };
    const detail = (ran.stderr.trim() || ran.stdout.trim() || "systemctl show failed").split("\n")[0]!;
    return { kind: "unknown", reason: detail };
  }
  const loadState = ran.stdout.trim();
  if (loadState === "not-found") return { kind: "absent" };
  if (loadState.length === 0) return { kind: "unknown", reason: "systemctl returned no LoadState" };
  // loaded, masked, error and bad-setting all prove a manager-known unit. Let
  // `systemctl start` provide the actionable refusal for a broken definition.
  return { kind: "installed" };
}

/**
 * The MainPID of an *active* {@link SYSTEMD_UNIT}, or `undefined` when the
 * unit is confirmed inactive, failed, or absent, or when ownership could not
 * be determined. Prefer {@link probeUnit} when the caller must distinguish
 * those two.
 */
export function systemdMainPid(unit = SYSTEMD_UNIT): number | undefined {
  const ownership = probeUnit(unit);
  return ownership.kind === "active" ? ownership.pid : undefined;
}

/**
 * The run-control endpoint the `worker` verbs may address.
 *
 * `record` is the pidfile's own answer: the record port is authoritative
 * because the daemon itself wrote it, and the record's `project` (when set)
 * rides along so callers keep refusing a record pinned to another project
 * exactly as before. `unit` is the systemd answer for a missing or stale
 * record: the unit is active, its MainPID is alive, and `/healthz` on the
 * probe port proved a conductor daemon serving `expectedProject` — the port
 * was located, not assumed. `unserved` is that same live unit when no port
 * proved itself: the daemon is running but its control address cannot be
 * located, so callers must refuse rather than guess. `unknown` is the
 * manager-failure answer: the unit's ownership could not be read (bus,
 * permission, timeout), so neither running nor stopped is provable and the
 * callers must refuse — collapsing that into `none` is how a control verb
 * lies "daemon is not running" while the unit is up. `none` is the confirmed
 * negative: no record, and no unit with a live MainPID.
 */
export type DaemonControlTarget =
  | { kind: "record"; pid: number; port: number; project?: string }
  | { kind: "unit"; pid: number; port: number }
  | { kind: "unserved"; pid: number; port: number }
  | { kind: "unknown"; reason: string }
  | { kind: "none" };

/**
 * The port a unit-run daemon is probed on when the pidfile is absent: the
 * generated unit always pins `--port` 8787 (`DEFAULT_PORT`), so that is the
 * default candidate. The port is verified by `/healthz` before it is used —
 * a listener is never trusted for being on the default port alone (#811).
 *
 * A test process redirects the probe with `OMP_CONDUCTOR_TEST_DAEMON_PORT`,
 * the same seam discipline as `OMP_CONDUCTOR_TEST_SYSTEMCTL_STATE`: a CLI
 * test cannot bind the real 8787 (the host's actual daemon may answer, the
 * #399 failure mode), so its fake unit daemon serves on an ephemeral port
 * that the state names.
 */
function controlProbePort(): number {
  if (process.env["NODE_ENV"] === "test") {
    const override = process.env["OMP_CONDUCTOR_TEST_DAEMON_PORT"];
    if (override !== undefined && override.trim() !== "") {
      const port = Number.parseInt(override.trim(), 10);
      if (Number.isInteger(port) && port > 0 && port <= 65535) return port;
    }
  }
  return DEFAULT_PORT;
}

/**
 * Resolve the running daemon's pid + port for the run-control verbs, or prove
 * the daemon is not running. The pidfile is authoritative while the process
 * it names is alive; when it is missing or stale, the systemd unit is the
 * liveness witness — the same consultation `status` performs (#716): an
 * active unit whose MainPID is alive proves the daemon runs even though no
 * record pins its port. The port is never assumed: `/healthz` on the probe
 * port must prove a conductor daemon answering for `expectedProject` before
 * that address is offered as something to operate (#811).
 *
 * Never throws — the caller decides whether `unserved` (a live unit with no
 * proved endpoint) or `none` (nothing running) is the answer to give.
 */
export async function daemonControlTarget(
  expectedProject?: string,
): Promise<DaemonControlTarget> {
  const record = livingDaemon();
  if (record !== undefined) {
    return {
      kind: "record",
      pid: record.pid,
      port: record.port,
      ...(record.project === undefined ? {} : { project: record.project }),
    };
  }
  const ownership = probeUnit();
  // A manager query the manager could not answer is not "no unit": collapsing
  // it into `none` is exactly how a control verb reports a live unit-owned
  // daemon stopped. The caller must refuse on the honest indeterminacy.
  if (ownership.kind === "unknown") return { kind: "unknown", reason: ownership.reason };
  if (ownership.kind !== "active" || !isAlive(ownership.pid)) return { kind: "none" };
  const port = controlProbePort();
  const health = await healthCheck(port);
  if (!health.ok || !healthServesProject(health.body, expectedProject)) {
    return { kind: "unserved", pid: ownership.pid, port };
  }
  return { kind: "unit", pid: ownership.pid, port };
}

/**
 * The reachable { pid, port } a run-control verb may operate, or the refusal
 * that names why it cannot. The refusal language is one shared place so
 * `worker` and `extend` cannot drift apart on the same endpoint questions:
 * `none` is the plain "not running"; a record pinned to another project is
 * the same refusal the verbs always gave; an unprovable live unit and an
 * unknown manager state both fail closed instead of contacting anything or
 * claiming the daemon stopped.
 */
export async function requireDaemonControl(
  expectedProject: string,
): Promise<{ pid: number; port: number }> {
  const target = await daemonControlTarget(expectedProject);
  if (target.kind === "none") throw new Error("daemon is not running");
  if (target.kind === "unknown") {
    throw new Error(
      `cannot determine whether ${SYSTEMD_UNIT} is running (${target.reason}) — ` +
        `refusing to send run control while liveness is unknown; ` +
        `retry when systemctl answers, or check \`systemctl status ${SYSTEMD_UNIT}\``,
    );
  }
  if (target.kind === "record" && target.project !== undefined && target.project !== expectedProject) {
    throw new Error(
      `daemon serves project "${target.project}", not requested project "${expectedProject}"`,
    );
  }
  if (target.kind === "unserved") {
    throw new Error(
      `daemon is running (${SYSTEMD_UNIT} pid ${target.pid}) but not serving /healthz for project "${expectedProject}" on port ${target.port} — no run-control endpoint could be located, so nothing was changed`,
    );
  }
  return { pid: target.pid, port: target.port };
}

// ---------------------------------------------------------------------------
// internals
// ---------------------------------------------------------------------------

/**
 * Whether systemd should stop this daemon.
 *
 * - `stop` — unit is active and owns `pid` (or there is no pidfile and the
 *   unit is the only candidate). Caller MUST go through systemctl; a failed
 *   manager call is terminal.
 * - `not-ours` — confirmed inactive/failed/absent unit, no systemd binary, or
 *   a unit whose MainPID is someone else. Caller may SIGTERM its own pidfile
 *   process (a failed unit owns nothing, so signalling its record is safe).
 * - `unknown` — ownership query failed. Caller MUST NOT signal.
 */
type SystemdStopDecision =
  | { kind: "stop"; mainPid: number }
  | { kind: "not-ours" }
  | { kind: "unknown"; reason: string };

function decideSystemdStop(pid: number | undefined): SystemdStopDecision {
  const ownership = probeUnit();
  if (ownership.kind === "unknown") return { kind: "unknown", reason: ownership.reason };
  // A failed unit owns no process (MainPID 0 — systemd has already reaped or
  // lost it), so a live record pid is a confirmed "not ours", exactly like an
  // inactive unit. Restoring the unit is restart's job, not stop's.
  if (ownership.kind === "inactive" || ownership.kind === "failed") return { kind: "not-ours" };
  if (pid !== undefined && ownership.pid !== pid) return { kind: "not-ours" };
  return { kind: "stop", mainPid: ownership.pid };
}

/**
 * `systemctl stop` for a pid we have already proven the unit owns.
 *
 * Throws on manager refusal/timeout — never returns "false" for the caller to
 * reinterpret as "try SIGTERM instead". That reinterpretation is how stop
 * bounced under `Restart=on-failure`.
 */
async function runSystemdStop(pid: number, timeoutMs?: number): Promise<void> {
  const ran = systemctl(["stop", SYSTEMD_UNIT]);
  if (!ran.ok) {
    // Record stays: the unit still owns the process, and clearing it would let
    // the next `start` believe the port is free while systemd still holds it.
    throw new Error(systemctlFailure("stop", ran));
  }
  // systemctl stop is synchronous for Type=simple, but a slow drain still
  // holds the old pid briefly and the next start would refuse against it.
  const deadline = Date.now() + (timeoutMs ?? STOP_TIMEOUT_MS);
  while (isAlive(pid) && Date.now() < deadline) await sleep(100);
  if (isAlive(pid)) {
    throw new Error(
      `daemon pid ${pid} still alive after systemctl stop ${SYSTEMD_UNIT} — ` +
        `check \`systemctl status ${SYSTEMD_UNIT}\``,
    );
  }
}

function systemctlFailure(
  verb: "stop" | "restart" | "reset-failed" | "start",
  ran: SystemctlResult,
): string {
  const detail = (ran.stderr.trim() || ran.stdout.trim() || "no output").split("\n")[0] ?? "no output";
  const guard =
    verb === "stop" || verb === "restart"
      ? `refusing to signal a unit-owned daemon (that is how Restart=on-failure turns stop into a bounce)`
      : `the unit was NOT restored, so the daemon would be unmanaged`;
  return (
    `systemctl ${verb} ${SYSTEMD_UNIT} failed: ${detail} — ${guard}; ` +
    `fix the unit or run \`systemctl ${verb} ${SYSTEMD_UNIT}\` yourself`
  );
}

function ownershipUnknown(verb: "start" | "stop" | "restart", reason: string): string {
  if (verb === "start") {
    return (
      `cannot determine whether ${SYSTEMD_UNIT} owns the daemon (${reason}) — ` +
      `refusing to start a detached competitor while ownership is unknown; ` +
      `retry when systemctl answers, or run \`systemctl start ${SYSTEMD_UNIT}\` yourself`
    );
  }
  return (
    `cannot determine whether ${SYSTEMD_UNIT} owns the daemon (${reason}) — ` +
    `refusing to ${verb} via signal while ownership is unknown ` +
    `(a raw SIGTERM under Restart=on-failure is a bounce); ` +
    `retry when systemctl answers, or run \`systemctl ${verb} ${SYSTEMD_UNIT}\` yourself`
  );
}

/**
 * Run one `systemctl` invocation. Captures output and never throws: the caller
 * classifies the result. `missing: true` means the binary is not on PATH — that
 * is a confirmed "no systemd on this host", not a transient query failure.
 *
 * The default shells out. Tests replace it with {@link setSystemctlForTest}
 * so the ownership decision is exercised without a real systemd.
 */
export type SystemctlResult = {
  ok: boolean;
  stdout: string;
  stderr: string;
  /** Binary not found (ENOENT). Distinct from a failed invocation of a present binary. */
  missing?: boolean;
  /** The client bound expired (spawnSync ETIMEDOUT). A state-changing verb
   *  that reports this is re-judged from the unit's observed state before it
   *  becomes a failure (#1112). */
  timedOut?: boolean;
};

export type SystemctlFn = (args: string[]) => SystemctlResult;

interface TestSystemctlState {
  installed?: boolean;
  mainPid?: number;
  activeState?: string;
  calls?: string[][];
  failures?: Record<string, string>;
}

/**
 * Child-process CLI tests cannot inject {@link setSystemctlForTest}. Under
 * NODE_ENV=test they therefore get a file-backed fake manager, or a refusal
 * when no fixture was supplied. The real binary is never reached from a test
 * process (#399).
 */
function testSystemctl(args: string[]): SystemctlResult {
  const path = process.env["OMP_CONDUCTOR_TEST_SYSTEMCTL_STATE"];
  if (path === undefined) {
    return {
      ok: false,
      stdout: "",
      stderr: "real systemctl is disabled under NODE_ENV=test; no fake manager state was supplied",
    };
  }
  try {
    const state = JSON.parse(readFileSync(path, "utf8")) as TestSystemctlState;
    const calls = Array.isArray(state.calls) ? state.calls : [];
    calls.push(args);
    state.calls = calls;
    const verb = args[0] ?? "";
    const failure = state.failures?.[verb];
    if (failure !== undefined) {
      writeFileSync(path, JSON.stringify(state));
      return { ok: false, stdout: "", stderr: failure };
    }
    let stdout = "";
    if (verb === "show") {
      stdout = args.includes("--property=LoadState")
        ? `${state.installed === false ? "not-found" : "loaded"}\n`
        : `${state.mainPid ?? 0}\n${state.activeState ?? "inactive"}\n`;
    } else if (verb === "is-active") {
      const activeState = state.activeState ?? "inactive";
      stdout = `${activeState}\n`;
      writeFileSync(path, JSON.stringify(state));
      return { ok: activeState === "active", stdout, stderr: "" };
    } else if (verb === "stop") {
      state.mainPid = 0;
      state.activeState = "inactive";
    } else if (verb === "reset-failed") {
      state.activeState = "inactive";
    }
    writeFileSync(path, JSON.stringify(state));
    return { ok: true, stdout, stderr: "" };
  } catch (err) {
    return {
      ok: false,
      stdout: "",
      stderr: `fake systemctl state failed: ${err instanceof Error ? err.message : String(err)}`,
    };
  }
}

/** Verbs whose completion is a unit job: systemd holds the client until the
 * stop/start finishes, which legitimately takes as long as the unit's own
 * graceful-shutdown budget. Everything else answers from manager state and
 * keeps the short probe bound (#1112). */
const STATE_CHANGING_VERBS: Record<string, true> = {
  start: true,
  stop: true,
  restart: true,
  "try-restart": true,
  reload: true,
  "reload-or-restart": true,
};

/**
 * Bound for probe verbs (`show`, `is-active`, `is-enabled`, …): a hung dbus
 * becomes `unknown` ownership instead of a hang. Callers must NOT treat that
 * timeout as "no unit". State-changing verbs get their own budget below — a
 * 15s bound on `restart` is what declared a healthy restart failed and rolled
 * a good release back (#1112).
 */
const SYSTEMCTL_PROBE_TIMEOUT_MS = 15_000;

/**
 * systemd's built-in `TimeoutStopSec`: how long any unit without an explicit
 * override may legitimately spend stopping before systemd itself escalates.
 * A client bound for a state-changing verb never sits below this floor —
 * smaller is structurally able to lose against a healthy, graceful stop.
 */
const SYSTEMD_DEFAULT_TIMEOUT_STOP_MS = 90_000;

/** Slack over the unit's own stop budget: the start half of a restart plus
 * scheduling delay, before the client's stopwatch means anything at all. */
const STATE_VERB_SLACK_MS = 30_000;

type RawSystemctl = (args: string[], timeoutMs: number) => SystemctlResult;

/** One unit-state reading, in the shape the post-timeout verdict needs. */
interface UnitStateSample {
  activeState?: string;
  /** `ExecMainStartTimestampMonotonic` — changes on every fresh start of the
   *  unit, which is what makes "the old instance" distinguishable. */
  startStampMonotonic?: string;
  /** `TimeoutStopUSec` — integer microseconds, or `infinity`. */
  timeoutStopUSec?: string;
}

const STATE_SAMPLE_PROPERTIES = [
  "--property=ActiveState",
  "--property=ExecMainStartTimestampMonotonic",
  "--property=TimeoutStopUSec",
] as const;

function sampleUnitState(unit: string, raw: RawSystemctl): UnitStateSample | undefined {
  const ran = raw(
    ["show", unit, "--value", ...STATE_SAMPLE_PROPERTIES],
    SYSTEMCTL_PROBE_TIMEOUT_MS,
  );
  if (!ran.ok) return undefined;
  const values = ran.stdout.split("\n");
  const pick = (index: number): string | undefined => {
    const value = values[index]?.trim();
    return value === undefined || value === "" ? undefined : value;
  };
  return {
    activeState: pick(0),
    startStampMonotonic: pick(1),
    timeoutStopUSec: pick(2),
  };
}

/**
 * The client bound for one state-changing verb: at least the unit's own
 * `TimeoutStopUSec` — read live, because any fixed constant re-creates this
 * bug on whatever host drains slower than it — and never below systemd's own
 * default stop budget. An unreadable budget gets the documented fallback.
 */
function stateVerbTimeoutMs(sample: UnitStateSample | undefined): number {
  let stopMs = SYSTEMD_DEFAULT_TIMEOUT_STOP_MS;
  const usec =
    sample?.timeoutStopUSec === undefined ? Number.NaN : Number.parseInt(sample.timeoutStopUSec, 10);
  if (Number.isFinite(usec) && usec > 0) {
    stopMs = Math.max(Math.ceil(usec / 1000), stopMs);
  }
  return stopMs + STATE_VERB_SLACK_MS;
}

/**
 * The stopwatch lost; the unit's own state decides. A state-changing verb that
 * outlived the client bound succeeded exactly when the manager shows the
 * verb's goal reached: not-active for `stop`, active with a FRESH start
 * timestamp for the start/restart family — an unchanged timestamp means the
 * old instance is still the running one. Anything else stays a failure that
 * carries both facts.
 */
function stateVerbTimeoutVerdict(
  verb: string,
  unit: string,
  before: UnitStateSample | undefined,
  after: UnitStateSample | undefined,
  timedOut: SystemctlResult,
  timeoutMs: number,
): SystemctlResult {
  const base =
    `systemctl ${verb} ${unit} exceeded its ${Math.round(timeoutMs / 1000)}s client bound ` +
    `(${timedOut.stderr.trim() || "client timeout"})`;
  if (after?.activeState === undefined) {
    return {
      ...timedOut,
      stderr:
        `${base}, and the unit's state could not be re-read afterwards — ` +
        `the outcome is unproven, not failed`,
    };
  }
  if (verb === "stop") {
    if (after.activeState !== "active" && after.activeState !== "deactivating") {
      // Goal state reached: the stop completed, whatever the stopwatch said.
      return { ok: true, stdout: "", stderr: "" };
    }
    return {
      ...timedOut,
      stderr: `${base}; the unit is still ${after.activeState} — the stop did not complete`,
    };
  }
  const freshStart =
    before !== undefined &&
    after.startStampMonotonic !== undefined &&
    after.startStampMonotonic !== before.startStampMonotonic;
  if (after.activeState === "active" && freshStart) {
    // The unit restarted; the client simply gave up first.
    return { ok: true, stdout: "", stderr: "" };
  }
  const why =
    after.activeState !== "active"
      ? `the unit is ${after.activeState}`
      : before === undefined
        ? "the unit is active but no pre-call sample exists to prove the start is fresh"
        : "its start timestamp is unchanged, so the running instance predates the call";
  return { ...timedOut, stderr: `${base}; ${why} — the ${verb} did not complete` };
}

/**
 * Run one systemctl invocation with #1112 semantics: probes keep the short
 * hung-dbus bound, state-changing verbs wait out the unit's own stop budget,
 * and a state-changing verb that outlives even that is judged by the unit's
 * observed state rather than the client's stopwatch. Exported for tests,
 * which inject their own {@link RawSystemctl}.
 */
export function runSystemctlInvocation(args: string[], raw: RawSystemctl): SystemctlResult {
  const verb = args[0] ?? "";
  if (STATE_CHANGING_VERBS[verb] !== true) return raw(args, SYSTEMCTL_PROBE_TIMEOUT_MS);
  const unit = args.length > 1 && !args[1]!.startsWith("-") ? args[1]! : SYSTEMD_UNIT;
  const before = sampleUnitState(unit, raw);
  const timeoutMs = stateVerbTimeoutMs(before);
  const ran = raw(args, timeoutMs);
  if (ran.ok || !ran.timedOut) return ran;
  return stateVerbTimeoutVerdict(verb, unit, before, sampleUnitState(unit, raw), ran, timeoutMs);
}

function spawnSystemctl(args: string[], timeoutMs: number): SystemctlResult {
  try {
    const res = spawnSync("systemctl", args, {
      encoding: "utf8",
      timeout: timeoutMs,
      env: process.env,
    });
    if (res.error) {
      const err = res.error as NodeJS.ErrnoException;
      return {
        ok: false,
        stdout: "",
        stderr: err.message,
        ...(err.code === "ENOENT" ? { missing: true } : {}),
        ...(err.code === "ETIMEDOUT" ? { timedOut: true } : {}),
      };
    }
    return {
      ok: res.status === 0,
      stdout: res.stdout ?? "",
      stderr: res.stderr ?? "",
    };
  } catch (err) {
    return { ok: false, stdout: "", stderr: err instanceof Error ? err.message : String(err) };
  }
}

function defaultSystemctl(args: string[]): SystemctlResult {
  if (process.env["NODE_ENV"] === "test") return testSystemctl(args);
  return runSystemctlInvocation(args, spawnSystemctl);
}

let systemctl: SystemctlFn = defaultSystemctl;

/** Shared manager runner. Production reaches the hardcoded binary; test
 * processes reach only the explicit fake or the fail-closed refusal. */
export function runSystemctl(args: string[]): SystemctlResult {
  return systemctl(args);
}

/** Test-only: replace the `systemctl` runner. Pass `undefined` to restore. */
export function setSystemctlForTest(fn: SystemctlFn | undefined): void {
  systemctl = fn ?? defaultSystemctl;
}


function sleep(ms: number): Promise<void> {
  return new Promise<void>((resolve) => {
    const t = setTimeout(resolve, ms);
    t.unref?.();
  });
}

/** `SIGTERM`, poll until dead, then `SIGKILL`. Returns whether the pid is gone. */
async function terminate(pid: number, timeoutMs: number): Promise<boolean> {
  if (!isAlive(pid)) return true;
  if (!signal(pid, "SIGTERM")) return !isAlive(pid);

  const deadline = Date.now() + timeoutMs;
  while (isAlive(pid) && Date.now() < deadline) await sleep(100);
  if (!isAlive(pid)) return true;

  signal(pid, "SIGKILL");
  // SIGKILL is not instant — the kernel still has to reap it — so give the
  // pid a moment to disappear before anyone reads liveness again.
  const hardDeadline = Date.now() + 2_000;
  while (isAlive(pid) && Date.now() < hardDeadline) await sleep(50);
  return !isAlive(pid);
}

/** Whether a raw signal should be delivered to a pid. */
export type SignalFn = (pid: number, sig: NodeJS.Signals) => boolean;

/**
 * The default signal sender. Production reaches the real `kill`. Under
 * NODE_ENV=test it dispatches to the fail-closed refusal instead: a test
 * process must never signal a pid it did not explicitly authorise, and the
 * default authorises nothing (#399).
 */
function defaultSignal(pid: number, sig: NodeJS.Signals): boolean {
  if (process.env["NODE_ENV"] === "test") return testSignal(pid, sig);
  try {
    process.kill(pid, sig);
    return true;
  } catch (err) {
    return (err as NodeJS.ErrnoException).code === "ESRCH";
  }
}

/**
 * Child-process CLI tests cannot inject {@link setSignalForTest}. Under
 * NODE_ENV=test they therefore get a refusal: a test process may only signal
 * a pid its own code explicitly authorised, and a child has authorised none.
 * Without this, a child `stop` that reads the operator's real daemon.json
 * (an *inherited* runtime record) would SIGTERM the installed daemon — the
 * #399 incident.
 */
function testSignal(_pid: number, _sig: NodeJS.Signals): boolean {
  return false;
}

let signalFn: SignalFn = defaultSignal;

/**
 * Test-only: replace the raw signal sender. Pass `undefined` to restore.
 * A test that intentionally spawned a helper may authorise signals to exactly
 * that helper pid; every other pid is refused. Ownership is never inferred
 * from writing a record — a pidfile fixture authorises nothing (#399).
 */
export function setSignalForTest(fn: SignalFn | undefined): void {
  signalFn = fn ?? defaultSignal;
}

/** Sends a signal, treating "already gone" as success. Returns false if the send failed. */
function signal(pid: number, sig: NodeJS.Signals): boolean {
  return signalFn(pid, sig);
}

/**
 * The tail of the daemon log, formatted for an error message. Reads the last
 * few KB by offset rather than the whole file, because the log is append-only
 * across every boot and can be arbitrarily long.
 */
function tailLog(path: string): string {
  let text: string;
  let truncated = false;
  try {
    const size = statSync(path).size;
    const want = Math.min(size, LOG_TAIL_BYTES);
    if (want === 0) return ` — nothing was written to ${path}`;
    truncated = want < size;
    const fd = openSync(path, "r");
    try {
      const buf = Buffer.allocUnsafe(want);
      const read = readSync(fd, buf, 0, want, size - want);
      text = buf.subarray(0, read).toString("utf8");
    } finally {
      closeSync(fd);
    }
  } catch {
    return ` — see ${path}`;
  }
  // Reading from an offset lands mid-line; that fragment is an artefact, not
  // output. Reading the whole file does not, so keep line one in that case.
  const all = text.split("\n");
  const lines = (truncated ? all.slice(1) : all).filter((l) => l.trim().length > 0);
  if (lines.length === 0) return ` — see ${path}`;
  return `:\n${lines.slice(-10).join("\n")}\n(full log: ${path})`;
}