diff --git a/packages/runtime/src/data-governance.ts b/packages/runtime/src/data-governance.ts index a1438c071..554d999e1 100644 --- a/packages/runtime/src/data-governance.ts +++ b/packages/runtime/src/data-governance.ts @@ -554,6 +554,8 @@ export function controllerOwnedGovernancePaths(projectRoot: string, runRoot: str path.join(projectRoot, ".smithers", "workflows"), // `resume --refresh-controller` renders each refreshed controller here. path.join(projectRoot, ".smithers", "continuations"), + // The supervisor's relaunch of a dead engine logs its own output here. + path.join(projectRoot, ".smithers", "logs"), // The workflow engine opens its SQLite database in the target root, so a // launched run leaves engine state in the governed worktree. path.join(projectRoot, "smithers.db"), diff --git a/packages/runtime/src/smithers-executable-capability.ts b/packages/runtime/src/smithers-executable-capability.ts index 24f4793ec..06f13cda9 100644 --- a/packages/runtime/src/smithers-executable-capability.ts +++ b/packages/runtime/src/smithers-executable-capability.ts @@ -4,6 +4,18 @@ import path from "node:path"; const SMITHERS_EXECUTABLE_CAPABILITY: unique symbol = Symbol("ultrafuzz.smithers-executable-capability"); const TARGET_LOCAL_DELEGATION_ANCHOR = "if (!delegateToLocalCliIfPresent()) {"; +/** + * Bun reads `bunfig.toml` (and runs its `preload` list) and `.env` from its + * working directory, which for every controller process is the target + * repository. A controller Bun process without execution-snapshot startup + * controls runs with these flags, so target configuration never reaches it. + */ +export const BUN_TARGET_CONFIGURATION_GUARD_ARGS: readonly string[] = [ + "--config=/dev/null", + "--no-env-file", + "--no-install", + "--no-addons" +]; interface FileIdentity { path: string; @@ -224,18 +236,22 @@ export function acquireSmithersExecutableAnchor( if (bunControls !== undefined) for (const [name, identity] of Object.entries(capability.bunStartup!)) assertPathIdentity(bunControls[name as keyof typeof bunControls], identity, `Bun workflow runner ${name}`); + // An unsealed native continuation has no startup controls but still runs + // in the target repository. const interpreterArguments = - capability.interpreter.runtime === "bun" && bunControls !== undefined - ? [ - `--config=${bunControls!.config}`, - `--env-file=${bunControls!.environment}`, - "--no-env-file", - "--no-install", - "--no-addons", - "--preserve-symlinks-main", - `--preload=${bunControls!.confinement}` - ] - : []; + capability.interpreter.runtime !== "bun" + ? [] + : bunControls === undefined + ? BUN_TARGET_CONFIGURATION_GUARD_ARGS + : [ + `--config=${bunControls.config}`, + `--env-file=${bunControls.environment}`, + "--no-env-file", + "--no-install", + "--no-addons", + "--preserve-symlinks-main", + `--preload=${bunControls.confinement}` + ]; let closed = false; const assertCurrent = (): void => { if (closed) throw new Error("workflow runner executable anchor is already closed"); diff --git a/packages/runtime/src/smithers.ts b/packages/runtime/src/smithers.ts index a7b3c9c9c..b894e048f 100644 --- a/packages/runtime/src/smithers.ts +++ b/packages/runtime/src/smithers.ts @@ -105,6 +105,7 @@ import { assertExecutableOutsideRoot, bindOperatorSmithersExecutableCapability, bindSmithersExecutableCapability, + BUN_TARGET_CONFIGURATION_GUARD_ARGS, nativeOperatorSmithersNodePath, smithersExecutableCapability, type SmithersExecutableAnchor @@ -184,6 +185,10 @@ const operatorControllerProjects = new Map process.platform === "darwin" ? ultrafuzzVolfsRoot(descriptor) : "/proc/self/fd/3";`; +// The startup controls exist only under an inherited descriptor. A supervisor +// that holds none (every native continuation) passes the Bun guard instead: +// the `/proc/self/fd/3` paths would not exist in its relaunched engine. const SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_PATCH = `${SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_LOCAL_ROOT_HELPERS} ${SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_PRESERVE_SYMLINKS_PREDECESSOR_PATCH.replace( RESUME_SNAPSHOT_INLINE_BUN_STARTUP_ARGS, - '[...(process.versions.bun ? ["--config=" + snapshotChildRoot + "/controls/bunfig.toml", "--env-file=" + snapshotChildRoot + "/controls/bun-empty.env", "--no-env-file", "--no-install", "--no-addons", "--preserve-symlinks-main", "--preload=" + snapshotChildRoot + "/controls/bun-module-confinement.js"] : []), ...args.map(rewriteSnapshotArgument)]' + `[...(snapshotDescriptor === undefined ? ${SMITHERS_BUN_GUARD_ARGS} : process.versions.bun ? ["--config=" + snapshotChildRoot + "/controls/bunfig.toml", "--env-file=" + snapshotChildRoot + "/controls/bun-empty.env", "--no-env-file", "--no-install", "--no-addons", "--preserve-symlinks-main", "--preload=" + snapshotChildRoot + "/controls/bun-module-confinement.js"] : []), ...args.map(rewriteSnapshotArgument)]` )}`; const SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_PREDECESSORS = [ SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_PREDECESSOR_PATCH, SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_PRESERVE_SYMLINKS_PREDECESSOR_PATCH, SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_UNSCOPED_HELPER_PREDECESSOR_PATCH ] as const; +// A supervisor relaunch runs exactly this argv, so without the patch the +// relaunched engine writes its events to `/.smithers/executions//logs` +// instead of the run's `--log-dir`, which the supervisor spawn patch hands down. +const SMITHERS_CLI_RESUME_LOG_DIR_SOURCE = + ' return ["up", target.workflowPath, "--resume", "--run-id", runId, "-d", "--force"];'; +const SMITHERS_CLI_RESUME_LOG_DIR_PATCH = + ' return ["up", target.workflowPath, "--resume", "--run-id", runId, "-d", "--force", ...(process.env.ULTRAFUZZ_SMITHERS_LOG_DIR ? ["--log-dir", process.env.ULTRAFUZZ_SMITHERS_LOG_DIR] : [])];'; +// Upstream relaunches a workflow file from `dirname(workflowPath)`, but +// `createSmithers` opens the database relative to the working directory, and +// the generated workflow resolves its task paths from it too. The relaunch +// therefore starts in the rootDir the engine persisted, read the way +// `parsePersistedRootDir` reads it for `up --resume`. +const SMITHERS_CLI_SUPERVISOR_RESUME_ROOT_SOURCE = ` const direct = resolveResumeTarget(run, { workflowExists: options.deps.workflowExists }); + if (direct) return direct;`; +const SMITHERS_CLI_SUPERVISOR_RESUME_ROOT_PATCH = ` const direct = resolveResumeTarget(run, { workflowExists: options.deps.workflowExists }); + if (direct?.kind === "workflow-file") { + const persisted = + run.configJson !== undefined + ? run + : yield* Effect.promise(() => Promise.resolve(options.adapter.getRun(run.runId))).pipe( + Effect.catchDefect(() => Effect.succeed(null)), + ); + let rootDir; + try { + rootDir = JSON.parse(persisted?.configJson ?? "null")?.rootDir; + } catch {} + return typeof rootDir === "string" && rootDir.length > 0 ? { ...direct, cwd: rootDir } : direct; + } + if (direct) return direct;`; // 0.35.0 reflowed this import across multiple lines and added `watch`; // `realpathSync` is still absent, so the CLI still cannot compare a workflow // path against its persisted generation without this patch. @@ -2525,6 +2564,8 @@ export type SmithersCompatibilityPatchId = | "detached_snapshot_transfer" | "supervisor_descriptor" | "resume_snapshot_transfer" + | "resume_log_dir" + | "supervisor_resume_root" | "terminal_state_restore" | "skip_predicate_rerender" | "skip_marks_predicates_stale" @@ -2653,6 +2694,24 @@ export const SMITHERS_COMPATIBILITY_PATCHES: readonly SmithersCompatibilityPatch patchedFamilyMarkers: ["ULTRAFUZZ_SNAPSHOT_INHERITED_DESCRIPTOR"], upstreamAbsent: ["ULTRAFUZZ_SNAPSHOT_INHERITED_DESCRIPTOR"] }, + { + id: "resume_log_dir", + packageName: "@smthrs/cli", + sourceRelativePath: "src/resume-detached.js", + patchable: SMITHERS_CLI_RESUME_LOG_DIR_SOURCE, + patched: SMITHERS_CLI_RESUME_LOG_DIR_PATCH, + // Upstream forwarding a log directory to the relaunch retires this patch. + upstreamAbsent: ["--log-dir"] + }, + { + id: "supervisor_resume_root", + packageName: "@smthrs/cli", + sourceRelativePath: "src/supervisor.js", + patchable: SMITHERS_CLI_SUPERVISOR_RESUME_ROOT_SOURCE, + patched: SMITHERS_CLI_SUPERVISOR_RESUME_ROOT_PATCH, + // Upstream reading the persisted root for a relaunch retires this patch. + upstreamAbsent: ["rootDir", "parsePersistedRootDir"] + }, { id: "terminal_state_restore", packageName: "@smthrs/scheduler", @@ -7016,11 +7075,13 @@ export function applySmithersCompatibilityPatches(projectRoot: string): void { const cliSource = path.join(packageRoot, "src", "index.js"); const observabilitySource = path.join(packageRoot, "src", "observability-helpers.js"); const resumeDetachedSource = path.join(packageRoot, "src", "resume-detached.js"); + const supervisorSource = path.join(packageRoot, "src", "supervisor.js"); assertRegularFileInside(nodeModules, packageJson, "installed Smithers CLI package metadata"); assertRegularFileInside(nodeModules, runnerSource, "installed Smithers public entrypoint"); assertRegularFileInside(nodeModules, cliSource, "installed Smithers CLI implementation"); assertRegularFileInside(nodeModules, observabilitySource, "installed Smithers observability implementation"); assertRegularFileInside(nodeModules, resumeDetachedSource, "installed Smithers detached resume implementation"); + assertRegularFileInside(nodeModules, supervisorSource, "installed Smithers supervisor implementation"); const metadata = readPackageManagerOwnedManifestEnvelope(packageJson, "installed Smithers CLI package manifest"); if (optionalPackageManifestString(metadata, "version", packageJson) !== SMITHERS_VERSION) { throw new Error(`installed Smithers CLI package version must be ${SMITHERS_VERSION}`); @@ -7075,16 +7136,30 @@ export function applySmithersCompatibilityPatches(projectRoot: string): void { "lifecycle trace summary visibility" ) ); - const resumeDetachedContents = fs.readFileSync(resumeDetachedSource, "utf8"); + const resumeDetachedContents = applyRequiredSmithersPatch( + fs.readFileSync(resumeDetachedSource, "utf8"), + SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_SOURCE, + SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_PATCH, + "detached resume execution snapshot transfer", + SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_PREDECESSORS, + ["ULTRAFUZZ_SNAPSHOT_INHERITED_DESCRIPTOR"] + ); writeFileDurable( resumeDetachedSource, applyRequiredSmithersPatch( resumeDetachedContents, - SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_SOURCE, - SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_PATCH, - "detached resume execution snapshot transfer", - SMITHERS_CLI_RESUME_SNAPSHOT_TRANSFER_PREDECESSORS, - ["ULTRAFUZZ_SNAPSHOT_INHERITED_DESCRIPTOR"] + SMITHERS_CLI_RESUME_LOG_DIR_SOURCE, + SMITHERS_CLI_RESUME_LOG_DIR_PATCH, + "detached resume log directory" + ) + ); + writeFileDurable( + supervisorSource, + applyRequiredSmithersPatch( + fs.readFileSync(supervisorSource, "utf8"), + SMITHERS_CLI_SUPERVISOR_RESUME_ROOT_SOURCE, + SMITHERS_CLI_SUPERVISOR_RESUME_ROOT_PATCH, + "supervisor relaunch root" ) ); diff --git a/packages/runtime/test/data-governance.test.ts b/packages/runtime/test/data-governance.test.ts index 868b34cae..a15259fec 100644 --- a/packages/runtime/test/data-governance.test.ts +++ b/packages/runtime/test/data-governance.test.ts @@ -459,6 +459,18 @@ test("a refreshed controller rendered into the target leaves its governed identi assert.equal(initial.dirty, false); assert.deepEqual(targetIdentity(root, owned), initial); }); +test("a supervisor relaunch's log in the target leaves its governed identity unchanged", () => { + const root = repository(), + owned = controllerOwnedGovernancePaths(root, path.join(root, ".ultrafuzz", "runs", "relaunched-run")), + initial = targetIdentity(root, owned); + // Where Smithers' resumeRunDetached writes the output of the relaunch it starts in the run's root. + const log = path.join(root, ".smithers", "logs", "relaunched-run.log"); + fs.mkdirSync(path.dirname(log), { recursive: true }); + fs.writeFileSync(log, "relaunch\n"); + + assert.equal(initial.dirty, false); + assert.deepEqual(targetIdentity(root, owned), initial); +}); test("governance precedes preflight and rejects a policy mutated during it", async () => { const project = temporaryRoot("ufz-governance-plan-"); assert.equal(initProject({ projectRoot: project, force: true }).ok, true); diff --git a/packages/runtime/test/patched-smithers-runner.ts b/packages/runtime/test/patched-smithers-runner.ts new file mode 100644 index 000000000..9f1497618 --- /dev/null +++ b/packages/runtime/test/patched-smithers-runner.ts @@ -0,0 +1,39 @@ +import assert from "node:assert/strict"; +import fs from "node:fs"; +import { createRequire } from "node:module"; +import path from "node:path"; + +import { SMITHERS_COMPATIBILITY_PATCHES } from "../src/smithers.js"; + +/** + * The pinned Smithers runner with `patches` applied, built below `copy`; returns the runner's package root. + * + * The pnpm store is shared by every checkout on the machine, so it is never written. Every Smithers + * package is copied, and so is the store's hoisted `node_modules`, whose relative links then resolve + * inside the copy: each Smithers module loads once, from the copy. Every other package links back to + * the store. + */ +export function patchedSmithersRunner(copy: string, patches = SMITHERS_COMPATIBILITY_PATCHES): string { + const runner = path.dirname(path.dirname(fs.realpathSync(createRequire(import.meta.url).resolve("smthrs")))); + const store = path.resolve(runner, "..", "..", ".."); + const applied = new Set(); + fs.mkdirSync(copy, { recursive: true }); + for (const entry of fs.readdirSync(store)) { + if (entry !== "node_modules" && !entry.startsWith("smthrs@") && !entry.startsWith("@smthrs+")) { + fs.symlinkSync(path.join(store, entry), path.join(copy, entry)); + continue; + } + fs.cpSync(path.join(store, entry), path.join(copy, entry), { recursive: true, verbatimSymlinks: true }); + for (const patch of patches) { + const home = path.join(copy, entry, "node_modules", ...patch.packageName.split("/")); + if (!fs.existsSync(home) || fs.lstatSync(home).isSymbolicLink()) continue; + const source = path.join(home, ...patch.sourceRelativePath.split("/")); + const parts = fs.readFileSync(source, "utf8").split(patch.patchable); + assert.equal(parts.length, 2, `${patch.id} no longer anchors in ${source}`); + fs.writeFileSync(source, parts.join(patch.patched)); + applied.add(patch.id); + } + } + assert.deepEqual([...applied].sort(), patches.map((patch) => patch.id).sort(), "a patched module was not copied"); + return path.join(copy, path.relative(store, runner)); +} diff --git a/packages/runtime/test/runtime.test.ts b/packages/runtime/test/runtime.test.ts index c3b96850c..d58a3b7b7 100644 --- a/packages/runtime/test/runtime.test.ts +++ b/packages/runtime/test/runtime.test.ts @@ -17625,6 +17625,7 @@ const sleep = (milliseconds) => Atomics.wait(sleepArray, 0, 0, milliseconds); async function launchSupervisor(child, logFile, workflowArgument) { const cliPath = process.argv[1]; const supervisorArgs = [cliPath, "supervisor", workflowArgument]; + const options = {}; let supervisorPid; const fail = (failure) => ({ failure }); ${supervisorPatch.patched} @@ -17945,6 +17946,7 @@ testWhen(process.platform !== "win32" && fs.existsSync("/proc/self/fd"))( const configPath = path.join(root, "controls", "ultrafuzz.toml"); const scriptPath = path.join(root, "descriptor-generations.mjs"); const logPath = path.join(coordinationRoot, "detached.log"); + const runLogDir = path.join(coordinationRoot, "run-logs"); const launcherRecordPath = path.join(coordinationRoot, "launcher.json"); const supervisorRecordPath = path.join(coordinationRoot, "supervisor.json"); const resumeLaunchPath = path.join(coordinationRoot, "resume-launch.json"); @@ -18045,6 +18047,7 @@ ${detachedTransferPatch.patched} function launchSupervisor() { const supervisorArgs = [process.argv[1], "supervisor"]; const logFile = process.env.UFZ_LOG_PATH; + const options = { logDir: process.env.UFZ_RUN_LOG_DIR }; let supervisorPid; ${supervisorPatch.patched} return supervisorPid; @@ -18131,7 +18134,7 @@ if (phase === "supervisor") { writeFileSync(process.env.UFZ_RESUME_LAUNCH_PATH, JSON.stringify({ logged: logged.args, ignored: ignored.args })); writeFileSync( process.env.UFZ_SUPERVISOR_RECORD_PATH, - JSON.stringify({ ...ownEvidence, ...modules, logged_pid: logged.pid, ignored_pid: ignored.pid, reused_descriptor: reusedDescriptor, reused_startup_descriptor: reusedStartupDescriptor }) + JSON.stringify({ ...ownEvidence, ...modules, logged_pid: logged.pid, ignored_pid: ignored.pid, reused_descriptor: reusedDescriptor, reused_startup_descriptor: reusedStartupDescriptor, smithers_log_dir: process.env.ULTRAFUZZ_SMITHERS_LOG_DIR }) ); process.exit(0); } @@ -18178,6 +18181,7 @@ if (phase === "engine" || phase === "resume-logged" || phase === "resume-ignored ULTRAFUZZ_CONFIG_PATH: path.join(controllerRoot, "controls", "ultrafuzz.toml"), ULTRAFUZZ_WORKFLOW_PERSISTED_PATH: workflowPath, UFZ_LOG_PATH: logPath, + UFZ_RUN_LOG_DIR: runLogDir, UFZ_LAUNCHER_RECORD_PATH: launcherRecordPath, UFZ_SUPERVISOR_RECORD_PATH: supervisorRecordPath, UFZ_RESUME_LAUNCH_PATH: resumeLaunchPath, @@ -18293,6 +18297,8 @@ if (phase === "engine" || phase === "resume-logged" || phase === "resume-ignored assert.equal(supervisorEvidence.reused_descriptor, supervisorEvidence.process_descriptor); assert.equal(supervisorEvidence.relative_module, "sealed-relative"); assert.match(String(supervisorEvidence.ambient_error), /outside its sealed snapshot/u); + // The sealed supervisor keeps the launch's --log-dir for the relaunch argv (`resume_log_dir`). + assert.equal(supervisorEvidence.smithers_log_dir, runLogDir); addEvidencePids(pids, supervisorEvidence, "logged_pid", "ignored_pid"); const resumeLaunch = JSON.parse(fs.readFileSync(resumeLaunchPath, "utf8")) as { diff --git a/packages/runtime/test/smithers-executable-capability.test.ts b/packages/runtime/test/smithers-executable-capability.test.ts index 3d44de609..f88003dc8 100644 --- a/packages/runtime/test/smithers-executable-capability.test.ts +++ b/packages/runtime/test/smithers-executable-capability.test.ts @@ -297,6 +297,42 @@ test( } ); +test( + "native operator continuations run Bun without the target repository's bunfig.toml or .env", + { skip: !bunAvailable || process.platform === "win32" || !fs.existsSync("/proc/self/fd") }, + async () => { + const root = temporaryDirectory("ufz-native-operator-bun-config-"), + operatorRoot = path.join(root, "operator"), + targetRoot = path.join(root, "target"), + runner = path.join(operatorRoot, ".smithers", "node_modules", "smthrs", "src", "bin", "smithers.js"), + preloaded = path.join(root, "target-preload-ran"); + fs.mkdirSync(path.dirname(runner), { recursive: true }); + fs.mkdirSync(targetRoot); + // Bun reads both files from its working directory, which is the target repository. + fs.writeFileSync(path.join(targetRoot, "bunfig.toml"), 'preload = ["./preload.js"]\n'); + fs.writeFileSync( + path.join(targetRoot, "preload.js"), + `require("node:fs").writeFileSync(${JSON.stringify(preloaded)}, "");\n` + ); + fs.writeFileSync(path.join(targetRoot, ".env"), "TARGET_DOTENV=loaded\n"); + writeExecutable( + runner, + "#!/usr/bin/env bun\nconsole.log(JSON.stringify({ dotenv: process.env.TARGET_DOTENV ?? null }));\n" + ); + const env = bindOperatorSmithersExecutableCapability({}, runner, operatorRoot, () => {}, targetRoot, true); + + const result = await runSmithersInspectionCommand({ + args: ["inspect", "fixture", "--format", "json"], + projectRoot: targetRoot, + env + }); + + assert.equal(result.ok, true, result.error); + assert.deepEqual(result.json, { dotenv: null }); + assert.equal(fs.existsSync(preloaded), false); + } +); + test("streaming Smithers commands keep the executable anchor through child close", async () => { const root = temporaryDirectory("ufz-runner-stream-"); const runner = nodeRunner(root, "console.log(JSON.stringify({ sequence: 1 }));\n"); diff --git a/packages/runtime/test/smithers-resume-reopen.integration.test.ts b/packages/runtime/test/smithers-resume-reopen.integration.test.ts index 29ecfb2da..d865a74a6 100644 --- a/packages/runtime/test/smithers-resume-reopen.integration.test.ts +++ b/packages/runtime/test/smithers-resume-reopen.integration.test.ts @@ -6,6 +6,7 @@ import test from "node:test"; import { fileURLToPath } from "node:url"; import { SMITHERS_COMPATIBILITY_PATCHES } from "../src/smithers.js"; +import { patchedSmithersRunner } from "./patched-smithers-runner.js"; import { temporaryRoot } from "./temporary-root.js"; interface Inspection { @@ -98,7 +99,13 @@ let sharedRunner: string | undefined; function startCampaign(prefix: string, sideTask: keyof typeof SIDE_TASKS = "independent") { const root = temporaryRoot(prefix); - const runner = (sharedRunner ??= patchedSmithersRunner(temporaryRoot("ufz-patched-smithers-"))); + // Ultrafuzz's scheduler patches and resume hydration together decide what a resumed session runs again. + const runner = (sharedRunner ??= patchedSmithersRunner( + temporaryRoot("ufz-patched-smithers-"), + SMITHERS_COMPATIBILITY_PATCHES.filter( + (patch) => patch.packageName === "@smthrs/scheduler" || patch.id === "resume_hydration" + ) + )); const runId = `${path.basename(root)}-${process.pid}`; const workflow = path.join(root, ".smithers", "workflows", "reopen.tsx"); fs.mkdirSync(path.dirname(workflow), { recursive: true }); @@ -142,38 +149,6 @@ function startCampaign(prefix: string, sideTask: keyof typeof SIDE_TASKS = "inde }; } -// The pinned runner with Ultrafuzz's scheduler patches and resume hydration -// applied, which together decide what a resumed session runs again. The pnpm -// store is shared by every checkout on the machine, so it is never written: -// the Smithers packages (the only importers of the patched modules) are -// copied, and every other package links back to the store. -function patchedSmithersRunner(copy: string): string { - const runner = fs.realpathSync(path.join(runtimePackageRoot(), "node_modules", "smthrs")); - const store = path.resolve(runner, "..", "..", ".."); - const patches = SMITHERS_COMPATIBILITY_PATCHES.filter( - (patch) => patch.packageName === "@smthrs/scheduler" || patch.id === "resume_hydration" - ); - const applied = new Set(); - for (const entry of fs.readdirSync(store)) { - if (!entry.startsWith("smthrs@") && !entry.startsWith("@smthrs+")) { - fs.symlinkSync(path.join(store, entry), path.join(copy, entry)); - continue; - } - fs.cpSync(path.join(store, entry), path.join(copy, entry), { recursive: true, verbatimSymlinks: true }); - for (const patch of patches) { - const home = path.join(copy, entry, "node_modules", ...patch.packageName.split("/")); - if (!fs.existsSync(home) || fs.lstatSync(home).isSymbolicLink()) continue; - const source = path.join(home, ...patch.sourceRelativePath.split("/")); - const parts = fs.readFileSync(source, "utf8").split(patch.patchable); - assert.equal(parts.length, 2, `${patch.id} no longer anchors in ${source}`); - fs.writeFileSync(source, parts.join(patch.patched)); - applied.add(patch.id); - } - } - assert.deepEqual([...applied].sort(), patches.map((patch) => patch.id).sort(), "a patched module was not copied"); - return path.join(copy, path.relative(store, runner)); -} - // A producer -> consumer -> downstream chain of generated prepare/node/verify // triplets, skipped with the generated workflow's own predicates. The // producer's agent fails until `producer-may-succeed` exists, and preparation diff --git a/packages/runtime/test/smithers-supervisor-relaunch.integration.test.ts b/packages/runtime/test/smithers-supervisor-relaunch.integration.test.ts new file mode 100644 index 000000000..f82bc7684 --- /dev/null +++ b/packages/runtime/test/smithers-supervisor-relaunch.integration.test.ts @@ -0,0 +1,193 @@ +import assert from "node:assert/strict"; +import { execFileSync, spawnSync } from "node:child_process"; +import fs from "node:fs"; +import path from "node:path"; +import test from "node:test"; + +import { BUN_TARGET_CONFIGURATION_GUARD_ARGS } from "../src/smithers-executable-capability.js"; +import { patchedSmithersRunner } from "./patched-smithers-runner.js"; +import { temporaryRoot } from "./temporary-root.js"; + +interface TaskRecord { + task: string; + pid: number; + cwd: string; + dotenv: string | null; +} + +// An engine run from plain paths, as `ultrafuzz resume` runs it: the workflow file lives in a +// run-owned directory below the launch root, its packages resolve through NODE_PATH, and no snapshot +// descriptor is inherited. When the engine dies, only Smithers' own supervisor can finish the run +// without an operator. +test("the supervisor relaunch finishes a SIGKILLed engine's run from its launch root and log directory", async () => { + const root = temporaryRoot("ufz-supervisor-relaunch-"); + // Smithers anchors its store on a `.smithers` directory only below HOME. With the target outside + // HOME, the store is `smithers.db` in whatever directory the engine starts in. + const home = path.join(root, "home"); + const target = path.join(root, "target"); + const runRoot = path.join(target, ".ultrafuzz", "runs", "r1", "smithers"); + const workflow = path.join(runRoot, "workflows", "relaunch.tsx"); + const logDir = path.join(runRoot, "logs"); + const records = path.join(root, "tasks.jsonl"); + const preloaded = path.join(root, "target-preload-ran"); + fs.mkdirSync(home); + fs.mkdirSync(path.dirname(workflow), { recursive: true }); + execFileSync("git", ["init", "--quiet", "--initial-branch=main"], { cwd: target }); + // Bun loads both files from its working directory, which is the target repository. + fs.writeFileSync(path.join(target, "bunfig.toml"), 'preload = ["./preload.js"]\n'); + fs.writeFileSync( + path.join(target, "preload.js"), + `require("node:fs").appendFileSync(${JSON.stringify(preloaded)}, process.argv.join(" ") + "\\n");\n` + ); + fs.writeFileSync(path.join(target, ".env"), "TARGET_DOTENV=loaded\n"); + fs.writeFileSync(workflow, relaunchWorkflowSource(records)); + + const runner = patchedSmithersRunner(path.join(root, "runner")); + const runId = `supervisor-relaunch-${process.pid}`; + const cli = [...BUN_TARGET_CONFIGURATION_GUARD_ARGS, path.join(runner, "src", "bin", "smithers.js")]; + const smithers = (...args: string[]): string => { + const result = spawnSync("bun", [...cli, ...args], { + cwd: target, + encoding: "utf8", + env: { + HOME: home, + PATH: process.env.PATH, + ...(process.env.TMPDIR === undefined ? {} : { TMPDIR: process.env.TMPDIR }), + NODE_PATH: path.dirname(runner), + SMITHERS_MONITOR_SUPPRESS: "1", + SMITHERS_POST_FAILURE: "0", + ULTRAFUZZ_WORKFLOW_PERSISTED_PATH: workflow + }, + maxBuffer: 16 * 1024 * 1024, + timeout: 120_000 + }); + assert.equal(result.status, 0, [result.error, result.stdout, result.stderr].join("\n").slice(-4_000)); + return result.stdout; + }; + try { + smithers( + "up", + workflow, + "--detach", + "--run-id", + runId, + "--root", + target, + "--input", + "{}", + "--log-dir", + logDir, + "--format", + "json", + "--supervise", + "--supervise-interval", + "1s", + // Production's threshold (controller_lease_seconds). A relaunch that has not activated within + // it is claimed again as a second attempt. + "--supervise-stale-threshold", + "30s", + "--supervise-max-concurrent", + "1" + ); + const killed = await waitFor("the first attempt", () => taskRecords(records)[0]); + process.kill(killed.pid, "SIGKILL"); + const status = await waitFor("the run to end", () => { + const inspected = JSON.parse(smithers("inspect", runId, "--format", "json")) as { + status?: string; + run?: { status?: string }; + }; + const current = inspected.run?.status ?? inspected.status; + return current === "finished" || current === "failed" || current === "cancelled" ? current : undefined; + }); + + const events = smithers("events", runId, "--json") + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { type: string; payload: { resumeAttempt?: number } }); + assert.equal(status, "finished", events.map((event) => event.type).join(",")); + assert.deepEqual( + events.filter((event) => event.type === "RunAutoResumed").map((event) => event.payload.resumeAttempt), + [1], + "the first supervisor relaunch did not finish the run" + ); + const recorded = taskRecords(records); + assert.deepEqual( + recorded.map((record) => record.task), + ["interrupted", "interrupted", "after"] + ); + assert.notEqual(recorded[1]?.pid, killed.pid); + for (const record of recorded) { + assert.equal(record.cwd, target, "the relaunched engine did not start in the launch root"); + assert.equal(record.dotenv, null, "an engine loaded the target's .env"); + } + assert.equal(fs.existsSync(preloaded), false, "a controller process ran the target's bunfig.toml preload"); + // Both engines append to the run's log directory; none falls back to the default one. + const logged = fs + .readFileSync(path.join(logDir, "stream.ndjson"), "utf8") + .trim() + .split("\n") + .map((line) => (JSON.parse(line) as { type: string }).type); + assert.equal(logged.filter((type) => type === "RunStarted").length, 2); + assert.equal(logged.at(-1), "RunFinished"); + assert.equal(fs.existsSync(path.join(target, ".smithers", "executions")), false); + } finally { + // The supervisor exits once the run ends; a failed run can leave it and a hung engine behind. + // Both run the copied CLI below `root`, and `pkill -f` takes a regular expression. + spawnSync("pkill", ["-KILL", "-f", root.replace(/[\\^$.*+?()[\]{}|]/gu, "\\$&")]); + } +}); + +// The first attempt blocks until the test kills its engine; the relaunched attempt returns. +function relaunchWorkflowSource(records: string): string { + return `/** @jsxImportSource smthrs */ +import fs from "node:fs"; +import { createSmithers } from "smthrs"; +import { z } from "zod/v4"; + +const { Workflow, Sequence, Task, smithers, outputs } = createSmithers({ + input: z.object({}), + result: z.object({ value: z.string() }) +}); +const records = ${JSON.stringify(records)}; +const record = (task: string): boolean => { + const repeated = fs.existsSync(records) && fs.readFileSync(records, "utf8").includes('"task":"' + task + '"'); + const entry = { task, pid: process.pid, cwd: process.cwd(), dotenv: process.env.TARGET_DOTENV ?? null }; + fs.appendFileSync(records, JSON.stringify(entry) + "\\n"); + return repeated; +}; +export default smithers(() => ( + + + + {async () => { + if (!record("interrupted")) await new Promise(() => {}); + return { value: "interrupted" }; + }} + + + {() => (record("after"), { value: "after" })} + + + +)); +`; +} + +function taskRecords(file: string): TaskRecord[] { + if (!fs.existsSync(file)) return []; + return fs + .readFileSync(file, "utf8") + .split("\n") + .filter((line) => line.length > 0) + .map((line) => JSON.parse(line) as TaskRecord); +} + +// Long enough for the supervisor to give up after three relaunches, so a regression reports its events. +async function waitFor(label: string, probe: () => T | undefined): Promise { + for (const deadline = Date.now() + 180_000; Date.now() < deadline;) { + const value = probe(); + if (value !== undefined) return value; + await new Promise((resolve) => setTimeout(resolve, 250)); + } + throw new Error(`timed out waiting for ${label}`); +}