import { spawnSync } from "node:child_process";
import { createHash } from "node:crypto";
import { closeSync, copyFileSync, existsSync, mkdirSync, openSync, readFileSync, readdirSync, realpathSync, statSync, unlinkSync, writeFileSync } from "node:fs";
import { homedir } from "node:os";
import { basename, dirname, join, normalize, relative } from "node:path";
import { fileURLToPath } from "node:url";
import { loadRegistry, resolveAccess, resolveHosts } from "./buildbox-registry.mjs";
import { loadFleet } from "../../../../lib/fleet/loader.mjs";
import { sweepRemoteMirrors } from "./remote-build-gc.mjs";
import { auditHostSeat } from "./remote-seat-provision.mjs";
import { shouldSkipRemoteForSeatHost } from "./seat-host-gate.mjs";
import { probeMs, shq, sshOpts, sshTarget } from "./remote-ssh.mjs";
import { k3sConfig, tryK3sBuild } from "./k3s-remote-build.mjs";

export { auditSeatHosts } from "./remote-seat-provision.mjs";
export { probeMs, remoteTransport, shq, sshOpts, sshTarget } from "./remote-ssh.mjs";

const HOME = homedir();
// Prefer the spine shipped next to this module (deploy clone, worktree, or a
// buildbox mirror); the ~/.claude/lib copy has no repo sibling and uses the
// deploy clone.
const REPO_SPINE_CLI = join(dirname(fileURLToPath(import.meta.url)), "..", "..", "..", "..", "spine", "src", "cli.ts");
const DEPLOY_SPINE_CLI = join(HOME, ".local", "share", "overdeck", "deploy", "spine", "src", "cli.ts");

function dependencyResolves(fromDir, dep) {
  for (let dir = fromDir; ; dir = dirname(dir)) {
    if (existsSync(join(dir, "node_modules", dep, "package.json"))) return true;
    if (dirname(dir) === dir) return false;
  }
}

// A checked-out cli.ts with no installed dependencies exists but cannot run.
function spineCliRunnable(cliPath) {
  if (!existsSync(cliPath)) return false;
  const pkgRoot = dirname(dirname(cliPath));
  let deps;
  try {
    deps = Object.keys(JSON.parse(readFileSync(join(pkgRoot, "package.json"), "utf8")).dependencies || {});
  } catch {
    return false;
  }
  return deps.every((dep) => dependencyResolves(pkgRoot, dep));
}

export const SPINE_CLI = spineCliRunnable(REPO_SPINE_CLI) ? REPO_SPINE_CLI : DEPLOY_SPINE_CLI;
export const REMOTE_CONFIG_PATH = process.env.BUILD_REMOTE_CONFIG || join(HOME, ".claude", "build-remote.json");
const RUNNER_LOCAL = join(HOME, ".claude", "lib", "remote-runner.sh");
const SUPERVISOR_LOCAL = join(HOME, ".claude", "lib", "command-supervisor.mjs");
const RUNNER_REMOTE = ".rb/runner.sh";
const SUPERVISOR_REMOTE = ".rb/command-supervisor.mjs";
const TELEMETRY_MAX_AGE_MS = 90_000;
const TELEMETRY_REMOTE_PATH = ".local/state/overdeck/buildbox-telemetry.json";
const THERMAL_WARM_MARGIN_C = 10;
const THERMAL_WARM_PENALTY = 1;
const CLAIM_SETTLE_MS = 60_000;
const CLAIM_WEIGHT = 1;
const DEFAULT_REMOTE_CONFIG = {
  enabled: false, hosts: [], port: 22, ssh_user: "user", remote_root: "/home/user/builds",
  health_ttl_sec: 30, connect_timeout_sec: 6, max_remote_jobs: 6,
  ssh_probe_timeout_sec: 15, ssh_exec_timeout_sec: 1800, rsync_io_timeout_sec: 300,
  rsync_timeout_sec: 900, dispatch_timeout_sec: 120, doctor_host_timeout_sec: 30,
  local_step_timeout_sec: 120,
  grace_window_sec: 600, reconnect_interval_sec: 10, local_only: [],
  local_fallback: true, remote_wait_sec: 60, local_requeue_sec: 900, rsync_only: [],
  push_excludes: ["node_modules", ".pnpm-store", ".turbo", ".astro", ".cache", "dist", "build", ".next", "coverage", "playwright-report", "test-results", "target", ".tmpjail-work", ".rb-lockhash", ".rb-epoch", ".rb-origin"],
  pull_excludes: ["node_modules", ".git", ".pnpm-store", ".cache", "target", ".tmpjail-work", ".rb-lockhash", ".rb-epoch", ".rb-origin", ".rb-overlay-manifest"],
  ship_ignored: [".dev.vars", ".env", ".env.local"],
};

// Work that reaches this module runs on a buildbox unless it matches one of these.
// The list is the whole local-execution policy: nothing else stays on the workstation.
const INSTALL_VERBS = ["install", "i", "ci", "add", "remove", "rm", "update", "up", "link", "dedupe", "prune", "rebuild"];
const PACKAGE_MANAGERS = ["pnpm", "pnpm.js", "npm", "npm-cli.js", "yarn", "yarn.js", "bun"];
const SERVICE_VERBS = ["dev", "serve", "preview", "start"];
export const LOCAL_EXEMPTIONS = [
  {
    id: "deps-install",
    why: "an install has to materialize node_modules in this tree; node_modules is in pull_excludes and the pnpm store is a local hardlink farm, so a remote install lands nothing here (measured: 15s warm local, 330M / 67760 files)",
    match: (base, args) => PACKAGE_MANAGERS.includes(base) && args.some((a) => INSTALL_VERBS.includes(a)),
  },
  {
    id: "python-install",
    why: "pip installs into a local venv/--target directory and builds wheels against this machine's Python ABI",
    match: (base, args) => (["pip", "pip3", "uv"].includes(base) && args.includes("install"))
      || (base.startsWith("python") && args[0] === "-m" && args[1] === "pip" && args.includes("install")),
  },
  {
    id: "native-build",
    why: "native addons compile against this machine's Node ABI and libc; a remote artifact would not load here",
    match: (base, args) => base === "node-gyp" || args.includes("--build-from-source"),
  },
  {
    id: "local-service",
    why: "long-running and bound to a local port; a remote job never returns and the port is not reachable from here",
    match: (_base, args) => args.some((a) => SERVICE_VERBS.includes(a) || a === "--watch" || a.startsWith("--watch=")),
  },
];

export function localExemption(argv, patterns = []) {
  const base = basename(String(argv[0] ?? ""));
  const args = argv.slice(1);
  for (const rule of LOCAL_EXEMPTIONS) if (rule.match(base, args)) return rule;
  for (const pattern of patterns) {
    if (argv.some((tok) => tok.includes(pattern))) return { id: `config:${pattern}`, why: "operator override in build-remote.json" };
  }
  return null;
}

// Races between concurrent jobs on one mirror (a newer push re-stamps the
// epoch mid-job), not box-down states — a fresh attempt usually lands.
export const TRANSIENT_REMOTE_REASONS = new Set(["epoch-mismatch", "fenced", "pull-failed", "push-failed"]);

export class RemoteConfigUnavailableError extends Error {
  constructor(message) {
    super(message);
    this.name = "RemoteConfigUnavailableError";
  }
}

export class RemoteConfigInvalidError extends Error {
  constructor(message) {
    super(message);
    this.name = "RemoteConfigInvalidError";
  }
}

// The fleet declaration is the only source of build-host membership; the registry
// supplies their state and spill order. An empty result raises rather than degrading
// into a local run.
export function registryBuildHosts() {
  try {
    const fleet = loadFleet();
    const declared = new Set(
      fleet.nodesByRole("builder").filter((node) => node.transport === "ssh").map((node) => node.host_ref),
    );
    const hosts = resolveHosts(loadRegistry(), { order: "build" }).filter((name) => declared.has(name));
    if (hosts.length === 0) {
      throw new Error('no declared fleet builder is reachable in registry order "build"; refusing to run locally');
    }
    return hosts;
  } catch (err) {
    throw new RemoteConfigUnavailableError(err.message);
  }
}

// The k3s hostPath volume mounts the box's scratch disk at its raw mountpoint;
// DEFAULT_REMOTE_CONFIG.remote_root is a bind-mounted subdirectory of that same
// disk, so the mirror a push lands at "<remote_root>/<mirror>" is visible inside
// the pod at "<K3S_HOSTPATH_ROOT>/<basename(remote_root)>/<mirror>".
export const K3S_HOSTPATH_ROOT = "/var/lib/buildbox";

// Fans the existing git-mirror push (syncPushGit, unmodified) out to every
// registry-listed reachable buildbox, sequentially, tolerating individual host
// failures — never chooseHost. Callers restrict the k3s job to `succeeded` via
// nodeAffinity, computed from this push's live results, not cached health state.
export function pushMirrorFanout(root, mirror, stateDir, log = () => {}, deps = {}) {
  const loadReg = deps.loadRegistry ?? loadRegistry;
  const buildHosts = deps.registryBuildHosts ?? registryBuildHosts;
  const resolveAcc = deps.resolveAccess ?? resolveAccess;
  const pushGit = deps.syncPushGit ?? syncPushGit;
  const prepareWorkspace = deps.prepareWorkspace ?? ((cfg, path) => ssh(cfg, `mkdir -p ${shq(path)}`, { step: "k3s-workspace-prep" }));
  const epochOf = deps.nextEpoch ?? nextEpoch;
  const registry = loadReg();
  const hostNames = buildHosts();
  const epoch = epochOf(stateDir, "k3s-fanout", mirror);
  const succeeded = [];
  const failed = [];
  for (const name of hostNames) {
    let access;
    try {
      access = resolveAcc(registry, name, "tailscale_ip");
    } catch (err) {
      failed.push({ host: name, reason: err.message });
      continue;
    }
    if (!access) {
      failed.push({ host: name, reason: "no tailscale_ip access configured" });
      continue;
    }
    const cfg = { ...DEFAULT_REMOTE_CONFIG, host: access.host, port: access.port, ssh_user: access.user, identity_file: access.identity_file };
    const mirrorPath = `${cfg.remote_root}/${mirror}`;
    const prepared = prepareWorkspace(cfg, mirrorPath);
    if (prepared?.status !== 0) {
      failed.push({ host: name, reason: `workspace-prep-failed${prepared?.stderr ? `: ${String(prepared.stderr).trim()}` : ""}` });
      continue;
    }
    const result = pushGit(cfg, root, stateDir, mirror, mirrorPath, epoch, log);
    if (result === "no-git" || result === null) {
      failed.push({ host: name, reason: result === "no-git" ? "no-git" : "push-failed" });
      continue;
    }
    succeeded.push(name);
  }
  return { epoch, succeeded, failed };
}

function spineHookEnv() {
  return { ...process.env, FT_FROM_HOOK: "1", OVERDECK_SPINE_REPORT: process.env.OVERDECK_SPINE_REPORT ?? "1" };
}

function spineEnvelopeLine(stdout) {
  const line = String(stdout || "").trim().split("\n").map((entry) => entry.trim()).find(Boolean);
  if (!line) return null;
  try { return JSON.parse(line); } catch { return null; }
}

