From 1519f38aca48170151b4372ff232dc872e7e7c9f Mon Sep 17 00:00:00 2001 From: Ryan Rozich Date: Sat, 17 Oct 2026 03:09:22 -0500 Subject: [PATCH] =?UTF-8?q?feat(cli):=20CTC-4646=20=E2=80=94=20a=20self-ho?= =?UTF-8?q?sted=20runner=20moves=20to=20the=20approved=20release=20on=20it?= =?UTF-8?q?s=20own,=20between=20phases?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `catalyst runner update` reads the release Catalyst approved for self-hosted machines (GET /api/v1/hosts/desired-release, read with the host's own credential from inside a one-off supervisor container, so the secret never leaves Docker). It moves only when the release parses strictly (three digests from the three Catalyst repositories, never a tag or latest, release id matching the digests, a generation newer than the last applied), no phase container exists for this host, all three images pull natively by digest, and, on a Mac with private thoughts storage, that storage re-attests on the new supervisor with the same authority. Between phases is enforced, not sampled: after the pulls it stops the supervisor so it admits nothing, looks again, and if a phase started in between it brings the supervisor back on its old images and waits. A release counts as applied only when the running containers are its images and the supervisor logged host_ready since it started, so a .env an interrupted run left behind is reconciled, never trusted. Every path that does not end on the new release attests Mac thoughts storage back onto the images the machine keeps. The lock is taken over only when its owner process is gone. Then it rewrites only the three image lines in .env (keeping the previous file as .env.release-previous), runs docker compose up, and waits up to five minutes for the new supervisor to log host_ready. If it does not, the previous file and images come back and the run exits 1. Every other answer keeps the current images. release-state.json records the applied generation and the last outcome. After a verified update it removes the images it replaced (a runner image is about 9 GB against the 25 GiB free-disk floor); a rollback or failure keeps them. `catalyst runner auto-update on|off|status` schedules the check every 15 minutes (a LaunchAgent on macOS, on calendar minutes 0/15/30/45 because a StartInterval job sat pended for hours on a busy Mac, a systemd user timer on Linux, which says when lingering is off). Setup turns it on after the runner starts; a failure there is reported and never fails setup. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LGe7UXEGNZW1GenNuUZRDU Change-Id: I9624b6ebe5c2ca63f14cb48f9fff34e2deaec99f --- src/args.ts | 6 + src/capabilities.ts | 1 + src/cli.ts | 5 + src/onboard-runner.ts | 54 ++-- src/onboard-runtime.ts | 4 + src/runner-auto-update.ts | 199 +++++++++++++ src/runner-command.ts | 93 ++++++ src/runner-update.ts | 596 +++++++++++++++++++++++++++++++++++++ test/runner-update.test.ts | 435 +++++++++++++++++++++++++++ 9 files changed, 1376 insertions(+), 17 deletions(-) create mode 100644 src/runner-auto-update.ts create mode 100644 src/runner-command.ts create mode 100644 src/runner-update.ts create mode 100644 test/runner-update.test.ts diff --git a/src/args.ts b/src/args.ts index e294f17b..2c7fbb47 100644 --- a/src/args.ts +++ b/src/args.ts @@ -52,6 +52,11 @@ const GLOBAL_FLAGS: FlagTable = { /** Per-verb flag tables. A verb absent here accepts only the global flags. */ export const FLAG_TABLES: Record = { hosts: { yes: { value: false, help: "confirm removal without a terminal" } }, + runner: { + check: { value: false, help: "read the approved release and say what would change, without changing it" }, + auto: { value: false, help: "one-line output for the scheduled run" }, + dir: { value: true, help: "the runner directory (default: the one setup created)" }, + }, login: { "start-replica": { value: false, @@ -532,6 +537,7 @@ export const FLAG_TABLES: Record = { export const VERB_USAGE: Record = { hosts: "hosts [list] | hosts remove [--yes] | hosts rename [--json]", + runner: "runner update [--check] [--dir ] | runner auto-update [status|on|off]", setup: "setup [onboard options] (sets up Catalyst step by step; same as catalyst onboard) | setup --engine --engine-sha256 [-- ] (run by install.sh; install.sh --help lists the options)", login: diff --git a/src/capabilities.ts b/src/capabilities.ts index 4ffbeaa6..c5ca9064 100644 --- a/src/capabilities.ts +++ b/src/capabilities.ts @@ -41,6 +41,7 @@ const agent = (method: "GET" | "POST", name: string) => ({ method, path: `/api/v export const CAPABILITIES: readonly Capability[] = [ {verb:"hosts",does:"list this account's machines, current use and live capacity",needs:"member",routes:[],since:"0.16.0"}, {verb:"hosts remove|rename",does:"remove or rename a self-hosted machine in this account",needs:"admin",routes:[],since:"0.16.0"}, + {verb:"runner update|auto-update",does:"move this machine's self-hosted runner to the release Catalyst approved, and check for one every 15 minutes",needs:"member",routes:[],since:"0.16.16"}, // CTC-4680: plain `catalyst setup` runs the onboard flow, so install.sh may hand off to it. install.sh // reads this flag rather than `setup --help`, which already exits 0 on 0.15.x. { verb: "setup", does: "set up Catalyst step by step, or show install.sh's plan, progress and sign-in in this terminal", needs: "member", routes: [], since: "0.15.0", engineProtocol: 1, bootstrapHandoff: true, bootstrapHandoffVersion: 2 }, diff --git a/src/cli.ts b/src/cli.ts index 28c084da..8c721242 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -1,4 +1,5 @@ import { cmdHosts, type HostsDeps } from "./hosts.js"; +import { cmdRunner, type RunnerCommandDeps } from "./runner-command.js"; import { cmdSeat } from "./seat.js"; // cli.ts — the `catalyst` dispatcher (npm: @catalyst-cloud/cli). Verbs live in // their own modules; this file keeps every export the 0.1 tests and the bin import. @@ -207,6 +208,7 @@ export function usageText(): string { " catalyst status | notice | me | ready | accounts", " catalyst seat checkout --size big|med --harness claude,codex,opencode [--session-id ID] [--stage STAGE]", " catalyst hosts [list] | hosts remove [--yes] | hosts rename ", + " catalyst runner update [--check] | runner auto-update [status|on|off] (move this machine's runner to the approved release)", " catalyst mcp add|list|remove (vault references only)", " catalyst contract [--refresh] [--path ]", " catalyst query |pulls|pull |projects|cycles|search |changes --since >", @@ -267,6 +269,7 @@ export interface MainDeps { var?: VarDeps; connections?: ConnectionsDeps; hosts?: HostsDeps; + runner?: RunnerCommandDeps; identity?: IdentityDeps; team?: TeamDeps; legacy?: LegacyDeps; @@ -749,6 +752,8 @@ export async function main( return await cmdIdentity(args, ctx, deps.identity ?? {}); case "hosts": return await cmdHosts(args, ctx, deps.hosts); + case "runner": + return await cmdRunner(args, ctx, deps.runner); case "connections": return await cmdConnections(args, ctx, deps.connections ?? {}); default: diff --git a/src/onboard-runner.ts b/src/onboard-runner.ts index aee51aee..016e2e7c 100644 --- a/src/onboard-runner.ts +++ b/src/onboard-runner.ts @@ -69,7 +69,7 @@ const CUSTODY_KEYS = [ ] as const; /** A persisted authority selects bindings, never substitutes for a fresh native proof. */ -function persistedCustody(dir: string, values: Record | null): "absent" | "invalid" | "installed" { +export function persistedCustody(dir: string, values: Record | null): "absent" | "invalid" | "installed" { const present = CUSTODY_KEYS.map(key => values?.[key] ?? ""); if (present.every(value => !value)) return "absent"; if (present.some(value => !value)) return "invalid"; @@ -186,7 +186,8 @@ export type RunnerExec = ( options: RunnerExecOptions, ) => Promise<{ code: number; stdout: string; stderr?: string }>; -const spawnDocker: RunnerExec = (args, options) => +/** The docker CLI as a child process. Exported for `catalyst runner update`. */ +export const spawnDocker: RunnerExec = (args, options) => new Promise((resolve, reject) => { const child = spawn("docker", args, { env: options.env, @@ -221,6 +222,23 @@ function composeErrorText(raw: string | undefined): string | undefined { return text !== "" ? text : undefined; } +/** The Compose arguments for the installation in `dir`, with the Darwin custody file when its + * persisted authority is installed. Null when that authority is only partly present. */ +export function runnerComposeArgs(dir: string): string[] | null { + const custody = persistedCustody(dir, readEnvFile(join(dir, ".env"))); + if (custody === "invalid") return null; + return [ + "compose", + "--project-name", + RUNNER_PROJECT, + "--file", + join(dir, "compose.yaml"), + ...(custody === "installed" ? ["--file", join(dir, "compose.darwin-thoughts.yaml")] : []), + "--env-file", + join(dir, ".env"), + ]; +} + /** The real engine, through the docker CLI. Compose reads `.env` only: the caller's CATALYST_ and * DOCKER_SOCKET_ variables are removed, because Compose prefers an exported value over the file. */ export function dockerRunnerEngine( @@ -242,20 +260,7 @@ export function dockerRunnerEngine( return { code: 1, stdout: "" }; } }; - const compose = (dir: string): string[] | null => { - const custody = persistedCustody(dir, readEnvFile(join(dir, ".env"))); - if (custody === "invalid") return null; - return [ - "compose", - "--project-name", - RUNNER_PROJECT, - "--file", - join(dir, "compose.yaml"), - ...(custody === "installed" ? ["--file", join(dir, "compose.darwin-thoughts.yaml")] : []), - "--env-file", - join(dir, ".env"), - ]; - }; + const compose = runnerComposeArgs; const supervisorShell = (dir: string, script: string): string[] | null => { const files = compose(dir); if (!files) return null; @@ -682,13 +687,18 @@ function parseHosts(body: unknown): ListedHost[] | null { } /** `KEY=value` lines; a single-quoted value is unquoted. Anything else is ignored. */ -function readEnvFile(path: string): Record | null { +export function readEnvFile(path: string): Record | null { let text: string; try { text = readFileSync(path, "utf8"); } catch { return null; } + return parseEnvText(text); +} + +/** `KEY=value` lines of `.env` text, as readEnvFile reads them. */ +export function parseEnvText(text: string): Record { const values: Record = {}; for (const line of text.split("\n")) { const match = /^([A-Z][A-Z0-9_]*)=(.*)$/.exec(line); @@ -734,6 +744,9 @@ export interface OnboardRunnerInput { chooseName?: (defaultName: string) => Promise; message?: (text: string) => void; sleep?: (ms: number) => Promise; + /** CTC-4646: turns on the runner's update schedule after it starts; returns the line to show. + * Never fails setup: an update schedule is a convenience, and `catalyst runner update` works without it. */ + autoUpdate?: (dir: string) => string; /** How long a check waits for the host to enroll and advertise. */ waitMs?: number; pollMs?: number; @@ -1310,6 +1323,13 @@ export function onboardRunnerAdapter(input: OnboardRunnerInput = {}, return { state: "failed", reason: "runner_compose_failed", evidence: { ...selected, project: RUNNER_PROJECT, ...(up.error ? { composeError: up.error } : {}) } }; acted = true; + if (input.autoUpdate) { + try { + input.message?.(input.autoUpdate(p.dir)); + } catch { + input.message?.("Automatic runner updates could not be turned on. Run catalyst runner auto-update on to try again."); + } + } return { state: "done" }; }, }; diff --git a/src/onboard-runtime.ts b/src/onboard-runtime.ts index e1a79fae..f0e9790a 100644 --- a/src/onboard-runtime.ts +++ b/src/onboard-runtime.ts @@ -54,6 +54,7 @@ import type { import type { OnboardUi } from "./onboard-ui.js"; import { TIMED_OUT_REASONS } from "./setup-onboard-copy.js"; import { fetchMe } from "./transport.js"; +import { enableRunnerAutoUpdate } from "./runner-command.js"; export interface OnboardRuntimeHooks { beforeStep?: (id: OnboardStepId) => Promise; @@ -71,6 +72,8 @@ export interface OnboardRuntimeHooks { realHome?: () => string; skillNames?: readonly string[]; runnerEngine?: RunnerEngine; + /** Turns on the runner's update schedule once it runs. Without an injected engine, the real one. */ + runnerAutoUpdate?: (dir: string) => string; /** How the daily update step reads this computer's scheduler; tests replace it. */ scheduler?: HousekeepingOptions; } @@ -466,6 +469,7 @@ export function createOnboardRuntime( : undefined, sleep: hooks.sleep, message: (text) => (hooks.ui ? hooks.ui.message(text) : ctx.stderr(text)), + autoUpdate: hooks.runnerAutoUpdate ?? (hooks.runnerEngine ? undefined : (dir) => enableRunnerAutoUpdate(ctx, dir)), }), "linear.adopt": onboardWorkflowVerificationAdapter({ message: (text) => (hooks.ui ? hooks.ui.message(text) : ctx.stderr(text)), diff --git a/src/runner-auto-update.ts b/src/runner-auto-update.ts new file mode 100644 index 00000000..12d650e5 --- /dev/null +++ b/src/runner-auto-update.ts @@ -0,0 +1,199 @@ +import { spawnSync } from "node:child_process"; +import { existsSync, mkdirSync, readFileSync, renameSync, unlinkSync, writeFileSync } from "node:fs"; +import { dirname, join } from "node:path"; + +// The schedule that runs `catalyst runner update --auto` every 15 minutes (CTC-4646): a LaunchAgent +// on macOS, a systemd user timer on Linux. It runs as the person who set the runner up, with the +// docker command they use, and only ever calls this CLI. Turning it off removes exactly these files. + +export const AUTO_UPDATE_LABEL = "dev.catalystcloud.runner-update"; +export const AUTO_UPDATE_UNIT = "catalyst-runner-update"; +export const AUTO_UPDATE_INTERVAL_SECONDS = 900; +/** The minutes past each hour the macOS schedule runs at: every 15 minutes. */ +const CALENDAR_MINUTES = [0, 15, 30, 45] as const; + +export type SchedulerRun = (cmd: string, args: string[]) => { status: number; stdout: string; stderr: string }; + +export interface AutoUpdateDeps { + platform?: NodeJS.Platform; + home: string; + uid?: number; + /** The `catalyst` executable the schedule runs. */ + cli: string; + /** PATH for the scheduled run: it must find docker and node. */ + path: string; + run?: SchedulerRun; +} + +export type AutoUpdateStatus = + | { state: "on"; detail: string; path: string } + | { state: "off"; detail: string } + | { state: "unsupported"; detail: string }; + +/** `su` and many ssh sessions leave XDG_RUNTIME_DIR unset, so `systemctl --user` cannot find a + * running user manager. Its socket is at /run/user/; use that when it exists. */ +function schedulerEnv(): NodeJS.ProcessEnv { + const uid = process.getuid?.(); + const runtime = uid === undefined ? undefined : `/run/user/${uid}`; + return !process.env.XDG_RUNTIME_DIR && runtime && existsSync(runtime) + ? { ...process.env, XDG_RUNTIME_DIR: runtime } + : process.env; +} + +const defaultRun: SchedulerRun = (cmd, args) => { + const r = spawnSync(cmd, args, { encoding: "utf8", timeout: 15_000, env: schedulerEnv() }); + return { status: r.status ?? 1, stdout: r.stdout ?? "", stderr: r.error ? r.error.message : (r.stderr ?? "") }; +}; + +const xml = (value: string) => + value.replace(/&/g, "&").replace(//g, ">").replace(/"/g, """); +/** systemd's own quoting for one ExecStart/Environment word. */ +const unitWord = (value: string) => `"${value.replace(/\\/g, "\\\\").replace(/"/g, '\\"').replace(/%/g, "%%")}"`; + +export function launchAgentPath(home: string): string { + return join(home, "Library", "LaunchAgents", `${AUTO_UPDATE_LABEL}.plist`); +} +export function systemdUnitPaths(home: string): { service: string; timer: string } { + const base = join(home, ".config", "systemd", "user"); + return { service: join(base, `${AUTO_UPDATE_UNIT}.service`), timer: join(base, `${AUTO_UPDATE_UNIT}.timer`) }; +} + +export function launchAgentPlist(deps: Pick, dir: string): string { + const args = [deps.cli, "runner", "update", "--auto", "--dir", dir]; + return [ + '', + '', + '', + `Label${AUTO_UPDATE_LABEL}`, + `ProgramArguments${args.map((a) => `${xml(a)}`).join("")}`, + `EnvironmentVariablesPATH${xml(deps.path)}HOME${xml(deps.home)}`, + // A calendar schedule, not StartInterval: on a busy Mac launchd held a StartInterval job as + // "pended nondemand spawn" for hours (mini-2, 2026-10-17), and RunAtLoad was pended the same + // way. The calendar entries fired on time on that same machine. + `StartCalendarInterval${CALENDAR_MINUTES.map((m) => `Minute${m}`).join("")}`, + "ProcessTypeStandard", + `StandardOutPath${xml(join(dir, "update.log"))}`, + `StandardErrorPath${xml(join(dir, "update.log"))}`, + "", + "", + ].join("\n"); +} + +export function systemdUnits(deps: Pick, dir: string): { service: string; timer: string } { + return { + service: [ + "[Unit]", + "Description=Catalyst runner update", + "", + "[Service]", + "Type=oneshot", + `Environment=${unitWord(`PATH=${deps.path}`)}`, + `ExecStart=${[deps.cli, "runner", "update", "--auto", "--dir", dir].map(unitWord).join(" ")}`, + "", + ].join("\n"), + timer: [ + "[Unit]", + "Description=Check for an approved Catalyst runner release every 15 minutes", + "", + "[Timer]", + "OnBootSec=5min", + `OnUnitActiveSec=${AUTO_UPDATE_INTERVAL_SECONDS / 60}min`, + "RandomizedDelaySec=60", + "Persistent=true", + "", + "[Install]", + "WantedBy=timers.target", + "", + ].join("\n"), + }; +} + +function writeFileAtomic(path: string, text: string): void { + mkdirSync(dirname(path), { recursive: true }); + const temp = `${path}.${process.pid}.tmp`; + writeFileSync(temp, text, { mode: 0o644 }); + renameSync(temp, path); +} + +export function autoUpdateStatus(deps: AutoUpdateDeps): AutoUpdateStatus { + const platform = deps.platform ?? process.platform; + const run = deps.run ?? defaultRun; + if (platform === "darwin") { + const path = launchAgentPath(deps.home); + if (!existsSync(path)) return { state: "off", detail: "no_launch_agent" }; + const loaded = run("launchctl", ["print", `gui/${deps.uid ?? process.getuid?.() ?? 0}/${AUTO_UPDATE_LABEL}`]).status === 0; + return loaded ? { state: "on", detail: "launch_agent_loaded", path } : { state: "off", detail: "launch_agent_not_loaded" }; + } + if (platform === "linux") { + const { timer } = systemdUnitPaths(deps.home); + if (!existsSync(timer)) return { state: "off", detail: "no_timer" }; + const enabled = run("systemctl", ["--user", "is-enabled", `${AUTO_UPDATE_UNIT}.timer`]); + return enabled.status === 0 ? { state: "on", detail: "timer_enabled", path: timer } : { state: "off", detail: "timer_not_enabled" }; + } + return { state: "unsupported", detail: `platform_${platform}` }; +} + +/** Install and start the schedule. Rewrites the files every time, so a moved CLI or PATH is picked up. */ +export function enableAutoUpdate(deps: AutoUpdateDeps, dir: string): AutoUpdateStatus { + const platform = deps.platform ?? process.platform; + const run = deps.run ?? defaultRun; + if (platform === "darwin") { + const uid = deps.uid ?? process.getuid?.() ?? 0; + const path = launchAgentPath(deps.home); + const text = launchAgentPlist(deps, dir); + const unchanged = existsSync(path) && readFileSync(path, "utf8") === text; + const domain = `gui/${uid}`; + const loaded = run("launchctl", ["print", `${domain}/${AUTO_UPDATE_LABEL}`]).status === 0; + if (unchanged && loaded) return { state: "on", detail: "launch_agent_loaded", path }; + if (loaded) run("launchctl", ["bootout", `${domain}/${AUTO_UPDATE_LABEL}`]); + writeFileAtomic(path, text); + const boot = run("launchctl", ["bootstrap", domain, path]); + return boot.status === 0 + ? { state: "on", detail: "launch_agent_loaded", path } + : { state: "off", detail: `launchctl_bootstrap_failed:${boot.stderr.trim().slice(0, 160)}` }; + } + if (platform === "linux") { + if (run("systemctl", ["--user", "show-environment"]).status !== 0) + return { state: "unsupported", detail: "no_systemd_user_manager" }; + const paths = systemdUnitPaths(deps.home); + const units = systemdUnits(deps, dir); + writeFileAtomic(paths.service, units.service); + writeFileAtomic(paths.timer, units.timer); + run("systemctl", ["--user", "daemon-reload"]); + const enabled = run("systemctl", ["--user", "enable", "--now", `${AUTO_UPDATE_UNIT}.timer`]); + if (enabled.status !== 0) + return { state: "off", detail: `systemctl_enable_failed:${enabled.stderr.trim().slice(0, 160)}` }; + // Without lingering, a user timer stops when this person logs out of a headless machine. + const linger = run("loginctl", ["show-user", String(deps.uid ?? process.getuid?.() ?? 0), "--property=Linger"]); + return { state: "on", detail: /Linger=yes/.test(linger.stdout) ? "timer_enabled" : "timer_enabled_without_linger", path: paths.timer }; + } + return { state: "unsupported", detail: `platform_${platform}` }; +} + +/** Stop and remove the schedule's own files. Nothing else. */ +export function disableAutoUpdate(deps: AutoUpdateDeps): AutoUpdateStatus { + const platform = deps.platform ?? process.platform; + const run = deps.run ?? defaultRun; + if (platform === "darwin") { + const uid = deps.uid ?? process.getuid?.() ?? 0; + run("launchctl", ["bootout", `gui/${uid}/${AUTO_UPDATE_LABEL}`]); + const path = launchAgentPath(deps.home); + if (existsSync(path)) unlinkSync(path); + return { state: "off", detail: "launch_agent_removed" }; + } + if (platform === "linux") { + run("systemctl", ["--user", "disable", "--now", `${AUTO_UPDATE_UNIT}.timer`]); + const paths = systemdUnitPaths(deps.home); + for (const path of [paths.timer, paths.service]) if (existsSync(path)) unlinkSync(path); + run("systemctl", ["--user", "daemon-reload"]); + return { state: "off", detail: "timer_removed" }; + } + return { state: "unsupported", detail: `platform_${platform}` }; +} + +/** PATH for the scheduled run: the CLI's and node's directories first, then the usual docker homes. */ +export function scheduledPath(cli: string, nodeExecutable: string, current: string | undefined): string { + const dirs = [dirname(cli), dirname(nodeExecutable), "/opt/homebrew/bin", "/usr/local/bin", "/usr/bin", "/bin", + ...(current ?? "").split(":")]; + return [...new Set(dirs.filter((d) => d.startsWith("/")))].join(":"); +} diff --git a/src/runner-command.ts b/src/runner-command.ts new file mode 100644 index 00000000..a63a642b --- /dev/null +++ b/src/runner-command.ts @@ -0,0 +1,93 @@ +import { isAbsolute, resolve } from "node:path"; +import { flagBool, flagString, positionals, type ParsedArgs } from "./args.js"; +import type { Ctx } from "./config.js"; +import { UsageError } from "./errors.js"; +import { + autoUpdateStatus, + disableAutoUpdate, + enableAutoUpdate, + scheduledPath, + type AutoUpdateDeps, + type AutoUpdateStatus, +} from "./runner-auto-update.js"; +import { + dockerRunnerUpdateEngine, + readReleaseState, + runnerDir, + runRunnerUpdate, + updateResultLines, + type RunnerUpdateEngine, +} from "./runner-update.js"; + +// `catalyst runner update [--check] [--auto] [--dir ]` and +// `catalyst runner auto-update [status|on|off]` (CTC-4646). + +export interface RunnerCommandDeps { + engine?: RunnerUpdateEngine; + scheduler?: Partial; +} + +const USAGE = + "catalyst runner update [--check] [--dir ] | runner auto-update [status|on|off]"; + +function schedulerDeps(ctx: Ctx, deps: RunnerCommandDeps): AutoUpdateDeps { + const cli = deps.scheduler?.cli ?? resolve(process.argv[1] ?? "catalyst"); + return { + home: ctx.home, + cli, + path: scheduledPath(cli, process.execPath, ctx.env.PATH), + ...deps.scheduler, + }; +} + +function statusLine(status: AutoUpdateStatus): string { + if (status.state === "on") + return status.detail === "timer_enabled_without_linger" + ? "Automatic runner updates are on, every 15 minutes while you are logged in. To keep them running after you log out, run: sudo loginctl enable-linger $USER" + : "Automatic runner updates are on: this machine checks for an approved release every 15 minutes."; + if (status.state === "off") return `Automatic runner updates are off (${status.detail}). Turn them on with: catalyst runner auto-update on`; + if (status.detail === "no_systemd_user_manager") + return "Automatic runner updates need your systemd user manager, which is not running for you. Run: sudo loginctl enable-linger $USER, then catalyst runner auto-update on"; + return `Automatic runner updates are not available on this machine (${status.detail}). Run catalyst runner update yourself.`; +} + +/** Setup's call: turn the schedule on for `dir` and say what happened in one line. */ +export function enableRunnerAutoUpdate(ctx: Ctx, dir: string): string { + return statusLine(enableAutoUpdate(schedulerDeps(ctx, {}), dir)); +} + +export async function cmdRunner(args: ParsedArgs, ctx: Ctx, deps: RunnerCommandDeps = {}): Promise { + const parts = positionals(args); + const action = parts[0]; + const dirFlag = flagString(args, "dir"); + if (dirFlag !== undefined && !isAbsolute(dirFlag)) throw new UsageError("--dir must be an absolute path"); + const dir = () => dirFlag ?? runnerDir(ctx); + + if (action === "update" && parts.length === 1) { + const check = flagBool(args, "check"); + const auto = flagBool(args, "auto"); + const engine = deps.engine ?? dockerRunnerUpdateEngine({ env: ctx.env }); + const result = await runRunnerUpdate(engine, { dir: dir(), check }); + if (args.json) ctx.stdout(JSON.stringify(result)); + else if (auto) ctx.stdout(`${ctx.now().toISOString()} ${result.outcome} ${result.detail} generation=${result.generation ?? "none"}`); + else for (const line of updateResultLines(result)) ctx.stdout(line); + return result.exitCode; + } + + if (action === "auto-update" && parts.length <= 2) { + const mode = parts[1] ?? "status"; + const scheduler = schedulerDeps(ctx, deps); + let status: AutoUpdateStatus; + if (mode === "on") status = enableAutoUpdate(scheduler, dir()); + else if (mode === "off") status = disableAutoUpdate(scheduler); + else if (mode === "status") status = autoUpdateStatus(scheduler); + else throw new UsageError(USAGE); + if (args.json) { + const state = mode === "status" && dirFlag !== undefined ? readReleaseState(dirFlag) : undefined; + ctx.stdout(JSON.stringify({ ...status, ...(state ? { release: state } : {}) })); + } else ctx.stdout(statusLine(status)); + return status.state === "on" || mode !== "on" ? 0 : 1; + } + + throw new UsageError(USAGE); +} diff --git a/src/runner-update.ts b/src/runner-update.ts new file mode 100644 index 00000000..2648cf71 --- /dev/null +++ b/src/runner-update.ts @@ -0,0 +1,596 @@ +import { closeSync, existsSync, openSync, readFileSync, renameSync, statSync, unlinkSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import type { Ctx } from "./config.js"; +import { CliError } from "./errors.js"; +import { onboardStateRoot } from "./onboard.js"; +import { + dockerRunnerEngine, + parseEnvText, + persistedCustody, + prepareRunnerImages, + readEnvFile, + runnerComposeArgs, + spawnDocker, + type RunnerEngine, + type RunnerExec, +} from "./onboard-runner.js"; + +// `catalyst runner update`: move this machine's self-hosted runner to the release Catalyst approved +// (CTC-4646). The cloud names one approved runner, supervisor and watchdog by immutable digest with +// a monotonic generation (catalyst-cloud CTC-5926, GET /api/v1/hosts/desired-release, read with the +// host's own credential). This command reads it from inside a one-off supervisor container, so the +// host secret never leaves Docker, then moves only when every check passes: +// +// - the release parses strictly: three digests from the three Catalyst repositories, never a +// tag or `latest`, and a generation newer than the one this machine last applied; +// - no phase container exists for this host; +// - all three images are pulled by digest, natively for this engine, before anything changes; +// - on a Mac with private thoughts storage, that storage is re-attested on the new supervisor; +// - with the supervisor stopped, so it admits no new work, there is still no phase container. +// If one started in between, the supervisor comes back as it was and the update waits. +// +// Then it rewrites the three image lines in `.env`, keeps the previous file beside it, and runs +// `docker compose up`. The release counts as applied only when the running containers are its +// images and the supervisor logged `host_ready` since it started; a `.env` that already names the +// release is reconciled the same way, never trusted. Otherwise it restores the previous file, the +// thoughts storage attestation and the previous images. The outcome is kept in +// `release-state.json` for `--check` and status readers. + +export const RELEASE_STATE_FILE = "release-state.json"; +export const PREVIOUS_ENV_FILE = ".env.release-previous"; +const LOCK_FILE = "update.lock"; +const READY_DEADLINE_MS = 5 * 60_000; +const READY_POLL_MS = 5_000; +const ATTEMPT_HOST_LABEL = "dev.catalystcloud.attempt.host-id"; +const DESIRED_RELEASE_PATH = "/api/v1/hosts/desired-release"; + +export const RELEASE_COMPONENTS = ["runner", "supervisor", "watchdog"] as const; +export type ReleaseComponent = (typeof RELEASE_COMPONENTS)[number]; +/** The only repositories a release may name, exactly as catalyst-cloud's HOST_RELEASE_REPOSITORIES. */ +export const RELEASE_REPOSITORIES: Readonly> = { + runner: "ghcr.io/coalesce-labs/catalyst-runner", + supervisor: "ghcr.io/coalesce-labs/catalyst-supervisor", + watchdog: "ghcr.io/coalesce-labs/catalyst-deadline-watchdog", +}; +export const RELEASE_ENV_KEYS: Readonly> = { + runner: "CATALYST_RUNNER_IMAGE", + supervisor: "CATALYST_SUPERVISOR_IMAGE", + watchdog: "CATALYST_WATCHDOG_IMAGE", +}; + +export interface DesiredRelease { + releaseId: string; + generation: number; + approvedAtMs: number; + images: Record; +} + +/** What one desired-release read returned, as the container saw it. */ +export type DesiredReleaseRead = + | { kind: "answered"; status: number; body: unknown; hostId: string } + | { kind: "unreachable"; detail: string; hostId: string } + | { kind: "no_credential" }; + +const isRecord = (value: unknown): value is Record => + value !== null && typeof value === "object" && !Array.isArray(value); +const posInt = (value: unknown): value is number => + typeof value === "number" && Number.isSafeInteger(value) && value > 0; +const DIGEST = /^sha256:[0-9a-f]{64}$/; +const SHA40 = /^[0-9a-f]{40}$/; + +/** A reference this machine may pull for `component`: its repository, at a digest, with no tag. */ +export function releaseReference(component: ReleaseComponent, value: unknown): string | null { + if (typeof value !== "string" || value.length > 300) return null; + const at = value.indexOf("@"); + if (at === -1 || value.slice(0, at) !== RELEASE_REPOSITORIES[component]) return null; + return DIGEST.test(value.slice(at + 1)) ? value : null; +} + +const shortDigest = (reference: string) => reference.slice(reference.indexOf("@sha256:") + 8).slice(0, 16); + +/** The host-readable release, parsed as strictly as catalyst-cloud's parseHostDesiredRelease. */ +export function parseDesiredRelease(input: unknown): DesiredRelease | null { + if (!isRecord(input) || input.version !== 1) return null; + if (typeof input.releaseId !== "string" || !posInt(input.generation) || !posInt(input.approvedAtMs)) return null; + const images = {} as DesiredRelease["images"]; + for (const component of RELEASE_COMPONENTS) { + const value = input[component]; + if (!isRecord(value)) return null; + const reference = releaseReference(component, value.reference); + if (!reference || value.digest !== reference.slice(reference.indexOf("@") + 1)) return null; + if (typeof value.sourceSha !== "string" || !SHA40.test(value.sourceSha)) return null; + const tag = `${RELEASE_REPOSITORIES[component]}:${value.sourceSha}${component === "runner" ? "-multiarch" : ""}`; + if (value.tag !== tag) return null; + images[component] = { reference, sourceSha: value.sourceSha }; + } + const id = `hr-${shortDigest(images.runner.reference)}-${shortDigest(images.supervisor.reference)}-${shortDigest(images.watchdog.reference)}`; + if (input.releaseId !== id) return null; + return { releaseId: id, generation: input.generation, approvedAtMs: input.approvedAtMs, images }; +} + +export type ReleaseDecision = + | { action: "update"; release: DesiredRelease } + | { + action: "retain"; + reason: "no_credential" | "read_unreachable" | "release_unavailable" | "response_invalid" | "credential_refused" | "not_newer"; + /** True only for a revoked host: asking again cannot help. */ + terminal: boolean; + }; + +/** Whether to move. Every answer except a valid, strictly newer generation keeps the current images. */ +export function decideRelease(appliedGeneration: number | null, read: DesiredReleaseRead): ReleaseDecision { + const retain = (reason: Extract["reason"], terminal = false): ReleaseDecision => + ({ action: "retain", reason, terminal }); + if (read.kind === "no_credential") return retain("no_credential"); + if (read.kind === "unreachable") return retain("read_unreachable"); + if (read.status === 401 || read.status === 403) + return retain("credential_refused", isRecord(read.body) && read.body.refusal === "host_revoked" && read.body.terminal === true); + if (read.status !== 200) return retain("release_unavailable"); + const release = parseDesiredRelease(isRecord(read.body) && read.body.ok === true ? read.body.release : null); + if (!release) return retain("response_invalid"); + if (appliedGeneration !== null && release.generation <= appliedGeneration) return retain("not_newer"); + return { action: "update", release }; +} + +export interface ReleaseState { + version: 1; + /** The approved generation these images came from; null until the first one is applied. */ + generation: number | null; + releaseId: string | null; + appliedAtMs: number | null; + /** The images before the last applied release, for a person who wants to go back. */ + previous: Record | null; + last: { atMs: number; outcome: UpdateOutcome; detail: string; generation: number | null } | null; +} + +export function readReleaseState(dir: string): ReleaseState { + const empty: ReleaseState = { version: 1, generation: null, releaseId: null, appliedAtMs: null, previous: null, last: null }; + try { + const parsed: unknown = JSON.parse(readFileSync(join(dir, RELEASE_STATE_FILE), "utf8")); + if (!isRecord(parsed) || parsed.version !== 1) return empty; + return { + ...empty, + generation: posInt(parsed.generation) ? parsed.generation : null, + releaseId: typeof parsed.releaseId === "string" ? parsed.releaseId : null, + appliedAtMs: posInt(parsed.appliedAtMs) ? parsed.appliedAtMs : null, + previous: isRecord(parsed.previous) ? (parsed.previous as ReleaseState["previous"]) : null, + last: isRecord(parsed.last) ? (parsed.last as ReleaseState["last"]) : null, + }; + } catch { + return empty; + } +} + +/** Write a private file by rename, so a reader never sees half of it. */ +function writePrivateAtomic(path: string, text: string): void { + const temp = `${path}.${process.pid}.tmp`; + writeFileSync(temp, text, { mode: 0o600 }); + renameSync(temp, path); +} + +function writeReleaseState(dir: string, state: ReleaseState): void { + writePrivateAtomic(join(dir, RELEASE_STATE_FILE), `${JSON.stringify(state, null, 2)}\n`); +} + +/** `.env` with each `KEY=` line for `values` replaced in place (or appended), every other line kept. */ +export function rewriteEnvText(text: string, values: Readonly>): string { + const pending = new Map(Object.entries(values)); + const lines = text.split("\n").map((line) => { + const key = /^([A-Z][A-Z0-9_]*)=/.exec(line)?.[1]; + if (key === undefined || !pending.has(key)) return line; + const value = pending.get(key)!; + pending.delete(key); + return `${key}=${value}`; + }); + if (pending.size > 0) { + if (lines.length > 0 && lines[lines.length - 1] === "") lines.pop(); + for (const [key, value] of pending) lines.push(`${key}=${value}`); + lines.push(""); + } + return lines.join("\n"); +} + +export type UpdateOutcome = + | "current" + | "recorded" + | "updated" + | "planned" + | "deferred_busy" + | "retained" + | "rolled_back" + | "failed"; + +export interface UpdateResult { + outcome: UpdateOutcome; + detail: string; + generation: number | null; + releaseId?: string; + /** Exit code: 0 for every outcome a later run can improve on its own, 1 when a person should look. */ + exitCode: number; +} + +/** One Compose service as the Engine reports it. */ +export interface ServiceState { + image: string; + running: boolean; + startedAtMs: number; +} + +/** Everything the update touches outside its own files. The docker one is below; tests use fakes. */ +export interface RunnerUpdateEngine { + desiredRelease(dir: string, signal?: AbortSignal): Promise; + /** Attempt containers for this host in any state, or null when Docker could not be read. */ + attemptContainers(dir: string, hostId: string, signal?: AbortSignal): Promise; + /** Pull and verify each image natively for this engine. Null when ready, else why not. */ + prepare(images: string[], signal?: AbortSignal): Promise; + /** Attest Mac private thoughts storage on these images; returns the authority it holds, or null. */ + custody(dir: string, images: Record, signal?: AbortSignal): Promise; + /** The supervisor and watchdog containers, or null when Docker could not be read. */ + services(dir: string, signal?: AbortSignal): Promise<{ supervisor: ServiceState | null; watchdog: ServiceState | null } | null>; + /** Stop the supervisor so it takes no new work. Phase containers are left running (re-adopted on start). */ + stopSupervisor(dir: string, signal?: AbortSignal): Promise; + composeUp(dir: string, signal?: AbortSignal): Promise; + /** True once the supervisor container has logged `host_ready` since it last started. */ + supervisorReadySinceStart(dir: string, signal?: AbortSignal): Promise; + /** Remove these local images, best effort. The engine refuses one a container still uses. */ + removeImages(images: string[], signal?: AbortSignal): Promise; +} + +export interface RunnerUpdateOptions { + dir: string; + /** Plan only: read and decide, change nothing. */ + check?: boolean; + now?: () => number; + sleep?: (ms: number) => Promise; + readyDeadlineMs?: number; + signal?: AbortSignal; +} + +/** One serialized update run. A second concurrent run returns `retained` with `update_running`. */ +export async function runRunnerUpdate(engine: RunnerUpdateEngine, options: RunnerUpdateOptions): Promise { + const { dir } = options; + const now = options.now ?? Date.now; + const envPath = join(dir, ".env"); + if (!existsSync(envPath) || !existsSync(join(dir, "compose.yaml"))) + return { outcome: "failed", detail: "runner_not_installed", generation: null, exitCode: 1 }; + const lock = options.check ? null : takeLock(dir, now()); + if (!options.check && lock === null) + return { outcome: "retained", detail: "update_running", generation: readReleaseState(dir).generation, exitCode: 0 }; + try { + const result = await update(engine, options, now); + if (!options.check) { + const state = readReleaseState(dir); + writeReleaseState(dir, { ...state, last: { atMs: now(), outcome: result.outcome, detail: result.detail, generation: result.generation } }); + } + return result; + } finally { + if (lock !== null) releaseLock(dir, lock); + } +} + +type Images = Record; +const imagesOf = (env: Record): Images => + Object.fromEntries(RELEASE_COMPONENTS.map((c) => [c, env[RELEASE_ENV_KEYS[c]] ?? ""])) as Images; + +/** The running containers are these images, and the supervisor reported ready since it started. */ +async function running(engine: RunnerUpdateEngine, dir: string, images: Images, signal?: AbortSignal): Promise { + const live = await engine.services(dir, signal); + if (!live?.supervisor?.running || live.supervisor.image !== images.supervisor) return false; + if (!live.watchdog?.running || live.watchdog.image !== images.watchdog) return false; + return engine.supervisorReadySinceStart(dir, signal); +} + +async function update(engine: RunnerUpdateEngine, options: RunnerUpdateOptions, now: () => number): Promise { + const { dir, signal } = options; + const envPath = join(dir, ".env"); + const state = readReleaseState(dir); + const done = (outcome: UpdateOutcome, detail: string, extra: Partial = {}): UpdateResult => + ({ outcome, detail, generation: state.generation, exitCode: 0, ...extra }); + + const read = await engine.desiredRelease(dir, signal); + const decision = decideRelease(state.generation, read); + if (decision.action === "retain") + return decision.reason === "not_newer" + ? done("current", "not_newer") + : done("retained", decision.reason, { exitCode: decision.reason === "credential_refused" || decision.reason === "no_credential" ? 1 : 0 }); + const { release } = decision; + const target = Object.fromEntries(RELEASE_COMPONENTS.map((c) => [c, release.images[c].reference])) as Images; + const envText = readFileSync(envPath, "utf8"); + const env = readEnvFile(envPath) ?? {}; + const current = imagesOf(env); + const changed = RELEASE_COMPONENTS.filter((c) => current[c] !== target[c]); + const tagged = { releaseId: release.releaseId }; + const record = (outcome: UpdateOutcome, detail: string, previous: Images | null): UpdateResult => { + writeReleaseState(dir, { ...state, generation: release.generation, releaseId: release.releaseId, appliedAtMs: now(), previous }); + return { outcome, detail, generation: release.generation, exitCode: 0, ...tagged }; + }; + + if (options.check) + return done("planned", changed.length > 0 ? `would_update:${changed.join(",")}` : "already_configured", { ...tagged, generation: release.generation }); + // `.env` naming the release proves nothing on its own: an earlier run may have stopped after writing + // it. Only the running containers, ready, prove the release is applied. + if (changed.length === 0 && (await running(engine, dir, target, signal))) + return record("recorded", "already_running_release", state.previous); + + // What a failure goes back to: today's file, or for a half-applied file the one saved before it. + let rollbackText: string | null = envText; + if (changed.length === 0) { + try { + rollbackText = readFileSync(join(dir, PREVIOUS_ENV_FILE), "utf8"); + } catch { + rollbackText = null; + } + } + const rollbackImages = rollbackText === null ? null : imagesOf(parseEnvText(rollbackText)); + + const hostId = read.kind === "answered" ? read.hostId : ""; + const busy = await engine.attemptContainers(dir, hostId, signal); + if (busy === null) return done("retained", "docker_unavailable"); + if (busy > 0) return done("deferred_busy", `attempt_containers:${busy}`, tagged); + + const unprepared = await engine.prepare(RELEASE_COMPONENTS.map((c) => target[c]), signal); + if (unprepared !== null) return done("retained", unprepared, tagged); + + // Mac private thoughts storage records the supervisor image its helper runs. Every path that does + // not end on the new release attests it back onto the images the machine keeps. + const custodyInstalled = persistedCustody(dir, env) === "installed"; + const restoreCustody = async (images: Images | null): Promise => + !custodyInstalled || images === null || (await engine.custody(dir, images, signal)) === env.CATALYST_THOUGHTS_CUSTODY_AUTHORITY; + if (custodyInstalled) { + const authority = await engine.custody(dir, target, signal); + // Re-attestation keeps the installed authority. A different one would need setup, not an update. + if (authority === null || authority !== env.CATALYST_THOUGHTS_CUSTODY_AUTHORITY) { + const restored = await restoreCustody(rollbackImages ?? current); + return done(restored ? "retained" : "failed", restored ? "custody_reattest_failed" : "custody_restore_failed", + { ...tagged, exitCode: restored ? 0 : 1 }); + } + } + + // The boundary between phases: stop the supervisor so it admits nothing, then look again. A phase + // that started since the first look keeps running, the supervisor comes back, and the update waits. + if (!(await engine.stopSupervisor(dir, signal))) { + await engine.composeUp(dir, signal); + await restoreCustody(rollbackImages ?? current); + return done("retained", "supervisor_stop_failed", tagged); + } + const busyAfterStop = await engine.attemptContainers(dir, hostId, signal); + if (busyAfterStop !== 0) { + await engine.composeUp(dir, signal); + await restoreCustody(rollbackImages ?? current); + return busyAfterStop === null + ? done("retained", "docker_unavailable", tagged) + : done("deferred_busy", `attempt_containers:${busyAfterStop}`, tagged); + } + + // From here the machine changes. The previous file is the rollback. + if (changed.length > 0) writePrivateAtomic(join(dir, PREVIOUS_ENV_FILE), envText); + writePrivateAtomic(envPath, rewriteEnvText(envText, + Object.fromEntries(RELEASE_COMPONENTS.map((c) => [RELEASE_ENV_KEYS[c], target[c]])))); + if ((await engine.composeUp(dir, signal)) && (await waitReady(engine, dir, target, options, now))) { + // Only once the new release is running and ready: the replaced images free their disk (a runner + // image is several GB, against a free-space floor that decides whether the machine takes work). + // A rollback or failure never reaches here, so the previous images stay for it. + if (changed.length > 0) await engine.removeImages(changed.map((c) => current[c]).filter((ref) => ref !== ""), signal); + return record("updated", changed.length > 0 ? `updated:${changed.join(",")}` : "reconciled", changed.length > 0 ? current : state.previous); + } + + if (rollbackText === null || rollbackImages === null) + return done("failed", "new_supervisor_not_ready_no_previous", { ...tagged, exitCode: 1 }); + writePrivateAtomic(envPath, rollbackText); + const custodyBack = await restoreCustody(rollbackImages); + const restored = (await engine.composeUp(dir, signal)) && (await waitReady(engine, dir, rollbackImages, options, now)); + return done(restored && custodyBack ? "rolled_back" : "failed", + !restored ? "rollback_not_ready" : custodyBack ? "new_supervisor_not_ready" : "custody_restore_failed", + { ...tagged, exitCode: 1 }); +} + +async function waitReady( + engine: RunnerUpdateEngine, + dir: string, + images: Images, + options: RunnerUpdateOptions, + now: () => number, +): Promise { + const sleep = options.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))); + const deadline = now() + (options.readyDeadlineMs ?? READY_DEADLINE_MS); + for (;;) { + if (options.signal?.aborted) return false; + if (await running(engine, dir, images, options.signal)) return true; + if (now() >= deadline) return false; + await sleep(READY_POLL_MS); + } +} + +const alive = (pid: number): boolean => { + try { + process.kill(pid, 0); + return true; + } catch (error) { + // EPERM: the process exists but belongs to someone else. + return (error as NodeJS.ErrnoException).code === "EPERM"; + } +}; + +/** An exclusive lock file holding the owner's pid. It is taken over only when that process is gone, + * never by age: one run can legitimately take longer than any fixed limit (three image pulls). */ +function takeLock(dir: string, nowMs: number): string | null { + const path = join(dir, LOCK_FILE); + const token = `${process.pid} ${nowMs}\n`; + for (let attempt = 0; attempt < 2; attempt++) { + try { + const fd = openSync(path, "wx", 0o600); + writeFileSync(fd, token); + closeSync(fd); + return token; + } catch { + try { + const owner = Number(readFileSync(path, "utf8").split(" ")[0]); + if (Number.isSafeInteger(owner) && owner > 0 && alive(owner)) return null; + unlinkSync(path); + } catch { + return null; + } + } + } + return null; +} + +function releaseLock(dir: string, token: string): void { + const path = join(dir, LOCK_FILE); + try { + if (readFileSync(path, "utf8") === token) unlinkSync(path); + } catch { + // Already gone. + } +} + +/** Runs inside the supervisor image with its own environment. Prints one JSON object, never the + * secret: the bearer is built and sent from inside the container. */ +const DESIRED_RELEASE_SCRIPT = `(async () => { + const out = (v) => process.stdout.write(JSON.stringify(v)); + let state; + try { state = JSON.parse(require("node:fs").readFileSync(process.env.CATALYST_HOST_CREDENTIAL_FILE, "utf8")); } + catch { return out({ kind: "no_credential" }); } + const e = state && state.enrollment; + const base = process.env.CATALYST_MIRROR_URL; + if (!state || state.version !== 1 || typeof state.secret !== "string" || !e || + typeof e.tenant !== "string" || typeof e.hostId !== "string" || !base) return out({ kind: "no_credential" }); + const tenant = Buffer.from(e.tenant, "utf8").toString("base64url"); + try { + const response = await fetch(new URL(${JSON.stringify(DESIRED_RELEASE_PATH)}, base).href, { + redirect: "error", signal: AbortSignal.timeout(15000), + headers: { authorization: "Bearer ctc_host_" + tenant + "." + e.hostId + "." + state.secret, accept: "application/json" }, + }); + let body = null; + try { body = await response.json(); } catch {} + out({ kind: "answered", status: response.status, body, hostId: e.hostId }); + } catch (error) { + out({ kind: "unreachable", detail: String((error && error.name) || "fetch failed"), hostId: e.hostId }); + } +})();`; + +const HOST_ID = /^[A-Za-z0-9._:-]{1,128}$/; + +/** The real engine: the docker CLI and the setup engine's own pull, custody and Compose steps. */ +export function dockerRunnerUpdateEngine( + deps: { exec?: RunnerExec; env?: NodeJS.ProcessEnv; setup?: RunnerEngine } = {}, +): RunnerUpdateEngine { + const exec = deps.exec ?? spawnDocker; + const env = Object.fromEntries(Object.entries(deps.env ?? process.env) + .filter(([key]) => !key.startsWith("CATALYST_") && !key.startsWith("DOCKER_SOCKET_"))); + const setup = deps.setup ?? dockerRunnerEngine({ exec, env: deps.env }); + const docker = async (args: string[], options: { cwd?: string; signal?: AbortSignal; timeoutMs?: number } = {}) => { + try { + return await exec(args, { env, ...options }); + } catch { + return { code: 1, stdout: "" }; + } + }; + return { + async desiredRelease(dir, signal) { + const compose = runnerComposeArgs(dir); + if (!compose) return { kind: "no_credential" }; + const read = await docker([...compose, "run", "--rm", "-T", "--no-deps", "--entrypoint", "bun", "supervisor", "-e", DESIRED_RELEASE_SCRIPT], + { cwd: dir, signal, timeoutMs: 120_000 }); + if (read.code !== 0) return { kind: "no_credential" }; + try { + const parsed: unknown = JSON.parse(read.stdout); + if (!isRecord(parsed)) return { kind: "no_credential" }; + const hostId = typeof parsed.hostId === "string" && HOST_ID.test(parsed.hostId) ? parsed.hostId : null; + if (parsed.kind === "answered" && hostId && typeof parsed.status === "number") + return { kind: "answered", status: parsed.status, body: parsed.body ?? null, hostId }; + if (parsed.kind === "unreachable" && hostId) + return { kind: "unreachable", detail: String(parsed.detail ?? "fetch failed").slice(0, 80), hostId }; + return { kind: "no_credential" }; + } catch { + return { kind: "no_credential" }; + } + }, + async attemptContainers(_dir, hostId, signal) { + if (!HOST_ID.test(hostId)) return null; + const listed = await docker(["ps", "--all", "--quiet", "--filter", `label=${ATTEMPT_HOST_LABEL}=${hostId}`], { signal }); + return listed.code === 0 ? listed.stdout.split("\n").filter((line) => line.trim() !== "").length : null; + }, + async prepare(images, signal) { + const info = await setup.info(signal); + if (!info) return "docker_unavailable"; + const refused = await prepareRunnerImages(setup, images, info.arch, signal); + return refused === null ? null : `${refused.reason}${refused.image ? `:${refused.image}` : ""}`; + }, + async custody(dir, images, signal) { + if (!setup.nativeThoughtsCustody) return null; + try { + const result = await setup.nativeThoughtsCustody({ dir, supervisorImage: images.supervisor, runnerImage: images.runner }, signal); + return result.envAuthority; + } catch { + return null; + } + }, + async services(dir, signal) { + const compose = runnerComposeArgs(dir); + if (!compose) return null; + const service = async (name: string): Promise => { + const listed = await docker([...compose, "ps", "--all", "--quiet", name], { cwd: dir, signal }); + if (listed.code !== 0) return undefined; + const id = listed.stdout.trim(); + if (id === "") return null; + if (!/^[0-9a-f]{12,64}$/.test(id)) return undefined; + const inspected = await docker(["inspect", "--format", "{{.State.Running}}|{{.Config.Image}}|{{.State.StartedAt}}", id], { signal }); + const [isRunning, image, started] = inspected.stdout.trim().split("|"); + const startedAtMs = Date.parse(started ?? ""); + return inspected.code === 0 && image && Number.isFinite(startedAtMs) + ? { running: isRunning === "true", image, startedAtMs } : undefined; + }; + const [supervisor, watchdog] = [await service("supervisor"), await service("deadline-watchdog")]; + return supervisor === undefined || watchdog === undefined ? null : { supervisor, watchdog }; + }, + async stopSupervisor(dir, signal) { + const compose = runnerComposeArgs(dir); + if (!compose) return false; + return (await docker([...compose, "stop", "supervisor"], { cwd: dir, signal, timeoutMs: 120_000 })).code === 0; + }, + async composeUp(dir, signal) { + return (await setup.composeUp(dir, signal)).ok; + }, + async removeImages(images, signal) { + for (const ref of images) await docker(["image", "rm", ref], { signal, timeoutMs: 120_000 }); + }, + async supervisorReadySinceStart(dir, signal) { + const compose = runnerComposeArgs(dir); + if (!compose) return false; + const id = (await docker([...compose, "ps", "--quiet", "supervisor"], { cwd: dir, signal })).stdout.trim(); + if (!/^[0-9a-f]{12,64}$/.test(id)) return false; + const started = (await docker(["inspect", "--format", "{{.State.StartedAt}}", id], { signal })).stdout.trim(); + if (!Number.isFinite(Date.parse(started))) return false; + const logs = await docker(["logs", "--since", started, id], { signal, timeoutMs: 30_000 }); + return `${logs.stdout}\n${logs.stderr ?? ""}`.includes('"event":"host_ready"'); + }, + }; +} + +/** The runner directory setup uses. */ +export function runnerDir(ctx: Pick): string { + try { + return join(onboardStateRoot(ctx.home, ctx.env), "runner"); + } catch { + throw new CliError("The runner directory could not be found. Run catalyst setup --runner first.", "runner_directory_unavailable", 1); + } +} + +const OUTCOME_TEXT: Record = { + current: "The runner is on the approved release.", + recorded: "The runner was already on the approved release; recorded it.", + updated: "The runner moved to the approved release.", + planned: "Checked the approved release; nothing was changed.", + deferred_busy: "A phase is running here, so the update waits for the next check.", + retained: "Kept the current images.", + rolled_back: "The new supervisor did not become ready, so the previous images were restored.", + failed: "The update could not finish. Look at the runner with docker compose ps.", +}; + +export function updateResultLines(result: UpdateResult): string[] { + return [ + OUTCOME_TEXT[result.outcome], + `Release: ${result.releaseId ?? "none"}, generation ${result.generation ?? "none"} (${result.detail}).`, + ]; +} diff --git a/test/runner-update.test.ts b/test/runner-update.test.ts new file mode 100644 index 00000000..4d161b3e --- /dev/null +++ b/test/runner-update.test.ts @@ -0,0 +1,435 @@ +import { spawnSync } from "node:child_process"; +import { generateKeyPairSync } from "node:crypto"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, statSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { beforeEach, describe, expect, it } from "vitest"; +import { + decideRelease, + parseDesiredRelease, + PREVIOUS_ENV_FILE, + readReleaseState, + RELEASE_STATE_FILE, + rewriteEnvText, + runRunnerUpdate, + type DesiredReleaseRead, + type ReleaseComponent, + type RunnerUpdateEngine, +} from "../src/runner-update"; +import { + disableAutoUpdate, + enableAutoUpdate, + launchAgentPath, + launchAgentPlist, + systemdUnitPaths, + systemdUnits, + type SchedulerRun, +} from "../src/runner-auto-update"; +import { canonicalDarwinThoughtsJson } from "../vendor/self-host/darwin-thoughts-custody/verifier.mjs"; + +const hex = (c: string) => c.repeat(64); +const sha = (c: string) => c.repeat(40); +const REPO = { + runner: "ghcr.io/coalesce-labs/catalyst-runner", + supervisor: "ghcr.io/coalesce-labs/catalyst-supervisor", + watchdog: "ghcr.io/coalesce-labs/catalyst-deadline-watchdog", +} as const; +const ref = (component: ReleaseComponent, c: string) => `${REPO[component]}@sha256:${hex(c)}`; + +function release(over: { generation?: number; digits?: [string, string, string]; mutate?: (r: Record) => void } = {}) { + const [r, s, w] = over.digits ?? ["1", "2", "3"]; + const image = (component: ReleaseComponent, c: string, src: string) => ({ + reference: ref(component, c), + digest: `sha256:${hex(c)}`, + tag: `${REPO[component]}:${sha(src)}${component === "runner" ? "-multiarch" : ""}`, + sourceSha: sha(src), + }); + const body: Record = { + version: 1, + releaseId: `hr-${hex(r).slice(0, 16)}-${hex(s).slice(0, 16)}-${hex(w).slice(0, 16)}`, + generation: over.generation ?? 1, + approvedAtMs: 1_800_000_000_000, + runner: image("runner", r, "a"), + supervisor: image("supervisor", s, "b"), + watchdog: image("watchdog", w, "b"), + }; + over.mutate?.(body); + return body; +} +const answered = (body: unknown, status = 200): DesiredReleaseRead => + ({ kind: "answered", status, body: status === 200 ? { ok: true, release: body } : body, hostId: "host_abc" }); + +describe("parse and decide: only a valid, strictly newer approved release moves a machine", () => { + it("POSITIVE CONTROL: a well-formed release parses and a first one is an update", () => { + expect(parseDesiredRelease(release())).toMatchObject({ generation: 1, images: { runner: { reference: ref("runner", "1") } } }); + expect(decideRelease(null, answered(release()))).toMatchObject({ action: "update" }); + }); + + it.each([ + ["a tag instead of a digest", (b: Record) => ((b.runner as Record).reference = `${REPO.runner}:latest`)], + ["another repository", (b: Record) => ((b.supervisor as Record).reference = `ghcr.io/evil/catalyst-supervisor@sha256:${hex("2")}`)], + ["a digest that disagrees with the reference", (b: Record) => ((b.watchdog as Record).digest = `sha256:${hex("9")}`)], + ["a release id that does not match the digests", (b: Record) => (b.releaseId = "hr-0000000000000000-0000000000000000-0000000000000000")], + ["a tag that does not match its source commit", (b: Record) => ((b.runner as Record).tag = `${REPO.runner}:${sha("a")}`)], + ["another wire version", (b: Record) => (b.version = 2)], + ])("NEGATIVE: %s is refused and keeps the current images", (_label, mutate) => { + const body = release({ mutate }); + expect(parseDesiredRelease(body)).toBeNull(); + expect(decideRelease(null, answered(body))).toEqual({ action: "retain", reason: "response_invalid", terminal: false }); + }); + + it("an equal or older generation is not newer", () => { + expect(decideRelease(3, answered(release({ generation: 3 })))).toMatchObject({ action: "retain", reason: "not_newer" }); + expect(decideRelease(3, answered(release({ generation: 2 })))).toMatchObject({ action: "retain", reason: "not_newer" }); + expect(decideRelease(3, answered(release({ generation: 4 })))).toMatchObject({ action: "update" }); + }); + + it("unavailable, unreachable and refused answers retain; only a revoked host is terminal", () => { + expect(decideRelease(null, answered({ ok: false, refusal: "release_unavailable" }, 503))).toMatchObject({ reason: "release_unavailable" }); + expect(decideRelease(null, { kind: "unreachable", detail: "TimeoutError", hostId: "h" })).toMatchObject({ reason: "read_unreachable" }); + expect(decideRelease(null, answered({ ok: false, refusal: "host_revoked", terminal: true }, 401))).toEqual({ action: "retain", reason: "credential_refused", terminal: true }); + expect(decideRelease(null, answered({ ok: false, refusal: "credential_retired", terminal: false }, 401))).toMatchObject({ terminal: false }); + expect(decideRelease(null, { kind: "no_credential" })).toMatchObject({ reason: "no_credential" }); + }); +}); + +describe("rewriteEnvText changes only the three image lines", () => { + it("keeps comments, order and every other value", () => { + const text = "# header\nCATALYST_SLOTS=1\nCATALYST_RUNNER_IMAGE=old\nCATALYST_SUPERVISOR_IMAGE=old2\n"; + expect(rewriteEnvText(text, { CATALYST_RUNNER_IMAGE: "new", CATALYST_SUPERVISOR_IMAGE: "new2", CATALYST_WATCHDOG_IMAGE: "w" })) + .toBe("# header\nCATALYST_SLOTS=1\nCATALYST_RUNNER_IMAGE=new\nCATALYST_SUPERVISOR_IMAGE=new2\nCATALYST_WATCHDOG_IMAGE=w\n"); + }); +}); + +const ORIGINAL_ENV = [ + "# setup wrote this", + "CATALYST_HOST_FREE_FLOOR_GIB=25", + `CATALYST_SUPERVISOR_IMAGE=${ref("supervisor", "a")}`, + `CATALYST_WATCHDOG_IMAGE=${ref("watchdog", "b")}`, + `CATALYST_RUNNER_IMAGE=${ref("runner", "c")}`, + "", +].join("\n"); + +function install(env = ORIGINAL_ENV): string { + const dir = mkdtempSync(join(tmpdir(), "runner-update-")); + writeFileSync(join(dir, ".env"), env, { mode: 0o600 }); + writeFileSync(join(dir, "compose.yaml"), "name: catalyst-host\n"); + return dir; +} + +type Images = Record; +const imagesIn = (text: string): Images => { + const get = (key: string) => new RegExp(`^${key}=(.*)$`, "m").exec(text)?.[1] ?? ""; + return { runner: get("CATALYST_RUNNER_IMAGE"), supervisor: get("CATALYST_SUPERVISOR_IMAGE"), watchdog: get("CATALYST_WATCHDOG_IMAGE") }; +}; +const TARGET: Images = { runner: ref("runner", "1"), supervisor: ref("supervisor", "2"), watchdog: ref("watchdog", "3") }; +const OLD: Images = { runner: ref("runner", "c"), supervisor: ref("supervisor", "a"), watchdog: ref("watchdog", "b") }; + +interface FakeOptions { + read?: DesiredReleaseRead; + /** Attempt counts, one per look (the first look, then the look after the supervisor stops). */ + busy?: (number | null)[]; + prepare?: string | null; + composeUp?: boolean[]; + /** Whether a supervisor on this image reports host_ready. Default: always. */ + ready?: (supervisorImage: string) => boolean; + /** The authority custody returns for these images. Default: the installed one, unchanged. */ + custody?: (images: Images) => string | null; + /** What is running before the update. Default: what `.env` names. */ + live?: Images; + stop?: boolean; +} +/** A machine: what runs is what `.env` named at the last successful compose up. */ +function fakeEngine(dir: string, options: FakeOptions = {}) { + const calls: string[] = []; + let live: Images = options.live ?? imagesIn(readFileSync(join(dir, ".env"), "utf8")); + let stopped = false; + let ups = 0; + let looks = 0; + const engine: RunnerUpdateEngine = { + async desiredRelease() { calls.push("read"); return options.read ?? answered(release()); }, + async attemptContainers(_d, hostId) { calls.push(`attempts:${hostId}`); return options.busy ? (options.busy[looks++] ?? 0) : 0; }, + async prepare(images) { calls.push(`prepare:${images.length}`); return options.prepare ?? null; }, + async custody(_d, images) { calls.push(`custody:${images.supervisor}`); return options.custody ? options.custody(images) : "AUTH"; }, + async services() { + return { + supervisor: { image: live.supervisor, running: !stopped, startedAtMs: 1 }, + watchdog: { image: live.watchdog, running: true, startedAtMs: 1 }, + }; + }, + async stopSupervisor() { calls.push("stop"); if (options.stop === false) return false; stopped = true; return true; }, + async composeUp() { + calls.push("up"); + if (!(options.composeUp?.[ups++] ?? true)) return false; + live = imagesIn(readFileSync(join(dir, ".env"), "utf8")); + stopped = false; + return true; + }, + async supervisorReadySinceStart() { return options.ready ? options.ready(live.supervisor) : true; }, + async removeImages(images) { calls.push(`rm:${images.join(",")}`); }, + }; + return { engine, calls, live: () => live }; +} +const fastClock = () => { + let t = 1_000_000; + return { now: () => t, sleep: async (ms: number) => { t += ms; } }; +}; + +describe("runRunnerUpdate", () => { + let dir: string; + beforeEach(() => { dir = install(); }); + + it("POSITIVE CONTROL: idle machine, newer release → pulls, stops the supervisor, looks again, rewrites three lines, starts, waits for host_ready, records", async () => { + const { engine, calls, live } = fakeEngine(dir); + const result = await runRunnerUpdate(engine, { dir, ...fastClock() }); + expect(result).toMatchObject({ outcome: "updated", generation: 1, exitCode: 0 }); + expect(calls).toEqual(["read", "attempts:host_abc", "prepare:3", "stop", "attempts:host_abc", "up", + `rm:${OLD.runner},${OLD.supervisor},${OLD.watchdog}`]); + expect(live()).toEqual(TARGET); + const env = readFileSync(join(dir, ".env"), "utf8"); + expect(imagesIn(env)).toEqual(TARGET); + expect(env).toContain("CATALYST_HOST_FREE_FLOOR_GIB=25\n"); + expect(env.startsWith("# setup wrote this\n")).toBe(true); + expect(statSync(join(dir, ".env")).mode & 0o777).toBe(0o600); + expect(readFileSync(join(dir, PREVIOUS_ENV_FILE), "utf8")).toBe(ORIGINAL_ENV); + expect(readReleaseState(dir)).toMatchObject({ generation: 1, previous: OLD, last: { outcome: "updated" } }); + expect(existsSync(join(dir, "update.lock"))).toBe(false); + }); + + it("the next run with the same generation changes nothing", async () => { + const first = fakeEngine(dir); + await runRunnerUpdate(first.engine, { dir, ...fastClock() }); + const second = fakeEngine(dir); + expect(await runRunnerUpdate(second.engine, { dir, ...fastClock() })).toMatchObject({ outcome: "current", generation: 1 }); + expect(second.calls).toEqual(["read"]); + }); + + it("a phase container on this host at the first look defers, and nothing is stopped or written", async () => { + const { engine, calls } = fakeEngine(dir, { busy: [1] }); + expect(await runRunnerUpdate(engine, { dir, ...fastClock() })).toMatchObject({ outcome: "deferred_busy", exitCode: 0 }); + expect(calls).not.toContain("stop"); + expect(readFileSync(join(dir, ".env"), "utf8")).toBe(ORIGINAL_ENV); + expect(readReleaseState(dir).generation).toBeNull(); + }); + + it("a phase that started while images were pulled: the supervisor comes back on its old images and the update waits", async () => { + const { engine, calls, live } = fakeEngine(dir, { busy: [0, 1] }); + expect(await runRunnerUpdate(engine, { dir, ...fastClock() })).toMatchObject({ outcome: "deferred_busy", detail: "attempt_containers:1" }); + expect(calls.slice(-3)).toEqual(["stop", "attempts:host_abc", "up"]); + expect(live()).toEqual(OLD); + expect(readFileSync(join(dir, ".env"), "utf8")).toBe(ORIGINAL_ENV); + }); + + it("an image that cannot be pulled natively keeps the current images", async () => { + const { engine, calls } = fakeEngine(dir, { prepare: "runner_image_emulated" }); + expect(await runRunnerUpdate(engine, { dir, ...fastClock() })).toMatchObject({ outcome: "retained", detail: "runner_image_emulated" }); + expect(calls).not.toContain("stop"); + expect(readFileSync(join(dir, ".env"), "utf8")).toBe(ORIGINAL_ENV); + }); + + it("a new supervisor that never reports host_ready is rolled back to the previous images", async () => { + const { engine, calls, live } = fakeEngine(dir, { ready: (image) => image === OLD.supervisor }); + const result = await runRunnerUpdate(engine, { dir, ...fastClock(), readyDeadlineMs: 20_000 }); + expect(result).toMatchObject({ outcome: "rolled_back", exitCode: 1, generation: null }); + expect(readFileSync(join(dir, ".env"), "utf8")).toBe(ORIGINAL_ENV); + expect(calls.filter((c) => c === "up")).toHaveLength(2); + expect(live()).toEqual(OLD); + // The images it would go back to are kept. + expect(calls.some((c) => c.startsWith("rm:"))).toBe(false); + // Not recorded as applied, so the next check tries the same release again. + expect(readReleaseState(dir)).toMatchObject({ generation: null, last: { outcome: "rolled_back" } }); + }); + + it("a failed compose up is rolled back too", async () => { + const { engine } = fakeEngine(dir, { composeUp: [false, true] }); + expect(await runRunnerUpdate(engine, { dir, ...fastClock() })).toMatchObject({ outcome: "rolled_back" }); + expect(readFileSync(join(dir, ".env"), "utf8")).toBe(ORIGINAL_ENV); + }); + + it("a watchdog-only release is applied: the stopped supervisor restarts and reports ready again", async () => { + const onlyWatchdog = install(ORIGINAL_ENV.replace(OLD.supervisor, TARGET.supervisor).replace(OLD.runner, TARGET.runner)); + const { engine, live, calls } = fakeEngine(onlyWatchdog); + expect(await runRunnerUpdate(engine, { dir: onlyWatchdog, ...fastClock() })).toMatchObject({ outcome: "updated", detail: "updated:watchdog" }); + expect(live()).toEqual(TARGET); + // Only the replaced image is removed; the runner and supervisor it still runs are untouched. + expect(calls.filter((c) => c.startsWith("rm:"))).toEqual([`rm:${OLD.watchdog}`]); + }); + + it("a machine already running the release, ready, records the generation without restarting anything", async () => { + const same = install(ORIGINAL_ENV.replace(OLD.supervisor, TARGET.supervisor).replace(OLD.watchdog, TARGET.watchdog).replace(OLD.runner, TARGET.runner)); + const { engine, calls } = fakeEngine(same); + expect(await runRunnerUpdate(engine, { dir: same, ...fastClock() })).toMatchObject({ outcome: "recorded", generation: 1 }); + expect(calls).toEqual(["read"]); + }); + + describe("an earlier run that stopped after writing .env is reconciled, never trusted", () => { + const halfApplied = () => { + const d = install(ORIGINAL_ENV.replace(OLD.supervisor, TARGET.supervisor).replace(OLD.watchdog, TARGET.watchdog).replace(OLD.runner, TARGET.runner)); + writeFileSync(join(d, PREVIOUS_ENV_FILE), ORIGINAL_ENV, { mode: 0o600 }); + return d; + }; + + it("old containers still running: it starts the release, waits for ready, then records", async () => { + const d = halfApplied(); + const { engine, calls, live } = fakeEngine(d, { live: OLD }); + expect(await runRunnerUpdate(engine, { dir: d, ...fastClock() })).toMatchObject({ outcome: "updated", detail: "reconciled", generation: 1 }); + expect(calls).toContain("up"); + expect(live()).toEqual(TARGET); + }); + + it("the release never becomes ready: it goes back to the file saved before it", async () => { + const d = halfApplied(); + const { engine, live } = fakeEngine(d, { live: OLD, ready: (image) => image === OLD.supervisor }); + expect(await runRunnerUpdate(engine, { dir: d, ...fastClock(), readyDeadlineMs: 20_000 })).toMatchObject({ outcome: "rolled_back", generation: null }); + expect(readFileSync(join(d, ".env"), "utf8")).toBe(ORIGINAL_ENV); + expect(live()).toEqual(OLD); + }); + }); + + it("--check reads and plans, and writes nothing", async () => { + const { engine, calls } = fakeEngine(dir); + expect(await runRunnerUpdate(engine, { dir, check: true, ...fastClock() })).toMatchObject({ outcome: "planned", detail: "would_update:runner,supervisor,watchdog" }); + expect(calls).toEqual(["read"]); + expect(existsSync(join(dir, RELEASE_STATE_FILE))).toBe(false); + expect(readFileSync(join(dir, ".env"), "utf8")).toBe(ORIGINAL_ENV); + }); + + it("a revoked host retains and asks a person to look", async () => { + const { engine } = fakeEngine(dir, { read: answered({ ok: false, refusal: "host_revoked", terminal: true }, 401) }); + expect(await runRunnerUpdate(engine, { dir, ...fastClock() })).toMatchObject({ outcome: "retained", detail: "credential_refused", exitCode: 1 }); + }); + + describe("the lock", () => { + it("a live owner keeps it, however old", async () => { + writeFileSync(join(dir, "update.lock"), `${process.pid} 1\n`); + const { engine, calls } = fakeEngine(dir); + expect(await runRunnerUpdate(engine, { dir, ...fastClock() })).toMatchObject({ outcome: "retained", detail: "update_running" }); + expect(calls).toEqual([]); + }); + + it("a dead owner's lock is taken over", async () => { + const dead = spawnSync(process.execPath, ["-e", "process.stdout.write(String(process.pid))"], { encoding: "utf8" }).stdout; + writeFileSync(join(dir, "update.lock"), `${dead} 1\n`); + const { engine } = fakeEngine(dir); + expect(await runRunnerUpdate(engine, { dir, ...fastClock() })).toMatchObject({ outcome: "updated" }); + expect(existsSync(join(dir, "update.lock"))).toBe(false); + }); + }); + + describe("Mac private thoughts storage", () => { + function custodyInstall(): { dir: string; envAuthority: string } { + // A runner directory where setup puts it, so the authority's native home is a real home. + const base = mkdtempSync(join(tmpdir(), "custody-")); + const d = join(base, "home", ".local", "runner"); + mkdirSync(d, { recursive: true }); + writeFileSync(join(d, "compose.yaml"), "name: catalyst-host\n"); + const authority = { + version: 1 as const, installationId: "a".repeat(64), + publicKey: generateKeyPairSync("ed25519").publicKey.export({ type: "spki", format: "der" }).toString("base64"), + nativeUid: 501, nativeGid: 20, nativeHome: join(d, "..", ".."), endpoint: "unix:///fixture/docker.sock", daemonId: "fixture-daemon", + thoughtsRoot: join(d, "thoughts"), locksRoot: join(d, "locks"), + requestsRoot: join(d, "..", "darwin-thoughts-requests"), responsesRoot: join(d, "..", "darwin-thoughts-responses"), + }; + const envAuthority = canonicalDarwinThoughtsJson(authority); + writeFileSync(join(d, ".env"), `${ORIGINAL_ENV}CATALYST_THOUGHTS_CUSTODY_AUTHORITY='${envAuthority}'\nCATALYST_THOUGHTS_CUSTODY_REQUESTS_DIR=${authority.requestsRoot}\nCATALYST_THOUGHTS_CUSTODY_RESPONSES_DIR=${authority.responsesRoot}\n`, { mode: 0o600 }); + return { dir: d, envAuthority }; + } + const custodyCalls = (calls: string[]) => calls.filter((c) => c.startsWith("custody:")); + + it("POSITIVE CONTROL: installed storage is re-attested on the new supervisor before the supervisor stops, and the authority line is kept", async () => { + const { dir: d, envAuthority } = custodyInstall(); + const { engine, calls } = fakeEngine(d, { custody: () => envAuthority }); + expect(await runRunnerUpdate(engine, { dir: d, ...fastClock() })).toMatchObject({ outcome: "updated" }); + expect(custodyCalls(calls)).toEqual([`custody:${TARGET.supervisor}`]); + expect(calls.indexOf(`custody:${TARGET.supervisor}`)).toBeLessThan(calls.indexOf("stop")); + expect(readFileSync(join(d, ".env"), "utf8")).toContain(`CATALYST_THOUGHTS_CUSTODY_AUTHORITY='${envAuthority}'`); + }); + + it.each([["fails", null], ["returns a different authority", "{}"]])("NEGATIVE: when re-attestation %s, storage is attested back on the old supervisor and nothing else changes", async (_label, answer) => { + const { dir: d, envAuthority } = custodyInstall(); + const before = readFileSync(join(d, ".env"), "utf8"); + const { engine, calls } = fakeEngine(d, { custody: (images) => (images.supervisor === TARGET.supervisor ? answer : envAuthority) }); + expect(await runRunnerUpdate(engine, { dir: d, ...fastClock() })).toMatchObject({ outcome: "retained", detail: "custody_reattest_failed" }); + expect(custodyCalls(calls)).toEqual([`custody:${TARGET.supervisor}`, `custody:${OLD.supervisor}`]); + expect(calls).not.toContain("stop"); + expect(readFileSync(join(d, ".env"), "utf8")).toBe(before); + }); + + it("NEGATIVE: if storage cannot be attested back either, the run fails and asks a person to look", async () => { + const { dir: d } = custodyInstall(); + const { engine } = fakeEngine(d, { custody: () => null }); + expect(await runRunnerUpdate(engine, { dir: d, ...fastClock() })).toMatchObject({ outcome: "failed", detail: "custody_restore_failed", exitCode: 1 }); + }); + + it("a rollback attests storage back on the old supervisor too", async () => { + const { dir: d, envAuthority } = custodyInstall(); + const { engine, calls } = fakeEngine(d, { custody: () => envAuthority, ready: (image) => image === OLD.supervisor }); + expect(await runRunnerUpdate(engine, { dir: d, ...fastClock(), readyDeadlineMs: 20_000 })).toMatchObject({ outcome: "rolled_back" }); + expect(custodyCalls(calls)).toEqual([`custody:${TARGET.supervisor}`, `custody:${OLD.supervisor}`]); + }); + + it("a phase that started during preparation also attests storage back", async () => { + const { dir: d, envAuthority } = custodyInstall(); + const { engine, calls } = fakeEngine(d, { custody: () => envAuthority, busy: [0, 1] }); + expect(await runRunnerUpdate(engine, { dir: d, ...fastClock() })).toMatchObject({ outcome: "deferred_busy" }); + expect(custodyCalls(calls)).toEqual([`custody:${TARGET.supervisor}`, `custody:${OLD.supervisor}`]); + }); + + it("without custody there is no re-attestation", async () => { + const { engine, calls } = fakeEngine(dir); + await runRunnerUpdate(engine, { dir, ...fastClock() }); + expect(custodyCalls(calls)).toEqual([]); + }); + }); +}); + +describe("the update schedule", () => { + const recorder = () => { + const calls: string[] = []; + const run: SchedulerRun = (cmd, args) => { + calls.push([cmd, ...args].join(" ")); + if (cmd === "launchctl" && args[0] === "print") return { status: 113, stdout: "", stderr: "" }; + if (cmd === "loginctl") return { status: 0, stdout: "Linger=no\n", stderr: "" }; + return { status: 0, stdout: "", stderr: "" }; + }; + return { calls, run }; + }; + + it("macOS: a LaunchAgent every 15 minutes that runs only this CLI's update, with docker on PATH", () => { + const home = mkdtempSync(join(tmpdir(), "home-")); + const { calls, run } = recorder(); + const status = enableAutoUpdate({ platform: "darwin", home, uid: 501, cli: "/opt/homebrew/bin/catalyst", path: "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin", run }, "/Users/x/.local/state/catalyst/runner"); + expect(status).toMatchObject({ state: "on" }); + const plist = readFileSync(launchAgentPath(home), "utf8"); + expect(plist).toBe(launchAgentPlist({ cli: "/opt/homebrew/bin/catalyst", path: "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin", home }, "/Users/x/.local/state/catalyst/runner")); + expect(plist).toContain("/opt/homebrew/bin/catalystrunnerupdate--auto--dir"); + expect(plist).toContain("StartCalendarIntervalMinute0Minute15Minute30Minute45"); + expect(plist).not.toContain("StartInterval"); + expect(plist).toContain("ProcessTypeStandard"); + expect(calls).toContain(`launchctl bootstrap gui/501 ${launchAgentPath(home)}`); + disableAutoUpdate({ platform: "darwin", home, uid: 501, cli: "x", path: "x", run }); + expect(existsSync(launchAgentPath(home))).toBe(false); + expect(calls).toContain("launchctl bootout gui/501/dev.catalystcloud.runner-update"); + }); + + it("Linux: a systemd user timer, and it says when lingering is off", () => { + const home = mkdtempSync(join(tmpdir(), "home-")); + const { calls, run } = recorder(); + const status = enableAutoUpdate({ platform: "linux", home, uid: 1000, cli: "/usr/local/bin/catalyst", path: "/usr/local/bin:/usr/bin:/bin", run }, "/home/x/.local/state/catalyst/runner"); + expect(status).toMatchObject({ state: "on", detail: "timer_enabled_without_linger" }); + const units = systemdUnits({ cli: "/usr/local/bin/catalyst", path: "/usr/local/bin:/usr/bin:/bin" }, "/home/x/.local/state/catalyst/runner"); + expect(readFileSync(systemdUnitPaths(home).service, "utf8")).toBe(units.service); + expect(units.service).toContain('ExecStart="/usr/local/bin/catalyst" "runner" "update" "--auto" "--dir" "/home/x/.local/state/catalyst/runner"'); + expect(units.timer).toContain("OnUnitActiveSec=15min"); + expect(calls).toContain("systemctl --user enable --now catalyst-runner-update.timer"); + disableAutoUpdate({ platform: "linux", home, uid: 1000, cli: "x", path: "x", run }); + expect(existsSync(systemdUnitPaths(home).timer)).toBe(false); + }); + + it("Linux without a user manager is unsupported, and writes nothing", () => { + const home = mkdtempSync(join(tmpdir(), "home-")); + const run: SchedulerRun = () => ({ status: 1, stdout: "", stderr: "Failed to connect to bus" }); + expect(enableAutoUpdate({ platform: "linux", home, cli: "c", path: "/bin", run }, "/d")).toMatchObject({ state: "unsupported" }); + expect(existsSync(systemdUnitPaths(home).timer)).toBe(false); + }); +});