function runSpineCli(args, { timeoutMs, input } = {}) {
  const options = {
    encoding: "utf8",
    shell: false,
    timeout: timeoutMs,
    env: spineHookEnv(),
    input,
  };
  const wrapped = spawnSync("ft", ["bun", SPINE_CLI, ...args], options);

  // Only success or a spine envelope proves `ft` reached the spine CLI; on boxes
  // ~/.local/bin/ft is fewtok, which fails nonzero without an envelope.
  const wrapperReachedSpine =
    !wrapped.error && (wrapped.status === 0 || spineEnvelopeLine(wrapped.stdout) !== null);
  if (wrapperReachedSpine) return wrapped;

  return spawnSync("bun", [SPINE_CLI, ...args], options);
}

function bestEffortSpineReport(body) {
  try {
    const result = runSpineCli(["report", "--json"], { timeoutMs: 1500, input: `${JSON.stringify(body)}\n` });
    const envelope = spineEnvelopeLine(result.stdout);
    if (!envelope || envelope.ok !== true) {
      LOG(`job report rejected stage=${body.stage} detail=${envelope?.detail ?? envelope?.error ?? "no-envelope"}`);
    }
  } catch (err) {
    LOG(`job report failed stage=${body.stage} detail=${err?.message ?? err}`);
  }
}

export function hostEligibleCli(host, command, run = runSpineCli) {
  try {
    const result = run(["host-eligible", "--json", host, command], { timeoutMs: 1500 });
    if (result.error?.code === "ETIMEDOUT") {
      LOG(`host eligibility timeout host=${host} timeout_ms=1500`);
      return { eligible: true, reason: "hook-client-timeout" };
    }
    const envelope = spineEnvelopeLine(result.stdout);
    if (!envelope || result.status !== 0 || typeof envelope.eligible !== "boolean") {
      return { eligible: true, reason: "hook-client-unavailable" };
    }
    return envelope;
  } catch {
    return { eligible: true, reason: "hook-client-unavailable" };
  }
}

function emitJobReport(cfg, { key, mirror, argv, epoch, stage, startedAt, rc, finishedAt, missingCommand }) {
  const body = {
    source: "remote-build",
    host: cfg.host,
    key,
    mirror,
    repo: mirror,
    snapshot: String(epoch),
    argv: [...argv],
    attempt: 1,
    stage,
    startedAt,
    timeoutSec: Number(cfg.remote_job_timeout_sec),
  };
  if (stage === "finished") {
    body.rc = rc;
    body.finishedAt = finishedAt;
    if (missingCommand) body.missingCommand = missingCommand;
  }
  bestEffortSpineReport(body);
}

function readRemoteJobMeta(cfg, jobId) {
  const meta = ssh(cfg, `cat $HOME/.rb/jobs/${jobId}/meta.json`, { step: "job-meta" });
  if (meta.status !== 0) return null;
  try { return JSON.parse(meta.stdout); } catch { return null; }
}

export function loadRemoteConfig() {
  const configPath = REMOTE_CONFIG_PATH;
  try {
    statSync(configPath);
  } catch (err) {
    if (err?.code === "ENOENT") {
      bestEffortSpineReport({
        source: "remote-build",
        stage: "missing",
        configPath,
        observedAt: new Date().toISOString(),
      });
      return null;
    }
    throw new RemoteConfigUnavailableError(`config stat failed: ${err.message}`);
  }

  let result;
  try {
    // bun cold-start on a loaded box (parallel agent fleets) takes ~5s; 2s produced
    // spurious "parse-config timed out" gate failures.
    result = runSpineCli(["parse-config", "--json", configPath], { timeoutMs: 15000 });
  } catch (err) {
    throw new RemoteConfigUnavailableError(`parse-config spawn failed: ${err.message}`);
  }
  if (result.error?.code === "ETIMEDOUT") {
    throw new RemoteConfigUnavailableError("parse-config timed out");
  }

  if (result.status === 0) {
    const envelope = spineEnvelopeLine(result.stdout);
    if (!envelope || envelope.ok !== true || !envelope.config) {
      throw new RemoteConfigUnavailableError("parse-config returned non-contract stdout");
    }
    const config = envelope.config;
    const override = process.env.BUILD_REMOTE_LOCAL_FALLBACK === "1";
    if (override) config.local_fallback = true;
    // `enabled: false` is the off switch; an enabled config with no eligible host must
    // raise, because a null return reads as "disabled" and grants a local run.
    if (config.enabled !== false && !override) {
      config.hosts = registryBuildHosts();
    }
    const disabled = config.enabled === false || config.hosts.length === 0;
    bestEffortSpineReport({
      source: "remote-build",
      stage: "valid",
      configPath,
      observedAt: new Date().toISOString(),
      disabled,
      override,
    });
    if (disabled) return null;
    return config;
  }

  if (result.status === 65) {
    const envelope = spineEnvelopeLine(result.stdout);
    if (!envelope || envelope.ok !== false || !envelope.error) {
      throw new RemoteConfigUnavailableError("parse-config exit 65 with non-contract stdout");
    }
    const epoch = Math.floor(Date.now() / 1000);
    const preservedPath = `${configPath}.invalid-${epoch}`;
    copyFileSync(configPath, preservedPath);
    const override = process.env.BUILD_REMOTE_LOCAL_FALLBACK === "1";
    bestEffortSpineReport({
      source: "remote-build",
      stage: "invalid",
      kind: envelope.error,
      detail: envelope.detail || "",
      configPath,
      preservedPath,
      observedAt: new Date().toISOString(),
      override,
    });
    if (override) return null;
    throw new RemoteConfigInvalidError(`invalid remote config: ${envelope.detail || envelope.error}`);
  }

  throw new RemoteConfigUnavailableError(`parse-config failed with status ${result.status ?? "?"}`);
}

export function mirrorName(root) { return `${basename(root)}-${createHash("sha256").update(root).digest("hex").slice(0, 12)}`; }
export function deriveJobId(key, mirror) { return createHash("sha256").update(`${key}\0${mirror}`).digest("hex").slice(0, 24); }
export function compareEpoch(a, b) {
  if (!/^\d+$/.test(String(a)) || !/^\d+$/.test(String(b))) throw new Error(`invalid epoch: ${a}, ${b}`);
  const aa = BigInt(a); const bb = BigInt(b);
  return aa === bb ? 0 : (aa > bb ? 1 : -1);
}
export function parseHostTelemetry(text, nowMs) {
  if (!String(text ?? "").trim()) return { ok: false, reason: "telemetry-missing" };
  let parsed;
  try { parsed = JSON.parse(text); } catch { return { ok: false, reason: "telemetry-unparseable" }; }
  const { sampledAtEpochMs, load, cores, temperatures, ciJobsRunning } = parsed ?? {};
  if (!Number.isInteger(sampledAtEpochMs) || sampledAtEpochMs < 0
    || !Number.isFinite(load) || load < 0
    || !Number.isInteger(cores) || cores <= 0
    || !Number.isFinite(temperatures?.pkg) || !Number.isFinite(temperatures?.crit)) {
    return { ok: false, reason: "telemetry-unparseable" };
  }
  if (nowMs - sampledAtEpochMs > TELEMETRY_MAX_AGE_MS) return { ok: false, reason: "telemetry-stale" };
  if (sampledAtEpochMs - nowMs > TELEMETRY_MAX_AGE_MS) return { ok: false, reason: "telemetry-future" };
  const validCiJobs = Number.isInteger(ciJobsRunning) && ciJobsRunning >= 0;
  return {
    ok: true,
    load,
    cores,
    tempPkg: temperatures.pkg,
    tempCrit: temperatures.crit,
    sampledAtEpochMs,
    ciJobsRunning: validCiJobs ? ciJobsRunning : null,
    ciJobsReason: validCiJobs ? null : ciJobsRunning === undefined || ciJobsRunning === null
      ? "ci-jobs-missing"
      : "ci-jobs-unparseable",
  };
}
export function selectHosts(hosts, preferred, counts, cap, telemetry = {}, claims = {}) {
  // A measured CI job count consumes slots; a blind one never excludes the host.
  const ciMeasured = (h) => telemetry[h]?.ok === true && Number.isInteger(telemetry[h].ciJobsRunning);
  const available = hosts.filter((h) => Number.isInteger(counts[h])
    && counts[h] + (ciMeasured(h) ? telemetry[h].ciJobsRunning : 0) < cap);
  const originalOrder = new Map(hosts.map((host, index) => [host, index]));
  return available.filter((host) => !(telemetry[host]?.ok && telemetry[host].tempPkg >= telemetry[host].tempCrit)).sort((a, b) => {
    const aKnown = telemetry[a]?.ok === true;
    const bKnown = telemetry[b]?.ok === true;
    if (aKnown !== bKnown) return aKnown ? -1 : 1;
    const aClaims = Number.isInteger(claims[a]) ? claims[a] : 0;
    const bClaims = Number.isInteger(claims[b]) ? claims[b] : 0;
    const aScore = aKnown
      ? (telemetry[a].load + (ciMeasured(a) ? telemetry[a].ciJobsRunning : 0) + CLAIM_WEIGHT * aClaims) / telemetry[a].cores
        + (telemetry[a].tempCrit - telemetry[a].tempPkg <= THERMAL_WARM_MARGIN_C ? THERMAL_WARM_PENALTY : 0)
      : counts[a] + CLAIM_WEIGHT * aClaims;
    const bScore = bKnown
      ? (telemetry[b].load + (ciMeasured(b) ? telemetry[b].ciJobsRunning : 0) + CLAIM_WEIGHT * bClaims) / telemetry[b].cores
        + (telemetry[b].tempCrit - telemetry[b].tempPkg <= THERMAL_WARM_MARGIN_C ? THERMAL_WARM_PENALTY : 0)
      : counts[b] + CLAIM_WEIGHT * bClaims;
    if (aScore !== bScore) return aScore - bScore;
    if (a === preferred) return -1;
    if (b === preferred) return 1;
    return originalOrder.get(a) - originalOrder.get(b);
  });
}

function remoteGraceSec(cfg) {
  const seconds = Number(cfg.grace_window_sec ?? DEFAULT_REMOTE_CONFIG.grace_window_sec);
  return Number.isFinite(seconds) && seconds >= 0 ? seconds : DEFAULT_REMOTE_CONFIG.grace_window_sec;
}

// non-interactive ssh has no mise env; box node/pnpm are mise-installed
const MISE_ENV = `export PATH="$HOME/.cargo/bin:$HOME/.local/bin:$PATH"; [ -x "$HOME/.local/bin/mise" ] && eval "$("$HOME/.local/bin/mise" env -s bash)" 2>/dev/null; export PATH="$HOME/.local/bin:$PATH" COREPACK_ENABLE_DOWNLOAD_PROMPT=0; `;
function forHost(cfg, host) { return { ...cfg, host }; }
function safeHost(host) { return createHash("sha256").update(host).digest("hex").slice(0, 12); }
const MUX_TEMPLATE = join(HOME, ".ssh", "rb-%r@%h-%p");
let LOG = () => {};
export function setRemoteLog(fn) { LOG = typeof fn === "function" ? fn : () => {}; }
function timedOut(result) { return result?.error?.code === "ETIMEDOUT"; }
export function isTransportFailure(result) { return timedOut(result) || result?.status === 255; }
function didNotAnswer(result) {
  return timedOut(result) || (result?.status === 255 && /timed out|timeout|not responding/i.test(String(result.stderr || "")));
}
function execMs(cfg) { return (cfg.ssh_exec_timeout_sec ?? DEFAULT_REMOTE_CONFIG.ssh_exec_timeout_sec) * 1000; }
function positiveSeconds(value, fallback) {
  const seconds = Number(value ?? fallback);
  return Number.isFinite(seconds) && seconds > 0 ? seconds : fallback;
}
function stepTimeoutMs(cfg, requestedSec) {
  const requestedMs = positiveSeconds(requestedSec, DEFAULT_REMOTE_CONFIG.ssh_probe_timeout_sec) * 1000;
  if (!cfg?.dispatch_deadline_ms) return requestedMs;
  return Math.max(1, Math.min(requestedMs, cfg.dispatch_deadline_ms - Date.now()));
}
function timeoutFailure(step, host, timeoutMs) {
  return { status: null, error: Object.assign(new Error(`timed out after ${timeoutMs}ms`), { code: "ETIMEDOUT" }), step, host, timeoutMs };
}
function reportTimeout(step, cfg, timeoutMs) {
  const seconds = Math.max(1, Math.ceil(timeoutMs / 1000));
  const message = `remote dispatch timed out step=${step} host=${cfg.host ?? "fleet"} after=${seconds}s`;
  LOG(message);
  console.error(`local-gate: ${message}; run local-gate --remote-doctor or raise the matching build-remote timeout`);
}
function probeTimeoutSec(cfg) { return probeMs(cfg) / 1000; }
function muxSocket(cfg) { return join(HOME, ".ssh", `rb-${cfg.ssh_user}@${cfg.host}-${cfg.port}`); }
// ConnectTimeout covers new TCP connects only: a wedged ControlMaster accepts the
// channel open and never answers, so muxed calls hang past every ssh-level guard.
// A fresh unmuxed connection separates "master wedged" from "host down" — only the
// former may kill a master that other jobs are sharing.
function healMux(cfg, deps = {}) {
  const spawn = deps.spawnSync ?? spawnSync;
  const direct = spawn("ssh", [...sshOpts(cfg, { mux: false }), sshTarget(cfg), "true"], { encoding: "utf8", timeout: stepTimeoutMs(cfg, cfg.ssh_probe_timeout_sec) });
  if (direct.status !== 0) { LOG(`ssh deadline host=${cfg.host} direct=failed rc=${direct.status ?? direct.error?.code ?? "?"}`); return false; }
  spawn("ssh", ["-F", "/dev/null", "-o", `ControlPath=${MUX_TEMPLATE}`, "-p", String(cfg.port), "-O", "exit", sshTarget(cfg)], { encoding: "utf8", timeout: Math.min(5000, stepTimeoutMs(cfg, 5)) });
  try { unlinkSync(muxSocket(cfg)); } catch {}
  LOG(`mux reset host=${cfg.host}`);
  return true;
}
export function ssh(cfg, command, opts = {}, deps = {}) {
  const spawn = deps.spawnSync ?? spawnSync;
  const { timeoutSec, step = "ssh", reportTimeout: shouldReportTimeout = true, ...rest } = opts;
  const timeout = stepTimeoutMs(cfg, timeoutSec ?? cfg.ssh_probe_timeout_sec);
  if (cfg.dispatch_deadline_ms && cfg.dispatch_deadline_ms <= Date.now()) {
    const result = timeoutFailure(step, cfg.host ?? "fleet", timeout);
    if (shouldReportTimeout) reportTimeout(step, cfg, timeout);
    return result;
  }
  const run = () => spawn("ssh", [...sshOpts(cfg), sshTarget(cfg), command], { encoding: "utf8", timeout, ...rest });
  const result = run();
  if (!timedOut(result) || !healMux(cfg, deps)) {
    if (timedOut(result) && shouldReportTimeout) reportTimeout(step, cfg, timeout);
    return Object.assign(result, { step, host: cfg.host, timeoutMs: timeout });
  }
  const retried = run();
  if (timedOut(retried) && shouldReportTimeout) reportTimeout(step, cfg, timeout);
  return Object.assign(retried, { step, host: cfg.host, timeoutMs: timeout });
}
function fileHash(path) { return createHash("sha256").update(readFileSync(path)).digest("hex"); }
function localRunnerHash() { return fileHash(RUNNER_LOCAL); }
function localSupervisorHash() { return fileHash(SUPERVISOR_LOCAL); }

function deployRunner(cfg) {
  const timeoutSec = probeTimeoutSec(cfg);
  const prepare = ssh(cfg, `mkdir -p .rb/jobs ${shq(cfg.remote_root)}`, { timeoutSec, step: "runner-prep" });
  if (prepare.status !== 0) return { ok: false, unanswered: didNotAnswer(prepare) };
  const scpOpts = sshOpts(cfg).map((v) => v === "-p" ? "-P" : v);
  // Explicit destination paths: scp into a directory keeps the local basename, and
  // RUNNER_LOCAL's basename (remote-runner.sh) is not RUNNER_REMOTE (.rb/runner.sh).
  for (const [local, remote] of [[RUNNER_LOCAL, RUNNER_REMOTE], [SUPERVISOR_LOCAL, SUPERVISOR_REMOTE]]) {
    const timeout = stepTimeoutMs(cfg, cfg.ssh_probe_timeout_sec);
    const copied = spawnSync("scp", ["-q", ...scpOpts, local, `${sshTarget(cfg)}:${remote}`], { encoding: "utf8", timeout });
    if (timedOut(copied)) reportTimeout(`runner-copy:${basename(remote)}`, cfg, timeout);
    if (copied.status !== 0) return { ok: false, unanswered: didNotAnswer(copied) };
  }
  const orphan = `.rb/${basename(RUNNER_LOCAL)}`;
  const sweepOrphan = orphan === RUNNER_REMOTE ? "" : `rm -f ${shq(orphan)} && `;
  const installed = ssh(cfg, `${sweepOrphan}chmod +x ${RUNNER_REMOTE} && sha256sum ${RUNNER_REMOTE} ${SUPERVISOR_REMOTE}`, { timeoutSec, step: "runner-verify" });
  return { ok: installed.status === 0, unanswered: didNotAnswer(installed) };
}

export function markHostUnhealthy(cfg, stateDir, reason, log) {
  try {
    mkdirSync(stateDir, { recursive: true });
    writeFileSync(join(stateDir, `remote-health-${safeHost(cfg.host)}.json`), JSON.stringify({ host: cfg.host, ts: Date.now(), ok: false, runner_hash: localRunnerHash(), supervisor_hash: localSupervisorHash() }));
  } catch {}
  log(`host tripped host=${cfg.host} reason=${reason}`);
}

function remoteReadyResult(cfg, stateDir, log) {
  const cacheFile = join(stateDir, `remote-health-${safeHost(cfg.host)}.json`);
  const wantRunnerHash = localRunnerHash();
  const wantSupervisorHash = localSupervisorHash();
  try {
    const cached = JSON.parse(readFileSync(cacheFile, "utf8"));
    if (cached.host === cfg.host && cached.runner_hash === wantRunnerHash && cached.supervisor_hash === wantSupervisorHash && Date.now() - cached.ts < cfg.health_ttl_sec * 1000) {
      if (!cached.ok) log(`host cached-unhealthy host=${cfg.host} age_ms=${Date.now() - cached.ts}`);
      return { ready: cached.ok, unanswered: cached.unanswered === true };
    }
  } catch {}
  const timeoutSec = probeTimeoutSec(cfg);
  const probe = ssh(cfg, `sha256sum ${RUNNER_REMOTE} ${SUPERVISOR_REMOTE} 2>/dev/null || echo missing`, { timeoutSec, step: "runner-probe" });
  const hashes = String(probe.stdout || "").trim().split("\n").map((line) => line.trim().split(/\s+/)[0]);
  let ok = probe.status === 0 && hashes[0] === wantRunnerHash && hashes[1] === wantSupervisorHash;
  if (!ok && probe.status !== 0) log(`remote probe failed host=${cfg.host} status=${probe.status} err=${String(probe.stderr || probe.error || "").trim().slice(0, 160)}`);
  let unanswered = didNotAnswer(probe);
  // A transport-failed host cannot accept a deployment either. Do not spend another
  // full probe window rediscovering the same failure.
  if (!ok && isTransportFailure(probe)) {
    if (unanswered) log(`remote dispatch timeout host=${cfg.host} timeout_sec=${timeoutSec} error=host did not answer`);
  } else if (!ok) {
    const deployed = deployRunner(cfg);
    unanswered = deployed.unanswered;
    if (!deployed.ok) {
      log(`remote runner deploy failed host=${cfg.host}`); ok = false;
      if (unanswered) log(`remote dispatch timeout host=${cfg.host} timeout_sec=${timeoutSec} error=host did not answer`);
    } else {
      ok = true;
    }
  }
  mkdirSync(stateDir, { recursive: true });
  writeFileSync(cacheFile, JSON.stringify({ host: cfg.host, ts: Date.now(), ok, unanswered, runner_hash: wantRunnerHash, supervisor_hash: wantSupervisorHash }));
  return { ready: ok, unanswered };
}

export function remoteReady(cfg, stateDir, log) {
  return remoteReadyResult(cfg, stateDir, log).ready;
}

export function projectRoot(cwd, cfg = DEFAULT_REMOTE_CONFIG) {
  const timeout = stepTimeoutMs(cfg, cfg.local_step_timeout_sec ?? DEFAULT_REMOTE_CONFIG.local_step_timeout_sec);
  const result = spawnSync("git", ["rev-parse", "--show-toplevel"], { cwd, encoding: "utf8", timeout });
  if (timedOut(result)) reportTimeout("project-root", cfg, timeout);
  if (result.status === 0) return String(result.stdout).trim();
  // inside a .git dir --show-toplevel fails; the work tree is that dir's parent
  const gitDir = gitCommonDir(cwd, cfg);
  if (gitDir && basename(gitDir) === ".git" && existsSync(join(dirname(gitDir), ".git"))) return dirname(gitDir);
  return cwd;
}
function rsyncSsh(cfg) { return `ssh ${sshOpts(cfg).join(" ")}`; }
function pushExcludes(cfg, root) {
  const excludes = [...cfg.push_excludes];
  try { if (!statSync(join(root, ".git")).isDirectory()) excludes.push(".git"); } catch {}
  return excludes;
}
// rsync 24 = source files vanished mid-transfer — benign on a live tree.
// GNU timeout owns the process group so expiry cannot orphan rsync behind flock.
function flockedRsync(cfg, step, lockFile, args) {
  const io = cfg.rsync_io_timeout_sec ?? DEFAULT_REMOTE_CONFIG.rsync_io_timeout_sec;
  const timeoutMs = stepTimeoutMs(cfg, cfg.rsync_timeout_sec ?? DEFAULT_REMOTE_CONFIG.rsync_timeout_sec);
  const timeoutSec = Math.max(1, Math.ceil(timeoutMs / 1000));
  const result = spawnSync("timeout", ["--signal=TERM", "--kill-after=5s", `${timeoutSec}s`, "flock", "-w", String(timeoutSec), lockFile, "rsync", `--timeout=${io}`, ...args], {
    stdio: ["ignore", "ignore", "inherit"],
    timeout: timeoutMs + 10_000,
  });
  if ([30, 35, 124].includes(result.status) || timedOut(result)) {
    result.error ??= Object.assign(new Error(`timed out after ${timeoutSec}s`), { code: "ETIMEDOUT" });
    result.step = step;
    result.host = cfg.host;
    result.timeoutMs = timeoutMs;
    reportTimeout(step, cfg, timeoutMs);
  }
  if (result.status === 24) result.status = 0;
  return result;
}
function epochFile(stateDir, host, mirror) { return join(stateDir, `${mirror}.${safeHost(host)}.epoch`); }
function nextEpoch(stateDir, host, mirror) {
  const file = epochFile(stateDir, host, mirror);
  let previous = 0n;
  try { previous = BigInt(readFileSync(file, "utf8").trim()); } catch {}
  let value = BigInt(Date.now()) * 1000000n + process.hrtime.bigint() % 1000000n;
  if (value <= previous) value = previous + 1n;
  mkdirSync(stateDir, { recursive: true }); writeFileSync(file, String(value));
  return String(value);
}
// systemctl --user stop waits on the unit: control-plane deadline is too tight.
// The `fenced` marker is written before the stop — it is what tells the victim's
// own completion path that its SIGTERM rc is a fence, not a build failure.
export function fenceScript(mirrorPath, epoch) {
  return `const fs=require('fs'),cp=require('child_process'),R=process.env.HOME+'/.rb/jobs';for(const d of fs.readdirSync(R,{withFileTypes:true})){if(!d.isDirectory())continue;const j=R+'/'+d.name;try{const m=JSON.parse(fs.readFileSync(j+'/meta.json'));if(m.mirror===${JSON.stringify(mirrorPath)}&&BigInt(m.epoch)<BigInt(${JSON.stringify(epoch)})&&!fs.existsSync(j+'/rc')){fs.writeFileSync(j+'/fenced',${JSON.stringify(epoch)});cp.spawnSync('systemctl',['--user','stop','rb-'+d.name+'.service']);}}catch(e){if(cp.spawnSync('systemctl',['--user','is-active','--quiet','rb-'+d.name+'.service']).status===0){console.error(d.name+': '+e.message);process.exitCode=2}}}`;
}
function fenceOlderJobs(cfg, mirrorPath, epoch) {
  return ssh(cfg, `${MISE_ENV}node -e ${shq(fenceScript(mirrorPath, epoch))}`, { timeoutSec: cfg.ssh_exec_timeout_sec, step: "fence-jobs" });
}
// Top-level symlinks resolving outside the project root (e.g. .platform ->
// ~/Projects/platform). The pushed link would dangle remotely, so each target is
// mirrored to the SAME absolute path on the box — box and laptop share the
// /home/user layout. .worktrees/.wt-*/tmp are local worktree-junk conventions.
// Remote argv normalization: laptop-absolute paths don't exist on the box.
// Paths under the project root map into the mirror; known toolchain binaries
// (shim-resolved absolute paths) map to bare names for the remote mise PATH.
const TOOL_BINS = new Set(["node", "npm", "npx", "pnpm", "yarn", "bun", "cargo", "go", "rustc", "tsc", "vitest", "jest", "turbo", "esbuild", "webpack", "rollup", "parcel", "astro", "next", "vite", "playwright", "ng", "nx"]);
const SHELL_WRAPPERS = new Set(["bash", "sh", "dash", "zsh", "fish", "env", "nice", "timeout", "command"]);
const TRANSPORT_SCRIPTS = new Set(["remote-runner.sh", "runner.sh"]);

function embeddedCommandIdentity(token) {
  for (const word of String(token).split(/\s+/)) {
    if (!word || word.startsWith("-")) continue;
    const base = basename(word);
    if (!base || SHELL_WRAPPERS.has(base) || TRANSPORT_SCRIPTS.has(base)) continue;
    return base;
  }
  return null;
}

/** Admission/breaker command identity — must match controller deriveCommand(). */
export function deriveCommandIdentity(argv) {
  for (const raw of argv) {
    const token = String(raw).trim();
    if (!token || token.startsWith("-")) continue;
    if (token.includes(" ")) {
      const embedded = embeddedCommandIdentity(token);
      if (embedded) return embedded;
      continue;
    }
    const base = basename(token);
    if (!base || SHELL_WRAPPERS.has(base) || TRANSPORT_SCRIPTS.has(base)) continue;
    return base;
  }
  throw new Error("empty command basename");
}

// A wrapper's 127 cannot identify the failed program: `bash script` can fail for
// any command inside script.  Only attest an executable when the submitted argv
// names it directly, then confirm it in the buildbox environment before reporting.
function directCapabilityCandidate(argv) {
  const token = String(argv[0] ?? "").trim();
  if (!token || token.includes(" ")) return null;
  const base = basename(token);
  return !base || SHELL_WRAPPERS.has(base) || TRANSPORT_SCRIPTS.has(base) ? null : base;
}

function confirmMissingCommand(cfg, argv, sshImpl = ssh) {
  const command = directCapabilityCandidate(argv);
  if (!command) return { missingCommand: null, failureReason: null };
  // Only an explicit sentinel over a successful transport counts: an ssh failure,
  // a timeout, or a wrapper's own rc is not evidence that the capability is absent.
  const check = sshImpl(
    cfg,
    `${MISE_ENV}if command -v ${shq(command)} >/dev/null 2>&1; then printf 'rb-capability-v1:present\\n'; else printf 'rb-capability-v1:absent\\n'; fi`,
    { timeoutSec: probeTimeoutSec(cfg), step: `capability-check:${command}` },
  );
  if (check.status !== 0) {
    const detail = String(check.stderr || check.error?.message || "").trim();
    return {
      missingCommand: null,
      failureReason: `capability check command failed on ${cfg.host} for ${command}: rc=${check.status ?? check.error?.code ?? "unknown"}${detail ? ` (${detail})` : ""}`,
    };
  }
  const sentinel = String(check.stdout).trim();
  if (sentinel === "rb-capability-v1:absent") return { missingCommand: command, failureReason: null };
  if (sentinel === "rb-capability-v1:present") return { missingCommand: null, failureReason: null };
  return {
    missingCommand: null,
    failureReason: `capability check returned an invalid response on ${cfg.host} for ${command}: ${JSON.stringify(sentinel)}`,
  };
}

export function remoteArgv(argv, root, mirrorPath) {
  const roots = [root];
  try {
    const real = realpathSync(root);
    if (real !== root) roots.push(real);
  } catch {}
  const underRoot = (candidate) => {
    for (const r of roots) {
      if (candidate === r) return mirrorPath;
      if (candidate.startsWith(`${r}/`)) return mirrorPath + candidate.slice(r.length);
    }
    return null;
  };
  const mapped = argv.map((tok) => {
    if (!tok.startsWith("/")) return tok;
    const norm = normalize(tok);
    const lexical = underRoot(norm);
    if (lexical !== null) return lexical;
    if (TOOL_BINS.has(basename(norm))) return basename(norm);
    try {
      const aliased = underRoot(realpathSync(norm));
      if (aliased !== null) return aliased;
    } catch {}
    throw new Error(`remote dispatch target outside mirror: ${tok} (mirror root: ${mirrorPath})`);
  });
  if (mapped[0] === "node" && TOOL_BINS.has(mapped[1]) && mapped[1] !== "node") mapped.shift();
  return mapped;
}
export function externalLinkTargets(root) {
  const out = new Set();
  try {
    for (const entry of readdirSync(root, { withFileTypes: true })) {
      if (!entry.isSymbolicLink()) continue;
      try {
        const target = realpathSync(join(root, entry.name));
        if (target !== root && !target.startsWith(`${root}/`)) out.add(target);
      } catch {}
    }
  } catch {}
  return [...out];
}
// Ext target is a read-only dependency snapshot: keep dist/build outputs
// (consumers need built .d.ts/js), drop only VCS/caches/worktree junk.
const EXT_TARGET_EXCLUDES = ["node_modules", ".git", ".pnpm-store", ".cache", ".turbo", ".worktrees", ".wt-*", ".tmpjail-work", "tmp", "coverage", "playwright-report", "test-results"];
// A symlink may point at a plain file (a shared CLAUDE.md, an .npmrc). Mirroring
// that as a directory makes the remote `mkdir -p` create a directory AT the file's
// path and every later rsync of `<file>/` fail, permanently, for every job from
// that worktree.
// A directory target mirrors as a tree; a FILE target must mirror as a single file.
// `mkdir -p` on a file's own path creates a directory there and every later
// rsync of `<file>/` fails — permanently, for every job from that worktree.
export function extTargetSyncPlan(target, isDir) {
  if (isDir) {
    return {
      prep: `mkdir -p ${shq(target)}`,
      args: ["-az", "--delete", ...EXT_TARGET_EXCLUDES.flatMap((e) => ["--exclude", e]), `${target}/`, `${target}/`],
    };
  }
  return {
    prep: `if [ -d ${shq(target)} ]; then rm -rf ${shq(target)}; fi; mkdir -p ${shq(dirname(target))}`,
    args: ["-az", target, target],
  };
}
function syncExternalTarget(cfg, target, stateDir) {
  let isDir;
  try { isDir = statSync(target).isDirectory(); } catch (error) {
    return { status: null, error: `ext-target unreadable: ${error.message}` };
  }
  const plan = extTargetSyncPlan(target, isDir);
  const mk = ssh(cfg, plan.prep, { step: `external-prep:${target}` });
  if (mk.status !== 0) return mk;
  const [src, dest] = plan.args.slice(-2);
  const flags = plan.args.slice(0, -2);
  const hash = createHash("sha256").update(target).digest("hex").slice(0, 12);
  return flockedRsync(cfg, `rsync-external:${target}`, join(stateDir, `ext-${hash}.sync.lock`),
    [...flags, "-e", rsyncSsh(cfg), src, `${sshTarget(cfg)}:${dest}`]);
}
function sh(cmd, args, opts = {}) {
  const { cfg = DEFAULT_REMOTE_CONFIG, step = cmd, timeout: requestedTimeout, ...rest } = opts;
  const configuredMs = stepTimeoutMs(cfg, cfg.local_step_timeout_sec ?? DEFAULT_REMOTE_CONFIG.local_step_timeout_sec);
  const timeout = requestedTimeout ? Math.min(requestedTimeout, configuredMs) : configuredMs;
  const result = spawnSync(cmd, args, { encoding: "utf8", timeout, ...rest });
  if (timedOut(result)) reportTimeout(step, cfg, timeout);
  return Object.assign(result, { step, host: cfg.host ?? "fleet", timeoutMs: timeout });
}
let lastStepTransport = false;
let lastStepFailure = null;
function stepFail(log, step, result) {
  lastStepTransport = isTransportFailure(result);
  lastStepFailure = { step, host: result?.host, timedOut: timedOut(result) };
  const err = String(result.stderr || result.error || "").trim().slice(0, 300);
  log(`push-step-failed step=${step} rc=${result.status ?? "?"} err=${err}`);
  console.error(`local-gate: remote push failed at ${step}${err ? `: ${err}` : ` rc=${result.status}`}`);
  return null;
}
export function gitCommonDir(root, cfg = DEFAULT_REMOTE_CONFIG) {
  const result = sh("git", ["-C", root, "rev-parse", "--git-common-dir"], { cfg, step: "git-common-dir" });
  if (result.status !== 0) return null;
  const dir = String(result.stdout).trim();
  try { return realpathSync(dir.startsWith("/") ? dir : join(root, dir)); } catch { return null; }
}
export function repoKey(commonDir) {
  const name = basename(commonDir) === ".git" ? basename(join(commonDir, "..")) : basename(commonDir);
  return `${name}-${createHash("sha256").update(commonDir).digest("hex").slice(0, 12)}`;
}
export function snapshotCommit(root, stateDir, log, cfg = DEFAULT_REMOTE_CONFIG) {
  const head = sh("git", ["-C", root, "rev-parse", "--verify", "HEAD"], { cfg, step: "git-head" });
  if (head.status !== 0) return null;
  const headSha = String(head.stdout).trim();
  const dirty = sh("git", ["-C", root, "status", "--porcelain"], { cfg, step: "git-status" });
  if (dirty.status !== 0) return stepFail(log, "git-status", dirty);
  if (!String(dirty.stdout).trim()) return headSha;
  mkdirSync(stateDir, { recursive: true });
  const index = join(stateDir, `rb-index-${process.pid}`);
  const env = { ...process.env, GIT_INDEX_FILE: index };
  const idEnv = { ...env, GIT_AUTHOR_NAME: "rb", GIT_AUTHOR_EMAIL: "rb@local", GIT_COMMITTER_NAME: "rb", GIT_COMMITTER_EMAIL: "rb@local" };
  try {
    const read = sh("git", ["-C", root, "read-tree", "HEAD"], { env, cfg, step: "git-read-tree" });
    if (read.status !== 0) return stepFail(log, "git-read-tree", read);
    const add = sh("git", ["-C", root, "add", "-A", "--ignore-errors"], { env, cfg, step: "git-add" });
    if (add.status !== 0) return stepFail(log, "git-add", add);
    const tree = sh("git", ["-C", root, "write-tree"], { env, cfg, step: "git-write-tree" });
    if (tree.status !== 0) return stepFail(log, "git-write-tree", tree);
    const commit = sh("git", ["-C", root, "commit-tree", String(tree.stdout).trim(), "-p", headSha, "-m", "rb snapshot"], { env: idEnv, cfg, step: "git-commit-tree" });
    if (commit.status !== 0) return stepFail(log, "git-commit-tree", commit);
    return String(commit.stdout).trim();
  } finally { try { unlinkSync(index); } catch {} }
}
// gitignored files builds need (.dev.vars, .env) travel as a tar overlay, never
// as git objects; files inside ignored dirs (dist/, tmp/) are junk copies — skip.
export function overlayFiles(cfg, root) {
  const wantedList = cfg.ship_ignored || [];
  if (!wantedList.length) return [];
  // Scope the query to the wanted basenames at any depth; an unscoped ignored-file
  // listing includes node_modules (11MB+) and overflows the spawnSync buffer, which
  // would silently drop every overlay file.
  const exact = new Set(wantedList.filter((path) => path.includes("/")));
  const basenames = new Set(wantedList.filter((path) => !path.includes("/")));
  const pathspecs = wantedList.flatMap((path) => path.includes("/") ? [path] : [path, `:(glob)**/${path}`]);
  const listed = sh("git", ["-C", root, "ls-files", "-o", "-i", "--exclude-standard", "-z", "--", ...pathspecs], { maxBuffer: 64 * 1024 * 1024, cfg, step: "git-overlay-list" });
  if (listed.status !== 0) return [];
  const candidates = String(listed.stdout).split("\0").filter((path) => path && (exact.has(path) || basenames.has(basename(path))));
  if (!candidates.length) return [];
  const dirs = [...new Set(candidates.map((p) => p.includes("/") ? p.slice(0, p.lastIndexOf("/")) : ""))].filter(Boolean);
  const ignoredDirs = new Set();
  if (dirs.length) {
    const check = sh("git", ["-C", root, "check-ignore", "-z", "--stdin"], { input: dirs.join("\0"), cfg, step: "git-overlay-check" });
    for (const d of String(check.stdout || "").split("\0").filter(Boolean)) ignoredDirs.add(d);
  }
  return candidates.filter((path) => exact.has(path) || !path.includes("/") || !ignoredDirs.has(path.slice(0, path.lastIndexOf("/"))));
}
function bareUrl(cfg, barePath) { return `ssh://${cfg.ssh_user}@${cfg.host}:${cfg.port}${barePath}`; }
export function bareMirrorEnsureCommand(cfg, key, barePath, origin = "") {
  const configLock = `${barePath}/config.lock`;
  const staleCleanup = `lk=${shq(configLock)}; if [ -f "$lk" ]; then if command -v lsof >/dev/null 2>&1; then [ -z "$(lsof -t -- "$lk" 2>/dev/null)" ] || exit 90; elif command -v fuser >/dev/null 2>&1; then fuser "$lk" >/dev/null 2>&1 && exit 90; fi; rm -f "$lk"; fi`;
  return [
    `mkdir -p ~/.rb ${shq(`${cfg.remote_root}/.repos`)}`,
    `exec 9>~/.rb/bare-${key}.lock`,
    "flock -w 600 9",
    `{ [ -d ${shq(`${barePath}/objects`)} ] || git init --bare -q ${shq(barePath)}; }`,
    staleCleanup,
    `git -C ${shq(barePath)} config receive.shallowUpdate true`,
    ...(origin ? [`printf %s ${shq(origin)} > ${shq(`${barePath}/.rb-origin`)}`] : []),
  ].join(" && ");
}
// bulk object transfer gets a dedicated connection: the shared mux channel drops mid-pack
function gitSshEnv(cfg) {
  return { ...process.env, GIT_SSH_COMMAND: `ssh ${sshOpts(cfg, { mux: false }).join(" ")}` };
}
function syncPushGit(cfg, root, stateDir, mirror, mirrorPath, epoch, log) {
  // deny-all-gitignore repos (build inputs untracked by design) need the full-tree path
  if ((cfg.rsync_only || []).some((p) => root === p || root.startsWith(`${p}/`))) return "no-git";
  const commonDir = gitCommonDir(root, cfg);
  if (!commonDir) return "no-git";
  if (String(sh("git", ["-C", root, "rev-parse", "--is-inside-work-tree"], { cfg, step: "git-worktree-check" }).stdout).trim() !== "true") return "no-git";
  const sha = snapshotCommit(root, stateDir, log, cfg);
  if (sha === null && sh("git", ["-C", root, "rev-parse", "--verify", "HEAD"], { cfg, step: "git-head-check" }).status !== 0) return "no-git";
  if (!sha) return null;
  const key = repoKey(commonDir);
  const barePath = `${cfg.remote_root}/.repos/${key}.git`;
  const ensure = ssh(cfg, bareMirrorEnsureCommand(cfg, key, barePath, commonDir), { timeoutSec: cfg.ssh_exec_timeout_sec, step: "bare-init" });
  if (ensure.status !== 0) return stepFail(log, "bare-init", ensure);
  const refMirror = mirror.replace(/[^A-Za-z0-9_-]/g, "-");
  const ref = `refs/rb/${refMirror}/${epoch}`;
  const refCutoff = String(BigInt(epoch) - 86400000n * 1000000n);
  const push = sh("git", ["-C", root, "push", "-q", "--no-verify", bareUrl(cfg, barePath), `${sha}:${ref}`], { env: gitSshEnv(cfg), cfg, step: "git-push", timeout: execMs(cfg) });
  if (push.status !== 0) return stepFail(log, "git-push", push);
  const overlay = overlayFiles(cfg, root);
  const ovRemote = `.rb/ov-${mirror}.tgz`;
  if (overlay.length) {
    const ovLocal = join(stateDir, `rb-ov-${process.pid}.tgz`);
    try {
      const tar = sh("tar", ["-czf", ovLocal, "-C", root, "--null", "-T", "-"], { input: `${overlay.join("\0")}\0`, cfg, step: "overlay-tar" });
      if (tar.status !== 0) return stepFail(log, "overlay-tar", tar);
      const scpOpts = sshOpts(cfg).map((v) => v === "-p" ? "-P" : v);
      const scp = sh("scp", ["-q", ...scpOpts, ovLocal, `${sshTarget(cfg)}:${ovRemote}`], { timeout: execMs(cfg), cfg, step: "overlay-scp" });
      if (scp.status !== 0) return stepFail(log, "overlay-scp", scp);
    } finally { try { unlinkSync(ovLocal); } catch {} }
  }
  const script = [
    "set -e",
    `exec 9>~/.rb/mat-${mirror}.lock`, "flock -w 600 9",
    `cd ${shq(mirrorPath)}`,
    // Everything below rewrites this tree (checkout -f, clean -ffd). If a seat was running here
    // when the box went down, its uncommitted work is still in the tree and this is the last
    // moment it exists. The reconciler parks it in a ref first. Inside the mat lock by design —
    // the boot unit cannot be relied on to win the race against an early dispatch.
    `if [ -x ~/.rb/seat-reconcile.sh ]; then ~/.rb/seat-reconcile.sh guard ${shq(mirrorPath)} || true; fi`,
    "[ -e .git ] || git init -q .",
    `git remote add origin ${shq(barePath)} 2>/dev/null || git remote set-url origin ${shq(barePath)}`,
    "mkdir -p .git/objects/info",
    `printf '%s\\n' ${shq(`${barePath}/objects`)} > .git/objects/info/alternates`,
    `git fetch -q origin ${shq(`+${ref}:refs/rb/in`)}`,
    "if [ -f .rb-overlay-manifest ]; then xargs -0 -r rm -f -- < .rb-overlay-manifest; rm -f .rb-overlay-manifest; fi",
    `git checkout -qf ${sha}`,
    "git clean -qffd -e .rb-lockhash",
    overlay.length
      ? `tar -xzf ~/${ovRemote} && tar -tzf ~/${ovRemote} | tr '\\n' '\\0' > .rb-overlay-manifest && rm -f ~/${ovRemote}`
      : `rm -f ~/${ovRemote}`,
    `git -C ${shq(barePath)} for-each-ref --format='%(refname)' ${shq(`refs/rb/${refMirror}/*`)} | while IFS= read -r r; do e=\${r##*/}; if [ "$e" -lt ${shq(refCutoff)} ] 2>/dev/null; then git -C ${shq(barePath)} update-ref -d "$r"; fi; done`,
    `printf %s ${shq(epoch)} > .rb-epoch`,
    `printf %s ${shq(root)} > .rb-origin`,
  ].join("\n");
  const materialize = ssh(cfg, script, { timeoutSec: cfg.ssh_exec_timeout_sec, step: "materialize" });
  if (materialize.status !== 0) return stepFail(log, "materialize", materialize);
  return epoch;
}
export function syncPush(cfg, root, stateDir, log = () => {}) {
  const mirror = mirrorName(root); const mirrorPath = `${cfg.remote_root}/${mirror}`; const epoch = nextEpoch(stateDir, cfg.host, mirror);
  const fence = ssh(cfg, `${MISE_ENV}mkdir -p ~/.rb/jobs ${shq(mirrorPath)} && node --version >/dev/null`, { step: "fence" });
  if (fence.status !== 0) return stepFail(log, "fence", fence);
  const stopped = fenceOlderJobs(cfg, mirrorPath, epoch);
  if (stopped.status !== 0) return stepFail(log, "fence-jobs", stopped);
  for (const target of externalLinkTargets(root)) {
    const synced = syncExternalTarget(cfg, target, stateDir);
    if (synced.status !== 0) return stepFail(log, `ext-target:${target}`, synced);
  }
  const viaGit = syncPushGit(cfg, root, stateDir, mirror, mirrorPath, epoch, log);
  if (viaGit !== "no-git") return viaGit;
  const excludes = pushExcludes(cfg, root).flatMap((e) => ["--exclude", e]);
  const result = flockedRsync(cfg, "rsync-push", join(stateDir, `${mirror}.${safeHost(cfg.host)}.sync.lock`), ["-az", "--delete", "-e", rsyncSsh(cfg), ...excludes, `${root}/`, `${sshTarget(cfg)}:${mirrorPath}/`]);
  if (result.status !== 0) return stepFail(log, "rsync-push", result);
  const stamp = ssh(cfg, `printf %s ${shq(epoch)} > ${shq(`${mirrorPath}/.rb-epoch`)} && printf %s ${shq(root)} > ${shq(`${mirrorPath}/.rb-origin`)}`, { step: "epoch-stamp" });
  if (stamp.status !== 0) return stepFail(log, "epoch-stamp", stamp);
  return epoch;
}
function syncPull(cfg, root, stateDir) {
  const mirror = mirrorName(root); const excludes = cfg.pull_excludes.flatMap((e) => ["--exclude", e]);
  return flockedRsync(cfg, "rsync-pull", join(stateDir, `${mirror}.${safeHost(cfg.host)}.sync.lock`), ["-auz", "-e", rsyncSsh(cfg), ...excludes, `${sshTarget(cfg)}:${cfg.remote_root}/${mirror}/`, `${root}/`]);
}

function pidAlive(pid) { try { process.kill(pid, 0); return true; } catch { return false; } }
function activeRemoteJobs(stateDir) {
  try { return readFileSync(join(stateDir, "remote-jobs"), "utf8").split("\n").map(Number).filter((p) => p && pidAlive(p)); } catch { return []; }
}
function saveRemoteJobs(stateDir, pids) { mkdirSync(stateDir, { recursive: true }); writeFileSync(join(stateDir, "remote-jobs"), `${pids.join("\n")}${pids.length ? "\n" : ""}`); }
export function readHostClaims(stateDir, host, sampledAtEpochMs) {
  const dir = join(stateDir, "host-claims", safeHost(host));
  let files;
  try { files = readdirSync(dir); } catch { return 0; }
  let count = 0;
  for (const name of files) {
    const path = join(dir, name);
    let claim;
    try { claim = JSON.parse(readFileSync(path, "utf8")); } catch {}
    const expired = !Number.isInteger(claim?.pid) || !Number.isFinite(claim?.ts)
      || !pidAlive(claim.pid)
      || (typeof sampledAtEpochMs === "number" && sampledAtEpochMs - claim.ts > CLAIM_SETTLE_MS);
    if (expired) {
      try { unlinkSync(path); } catch {}
    } else count += 1;
  }
  return count;
}
export function claimHost(stateDir, host) {
  const dir = join(stateDir, "host-claims", safeHost(host));
  mkdirSync(dir, { recursive: true });
  const ts = Date.now();
  const path = join(dir, `${process.pid}-${ts}-${process.hrtime.bigint()}.json`);
  const fd = openSync(path, "wx");
  try { writeFileSync(fd, JSON.stringify({ pid: process.pid, ts })); } finally { closeSync(fd); }
  return path;
}
export function releaseHostClaim(path) {
  try { if (path) unlinkSync(path); } catch {}
}
function sleepMs(ms) { Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); }
function acquireKeyFile(stateDir, key, argv) {
  const file = join(stateDir, `${key}.key`);
  try { const fd = openSync(file, "wx"); writeFileSync(fd, JSON.stringify({ pid: process.pid, command: argv, started_at: new Date().toISOString() })); return file; }
  catch {
    try { if (pidAlive(Number(JSON.parse(readFileSync(file, "utf8")).pid))) return null; } catch {}
    writeFileSync(file, JSON.stringify({ pid: process.pid, command: argv, started_at: new Date().toISOString() })); return file;
  }
}
function remoteJobCount(cfg) {
  const result = ssh(cfg, "set -o pipefail; systemctl --user list-units 'rb-*.service' --state=running --no-legend --plain 2>/dev/null | awk 'END { print NR }'", { timeoutSec: probeTimeoutSec(cfg), step: "job-count" });
  if (result.status !== 0 || !/^\d+$/.test(String(result.stdout).trim())) {
    return { count: null, unanswered: didNotAnswer(result) };
  }
  return { count: Number(String(result.stdout).trim()), unanswered: false };
}
function hostTelemetry(cfg) {
  const result = ssh(cfg, `cat ${TELEMETRY_REMOTE_PATH} 2>/dev/null || true`, { timeoutSec: probeTimeoutSec(cfg), step: "telemetry" });
  if (result.status !== 0) return { ok: false, reason: "telemetry-missing" };
  return parseHostTelemetry(result.stdout, Date.now());
}
function stickyFile(stateDir, mirror) { return join(stateDir, `${mirror}.host`); }
export function chooseHost(cfg, stateDir, mirror, log, command, deps = {}) {
  const eligible = deps.hostEligibleCli ?? hostEligibleCli;
  const ready = deps.remoteReadyResult ?? remoteReadyResult;
  const countJobs = deps.remoteJobCount ?? remoteJobCount;
  const readTelemetry = deps.hostTelemetry ?? hostTelemetry;
  const rankByLoad = process.env.BUILD_REMOTE_RANK_BY_LOAD !== "0";
  let preferred = ""; try { preferred = readFileSync(stickyFile(stateDir, mirror), "utf8").trim(); } catch {}
  const counts = {};
  const telemetry = {};
  const claims = {};
  const unansweredHosts = [];
  for (const host of cfg.hosts) {
    const verdict = eligible(host, command);
    if (!verdict.eligible) {
      log(`host skipped host=${host} reason=ineligible:${verdict.reason}`);
      continue;
    }
    const candidate = forHost(cfg, host);
    const readiness = ready(candidate, stateDir, log);
    if (!readiness.ready) {
      if (readiness.unanswered) unansweredHosts.push(host);
      log(`host skipped host=${host} reason=not-ready`);
      continue;
    }
    const jobs = countJobs(candidate);
    if (jobs.count === null) {
      if (jobs.unanswered) {
        unansweredHosts.push(host);
        log(`remote dispatch timeout host=${host} timeout_sec=${probeTimeoutSec(candidate)} error=host did not answer`);
      }
      log(`host skipped host=${host} reason=job-count-failed`);
    } else {
      counts[host] = jobs.count;
      if (rankByLoad && jobs.count < cfg.max_remote_jobs) {
        const sample = readTelemetry(candidate) ?? { ok: false, reason: "telemetry-missing" };
        telemetry[host] = sample;
        if (!sample.ok || !Number.isInteger(sample.ciJobsRunning)) {
          log(`host ci-jobs-blind host=${host} reason=${sample.ciJobsReason ?? sample.reason ?? "ci-jobs-missing"}`);
        }
        else if (sample.tempPkg >= sample.tempCrit) log(`host thermal-excluded host=${host} temp_pkg=${sample.tempPkg} temp_crit=${sample.tempCrit}`);
        claims[host] = readHostClaims(stateDir, host, sample.ok ? sample.sampledAtEpochMs : undefined);
      }
    }
  }
  const [host] = rankByLoad
    ? selectHosts(cfg.hosts, preferred, counts, cfg.max_remote_jobs, telemetry, claims)
    : selectHosts(cfg.hosts, preferred, counts, cfg.max_remote_jobs);
  if (!host) return { cfg: null, unansweredHosts };
  if (!rankByLoad) return { cfg: forHost(cfg, host), unansweredHosts };
  const claimFile = claimHost(stateDir, host);
  return { cfg: forHost(cfg, host), unansweredHosts, claimFile };
}
export function jobStateCommand(jobId) {
  const dir = `$HOME/.rb/jobs/${jobId}`;
  return `unit_state=$(systemctl --user is-active rb-${jobId}.service 2>/dev/null || true); case "$unit_state" in active|activating|reloading|deactivating) echo RUNNING;; *) if [ -f ${dir}/rc ]; then rc=$(cat ${dir}/rc); case "$rc" in ''|*[!0-9]*) echo BROKEN;; *) printf 'DONE %s\\n' "$rc";; esac; elif [ -d ${dir} ]; then echo BROKEN; else echo MISSING; fi;; esac`;
}
function jobState(cfg, jobId) {
  return ssh(cfg, jobStateCommand(jobId), { step: "job-state" });
}
function reconnectJobState(cfg, jobId) {
  const graceDeadline = Date.now() + cfg.grace_window_sec * 1000;
  const deadline = Math.min(graceDeadline, cfg.dispatch_deadline_ms ?? graceDeadline);
  for (;;) {
    const state = jobState(cfg, jobId);
    if (!isTransportFailure(state)) return state;
    if (Date.now() >= deadline) return state;
    sleepMs(Math.min(cfg.reconnect_interval_sec * 1000, deadline - Date.now()));
  }
}
function supervisorMode() {
  const mode = process.env.SUPERVISOR_MODE || "opaque";
  if (mode !== "opaque" && mode !== "declared") throw new Error(`invalid supervisor mode: ${mode}`);
  return mode;
}

const ENV_NAME = /^[A-Za-z_][A-Za-z0-9_]*$/;
export function jobEnvArgs(env) {
  return Object.entries(env ?? {}).map(([name, value]) => {
    if (!ENV_NAME.test(name)) throw new Error(`invalid remote env name: ${name}`);
    if (/[\n\0]/.test(String(value))) throw new Error(`invalid remote env value for ${name}`);
    return `--setenv=${shq(`${name}=${value}`)}`;
  });
}

function startJob(cfg, { jobId, key, mirrorPath, rel, epoch, argv, env, privateInput }, deps = {}) {
  const remoteSsh = deps.ssh ?? ssh;
  const dir = `$HOME/.rb/jobs/${jobId}`;
  const mode = supervisorMode();
  const input = privateInput === undefined ? null : Buffer.from(privateInput);
  const inputPath = `${dir}/input`;
  const inputSetup = input !== null
    ? `umask 077 && head -c ${input.length} > ${inputPath} && [ "$(wc -c < ${inputPath})" -eq ${input.length} ] && `
    : "";
  const inputEnv = input !== null ? `--setenv=BUILD_REMOTE_INPUT_FILE=$HOME/.rb/jobs/${jobId}/input ` : "";
  const runnerArgs = `${RUNNER_REMOTE} ${dir} ${shq(mirrorPath)} ${shq(rel)} ${argv.map(shq).join(" ")}`;
  const runnerCommand = input !== null
    ? `bash -c ${shq('trap \'rm -f -- "$BUILD_REMOTE_INPUT_FILE"\' EXIT; "$@"')} rb-private bash ${runnerArgs}`
    : `bash ${runnerArgs}`;
  const failureCleanup = input !== null ? `; rc=$?; [ "$rc" -eq 0 ] || rm -f ${inputPath}; exit "$rc"` : "";
  const meta = Buffer.from(JSON.stringify({ key, mirror: mirrorPath, epoch, argv, env: env ?? {}, supervisor_mode: mode, started_at: new Date().toISOString() })).toString("base64");
  const command = `rm -rf ${dir} && mkdir -p ${dir} && ${inputSetup}(systemctl --user reset-failed rb-${jobId}.service 2>/dev/null || true) && printf %s ${shq(meta)} | base64 -d > ${dir}/meta.json && systemd-run --user --collect --quiet --unit rb-${jobId} --setenv=SUPERVISOR_MODE=${shq(mode)} ${inputEnv}${jobEnvArgs(env).join(" ")} -- ${runnerCommand}${failureCleanup}`;
  return remoteSsh(cfg, command, { step: "start-job", ...(input !== null ? { input } : {}) });
}
// a log poll returns at once; the hold is a test hook that delays before exec
const REMOTE_WATCH_CHUNK_BYTES = 64 * 1024;
const REMOTE_WATCH_MAX_BUFFER_BYTES = 256 * 1024;
function watchSsh(cfg, command) {
  const args = ["ssh", ...sshOpts(cfg), sshTarget(cfg), command];
  const hold = Number(process.env.BUILD_REMOTE_WATCHER_HOLD_SEC || 0);
  const timeout = probeMs(cfg) + hold * 1000;
  const options = { timeout, maxBuffer: REMOTE_WATCH_MAX_BUFFER_BYTES };
  const result = process.env.BUILD_REMOTE_WATCHER_PID_FILE
    ? spawnSync("bash", ["-c", `echo $$ > "$1"; hold="$2"; shift 2; [ -n "$hold" ] && sleep "$hold"; exec "$@"`, "rb-watch", process.env.BUILD_REMOTE_WATCHER_PID_FILE, process.env.BUILD_REMOTE_WATCHER_HOLD_SEC || "", ...args], options)
    : spawnSync(args[0], args.slice(1), options);
  if (timedOut(result)) healMux(cfg);
  return result;
}
export function watchJob(cfg, jobId, log, deps = {}) {
  const poll = deps.watchSsh ?? watchSsh;
  const sleep = deps.sleep ?? sleepMs;
  const now = deps.now ?? Date.now;
  const reconnectMs = (() => {
    const seconds = Number(cfg.reconnect_interval_sec ?? DEFAULT_REMOTE_CONFIG.reconnect_interval_sec);
    return Number.isFinite(seconds) && seconds > 0 ? seconds * 1000 : 1000;
  })();
  const graceMs = remoteGraceSec(cfg) * 1000;
  let offset = 0;
  let failedSince = null;
  let missingSince = null;
  const transportExpired = () => {
    log(`remote reconnect grace expired host=${cfg.host} job=${jobId}`);
    return null;
  };
  for (;;) {
    const command = `d=$HOME/.rb/jobs/${jobId}; state=M; if [ -f "$d/rc" ]; then rc=$(cat "$d/rc"); case "$rc" in ''|*[!0-9]*) :;; *) state="C:$rc";; esac; fi; if [ "$state" = M ]; then unit_state=$(systemctl --user is-active rb-${jobId}.service 2>/dev/null || true); case "$unit_state" in active|activating|reloading|deactivating) state=R;; *) [ -d "$d" ] && state=I;; esac; fi; size=0; [ -f "$d/log" ] && size=$(wc -c < "$d/log"); available=$(( size - ${offset} )); [ "$available" -lt 0 ] && available=0; count="$available"; [ "$count" -gt ${REMOTE_WATCH_CHUNK_BYTES} ] && count=${REMOTE_WATCH_CHUNK_BYTES}; if [ "$count" -gt 0 ]; then dd if="$d/log" iflag=skip_bytes,count_bytes skip=${offset} count="$count" status=none || exit 1; fi; if [ "\${state#C:}" != "$state" ]; then printf '\\036RBCTL:%s\\036' "$state" >&2; elif [ "$state" = R ]; then printf '\\036RBCTL:R\\036' >&2; elif [ "$state" = T ]; then printf '\\036RBCTL:T\\036' >&2; elif [ "$state" = I ] && [ -d "$d" ]; then printf '\\036RBCTL:I\\036' >&2; else printf '\\036RBCTL:M\\036' >&2; fi`;
    const result = poll(cfg, command);
    const status = result.status ?? (result.signal ? 255 : 1);
    if (status === 0) {
      failedSince = null;
      const output = Buffer.isBuffer(result.stdout) ? result.stdout : Buffer.from(result.stdout || "");
      const control = (Buffer.isBuffer(result.stderr) ? result.stderr.toString("ascii") : String(result.stderr || "")).match(/\x1eRBCTL:(R|C:\d+|T|I|M)\x1e/)?.[1] ?? null;
      if (output.length) { process.stdout.write(output); offset += output.length; }
      // a full chunk means the log still has bytes past this offset: drain them
      // before reporting the exit code, or the job's tail is lost
      if (control?.startsWith("C:")) {
        if (output.length === REMOTE_WATCH_CHUNK_BYTES) continue;
        return { status: Number(control.slice(2)), classification: "remote-rc" };
      }
      if (control === "R" || control === "T") {
        missingSince = null;
      } else if (control) {
        const observed = now();
        missingSince ??= observed;
        if (observed - missingSince >= graceMs) {
          log(`remote watcher terminal host=${cfg.host} job=${jobId} classification=remote-job-disappeared status=254`);
          return { status: 254, classification: "remote-job-disappeared" };
        }
      }
    } else if (status === 255) {
      const observed = now();
      failedSince ??= observed;
      if (observed - failedSince >= graceMs) return transportExpired();
    } else {
      log(`remote watcher state failure host=${cfg.host} job=${jobId} status=${status}`);
      return null;
    }
    sleep(reconnectMs);
  }
}

function remoteCleanupPaths(paths) {
  if (!Array.isArray(paths)) throw new Error("remote cleanup paths must be an array");
  return [...new Set(paths)].map((path) => {
    if (typeof path !== "string" || path.startsWith("/") || path.includes("\\") || path.split("/").some((part) => !part || part === "." || part === "..")) {
      throw new Error(`invalid remote cleanup path: ${path}`);
    }
    return path;
  });
}

export function tryRemoteBuild({ key, argv, cwd, stateDir, log, env, cleanupRemote, privateInput }, deps = {}) {
  const seatGate = shouldSkipRemoteForSeatHost(process.env, deps);
  if (seatGate.error) {
    const reason = seatGate.reason ?? "seat-host-mismatch";
    log(`${reason} key=${key}`);
    return { ran: false, reason };
  }
  if (seatGate.skip) {
    return { ran: false, reason: "seat-local-host" };
  }
  if (deps.tryK3sBuild || k3sConfig().enabled) {
    const k3sExemption = localExemption(argv, deps.k3sDeps?.localOnly ?? k3sConfig().localOnly);
    if (k3sExemption) return { ran: false, reason: `local_only:${k3sExemption.id}` };
    const k3sRoot = projectRoot(cwd);
    const k3sMirror = mirrorName(k3sRoot);
    const fanout = (deps.pushMirrorFanout ?? pushMirrorFanout)(k3sRoot, k3sMirror, stateDir, log);
    if (fanout.succeeded.length > 0) {
      const k3sMirrorPath = join(K3S_HOSTPATH_ROOT, basename(DEFAULT_REMOTE_CONFIG.remote_root), k3sMirror);
      let k3sArgv;
      try {
        k3sArgv = remoteArgv(argv, k3sRoot, k3sMirrorPath);
      } catch (err) {
        log(`k3s fallback key=${key} reason=argv-outside-mirror detail=${err.message}`);
        k3sArgv = null;
      }
      if (k3sArgv) {
        const k3sAttempt = (deps.tryK3sBuild ?? tryK3sBuild)(
          { key, argv: k3sArgv, cwd, stateDir, log, env, workingDir: k3sMirrorPath, nodeNames: fanout.succeeded, epoch: fanout.epoch },
          deps.k3sDeps,
        );
        if (k3sAttempt) return k3sAttempt;
      }
    } else {
      log(`k3s fallback key=${key} reason=mirror-push-failed hosts=${fanout.failed.map((f) => f.host).join(",")}`);
    }
  }
  const loadConfig = deps.loadRemoteConfig ?? loadRemoteConfig;
  const choose = deps.chooseHost ?? chooseHost;
  const reconnect = deps.reconnectJobState ?? reconnectJobState;
  const readMeta = deps.readRemoteJobMeta ?? readRemoteJobMeta;
  const push = deps.syncPush ?? syncPush;
  const sweep = deps.sweepRemoteMirrors ?? sweepRemoteMirrors;
  const start = deps.startJob ?? startJob;
  const report = deps.emitJobReport ?? emitJobReport;
  const watch = deps.watchJob ?? watchJob;
  const remoteSsh = deps.ssh ?? ssh;
  const pull = deps.syncPull ?? syncPull;
  setRemoteLog(log);
  const cleanupPaths = remoteCleanupPaths(cleanupRemote ?? []);
  let cleanupCfg = null;
  let cleanupMirrorPath = "";
  let cleanupArmed = false;
  const loadedCfg = loadConfig(); if (!loadedCfg) return { ran: false, reason: "disabled" };
  const baseCfg = {
    ...loadedCfg,
    dispatch_deadline_ms: Date.now() + positiveSeconds(loadedCfg.dispatch_timeout_sec, DEFAULT_REMOTE_CONFIG.dispatch_timeout_sec) * 1000,
  };
  const originalArgv = [...argv];
  const command = deriveCommandIdentity(originalArgv);
  const exempt = localExemption(originalArgv, baseCfg.local_only);
  if (exempt) return { ran: false, reason: `local_only:${exempt.id}` };
  const root = projectRoot(cwd, baseCfg); const rel = relative(root, cwd) || "."; const mirror = mirrorName(root);
  const keyFile = acquireKeyFile(stateDir, key, argv);
  if (keyFile === null) { log(`deduped key=${key} via=remote`); console.log(`local-gate: deduped key=${key}`); return { ran: true, status: 0, deduped: true }; }
  saveRemoteJobs(stateDir, [...activeRemoteJobs(stateDir), process.pid]);
  let claimFile;
  const done = () => { releaseHostClaim(claimFile); saveRemoteJobs(stateDir, activeRemoteJobs(stateDir).filter((p) => p !== process.pid)); try { unlinkSync(keyFile); } catch {} };
  try {
    const selection = choose(baseCfg, stateDir, mirror, log, command);
    claimFile = selection?.claimFile;
    // Keep the dependency seam compatible with existing callers that inject a
    // selected config directly; the production selector also reports exactly
    // which hosts exhausted their response deadline.
    const cfg = Object.hasOwn(selection ?? {}, "cfg") ? selection.cfg : selection;
    if (!cfg) {
      const unansweredHosts = selection?.unansweredHosts ?? [];
      if (unansweredHosts.length) {
        const hosts = unansweredHosts.join(",");
        log(`remote fallback key=${key} reason=dispatch-timeout hosts=${hosts}`);
        return { ran: false, reason: `dispatch-timeout:${hosts}` };
      }
      return { ran: false, reason: "unreachable-or-full", step: "host-selection", host: "fleet" };
    }
    // a transport-class failure trips the host in the health cache the next
    // chooseHost reads, so the retry spills to the next host — never to local
    const trip = (reason, step, result) => {
      markHostUnhealthy(cfg, stateDir, reason, log);
      return { ran: false, reason, step, host: cfg.host, timedOut: timedOut(result) };
    };
    const mirrorPath = `${cfg.remote_root}/${mirror}`; const jobId = deriveJobId(key, mirror);
    const remoteCommand = remoteArgv(argv, root, mirrorPath);
    cleanupCfg = cfg;
    cleanupMirrorPath = mirrorPath;
    let state = reconnect(cfg, jobId);
    if (isTransportFailure(state)) return trip("transport-grace-expired", "job-state", state);
    let epoch;
    let startedAt;
    if (String(state.stdout).trim() === "RUNNING") {
      const meta = readMeta(cfg, jobId);
      if (!meta?.epoch || !meta.started_at) return { ran: false, reason: "invalid-job-meta" };
      epoch = meta.epoch;
      startedAt = meta.started_at;
      report(cfg, { key, mirror, argv: originalArgv, epoch, stage: "started", startedAt });
    } else if (["MISSING", "DONE", "BROKEN"].some((value) => String(state.stdout).trim().startsWith(value))) {
      lastStepTransport = false;
      lastStepFailure = null;
      cleanupArmed = cleanupPaths.length > 0;
      epoch = push(cfg, root, stateDir, log);
      if (!epoch) {
        const detail = { ran: false, reason: "push-failed", step: lastStepFailure?.step ?? "push", host: lastStepFailure?.host ?? cfg.host, timedOut: lastStepFailure?.timedOut ?? false };
        return lastStepTransport ? trip("push-failed", detail.step, detail.timedOut ? { error: { code: "ETIMEDOUT" } } : null) : detail;
      }
      sweep({ cfg, stateDir, ssh: remoteSsh, log });
      state = start(cfg, { jobId, key, mirrorPath, rel, epoch, argv: remoteCommand, env, privateInput }, { ssh: remoteSsh });
      if (isTransportFailure(state)) {
        cleanupArmed = false;
        state = reconnect(cfg, jobId);
        // start() already went out; whether the job took is unknown. Callers that must not
        // double-dispatch need that distinguished from the pre-dispatch expiry above.
        if (isTransportFailure(state)) return trip("transport-grace-expired-after-start", "start-confirmation", state);
        if (!String(state.stdout).trim().match(/^(RUNNING|DONE )/)) return { ran: false, reason: `start-ambiguous:${String(state.stdout).trim()}` };
      } else if (state.status !== 0) return { ran: false, reason: "start-failed" };
      else cleanupArmed = false;
      const meta = readMeta(cfg, jobId);
      if (!meta?.epoch || !meta.started_at) return { ran: false, reason: "invalid-job-meta" };
      epoch = meta.epoch;
      startedAt = meta.started_at;
      report(cfg, { key, mirror, argv: originalArgv, epoch, stage: "started", startedAt });
    } else return { ran: false, reason: `unexpected-job-state:${String(state.stdout).trim()}` };
    delete cfg.dispatch_deadline_ms;
    const watched = watch(cfg, jobId, log, { startedAt });
    if (watched === null) return trip("transport-grace-expired-after-start", "job-watch", { error: { code: "ETIMEDOUT" } });
    const { status, classification } = watched;
    if (["remote-rc", "remote-job-disappeared"].includes(classification)) cleanupArmed = cleanupPaths.length > 0;
    const finishedAt = new Date().toISOString();
    const capabilityCheck = status === 127
      ? confirmMissingCommand(cfg, originalArgv, remoteSsh)
      : { missingCommand: null, failureReason: null };
    const { missingCommand } = capabilityCheck;
    if (capabilityCheck.failureReason) log(capabilityCheck.failureReason);
    report(cfg, { key, mirror, argv: originalArgv, epoch, stage: "finished", startedAt, rc: status, finishedAt, missingCommand, classification });
    // A command that never started on the box is an infrastructure gap, not the
    // command's verdict: hand it back unran so the caller can decide locally.
    if (missingCommand) {
      log(`remote missing-command key=${key} host=${cfg.host} command=${missingCommand}`);
      return { ran: false, reason: "remote-command-missing", command: missingCommand, host: cfg.host };
    }
    if (classification !== "remote-rc") {
      log(`remote done key=${key} host=${cfg.host} status=${status} classification=${classification}`);
      return { ran: true, status, classification };
    }
    const epochCommand = `${MISE_ENV}cat ${shq(`${mirrorPath}/.rb-epoch`)} && printf '\\n' && node -e ${shq(`const fs=require('fs'),j=process.env.HOME+'/.rb/jobs/${jobId}';process.stdout.write(JSON.parse(fs.readFileSync(j+'/meta.json')).epoch+'\\n'+(fs.existsSync(j+'/fenced')?'1':'0'))`)}`;
    const current = remoteSsh(cfg, epochCommand, { step: "epoch-read" });
    if (current.status !== 0) return isTransportFailure(current)
      ? trip("epoch-read-failed", "epoch-read", current)
      : { ran: false, reason: "epoch-read-failed", step: "epoch-read", host: cfg.host };
    const [mirrorEpoch, jobEpoch, fenced] = String(current.stdout).trim().split("\n");
    // A fence stops the job before the fencing push stamps the new epoch, so the
    // epoch comparison below still reads equal and would call this a real failure.
    if (fenced === "1") return { ran: false, reason: "fenced" };
    try { if (compareEpoch(jobEpoch, mirrorEpoch) !== 0) return { ran: false, reason: "epoch-mismatch" }; } catch { return { ran: false, reason: "invalid-epoch" }; }
    const pulled = pull(cfg, root, stateDir);
    const pullOk = pulled === true || pulled?.status === 0;
    if (!pullOk && status === 0) return { ran: false, reason: "pull-failed", step: pulled?.step ?? "rsync-pull", host: pulled?.host ?? cfg.host, timedOut: timedOut(pulled) };
    mkdirSync(stateDir, { recursive: true }); writeFileSync(stickyFile(stateDir, mirror), `${cfg.host}\n`);
    log(`remote done key=${key} host=${cfg.host} status=${status}`); return { ran: true, status };
  } finally {
    if (cleanupArmed && cleanupCfg) {
      const targets = cleanupPaths.map((path) => shq(`${cleanupMirrorPath}/${path}`));
      const cleaned = remoteSsh(cleanupCfg, `rm -rf -- ${targets.join(" ")}`);
      if (cleaned.status !== 0) log(`remote overlay cleanup failed host=${cleanupCfg.host} key=${key}`);
    }
    done();
  }
}

export function remoteDoctor(deps = {}) {
  const spawn = deps.spawnSync ?? spawnSync;
  const readConfig = deps.readConfig ?? (() => JSON.parse(readFileSync(REMOTE_CONFIG_PATH, "utf8")));
  const getHosts = deps.registryBuildHosts ?? registryBuildHosts;
  const runnerExists = deps.runnerExists ?? existsSync;
  const print = deps.print ?? console.log;
  const audit = deps.auditHostSeat ?? auditHostSeat;
  const out = []; const pass = (s) => out.push(`PASS ${s}`); const fail = (s) => out.push(`FAIL ${s}`);
  setRemoteLog((message) => out.push(`INFO ${message}`));
  let raw = null; try { raw = readConfig(); } catch {}
  let hosts = [];
  if (!raw) fail(`config missing/invalid: ${REMOTE_CONFIG_PATH}`); else if (!raw.enabled) out.push("WARN remote offload disabled (enabled=false)");
  if (raw?.enabled) {
    try { hosts = getHosts(); } catch (err) { fail(`buildbox registry unusable: ${err.message}`); }
  }
  for (const tool of ["ssh", "scp", "rsync", "flock", "timeout"]) {
    spawn("which", [tool], { stdio: "ignore", timeout: 5000 }).status === 0 ? pass(`local tool ${tool}`) : fail(`local tool ${tool} missing`);
  }
  if (!runnerExists(RUNNER_LOCAL)) fail(`runner missing: ${RUNNER_LOCAL}`);
  if (!runnerExists(SUPERVISOR_LOCAL)) fail(`supervisor missing: ${SUPERVISOR_LOCAL}`);
  const cfg = raw ? { ...DEFAULT_REMOTE_CONFIG, ...raw, hosts } : null;
  const auditOptKeys = new Set(["version", "rev", "ghToken", "skipGhAuth", "gitLsRemoteUrl", "repoRoot", "manifestPath", "home", "minDiskGb", "ghTokenFingerprint", "ccpVersion"]);
  const auditOpts = {};
  for (const [key, value] of Object.entries(deps)) if (auditOptKeys.has(key)) auditOpts[key] = value;
  for (const host of hosts) {
    const startedAt = Date.now();
    const hostTimeoutSec = positiveSeconds(cfg.doctor_host_timeout_sec, DEFAULT_REMOTE_CONFIG.doctor_host_timeout_sec);
    const deadline = startedAt + hostTimeoutSec * 1000;
    const hostSpawn = (cmd, args, options = {}) => {
      const remaining = deadline - Date.now();
      if (remaining <= 0) return timeoutFailure("doctor-host", host, hostTimeoutSec * 1000);
      const requested = Number(options.timeout);
      const timeout = Number.isFinite(requested) && requested > 0 ? Math.min(requested, remaining) : remaining;
      return spawn(cmd, args, { ...options, timeout: Math.max(1, timeout) });
    };
    const hc = { ...forHost(cfg, host), dispatch_deadline_ms: deadline };
    const hostStart = out.length;
    const probe = ssh(hc, "echo ok", { step: "ssh-probe", reportTimeout: false }, { spawnSync: hostSpawn });
    if (probe.status !== 0) {
      if (timedOut(probe)) fail(`${host} step=ssh-probe timed out after ${hostTimeoutSec}s`);
      else fail(`ssh ${sshTarget(hc)} unreachable`);
      out.push(`INFO ${host} doctor status=FAIL elapsed_ms=${Date.now() - startedAt}`);
      continue;
    }
    pass(`ssh ${sshTarget(hc)}`);
    const jobs = ssh(hc, "set -o pipefail; systemctl --user list-units 'rb-*.service' --state=running --no-legend --plain 2>/dev/null | awk '{print $1}'", { step: "running-jobs", reportTimeout: false }, { spawnSync: hostSpawn });
    if (timedOut(jobs)) fail(`${host} step=running-jobs timed out after ${hostTimeoutSec}s`);
    else if (jobs.status !== 0) fail(`${host} running jobs unavailable`);
    else out.push(`INFO ${host} running jobs: ${String(jobs.stdout).trim() || "none"}`);
    const versions = ssh(hc, `export PATH="$HOME/.cargo/bin:$HOME/.local/bin:$PATH"; eval "$(mise env -s bash)" 2>/dev/null; echo node=$(node --version 2>/dev/null); echo pnpm=$(pnpm --version 2>/dev/null); echo bun=$(bun --version 2>/dev/null); command -v rsync >/dev/null && echo rsync=ok || echo rsync=MISSING`, { step: "tool-versions", reportTimeout: false }, { spawnSync: hostSpawn });
    if (timedOut(versions)) fail(`${host} step=tool-versions timed out after ${hostTimeoutSec}s`);
    const remote = Object.fromEntries(String(versions.stdout || "").trim().split("\n").filter(Boolean).map((line) => line.split("=")));
    for (const [tool, args] of [["node", ["--version"]], ["pnpm", ["--version"]], ["bun", ["--version"]]]) {
      // cwd=HOME: packageManager pins in a project cwd would skew the parity check
      const local = hostSpawn(tool, args, { encoding: "utf8", cwd: HOME }); const value = String(local.stdout || "").trim();
      value && remote[tool] === value ? pass(`${host} ${tool} parity ${value}`) : fail(`${host} ${tool} parity local=${value || "missing"} remote=${remote[tool] || "missing"}`);
    }
    remote.rsync === "ok" ? pass(`${host} remote rsync`) : fail(`${host} remote rsync missing`);
    const seatDeps = { ...deps, spawnSync: hostSpawn };
    for (const line of audit(hc, auditOpts, seatDeps)) out.push(`${line.level} ${line.msg}`);
    if (Date.now() >= deadline) fail(`${host} step=doctor-host timed out after ${hostTimeoutSec}s`);
    const hostFailed = out.slice(hostStart).some((line) => line.startsWith("FAIL"));
    out.push(`INFO ${host} doctor status=${hostFailed ? "FAIL" : "PASS"} elapsed_ms=${Date.now() - startedAt}`);
  }
  print(out.join("\n")); return out.some((line) => line.startsWith("FAIL")) ? 1 : 0;
}
