From 0402012a9504bde50416df543d5092c763868be1 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sat, 24 Oct 2026 19:16:42 +0800 Subject: [PATCH] fix(verifiers): share command ownership and preserve stdin Move command and retained resource lifetimes into a shared verifier owner, keeping package workspace release as a separate dependency. Migrate Pi and Hermes verification to that owner while preserving open pipes, explicit EOF, diagnostics, and retained labs. Require positive writable finalization before accepting EOF delivery. Cover premature close, opaque failures, late ownership delivery, mixed cleanup outcomes, exact retry, and real descendant cleanup with authored process boundaries. --- scripts/lib/package-command-owner.mts | 259 +----- scripts/lib/verifier-command-owner.mts | 276 ++++++ scripts/lib/verifier-file-snapshot-owner.mts | 6 +- scripts/lib/verifier-terminal-owner.mts | 14 +- scripts/lib/verify-commands.d.mts | 9 + scripts/lib/verify-commands.mjs | 13 + scripts/lib/verify-package-commands.d.mts | 3 - scripts/lib/verify-setup-terminal.d.mts | 4 +- scripts/verify-native-agent-client.mjs | 839 ++++++++++-------- src/process/OwnedCommand.test.ts | 439 +++++++++ src/process/OwnedCommand.ts | 110 ++- .../nativeVerifierCommandOwnership.test.ts | 235 +++++ .../packageVerifierCommandSemantics.test.ts | 10 +- .../process/verifierCommandSemantics.test.ts | 606 +++++++++++++ .../process/verifierTerminalOwner.test.ts | 12 +- 15 files changed, 2181 insertions(+), 654 deletions(-) create mode 100644 scripts/lib/verifier-command-owner.mts create mode 100644 scripts/lib/verify-commands.d.mts create mode 100644 scripts/lib/verify-commands.mjs create mode 100644 src/process/OwnedCommand.test.ts create mode 100644 tests/boundary/process/nativeVerifierCommandOwnership.test.ts create mode 100644 tests/boundary/process/verifierCommandSemantics.test.ts diff --git a/scripts/lib/package-command-owner.mts b/scripts/lib/package-command-owner.mts index faca37a940..db0ddeb90b 100644 --- a/scripts/lib/package-command-owner.mts +++ b/scripts/lib/package-command-owner.mts @@ -1,73 +1,27 @@ -import { randomUUID } from "node:crypto"; - import { - ArtifactResourceScope, - type ArtifactResourceOwner, -} from "../../src/artifacts/ArtifactResourceScope.js"; -import { AnalysisCleanupLease } from "../../src/artifacts/AnalysisCleanupLease.js"; -import type { RetainedResourceScope } from "../../src/application/AnalysisProvider.js"; -import { - prepareOwnedCommand, - runOwnedCommand, -} from "../../src/process/OwnedCommand.js"; -import { - spawnOwnedProviderProcess, - type OwnedProviderProcessSpawnOptions, - type ProviderProcessSnapshot, - type SpawnedOwnedProviderProcess, -} from "../../src/process/ProviderProcess.js"; -import { retainOwnedCommandCleanup } from "../../src/providers/OwnedCommandCleanup.js"; -import { WindowsOwnedProcess } from "../../src/windows/WindowsOwnedProcess.js"; + createVerifierCommands, + VerifierCommandExitError, + type VerifierCommands, + type VerifierCommandsOptions, + type VerifierCommandOptions, + type VerifierCommandOutput, + type VerifierCleanupOwner, +} from "./verifier-command-owner.mjs"; import { npmInvocation } from "./npm-invocation.mjs"; -import { createWindowsOwnedCommand } from "./windows-owned-command.mjs"; export interface PackageVerifierWorkspace { readonly path: string; close(): Promise; } -interface PackageCommandContext { - readonly signal?: AbortSignal | undefined; - readonly cwd?: string | undefined; - readonly env?: NodeJS.ProcessEnv | undefined; - readonly timeout?: number | undefined; - readonly windowsHide?: boolean | undefined; -} - -export type PackageCommandOptions = PackageCommandContext & - ( - | { - /** execFile-compatible capacity for each output stream. */ - readonly maxBuffer?: number | undefined; - readonly diagnosticBytes?: never; - } - | { - /** Canonical command capacity across both streams. */ - readonly diagnosticBytes: number; - readonly maxBuffer?: never; - } - ); - -export interface PackageCommandOutput { - readonly stdout: string; - readonly stderr: string; -} - -export interface PackageCleanupOwner { - close(): Promise; - readonly cleanupResources?: readonly string[] | undefined; -} - -export interface PackageVerifierCommands { - exec( - command: string, - arguments_: readonly string[], - options?: PackageCommandOptions, - ): Promise; +export interface PackageVerifierCommands extends Pick< + VerifierCommands, + "exec" +> { execNpm( arguments_: readonly string[], - options?: PackageCommandOptions, - ): Promise; + options?: VerifierCommandOptions, + ): Promise; run( command: string, arguments_: readonly string[], @@ -78,29 +32,10 @@ export interface PackageVerifierCommands { arguments_: readonly string[], environment?: NodeJS.ProcessEnv, ): Promise<{ readonly stdout: string; readonly status: number }>; - adopt(owner: PackageCleanupOwner, resource?: string): void; + adopt(owner: VerifierCleanupOwner, resource?: string): void; close(): Promise; } -/** An ordinary nonzero exit whose process and output cleanup both completed. */ -export class PackageCommandExitError extends Error { - readonly stdout: string; - readonly stderr: string; - constructor( - readonly command: string, - readonly arguments_: readonly string[], - readonly code: number, - readonly snapshot: ProviderProcessSnapshot, - ) { - super( - `Command failed: ${[command, ...arguments_].join(" ")}\n${snapshot.stderr.text}`, - ); - this.name = "PackageCommandExitError"; - this.stdout = snapshot.stdout.text; - this.stderr = snapshot.stderr.text; - } -} - /** The same command/workspace capability remains available after failed close. */ export class PackageVerifierCleanupFailure extends Error { readonly path: string; @@ -118,142 +53,27 @@ export class PackageVerifierCleanupFailure extends Error { } } -const launchPackageCommand = async ( - input: OwnedProviderProcessSpawnOptions, -): Promise => { - if (process.platform !== "win32") return spawnOwnedProviderProcess(input); - input.signal?.throwIfAborted(); - const child = createWindowsOwnedCommand({ - WindowsOwnedProcess, - command: input.command, - arguments: input.arguments, - cwd: input.cwd, - environment: { - ...input.hostEnvironment, - ...input.env, - REA_PROCESS_RUN_ID: input.runId, - }, - }); - // These labels describe the launch; cleanup authority is the exact native job. - return { - process: child, - ownership: { - runId: input.runId, - leaderPid: child.pid, - processGroupId: child.pid, - expectedParentPid: process.pid, - }, - cleanup: () => child.cleanup(), - }; -}; - -const commandLimits = (options: PackageCommandOptions) => { - const timeout = options.timeout ?? 0; - if (!Number.isSafeInteger(timeout) || timeout < 0) - throw new RangeError( - "Package command timeout must be a nonnegative integer", - ); - if (options.maxBuffer !== undefined && options.diagnosticBytes !== undefined) - throw new TypeError("Select a per-stream or aggregate command capacity"); - const capacity = options.diagnosticBytes ?? options.maxBuffer ?? 1024 * 1024; - if (typeof capacity !== "number" || Number.isNaN(capacity) || capacity < 0) - throw new RangeError("Package command output capacity must be nonnegative"); - if (process.platform === "win32" && options.windowsHide === false) - throw new Error( - "Owned package commands require hidden Windows process launch", - ); - return { - timeoutMs: timeout === 0 ? null : timeout, - diagnosticBytes: - options.diagnosticBytes === undefined - ? { stdout: capacity, stderr: capacity } - : capacity, - acceptNonZeroExit: true, - }; -}; - class PackageCommands implements PackageVerifierCommands { - readonly #resources = new ArtifactResourceScope(); - readonly #controller = new AbortController(); - readonly #adopted = new WeakMap(); - readonly #resourceOwner: RetainedResourceScope; - #closed = false; #workspaceReleaseStarted = false; #closing: Promise | undefined; constructor( readonly workspace: PackageVerifierWorkspace, - readonly launcher: ( - input: OwnedProviderProcessSpawnOptions, - ) => Promise, - ) { - this.#resourceOwner = { - retain: (owner, resource) => this.adopt(owner, resource), - run: (work) => this.#resources.run(work), - }; - } + private readonly commands: VerifierCommands, + ) {} exec( command: string, arguments_: readonly string[], - options: PackageCommandOptions = {}, - ): Promise { - if (this.#closed) - return Promise.reject(new Error("Package command owner is closed")); - return this.#resources.run(async () => { - try { - const limits = commandLimits(options); - const signal = - options.signal === undefined - ? this.#controller.signal - : AbortSignal.any([this.#controller.signal, options.signal]); - // Match execFile's command deadline while keeping native ownership - // preparation independently bounded and inside this resource lifetime. - await prepareOwnedCommand(signal); - const snapshot = await runOwnedCommand( - { - command, - arguments: arguments_, - cwd: options.cwd ?? process.cwd(), - hostEnvironment: options.env ?? process.env, - expectedCommand: null, - runId: `rea-package-command-${randomUUID()}`, - sweepTokenOwnedProcesses: true, - }, - limits, - { - signal, - launcher: this.launcher, - }, - ); - const output = { - stdout: snapshot.stdout.text, - stderr: snapshot.stderr.text, - }; - if (typeof snapshot.exitCode === "number" && snapshot.exitCode !== 0) - throw new PackageCommandExitError( - command, - arguments_, - snapshot.exitCode, - snapshot, - ); - return output; - } catch (cause: unknown) { - retainOwnedCommandCleanup( - cause, - this.#resourceOwner, - "package-verifier", - command, - ); - throw cause; - } - }); + options: VerifierCommandOptions = {}, + ): Promise { + return this.commands.exec(command, arguments_, options); } execNpm( arguments_: readonly string[], - options: PackageCommandOptions = {}, - ): Promise { + options: VerifierCommandOptions = {}, + ): Promise { const invocation = npmInvocation(arguments_, options.env ?? process.env); return this.exec(invocation.command, invocation.arguments, options); } @@ -277,30 +97,19 @@ class PackageCommands implements PackageVerifierCommands { status: 0, }; } catch (cause: unknown) { - if (cause instanceof PackageCommandExitError && cause.code === 1) - return { stdout: cause.stdout, status: 1 }; + if (cause instanceof VerifierCommandExitError && cause.code === 1) + return { stdout: cause.snapshot.stdout.text, status: 1 }; throw cause; } } - adopt(owner: PackageCleanupOwner, resource = this.workspace.path): void { + adopt(owner: VerifierCleanupOwner, resource = this.workspace.path): void { if (this.#workspaceReleaseStarted) throw new Error("Package workspace release has started"); - let retained = this.#adopted.get(owner); - if (retained === undefined) { - retained = { - kind: "lease", - resource, - lease: AnalysisCleanupLease.forCloseOwner(owner, resource), - }; - this.#adopted.set(owner, retained); - } - this.#resources.retain(retained); + this.commands.adopt(owner, resource); } close(): Promise { - this.#closed = true; - this.#controller.abort(new Error("Package command owner closed")); if (this.#closing !== undefined) return this.#closing; let attempt: Promise; attempt = this.#release().catch((cause: unknown) => { @@ -312,21 +121,21 @@ class PackageCommands implements PackageVerifierCommands { } async #release(): Promise { - await this.#resources.close(); + await this.commands.close(); this.#workspaceReleaseStarted = true; await this.workspace.close(); } } -/** Allocate one command lifetime whose workspace depends on completed cleanup. */ +/** Remove the package workspace only after its shared command owner releases. */ export async function createPackageVerifierCommands({ workspace, - launcher = launchPackageCommand, -}: { + launcher, +}: VerifierCommandsOptions & { readonly workspace: PackageVerifierWorkspace; - readonly launcher?: ( - input: OwnedProviderProcessSpawnOptions, - ) => Promise; }): Promise { - return new PackageCommands(workspace, launcher); + const commands = await createVerifierCommands( + launcher === undefined ? {} : { launcher }, + ); + return new PackageCommands(workspace, commands); } diff --git a/scripts/lib/verifier-command-owner.mts b/scripts/lib/verifier-command-owner.mts new file mode 100644 index 0000000000..636f508cc5 --- /dev/null +++ b/scripts/lib/verifier-command-owner.mts @@ -0,0 +1,276 @@ +import { randomUUID } from "node:crypto"; + +import type { RetainedResourceScope } from "../../src/application/AnalysisProvider.js"; +import { AnalysisCleanupLease } from "../../src/artifacts/AnalysisCleanupLease.js"; +import { + ArtifactResourceScope, + type ArtifactResourceOwner, +} from "../../src/artifacts/ArtifactResourceScope.js"; +import { + prepareOwnedCommand, + runOwnedCommand, +} from "../../src/process/OwnedCommand.js"; +import { + spawnOwnedProviderProcess, + type OwnedProviderProcessSpawnOptions, + type ProviderProcessSnapshot, + type SpawnedOwnedProviderProcess, +} from "../../src/process/ProviderProcess.js"; +import { retainOwnedCommandCleanup } from "../../src/providers/OwnedCommandCleanup.js"; +import { WindowsOwnedProcess } from "../../src/windows/WindowsOwnedProcess.js"; +import { createWindowsOwnedCommand } from "./windows-owned-command.mjs"; + +interface VerifierCommandContext { + readonly signal?: AbortSignal | undefined; + readonly cwd?: string | undefined; + readonly env?: NodeJS.ProcessEnv | undefined; + readonly timeout?: number | undefined; + readonly windowsHide?: boolean | undefined; + /** Acquire a real owned stdin pipe; closed-pipe sends EOF before awaiting output. */ + readonly stdin?: "pipe" | "closed-pipe"; +} + +export type VerifierCommandOptions = VerifierCommandContext & + ( + | { + /** execFile-compatible capacity for each output stream. */ + readonly maxBuffer?: number | undefined; + readonly diagnosticBytes?: never; + } + | { + /** Canonical command capacity across both streams. */ + readonly diagnosticBytes: number; + readonly maxBuffer?: never; + } + ); + +export interface VerifierCommandOutput { + readonly stdout: string; + readonly stderr: string; +} + +export interface VerifierCleanupOwner { + close(): Promise; + readonly cleanupResources?: readonly string[] | undefined; +} + +/** Command and adopted-resource ownership without filesystem removal authority. */ +export interface VerifierCommands { + exec( + command: string, + arguments_: readonly string[], + options?: VerifierCommandOptions, + ): Promise; + adopt(owner: VerifierCleanupOwner, resource: string): void; + readonly cleanupResources: readonly string[]; + close(): Promise; +} + +export interface VerifierCommandsOptions { + readonly launcher?: ( + input: OwnedProviderProcessSpawnOptions, + ) => Promise; +} + +/** An ordinary nonzero exit whose process and output cleanup both completed. */ +export class VerifierCommandExitError extends Error { + constructor( + readonly command: string, + readonly arguments_: readonly string[], + readonly code: number, + readonly snapshot: ProviderProcessSnapshot, + ) { + super( + `Command failed: ${[command, ...arguments_].join(" ")}\n${snapshot.stderr.text}`, + ); + this.name = "VerifierCommandExitError"; + } +} + +const launchVerifierCommand = async ( + input: OwnedProviderProcessSpawnOptions, +): Promise => { + if (process.platform !== "win32") return spawnOwnedProviderProcess(input); + input.signal?.throwIfAborted(); + const child = createWindowsOwnedCommand({ + WindowsOwnedProcess, + command: input.command, + arguments: input.arguments, + cwd: input.cwd, + environment: { + ...input.hostEnvironment, + ...input.env, + REA_PROCESS_RUN_ID: input.runId, + }, + }); + // These labels describe the launch; cleanup authority is the exact native job. + return { + process: child, + ownership: { + runId: input.runId, + leaderPid: child.pid, + processGroupId: child.pid, + expectedParentPid: process.pid, + }, + cleanup: () => child.cleanup(), + }; +}; + +const commandLimits = (options: VerifierCommandOptions) => { + if ( + options.stdin !== undefined && + options.stdin !== "pipe" && + options.stdin !== "closed-pipe" + ) + throw new TypeError( + "Verifier command stdin must be pipe or closed-pipe when supplied", + ); + const timeout = options.timeout ?? 0; + if (!Number.isSafeInteger(timeout) || timeout < 0) + throw new RangeError( + "Verifier command timeout must be a nonnegative integer", + ); + if (options.maxBuffer !== undefined && options.diagnosticBytes !== undefined) + throw new TypeError("Select a per-stream or aggregate command capacity"); + const capacity = options.diagnosticBytes ?? options.maxBuffer ?? 1024 * 1024; + if (typeof capacity !== "number" || Number.isNaN(capacity) || capacity < 0) + throw new RangeError( + "Verifier command output capacity must be nonnegative", + ); + if (process.platform === "win32" && options.windowsHide === false) + throw new Error( + "Owned verifier commands require hidden Windows process launch", + ); + return { + timeoutMs: timeout === 0 ? null : timeout, + diagnosticBytes: + options.diagnosticBytes === undefined + ? { stdout: capacity, stderr: capacity } + : capacity, + acceptNonZeroExit: true, + ...(options.stdin === undefined ? {} : { stdin: options.stdin }), + }; +}; + +class VerifierCommandOwner implements VerifierCommands { + readonly #resources = new ArtifactResourceScope(); + readonly #controller = new AbortController(); + readonly #adopted = new WeakMap< + VerifierCleanupOwner, + ArtifactResourceOwner + >(); + readonly #resourceOwner: RetainedResourceScope; + #closed = false; + #released = false; + #closing: Promise | undefined; + + constructor( + private readonly launcher: ( + input: OwnedProviderProcessSpawnOptions, + ) => Promise, + ) { + this.#resourceOwner = { + retain: (owner, resource) => this.adopt(owner, resource), + run: (work) => this.#resources.run(work), + }; + } + + get cleanupResources(): readonly string[] { + return this.#resources.cleanupResources; + } + + exec( + command: string, + arguments_: readonly string[], + options: VerifierCommandOptions = {}, + ): Promise { + if (this.#closed) + return Promise.reject(new Error("Verifier command owner is closed")); + return this.#resources.run(async () => { + try { + const limits = commandLimits(options); + const signal = + options.signal === undefined + ? this.#controller.signal + : AbortSignal.any([this.#controller.signal, options.signal]); + // Independently bounded native preparation stays outside the command's + // deadline and inside this admitted resource lifetime. + await prepareOwnedCommand(signal); + const snapshot = await runOwnedCommand( + { + command, + arguments: arguments_, + cwd: options.cwd ?? process.cwd(), + hostEnvironment: options.env ?? process.env, + expectedCommand: null, + runId: `rea-verifier-command-${randomUUID()}`, + sweepTokenOwnedProcesses: true, + }, + limits, + { signal, launcher: this.launcher }, + ); + if (typeof snapshot.exitCode === "number" && snapshot.exitCode !== 0) + throw new VerifierCommandExitError( + command, + arguments_, + snapshot.exitCode, + snapshot, + ); + return { + stdout: snapshot.stdout.text, + stderr: snapshot.stderr.text, + }; + } catch (cause: unknown) { + retainOwnedCommandCleanup( + cause, + this.#resourceOwner, + "verifier", + command, + ); + throw cause; + } + }); + } + + adopt(owner: VerifierCleanupOwner, resource: string): void { + if (this.#released) + throw new Error("Verifier command resource release has completed"); + let retained = this.#adopted.get(owner); + if (retained === undefined) { + retained = { + kind: "lease", + resource, + lease: AnalysisCleanupLease.forCloseOwner(owner, resource), + }; + this.#adopted.set(owner, retained); + } + // An admitted command can transfer a late preparation/cleanup capability + // while close is joining it. Acquisition must precede this scope's release. + this.#resources.retain(retained); + } + + close(): Promise { + this.#closed = true; + this.#controller.abort(new Error("Verifier command owner closed")); + if (this.#closing !== undefined) return this.#closing; + let attempt: Promise; + attempt = this.#resources + .close() + .then(() => { + this.#released = true; + }) + .catch((cause: unknown) => { + if (this.#closing === attempt) this.#closing = undefined; + throw cause; + }); + this.#closing = attempt; + return attempt; + } +} + +/** Allocate command ownership without acquiring or deleting an evidence root. */ +export async function createVerifierCommands({ + launcher = launchVerifierCommand, +}: VerifierCommandsOptions = {}): Promise { + return new VerifierCommandOwner(launcher); +} diff --git a/scripts/lib/verifier-file-snapshot-owner.mts b/scripts/lib/verifier-file-snapshot-owner.mts index 0e321f6665..e32fe9e6ef 100644 --- a/scripts/lib/verifier-file-snapshot-owner.mts +++ b/scripts/lib/verifier-file-snapshot-owner.mts @@ -2,7 +2,7 @@ import { lstat, open } from "node:fs/promises"; import { ArtifactResourceScope } from "../../src/artifacts/ArtifactResourceScope.js"; import { OwnedFileHandle } from "../../src/filesystem/OwnedFileHandle.js"; -import type { PackageVerifierCommands } from "./package-command-owner.mjs"; +import type { VerifierCommands } from "./verifier-command-owner.mjs"; import { withVerifierCleanup } from "./verifier-cleanup.mjs"; export type VerifierFileSnapshotEntry = @@ -19,7 +19,7 @@ export type VerifierFileSnapshotEntry = /** Capture source values while the command owner retains every in-flight file. */ export async function captureVerifierFileSnapshot( - commands: Pick, + commands: Pick, paths: readonly string[], ): Promise { const resources = new ArtifactResourceScope(); @@ -61,7 +61,7 @@ export async function captureVerifierFileSnapshot( /** Compare the same selected paths against their completed baseline facts. */ export async function assertVerifierFilesUnchanged( - commands: Pick, + commands: Pick, snapshot: readonly VerifierFileSnapshotEntry[], label: string, ): Promise { diff --git a/scripts/lib/verifier-terminal-owner.mts b/scripts/lib/verifier-terminal-owner.mts index 1c23ac41e9..d0a0055c4f 100644 --- a/scripts/lib/verifier-terminal-owner.mts +++ b/scripts/lib/verifier-terminal-owner.mts @@ -19,16 +19,16 @@ import { ProviderStartupDeadline } from "../../src/process/ProviderDeadline.js"; import { processCaptureOwnershipUnavailableReason } from "../../src/process/capture/ProcessCaptureCapability.js"; import { retainOwnedCommandCleanup } from "../../src/providers/OwnedCommandCleanup.js"; import type { - PackageCleanupOwner, - PackageVerifierCommands, -} from "./package-command-owner.mjs"; + VerifierCleanupOwner, + VerifierCommands, +} from "./verifier-command-owner.mjs"; import { closeVerifierResources, withVerifierCleanup, } from "./verifier-cleanup.mjs"; export interface VerifierTerminalOptions { - readonly commands: Pick; + readonly commands: Pick; readonly command: string; readonly args: readonly string[]; readonly cwd: string; @@ -79,7 +79,7 @@ export class VerifierTerminalCleanupFailure extends Error { readonly reason = "cleanup_incomplete"; constructor( - readonly cleanupOwner: PackageCleanupOwner, + readonly cleanupOwner: VerifierCleanupOwner, readonly output: string, message: string, options?: ErrorOptions, @@ -102,7 +102,7 @@ const decodedPrefix = (text: string, capacity: number): string => { return text.slice(0, end); }; -class VerifierTerminalOwner implements PackageCleanupOwner { +class VerifierTerminalOwner implements VerifierCleanupOwner { readonly #controller = new AbortController(); readonly #runId = randomUUID(); readonly #evidence = new ProcessOwnershipEvidenceOwner(); @@ -409,7 +409,7 @@ export async function runOwnedVerifierTerminal( "Verifier terminal output capacity must be a non-negative safe integer", ); const owner = new VerifierTerminalOwner(options, dependencies); - options.commands.adopt(owner); + options.commands.adopt(owner, `verifier terminal ${options.command}`); owner.start(); let release: Promise | undefined; const finish = (): Promise => (release ??= owner.release()); diff --git a/scripts/lib/verify-commands.d.mts b/scripts/lib/verify-commands.d.mts new file mode 100644 index 0000000000..72c9398441 --- /dev/null +++ b/scripts/lib/verify-commands.d.mts @@ -0,0 +1,9 @@ +export { createVerifierCommands } from "./verifier-command-owner.mjs"; +export type { + VerifierCommands, + VerifierCommandsOptions, + VerifierCommandOptions, + VerifierCommandOutput, + VerifierCleanupOwner, + VerifierCommandExitError, +} from "./verifier-command-owner.mjs"; diff --git a/scripts/lib/verify-commands.mjs b/scripts/lib/verify-commands.mjs new file mode 100644 index 0000000000..bbe9f7c64a --- /dev/null +++ b/scripts/lib/verify-commands.mjs @@ -0,0 +1,13 @@ +let implementation; + +/** Load the development command owner and its canonical process source graph. */ +export async function createVerifierCommands(options = {}) { + implementation ??= import("tsx/esm/api").then(({ tsImport }) => + tsImport( + new URL("./verifier-command-owner.mts", import.meta.url).href, + import.meta.url, + ), + ); + const module = await implementation; + return module.createVerifierCommands(options); +} diff --git a/scripts/lib/verify-package-commands.d.mts b/scripts/lib/verify-package-commands.d.mts index b05e709864..37df25292b 100644 --- a/scripts/lib/verify-package-commands.d.mts +++ b/scripts/lib/verify-package-commands.d.mts @@ -2,7 +2,4 @@ export { createPackageVerifierCommands } from "./package-command-owner.mjs"; export type { PackageVerifierCommands, PackageVerifierWorkspace, - PackageCommandOptions, - PackageCommandOutput, - PackageCleanupOwner, } from "./package-command-owner.mjs"; diff --git a/scripts/lib/verify-setup-terminal.d.mts b/scripts/lib/verify-setup-terminal.d.mts index 04a625a486..6900c7eae8 100644 --- a/scripts/lib/verify-setup-terminal.d.mts +++ b/scripts/lib/verify-setup-terminal.d.mts @@ -1,7 +1,7 @@ -import type { PackageVerifierCommands } from "./verify-package-commands.mjs"; +import type { VerifierCommands } from "./verifier-command-owner.mjs"; export interface SetupTerminalOptions { - readonly commands: Pick; + readonly commands: Pick; readonly command: string; readonly environment: NodeJS.ProcessEnv; readonly root: string; diff --git a/scripts/verify-native-agent-client.mjs b/scripts/verify-native-agent-client.mjs index d4ba0c8b43..9c05492d3b 100644 --- a/scripts/verify-native-agent-client.mjs +++ b/scripts/verify-native-agent-client.mjs @@ -8,7 +8,8 @@ import { TOOL_CONTRACTS, toolContract, } from "../dist/contracts/toolContracts.js"; -import { exec } from "./lib/verify-package-core.mjs"; +import { createVerifierCommands } from "./lib/verify-commands.mjs"; +import { withVerifierCleanup } from "./lib/verifier-cleanup.mjs"; import { assertJavaScriptApplicationInputPaths } from "./lib/javascript-application-paths.mjs"; import { createVerifierRun, completeVerifierRun } from "./lib/verifier-run.mjs"; import { createOpenAiModelFixture } from "./verify/clients/model-fixture.mjs"; @@ -43,86 +44,9 @@ const skillPath = join( "SKILL.md", ); const verifier = createVerifierRun(); -await Promise.all([ - mkdir(account), - mkdir(profile, { recursive: true }), - ...(stickyHermes ? [mkdir(selectedProfile, { recursive: true })] : []), - mkdir(target, { recursive: true }), -]); -await writeFile( - join(target, "app.js"), - 'export function result() { return { clientCompatibility: "REA_CLIENT_FIXTURE", count: 7 }; }\n', -); -const environment = { - PATH: process.env.PATH ?? "", - HOME: account, - USERPROFILE: account, - [client === "pi" ? "PI_CODING_AGENT_DIR" : "HERMES_HOME"]: profile, - PI_OFFLINE: "1", - REA_PROCESS_RUN_ID: verifier.run_id, -}; -const configPath = join( - selectedProfile, - client === "pi" ? "mcp.json" : "config.yaml", -); -if (stickyHermes) await writeFile(join(profile, "active_profile"), "coder\n"); -const original = - client === "pi" - ? JSON.stringify({ - mcpServers: { other: { command: "unrelated-server", enabled: false } }, - }) - : "# Keep caller settings.\nmcp_servers:\n other:\n command: unrelated-server\n enabled: false\n"; -await writeFile(configPath, original); -const clientVersion = ( - await exec(command, ["--version"], { env: environment, timeout: 30_000 }) -).stdout.trim(); -for (const args of [ - ["setup", "--client", client, "--dry-run"], - ["setup", "--client", client, "--yes"], - ["doctor", "--client", client, "--skill"], -]) { - const result = await exec( - process.execPath, - [join(runtimeRoot, "scripts/rea.mjs"), ...args, "--json"], - { env: environment, cwd: workspace, timeout: 60_000 }, - ); - const parsed = JSON.parse(result.stdout); - if (args[0] === "doctor") assert.equal(parsed.healthy, true); - else - assert.equal( - parsed.status, - args.includes("--dry-run") ? "planned" : "ready", - ); - await writeFile( - join(lab, `${args[0]}${args.includes("--dry-run") ? "-plan" : ""}.json`), - result.stdout, - ); -} -assert.equal(await readFile(`${configPath}.rea.backup`, "utf8"), original); -const installedConfiguration = await readFile(configPath, "utf8"); -assert(installedConfiguration.includes("unrelated-server")); -if (client === "hermes") - assert(installedConfiguration.includes("# Keep caller settings.")); -const repeated = await exec( - process.execPath, - [ - join(runtimeRoot, "scripts/rea.mjs"), - "setup", - "--client", - client, - "--yes", - "--json", - ], - { env: environment, cwd: workspace, timeout: 60_000 }, -); -assert.deepEqual(JSON.parse(repeated.stdout).appliedActions, []); -assert.equal(await readFile(configPath, "utf8"), installedConfiguration); -assert.equal(await readFile(`${configPath}.rea.backup`, "utf8"), original); -assert( - (await readFile(skillPath, "utf8")).includes( - "name: reverse-engineer-anything", - ), -); +let clientVersion = null; +let fixture; +let passed = false; let evidenceSeen = false; let skillLoaded = false; let skillRequested = false; @@ -132,331 +56,460 @@ let describeRequested = false; let finalSeen = false; let catalog; let artifactDigest; -const extractEvidence = (value) => { - if (typeof value === "string") { - if ( - value.startsWith("") - ) { - // Hermes wraps external MCP content with its own instruction banner. - // Decode only that observed envelope, retaining the original request artifact. - const start = value.indexOf("\n\n"); - if (start >= 0) - return extractEvidence( - value - .slice(start + 2, value.lastIndexOf("")) - .trim(), - ); - } - try { - return extractEvidence(JSON.parse(value)); - } catch { - return undefined; - } - } - if (value && typeof value === "object") { - if ( - typeof value.evidence_id === "string" && - value.normalized_result?.graph?.nodes - ) - return value; - for (const child of Object.values(value)) { - const found = extractEvidence(child); - if (found) return found; - } - } -}; -const validateEvidence = async (value) => { - const evidence = toolContract( - "analyze_javascript_application", - ).outputSchema.parse(value); - await assertJavaScriptApplicationInputPaths(evidence, target); - const shape = evidence.normalized_result.graph.nodes - .flatMap((n) => n.observations) - .map((o) => o.properties) - .find( - (p) => - p.semantic_role === "export-return-shapes" && - p.exported_name === "result", +const commands = await createVerifierCommands(); +await withVerifierCleanup(async () => { + await withVerifierCleanup(async () => { + await Promise.all([ + mkdir(account), + mkdir(profile, { recursive: true }), + ...(stickyHermes ? [mkdir(selectedProfile, { recursive: true })] : []), + mkdir(target, { recursive: true }), + ]); + await writeFile( + join(target, "app.js"), + 'export function result() { return { clientCompatibility: "REA_CLIENT_FIXTURE", count: 7 }; }\n', ); - assert(shape?.return_shape_coverage?.projection_complete); - const fields = shape.static_return_shapes.flatMap((s) => s.fields); - assert( - fields.some( - (f) => - f.path === "/clientCompatibility" && - f.state === "literal" && - f.value === "REA_CLIENT_FIXTURE", - ), - ); - assert( - fields.some( - (f) => f.path === "/count" && f.state === "literal" && f.value === 7, - ), - ); - const serialized = JSON.stringify(evidence); - await writeFile(join(lab, "evidence.json"), serialized); - artifactDigest = createHash("sha256").update(serialized).digest("hex"); - evidenceSeen = true; -}; -const fixture = await createOpenAiModelFixture({ - directory: lab, - prefix: `${client}-${mode}`, - onRequest: async ({ body, tools, requestIndex }) => { - assert( - requestIndex <= 6, - "Client did not complete the bounded fixture workflow.", + const environment = { + PATH: process.env.PATH ?? "", + HOME: account, + USERPROFILE: account, + [client === "pi" ? "PI_CODING_AGENT_DIR" : "HERMES_HOME"]: profile, + PI_OFFLINE: "1", + REA_PROCESS_RUN_ID: verifier.run_id, + }; + const configPath = join( + selectedProfile, + client === "pi" ? "mcp.json" : "config.yaml", ); - const messages = body.messages ?? []; - const toolMessages = messages.filter((m) => m.role === "tool"); - const serializedTools = JSON.stringify(toolMessages); - const call = (name, args) => ({ - role: "assistant", - content: null, - tool_calls: [ + if (stickyHermes) + await writeFile(join(profile, "active_profile"), "coder\n"); + const original = + client === "pi" + ? JSON.stringify({ + mcpServers: { + other: { command: "unrelated-server", enabled: false }, + }, + }) + : "# Keep caller settings.\nmcp_servers:\n other:\n command: unrelated-server\n enabled: false\n"; + await writeFile(configPath, original); + clientVersion = ( + await commands.exec(command, ["--version"], { + env: environment, + timeout: 30_000, + stdin: "pipe", + }) + ).stdout.trim(); + for (const args of [ + ["setup", "--client", client, "--dry-run"], + ["setup", "--client", client, "--yes"], + ["doctor", "--client", client, "--skill"], + ]) { + const result = await commands.exec( + process.execPath, + [join(runtimeRoot, "scripts/rea.mjs"), ...args, "--json"], { - id: `call_rea_${requestIndex}`, - type: "function", - function: { name, arguments: JSON.stringify(args) }, + env: environment, + cwd: workspace, + timeout: 60_000, + stdin: "pipe", }, - ], - }); - if ( - client === "pi" && - callRequested && - serializedTools.includes("REA_CLIENT_FIXTURE") - ) { - const fact = toolMessages - .flatMap((m) => { - const text = - typeof m.content === "string" - ? m.content - : JSON.stringify(m.content); - const marker = text.indexOf('{"REA_CLIENT_FIXTURE":true'); - if (marker < 0) return []; - const end = text.indexOf("}\n", marker); - try { - return [ - JSON.parse(text.slice(marker, end < 0 ? undefined : end + 1)), - ]; - } catch { - return []; - } - }) - .find((v) => Array.isArray(v.catalog)); - assert( - fact, - "Pi codemode must return its actual catalog and skill-read receipt.", ); - catalog = fact.catalog; - skillLoaded = fact.skill_loaded === true; - await validateEvidence(JSON.parse(await readFile(artifactPath, "utf8"))); - } else if (client === "hermes") { - const search = tools.find((t) => t.name === "tool_search"); - const exposed = tools - .filter((t) => t.name.startsWith("mcp__rea__")) - .map((t) => t.name.slice(10)); - const deferred = [ - ...(search?.description ?? "").matchAll(/^- mcp__rea__(\w+):/gmu), - ].map((m) => m[1]); - const native = [...new Set([...exposed, ...deferred])]; - // Hermes advertises MCP prompt utilities beside server tools. - if (native.length) - catalog = native.filter( - (name) => !["get_prompt", "list_prompts"].includes(name), + const parsed = JSON.parse(result.stdout); + if (args[0] === "doctor") assert.equal(parsed.healthy, true); + else + assert.equal( + parsed.status, + args.includes("--dry-run") ? "planned" : "ready", ); - skillLoaded ||= - serializedTools.includes(skillPath) && - serializedTools.includes("reverse-engineer-anything"); - for (const message of toolMessages) { - const evidence = extractEvidence(message.content); - if (evidence) await validateEvidence(evidence); + await writeFile( + join( + lab, + `${args[0]}${args.includes("--dry-run") ? "-plan" : ""}.json`, + ), + result.stdout, + ); + } + assert.equal(await readFile(`${configPath}.rea.backup`, "utf8"), original); + const installedConfiguration = await readFile(configPath, "utf8"); + assert(installedConfiguration.includes("unrelated-server")); + if (client === "hermes") + assert(installedConfiguration.includes("# Keep caller settings.")); + const repeated = await commands.exec( + process.execPath, + [ + join(runtimeRoot, "scripts/rea.mjs"), + "setup", + "--client", + client, + "--yes", + "--json", + ], + { + env: environment, + cwd: workspace, + timeout: 60_000, + stdin: "pipe", + }, + ); + assert.deepEqual(JSON.parse(repeated.stdout).appliedActions, []); + assert.equal(await readFile(configPath, "utf8"), installedConfiguration); + assert.equal(await readFile(`${configPath}.rea.backup`, "utf8"), original); + assert( + (await readFile(skillPath, "utf8")).includes( + "name: reverse-engineer-anything", + ), + ); + const extractEvidence = (value) => { + if (typeof value === "string") { + if ( + value.startsWith("") + ) { + // Hermes wraps external MCP content with its own instruction banner. + // Decode only that observed envelope, retaining the original request artifact. + const start = value.indexOf("\n\n"); + if (start >= 0) + return extractEvidence( + value + .slice(start + 2, value.lastIndexOf("")) + .trim(), + ); + } + try { + return extractEvidence(JSON.parse(value)); + } catch { + return undefined; + } } - } - if (mode === "chat" || evidenceSeen) { - finalSeen = true; - return { role: "assistant", content: "REA_CLIENT_COMPATIBILITY_OK" }; - } - if (client === "pi") { - const tool = tools.find((t) => t.name === "codemode"); + if (value && typeof value === "object") { + if ( + typeof value.evidence_id === "string" && + value.normalized_result?.graph?.nodes + ) + return value; + for (const child of Object.values(value)) { + const found = extractEvidence(child); + if (found) return found; + } + } + }; + const validateEvidence = async (value) => { + const evidence = toolContract( + "analyze_javascript_application", + ).outputSchema.parse(value); + await assertJavaScriptApplicationInputPaths(evidence, target); + const shape = evidence.normalized_result.graph.nodes + .flatMap((n) => n.observations) + .map((o) => o.properties) + .find( + (p) => + p.semantic_role === "export-return-shapes" && + p.exported_name === "result", + ); + assert(shape?.return_shape_coverage?.projection_complete); + const fields = shape.static_return_shapes.flatMap((s) => s.fields); assert( - tool, - `Native Pi codemode unavailable: ${tools.map((t) => t.name).join(", ")}`, + fields.some( + (f) => + f.path === "/clientCompatibility" && + f.state === "literal" && + f.value === "REA_CLIENT_FIXTURE", + ), ); assert( - JSON.stringify(messages).includes(skillPath), - "Pi must advertise the installed REA skill.", + fields.some( + (f) => f.path === "/count" && f.state === "literal" && f.value === 7, + ), ); - callRequested = true; - return call(tool.name, { - code: `await describeNamespace("rea"); const skill = await tools.read({path:${JSON.stringify(skillPath)}}); const result = await tools.mcp__rea__analyze_javascript_application({input_path:${JSON.stringify(target)},format:"directory"}); if(result.isError) throw new Error(JSON.stringify(result)); const evidence = result.structuredContent ?? JSON.parse(result.content.find(c=>c.type==="text").text); await tools.write({path:${JSON.stringify(artifactPath)},content:JSON.stringify(evidence)}); text({REA_CLIENT_FIXTURE:true,skill_loaded:typeof skill==="string"&&skill.includes("name: reverse-engineer-anything"),catalog:ALL_TOOLS.filter(t=>t.name.startsWith("mcp__rea__")).map(t=>t.name.slice(10))});`, - }); - } - if (!skillRequested) { - const tool = tools.find((t) => t.name === "skill_view"); - assert(tool, "Hermes skill_view must be available."); - skillRequested = true; - return call(tool.name, { name: "reverse-engineer-anything" }); - } - if (!searchRequested) { - const tool = tools.find((t) => t.name === "tool_search"); - assert(tool, "Hermes native deferred tool_search must be available."); - searchRequested = true; - return call(tool.name, { queries: ["analyze_javascript_application"] }); - } - assert( - serializedTools.includes("mcp__rea__analyze_javascript_application"), - "Hermes must discover the analysis tool.", - ); - if (!describeRequested) { - const tool = tools.find((t) => t.name === "tool_describe"); - assert(tool, "Hermes must load the actual parameter schema."); - describeRequested = true; - return call(tool.name, { - names: ["mcp__rea__analyze_javascript_application"], - }); - } - const tool = tools.find((t) => t.name === "tool_call"); - assert(tool, "Hermes native deferred tool_call must be available."); - callRequested = true; - return call(tool.name, { - calls: [ - { - name: "mcp__rea__analyze_javascript_application", - arguments: { input_path: target, format: "directory" }, - }, - ], - }); - }, -}); -if (client === "pi") { - await writeFile( - join(profile, "models.json"), - JSON.stringify({ - providers: { - "rea-fixture": { - baseUrl: fixture.baseUrl, - api: "openai-completions", - apiKey: "local-fixture-key", - models: [ + const serialized = JSON.stringify(evidence); + await writeFile(join(lab, "evidence.json"), serialized); + artifactDigest = createHash("sha256").update(serialized).digest("hex"); + evidenceSeen = true; + }; + fixture = await createOpenAiModelFixture({ + directory: lab, + prefix: `${client}-${mode}`, + onRequest: async ({ body, tools, requestIndex }) => { + assert( + requestIndex <= 6, + "Client did not complete the bounded fixture workflow.", + ); + const messages = body.messages ?? []; + const toolMessages = messages.filter((m) => m.role === "tool"); + const serializedTools = JSON.stringify(toolMessages); + const call = (name, args) => ({ + role: "assistant", + content: null, + tool_calls: [ { - id: "rea-client-fixture", - reasoning: false, - contextWindow: 200000, - maxTokens: 4096, + id: `call_rea_${requestIndex}`, + type: "function", + function: { name, arguments: JSON.stringify(args) }, }, ], - }, + }); + if ( + client === "pi" && + callRequested && + serializedTools.includes("REA_CLIENT_FIXTURE") + ) { + const fact = toolMessages + .flatMap((m) => { + const text = + typeof m.content === "string" + ? m.content + : JSON.stringify(m.content); + const marker = text.indexOf('{"REA_CLIENT_FIXTURE":true'); + if (marker < 0) return []; + const end = text.indexOf("}\n", marker); + try { + return [ + JSON.parse(text.slice(marker, end < 0 ? undefined : end + 1)), + ]; + } catch { + return []; + } + }) + .find((v) => Array.isArray(v.catalog)); + assert( + fact, + "Pi codemode must return its actual catalog and skill-read receipt.", + ); + catalog = fact.catalog; + skillLoaded = fact.skill_loaded === true; + await validateEvidence( + JSON.parse(await readFile(artifactPath, "utf8")), + ); + } else if (client === "hermes") { + const search = tools.find((t) => t.name === "tool_search"); + const exposed = tools + .filter((t) => t.name.startsWith("mcp__rea__")) + .map((t) => t.name.slice(10)); + const deferred = [ + ...(search?.description ?? "").matchAll(/^- mcp__rea__(\w+):/gmu), + ].map((m) => m[1]); + const native = [...new Set([...exposed, ...deferred])]; + // Hermes advertises MCP prompt utilities beside server tools. + if (native.length) + catalog = native.filter( + (name) => !["get_prompt", "list_prompts"].includes(name), + ); + skillLoaded ||= + serializedTools.includes(skillPath) && + serializedTools.includes("reverse-engineer-anything"); + for (const message of toolMessages) { + const evidence = extractEvidence(message.content); + if (evidence) await validateEvidence(evidence); + } + } + if (mode === "chat" || evidenceSeen) { + finalSeen = true; + return { role: "assistant", content: "REA_CLIENT_COMPATIBILITY_OK" }; + } + if (client === "pi") { + const tool = tools.find((t) => t.name === "codemode"); + assert( + tool, + `Native Pi codemode unavailable: ${tools.map((t) => t.name).join(", ")}`, + ); + assert( + JSON.stringify(messages).includes(skillPath), + "Pi must advertise the installed REA skill.", + ); + callRequested = true; + return call(tool.name, { + code: `await describeNamespace("rea"); const skill = await tools.read({path:${JSON.stringify(skillPath)}}); const result = await tools.mcp__rea__analyze_javascript_application({input_path:${JSON.stringify(target)},format:"directory"}); if(result.isError) throw new Error(JSON.stringify(result)); const evidence = result.structuredContent ?? JSON.parse(result.content.find(c=>c.type==="text").text); await tools.write({path:${JSON.stringify(artifactPath)},content:JSON.stringify(evidence)}); text({REA_CLIENT_FIXTURE:true,skill_loaded:typeof skill==="string"&&skill.includes("name: reverse-engineer-anything"),catalog:ALL_TOOLS.filter(t=>t.name.startsWith("mcp__rea__")).map(t=>t.name.slice(10))});`, + }); + } + if (!skillRequested) { + const tool = tools.find((t) => t.name === "skill_view"); + assert(tool, "Hermes skill_view must be available."); + skillRequested = true; + return call(tool.name, { name: "reverse-engineer-anything" }); + } + if (!searchRequested) { + const tool = tools.find((t) => t.name === "tool_search"); + assert(tool, "Hermes native deferred tool_search must be available."); + searchRequested = true; + return call(tool.name, { + queries: ["analyze_javascript_application"], + }); + } + assert( + serializedTools.includes("mcp__rea__analyze_javascript_application"), + "Hermes must discover the analysis tool.", + ); + if (!describeRequested) { + const tool = tools.find((t) => t.name === "tool_describe"); + assert(tool, "Hermes must load the actual parameter schema."); + describeRequested = true; + return call(tool.name, { + names: ["mcp__rea__analyze_javascript_application"], + }); + } + const tool = tools.find((t) => t.name === "tool_call"); + assert(tool, "Hermes native deferred tool_call must be available."); + callRequested = true; + return call(tool.name, { + calls: [ + { + name: "mcp__rea__analyze_javascript_application", + arguments: { input_path: target, format: "directory" }, + }, + ], + }); }, - }), - ); -} else { - const configured = await readFile(configPath, "utf8"); - await writeFile( - configPath, - configured + - `\nmodel:\n default: rea-client-fixture\n provider: custom\n base_url: ${fixture.baseUrl}\nagent:\n max_turns: 6\n`, - ); -} -const args = - client === "pi" - ? [ - "--offline", - "--provider", - "rea-fixture", - "--model", - "rea-client-fixture", - "--no-session", - "--no-context-files", - "--mode", - "json", - "-p", - mode === "chat" - ? "Say hello briefly." - : `Use REA to analyze ${target}.`, - ] - : [ - "--oneshot", - mode === "chat" - ? "Say hello briefly." - : `Use REA to analyze ${target}.`, - "--provider", - "custom", - "--model", - "rea-client-fixture", - "--reasoning", - "none", - ]; -let failure; -try { - const execution = exec(command, args, { - env: { - ...environment, - OPENAI_API_KEY: "local-fixture-key", - OPENAI_BASE_URL: fixture.baseUrl, - }, - cwd: workspace, - timeout: 120_000, - maxBuffer: 10 * 1024 * 1024, - }); - // Pi consumes piped input before its first prompt; an unused open pipe is not EOF. - execution.child.stdin.end(); - const result = await execution; - await writeFile(join(lab, "client.stdout"), result.stdout); - await writeFile(join(lab, "client.stderr"), result.stderr); - assert.equal(fixture.failure, undefined); - assert(finalSeen); - assert(result.stdout.includes("REA_CLIENT_COMPATIBILITY_OK")); - if (mode === "call") { - assert(evidenceSeen); - assert( - skillLoaded, - "Native client must actually read the installed skill.", - ); - assert.deepEqual( - catalog?.sort(), - TOOL_CONTRACTS.map((t) => t.name).sort(), - "The complete REA catalog must survive native client discovery.", - ); - } -} catch (cause) { - failure = cause; - if (typeof cause?.stdout === "string") - await writeFile(join(lab, "client.stdout"), cause.stdout); - if (typeof cause?.stderr === "string") - await writeFile(join(lab, "client.stderr"), cause.stderr); -} finally { - await fixture.close(); - const receipt = { - client, - clientVersion, - mode, - passed: failure === undefined, - artifacts: lab, - runtimeRoot, - host: { - platform: process.platform, - architecture: process.arch, - node: process.versions.node, - }, - modelFixture: - "loopback deterministic OpenAI-compatible endpoint; no live provider claim", - skillDiscovery: + }); + if (client === "pi") { + await writeFile( + join(profile, "models.json"), + JSON.stringify({ + providers: { + "rea-fixture": { + baseUrl: fixture.baseUrl, + api: "openai-completions", + apiKey: "local-fixture-key", + models: [ + { + id: "rea-client-fixture", + reasoning: false, + contextWindow: 200000, + maxTokens: 4096, + }, + ], + }, + }, + }), + ); + } else { + const configured = await readFile(configPath, "utf8"); + await writeFile( + configPath, + configured + + `\nmodel:\n default: rea-client-fixture\n provider: custom\n base_url: ${fixture.baseUrl}\nagent:\n max_turns: 6\n`, + ); + } + const args = client === "pi" - ? "native HOME shared personal directory; no explicit skill setting" - : "native HERMES_HOME personal directory", - stickyHermes, - catalogSize: catalog?.length, - evidenceSeen, - skillLoaded, - artifactDigest, - finalSeen, - requests: fixture.requests, - probes: fixture.probes, - fixtureFailure: fixture.failure, - verifier: await completeVerifierRun(verifier), - }; - await writeFile(join(lab, "receipt.json"), JSON.stringify(receipt, null, 2)); - console.log(JSON.stringify(receipt)); -} -if (failure !== undefined) throw failure; + ? [ + "--offline", + "--provider", + "rea-fixture", + "--model", + "rea-client-fixture", + "--no-session", + "--no-context-files", + "--mode", + "json", + "-p", + mode === "chat" + ? "Say hello briefly." + : `Use REA to analyze ${target}.`, + ] + : [ + "--oneshot", + mode === "chat" + ? "Say hello briefly." + : `Use REA to analyze ${target}.`, + "--provider", + "custom", + "--model", + "rea-client-fixture", + "--reasoning", + "none", + ]; + const result = await commands + .exec(command, args, { + env: { + ...environment, + OPENAI_API_KEY: "local-fixture-key", + OPENAI_BASE_URL: fixture.baseUrl, + }, + cwd: workspace, + timeout: 120_000, + maxBuffer: 10 * 1024 * 1024, + // Pi consumes piped input before its first prompt; an unused open pipe is not EOF. + stdin: "closed-pipe", + }) + .catch(async (cause) => { + const outputWriters = []; + if (typeof cause?.snapshot?.stdout?.text === "string") + outputWriters.push(() => + writeFile(join(lab, "client.stdout"), cause.snapshot.stdout.text), + ); + if (typeof cause?.snapshot?.stderr?.text === "string") + outputWriters.push(() => + writeFile(join(lab, "client.stderr"), cause.snapshot.stderr.text), + ); + return withVerifierCleanup(() => { + throw cause; + }, outputWriters); + }); + await withVerifierCleanup( + () => writeFile(join(lab, "client.stdout"), result.stdout), + [() => writeFile(join(lab, "client.stderr"), result.stderr)], + ); + assert.equal(fixture.failure, undefined); + assert(finalSeen); + assert(result.stdout.includes("REA_CLIENT_COMPATIBILITY_OK")); + if (mode === "call") { + assert(evidenceSeen); + assert( + skillLoaded, + "Native client must actually read the installed skill.", + ); + assert.deepEqual( + catalog?.sort(), + TOOL_CONTRACTS.map((t) => t.name).sort(), + "The complete REA catalog must survive native client discovery.", + ); + } + }, [ + () => commands.close(), + async () => { + await fixture?.close(); + }, + ]); + passed = true; +}, [ + async () => { + const receipt = { + client, + clientVersion, + mode, + passed, + artifacts: lab, + runtimeRoot, + host: { + platform: process.platform, + architecture: process.arch, + node: process.versions.node, + }, + modelFixture: + "loopback deterministic OpenAI-compatible endpoint; no live provider claim", + skillDiscovery: + client === "pi" + ? "native HOME shared personal directory; no explicit skill setting" + : "native HERMES_HOME personal directory", + stickyHermes, + catalogSize: catalog?.length, + evidenceSeen, + skillLoaded, + artifactDigest, + finalSeen, + requests: fixture?.requests, + probes: fixture?.probes, + fixtureFailure: fixture?.failure, + verifier: await completeVerifierRun(verifier), + }; + await withVerifierCleanup( + () => + writeFile(join(lab, "receipt.json"), JSON.stringify(receipt, null, 2)), + [ + () => { + console.log(JSON.stringify(receipt)); + }, + ], + ); + }, +]); diff --git a/src/process/OwnedCommand.test.ts b/src/process/OwnedCommand.test.ts new file mode 100644 index 0000000000..160e546e1d --- /dev/null +++ b/src/process/OwnedCommand.test.ts @@ -0,0 +1,439 @@ +import { EventEmitter } from "node:events"; +import { PassThrough, Writable } from "node:stream"; +import { expect, it, vi } from "vitest"; + +import { createDeferred } from "../../tests/fixtures/binarySession.js"; +import { OwnedCommandFailure, runOwnedCommand } from "./OwnedCommand.js"; +import type { ProcessCleanupResult } from "./ProcessOwnership.js"; + +class CommandChild extends EventEmitter { + readonly stdout = new PassThrough(); + readonly stderr = new PassThrough(); + readonly kill = vi.fn(() => false); + exitCode: number | null = null; + readonly signalCode = null; + + constructor(readonly stdin: Writable | null) { + super(); + } + + complete(): void { + if (this.exitCode !== null) return; + this.exitCode = 0; + this.stdout.end(); + this.stderr.end(); + this.emit("exit", 0, null); + this.emit("close", 0, null); + } +} + +const ownedInput = (input: Writable | null) => { + const child = new CommandChild(input); + const ownership = { + runId: "authored-command-input", + leaderPid: 41234, + processGroupId: 41234, + }; + const cleanup = vi.fn(async (): Promise => { + input?.destroy(); + child.complete(); + return { cleaned: true, signaled: false }; + }); + return { child, ownership, cleanup }; +}; + +const spawn = { + command: "authored-command", + arguments: [], + runId: "authored-command-input", +}; + +it("captures synchronous EOF output after supervision and interprets it before cleanup", async () => { + const stages: string[] = []; + const input = new Writable({ + final(callback) { + stages.push("input-final"); + fixture.child.stdout.write("EOF output"); + fixture.child.stderr.write("EOF diagnostic"); + fixture.child.complete(); + callback(); + }, + }); + const fixture = ownedInput(input); + const cleanup = fixture.cleanup.getMockImplementation()!; + fixture.cleanup.mockImplementation(async () => { + stages.push("cleanup"); + return cleanup(); + }); + const snapshot = await runOwnedCommand( + spawn, + { + timeoutMs: null, + diagnosticBytes: 1024, + stdin: "closed-pipe", + observeCompleted: () => stages.push("observed"), + }, + { + launcher: async (request) => { + expect(request.stdin).toBe("pipe"); + expect(input.writableEnded).toBe(false); + return { + process: fixture.child, + ownership: fixture.ownership, + cleanup: fixture.cleanup, + }; + }, + }, + ); + expect(snapshot).toMatchObject({ + exitCode: 0, + stdout: { text: "EOF output" }, + stderr: { text: "EOF diagnostic" }, + }); + expect(stages).toEqual(["input-final", "observed", "cleanup"]); + expect(fixture.child.kill).not.toHaveBeenCalled(); +}); + +it("waits for delayed input finalization after output closes", async () => { + const finalizing = createDeferred(); + let finishInput: () => void = () => {}; + const input = new Writable({ + final(callback) { + finishInput = callback; + fixture.child.complete(); + finalizing.resolve(); + }, + }); + const fixture = ownedInput(input); + const observeCompleted = vi.fn(); + const pending = runOwnedCommand( + spawn, + { + timeoutMs: null, + diagnosticBytes: 1024, + stdin: "closed-pipe", + observeCompleted, + }, + { + launcher: async () => ({ + process: fixture.child, + ownership: fixture.ownership, + cleanup: fixture.cleanup, + }), + }, + ); + await finalizing.promise; + await new Promise((resolve) => setImmediate(resolve)); + expect(observeCompleted).not.toHaveBeenCalled(); + expect(fixture.cleanup).not.toHaveBeenCalled(); + finishInput(); + await expect(pending).resolves.toMatchObject({ exitCode: 0 }); + expect(observeCompleted).toHaveBeenCalledOnce(); + expect(fixture.cleanup).toHaveBeenCalledOnce(); +}); + +it.each([2_000, null])( + "cancels delayed input finalization with the opaque caller reason (timeout %s)", + async (timeoutMs) => { + const controller = new AbortController(); + const reason = new OwnedCommandFailure("timeout", "opaque caller reason"); + const finalizing = createDeferred(); + const input = new Writable({ + final(_callback) { + fixture.child.complete(); + finalizing.resolve(); + }, + }); + const fixture = ownedInput(input); + const observeCompleted = vi.fn(); + const pending = runOwnedCommand( + spawn, + { + timeoutMs, + diagnosticBytes: 1024, + stdin: "closed-pipe", + observeCompleted, + }, + { + signal: controller.signal, + launcher: async () => ({ + process: fixture.child, + ownership: fixture.ownership, + cleanup: fixture.cleanup, + }), + }, + ); + const rejected = expect(pending).rejects.toMatchObject({ + reason: "cancelled", + cancellation: { reason }, + cleanupFailure: null, + }); + await finalizing.promise; + controller.abort(reason); + await rejected; + expect(observeCompleted).not.toHaveBeenCalled(); + expect(fixture.cleanup).toHaveBeenCalledOnce(); + expect(input.destroyed).toBe(true); + expect(fixture.child.kill).not.toHaveBeenCalled(); + }, +); + +it.each([undefined, null, new Error("synchronous end failure")])( + "preserves a synchronous input end rejection as a physical collection failure: %j", + async (reason) => { + const input = new Writable(); + vi.spyOn(input, "end").mockImplementation(() => { + throw reason; + }); + const fixture = ownedInput(input); + const failure: unknown = await runOwnedCommand( + spawn, + { timeoutMs: null, diagnosticBytes: 1024, stdin: "closed-pipe" }, + { + launcher: async () => ({ + process: fixture.child, + ownership: fixture.ownership, + cleanup: fixture.cleanup, + }), + }, + ).catch((cause: unknown) => cause); + if (!(failure instanceof OwnedCommandFailure)) throw failure; + expect(failure.reason).toBe("process"); + expect(failure.collectionFailure).toBe(true); + expect(failure.cause).toBeInstanceOf(OwnedCommandFailure); + expect(failure.cause).toHaveProperty("cause", reason); + expect(fixture.cleanup).toHaveBeenCalledOnce(); + }, +); + +it("keeps an asynchronous input error primary before same-turn cancellation", async () => { + const controller = new AbortController(); + const inputError = new Error("authored input failure"); + const reason = { request: "abort after physical input error" }; + const input = new Writable({ + final(_callback) { + setImmediate(() => { + input.emit("error", inputError); + controller.abort(reason); + }); + }, + }); + const fixture = ownedInput(input); + const failure: unknown = await runOwnedCommand( + spawn, + { timeoutMs: null, diagnosticBytes: 1024, stdin: "closed-pipe" }, + { + signal: controller.signal, + launcher: async () => ({ + process: fixture.child, + ownership: fixture.ownership, + cleanup: fixture.cleanup, + }), + }, + ).catch((cause: unknown) => cause); + if (!(failure instanceof OwnedCommandFailure)) throw failure; + expect(failure).toMatchObject({ + reason: "process", + collectionFailure: true, + processError: inputError, + }); + expect(failure.cancellation?.reason).toBe(reason); + expect(failure.cause).toHaveProperty("cause", inputError); + expect(fixture.cleanup).toHaveBeenCalledOnce(); +}); + +it("collects an asynchronous Writable final callback error before process cleanup", async () => { + const inputError = new Error("asynchronous final callback failure"); + const input = new Writable({ + final(callback) { + setImmediate(() => callback(inputError)); + }, + }); + const fixture = ownedInput(input); + const failure: unknown = await runOwnedCommand( + spawn, + { timeoutMs: null, diagnosticBytes: 1024, stdin: "closed-pipe" }, + { + launcher: async () => ({ + process: fixture.child, + ownership: fixture.ownership, + cleanup: fixture.cleanup, + }), + }, + ).catch((cause: unknown) => cause); + if (!(failure instanceof OwnedCommandFailure)) throw failure; + expect(failure.reason).toBe("process"); + expect(failure.collectionFailure).toBe(true); + expect(failure.processError).toBe(inputError); + expect(failure.cause).toHaveProperty("cause", inputError); + expect(fixture.cleanup).toHaveBeenCalledOnce(); +}); + +it("reports a premature input close as a collection failure", async () => { + const input = new Writable({ + final(_callback) { + input.destroy(); + }, + }); + const fixture = ownedInput(input); + await expect( + runOwnedCommand( + spawn, + { timeoutMs: null, diagnosticBytes: 1024, stdin: "closed-pipe" }, + { + launcher: async () => ({ + process: fixture.child, + ownership: fixture.ownership, + cleanup: fixture.cleanup, + }), + }, + ), + ).rejects.toMatchObject({ + reason: "process", + collectionFailure: true, + message: + "stdin stream failed: closed before writable finalization completed", + processError: { + message: "closed before writable finalization completed", + }, + }); + expect(fixture.cleanup).toHaveBeenCalledOnce(); +}); + +it("keeps late input errors observed through refused cleanup and the exact retry owner", async () => { + const controller = new AbortController(); + const reason = { request: "abort during final cleanup" }; + const inputError = new Error("late input failure"); + const input = new Writable({ + autoDestroy: false, + final(callback) { + callback(); + fixture.child.complete(); + }, + }); + const fixture = ownedInput(input); + let denied = true; + fixture.cleanup.mockImplementation(async () => { + if (denied) { + input.emit("error", inputError); + controller.abort(reason); + return { cleaned: false, reason: "authored input owner still held" }; + } + input.emit("error", new Error("input error during retry")); + input.destroy(); + return { cleaned: true, signaled: false }; + }); + let failure: OwnedCommandFailure | undefined; + try { + const cause: unknown = await runOwnedCommand( + spawn, + { timeoutMs: null, diagnosticBytes: 1024, stdin: "closed-pipe" }, + { + signal: controller.signal, + launcher: async () => ({ + process: fixture.child, + ownership: fixture.ownership, + cleanup: fixture.cleanup, + }), + }, + ).catch((cause: unknown) => cause); + if (!(cause instanceof OwnedCommandFailure)) throw cause; + failure = cause; + expect(failure).toMatchObject({ + reason: "process", + collectionFailure: true, + processError: inputError, + cleanupFailure: "authored input owner still held", + }); + expect(failure.cancellation?.reason).toBe(reason); + expect(failure.cleanupOwner?.launch.ownership).toBe(fixture.ownership); + expect(failure.cleanupOwner?.launch.process.stdin).toBe(input); + expect(failure.cause).toHaveProperty("cause", inputError); + denied = false; + await expect(failure.cleanupOwner?.stop()).resolves.toEqual({ + status: "verified-cleanup", + }); + expect(fixture.cleanup).toHaveBeenCalledTimes(2); + expect(input.destroyed).toBe(true); + expect(fixture.child.kill).not.toHaveBeenCalled(); + } finally { + denied = false; + await failure?.cleanupOwner?.stop(); + input.destroy(); + } +}); + +it("keeps an explicitly selected input pipe open until process cleanup", async () => { + const stages: string[] = []; + const input = new Writable(); + const ended = vi.spyOn(input, "end"); + const fixture = ownedInput(input); + const cleanup = fixture.cleanup.getMockImplementation()!; + fixture.cleanup.mockImplementation(async () => { + stages.push("cleanup"); + expect(input.writableEnded).toBe(false); + expect(input.writableFinished).toBe(false); + expect(input.destroyed).toBe(false); + return cleanup(); + }); + const snapshot = await runOwnedCommand( + spawn, + { + timeoutMs: 2_000, + diagnosticBytes: 1024, + stdin: "pipe", + observeCompleted: () => stages.push("observed"), + }, + { + launcher: async (request) => { + expect(request.stdin).toBe("pipe"); + setImmediate(() => { + stages.push("output"); + fixture.child.stdout.write("open input output"); + fixture.child.stderr.write("open input diagnostic"); + fixture.child.complete(); + }); + return { + process: fixture.child, + ownership: fixture.ownership, + cleanup: fixture.cleanup, + }; + }, + }, + ); + expect(snapshot).toMatchObject({ + exitCode: 0, + stdout: { text: "open input output" }, + stderr: { text: "open input diagnostic" }, + }); + expect(stages).toEqual(["output", "observed", "cleanup"]); + expect(ended).not.toHaveBeenCalled(); + expect(fixture.cleanup).toHaveBeenCalledOnce(); + expect(input.destroyed).toBe(true); + expect(fixture.child.kill).not.toHaveBeenCalled(); +}); + +it.each([undefined, "pipe"] as const)( + "leaves stdin untouched without the closed-pipe contract (spawn stdin %s)", + async (stdin) => { + const input = new Writable(); + const ended = vi.spyOn(input, "end"); + const fixture = ownedInput(input); + await runOwnedCommand( + { ...spawn, ...(stdin === undefined ? {} : { stdin }) }, + { timeoutMs: null, diagnosticBytes: 1024 }, + { + launcher: async (request) => { + expect(request.stdin).toBe(stdin); + setImmediate(() => fixture.child.complete()); + return { + process: fixture.child, + ownership: fixture.ownership, + cleanup: fixture.cleanup, + }; + }, + }, + ); + expect(ended).not.toHaveBeenCalled(); + }, +); diff --git a/src/process/OwnedCommand.ts b/src/process/OwnedCommand.ts index bb6a7219ea..ccdbf6fedb 100644 --- a/src/process/OwnedCommand.ts +++ b/src/process/OwnedCommand.ts @@ -3,6 +3,7 @@ import { systemProcessOwnershipHost, } from "./ProcessOwnershipObservation.js"; import { DarwinProcessOwnershipInspectionError } from "./DarwinProcessRunTokenReader.js"; +import { finished } from "node:stream"; import type { Readable, Writable } from "node:stream"; import { pipeline } from "node:stream/promises"; @@ -85,6 +86,8 @@ export const ownedCommandResources = ( interface OwnedCommandLimits { /** Retained output budget; stdout is excluded when a sink consumes it. */ readonly diagnosticBytes: ProcessDiagnosticBudget; + /** Acquire a pipe; closed-pipe sends EOF after supervision starts. */ + readonly stdin?: "pipe" | "closed-pipe"; /** Accept an ordinary nonzero exit after output closes without other process errors. */ readonly acceptNonZeroExit?: boolean; /** Interpret complete physical output synchronously before awaited final cleanup. */ @@ -226,6 +229,7 @@ const collectOwnedCommand = async ( try { launched = await launcher({ ...spawn, + ...(limits.stdin === undefined ? {} : { stdin: "pipe" }), signal, }); } catch (cause: unknown) { @@ -235,9 +239,11 @@ const collectOwnedCommand = async ( let outputLimitObservedBeforeInterruption = false; let processFailure: string | undefined; let processError: Error | undefined; + let processFailureCause: { readonly cause: unknown } | undefined; let processFailureBeforeInterruption: | { readonly message: string; + readonly cause?: unknown; readonly processError: Error | undefined; readonly collectionFailure: boolean; } @@ -291,10 +297,11 @@ const collectOwnedCommand = async ( ); // A command that fails or is interrupted destroys its sink with its stdout. const abandoned = new AbortController(); + const collectionSignal = AbortSignal.any([signal, abandoned.signal]); const consumed = consumeStdout( launched.process.stdout, limits.stdout, - AbortSignal.any([signal, abandoned.signal]), + collectionSignal, ); void consumed.then((failed) => { processFailure ??= failed; @@ -305,30 +312,63 @@ const collectOwnedCommand = async ( collectionFailure: true, }; }); + let inputFailure: + | { + readonly cause: unknown; + readonly message: string; + readonly processError: Error | undefined; + } + | undefined; + const inputCompleted = + limits.stdin === "closed-pipe" + ? finishCommandInput(launched.process.stdin, (cause) => { + inputFailure ??= { + cause, + message: `stdin stream failed: ${cause instanceof Error ? cause.message : String(cause)}`, + processError: cause instanceof Error ? cause : undefined, + }; + if (processFailure === undefined) { + processFailure = inputFailure.message; + processError = inputFailure.processError; + processFailureCause = { cause: inputFailure.cause }; + } + if (deadline.interruption === undefined) + processFailureBeforeInterruption ??= { + message: processFailure, + ...processFailureCause, + processError, + collectionFailure: true, + }; + abandoned.abort(); + }) + : Promise.resolve(); let failure: OwnedCommandFailure | undefined; - const throwInterruptionOrObservedFailure = ( + const interruptionOrObservedFailure = ( interruption: "cancelled" | "timeout" | undefined, snapshot: ProviderProcessSnapshot | null = null, - ): void => { + ): OwnedCommandFailure | undefined => { if (interruption === undefined) return; if (outputLimitObservedBeforeInterruption) - throw new OwnedCommandFailure( + return new OwnedCommandFailure( "output-limit", "Command diagnostic output exceeded its complete-output budget.", snapshot, ); if (processFailureBeforeInterruption !== undefined) - throw new OwnedCommandFailure( + return new OwnedCommandFailure( "process", processFailureBeforeInterruption.message, snapshot, null, { + ...("cause" in processFailureBeforeInterruption + ? { cause: processFailureBeforeInterruption.cause } + : {}), processError: processFailureBeforeInterruption.processError, collectionFailure: processFailureBeforeInterruption.collectionFailure, }, ); - throw new OwnedCommandFailure( + return new OwnedCommandFailure( interruption, interruption === "cancelled" ? "Command cancelled." @@ -342,7 +382,8 @@ const collectOwnedCommand = async ( }; try { while (!(await supervisor.waitForOutputClose(10))) { - throwInterruptionOrObservedFailure(deadline.interruption); + const interrupted = interruptionOrObservedFailure(deadline.interruption); + if (interrupted !== undefined) throw interrupted; if (exceeded) throw new OwnedCommandFailure( "output-limit", @@ -350,16 +391,22 @@ const collectOwnedCommand = async ( ); if (processFailure !== undefined) throw new OwnedCommandFailure("process", processFailure, null, null, { + ...processFailureCause, processError, collectionFailure: true, }); await waitForAbortableDelay(25, signal); } // Output has closed, but the sink may still be parsing what it buffered. - const sinkFailure = await waitForAbortable(consumed, signal); + const sinkFailure = await waitForAbortable(consumed, collectionSignal); if (sinkFailure !== ABORTED) processFailure ??= sinkFailure; + await waitForAbortable(inputCompleted, collectionSignal); const snapshot = supervisor.snapshot(); - throwInterruptionOrObservedFailure(deadline.interruption, snapshot); + const interrupted = interruptionOrObservedFailure( + deadline.interruption, + snapshot, + ); + if (interrupted !== undefined) throw interrupted; if (exceeded) throw new OwnedCommandFailure( "output-limit", @@ -383,7 +430,11 @@ const collectOwnedCommand = async ( "Command failed; inspect its exit status and diagnostics.", snapshot, null, - { processError, collectionFailure: processFailure !== undefined }, + { + ...processFailureCause, + processError, + collectionFailure: processFailure !== undefined, + }, ); limits.observeCompleted?.(snapshot); } catch (cause: unknown) { @@ -402,6 +453,15 @@ const collectOwnedCommand = async ( abandoned.abort(); const stopped = await supervisor.stop(); const snapshot = supervisor.snapshot(); + // An input error can arrive after finish while process cleanup is still pending. + if (failure === undefined && inputFailure !== undefined) + failure = + interruptionOrObservedFailure(deadline.interruption, snapshot) ?? + new OwnedCommandFailure("process", inputFailure.message, snapshot, null, { + cause: inputFailure.cause, + processError: inputFailure.processError, + collectionFailure: true, + }); const cancellation = failure?.cancellation ?? (callerSignal?.aborted ? { reason: callerSignal.reason } : undefined); @@ -458,6 +518,36 @@ const collectOwnedCommand = async ( return snapshot; }; +/** Observe completion before sending EOF; listeners stay with input through cleanup retries. */ +const finishCommandInput = ( + input: Writable | null | undefined, + failed: (cause: unknown) => void, +): Promise => { + if (input === null || input === undefined) { + failed(new Error("stdin stream is unavailable")); + return Promise.resolve(); + } + return new Promise((resolve) => { + const failure = (cause: unknown): void => { + failed(cause); + resolve(); + }; + input.on("error", failure); + finished(input, { readable: false }, (cause) => { + if (cause !== undefined && cause !== null) failure(cause); + // finished() also accepts ended, empty writes when close interrupts _final. + else if (!input.writableFinished) + failure(new Error("closed before writable finalization completed")); + else resolve(); + }); + try { + input.end(); + } catch (cause: unknown) { + failure(cause); + } + }); +}; + /** * Pipe stdout into its sink until `signal` aborts, which destroys both; settles * with a failure message, never rejects. diff --git a/tests/boundary/process/nativeVerifierCommandOwnership.test.ts b/tests/boundary/process/nativeVerifierCommandOwnership.test.ts new file mode 100644 index 0000000000..a11bce0f0a --- /dev/null +++ b/tests/boundary/process/nativeVerifierCommandOwnership.test.ts @@ -0,0 +1,235 @@ +import { randomUUID } from "node:crypto"; +import { readFile, writeFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { pathToFileURL } from "node:url"; +import { expect, it } from "vitest"; +import { z } from "zod"; + +import { cleanupOwnedProcessGroup } from "../../../src/process/ProcessOwnership.js"; +import { systemProcessOwnershipHost } from "../../../src/process/ProcessOwnershipObservation.js"; +import { liveProcesses } from "../../../src/process/ProcessOwnershipProcessTree.js"; +import { + ProviderProcessSupervisor, + spawnOwnedProviderProcess, +} from "../../../src/process/ProviderProcess.js"; +import { + createTestWorkspace, + removeTestWorkspace, +} from "../../support/workspace/workspaceFixture.js"; + +const moduleUrl = (path: string) => pathToFileURL(resolve(path)).href; + +const descendantSource = ` +import { appendFileSync, existsSync } from "node:fs"; +const finish = (reason) => { + appendFileSync(process.env.NATIVE_EVENTS, JSON.stringify({ kind: "descendant-exit", reason, labExists: existsSync(process.env.NATIVE_LAB) }) + "\\n"); + process.exit(0); +}; +process.on("SIGTERM", () => finish("SIGTERM")); +process.send({ pid: process.pid, runId: process.env.REA_PROCESS_RUN_ID }); +setInterval(() => { + if (existsSync(process.env.NATIVE_RELEASE)) finish("fixture-release"); +}, 25); +setTimeout(() => finish("fixture-deadline"), 20000); +`; + +const leaderSource = ` +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { appendFileSync, existsSync, fstatSync, writeFileSync } from "node:fs"; +const metadata = fstatSync(0); +const stdin = { fifo: metadata.isFIFO(), socket: metadata.isSocket(), character: metadata.isCharacterDevice(), file: metadata.isFile() }; +assert.ok(stdin.fifo || stdin.socket, "Native client stdin must retain an actual pipe or socket"); +assert.equal(stdin.character, false); +assert.equal(stdin.file, false); +let inputBytes = 0; +for await (const chunk of process.stdin) inputBytes += chunk.length; +assert.equal(inputBytes, 0); +process.once("SIGTERM", () => { + appendFileSync(process.env.NATIVE_EVENTS, JSON.stringify({ kind: "leader-exit", labExists: existsSync(process.env.NATIVE_LAB) }) + "\\n"); + process.kill(process.pid, "SIGTERM"); +}); +const child = spawn(process.execPath, [process.env.NATIVE_DESCENDANT], { + stdio: ["ignore", "ignore", "ignore", "ipc"], +}); +const acknowledged = await new Promise((resolve, reject) => { + child.once("message", resolve); + child.once("error", reject); +}); +writeFileSync(process.env.NATIVE_READY, JSON.stringify({ + pids: [process.pid, acknowledged.pid], stdin, stdinEof: true, inputBytes, + runId: process.env.REA_PROCESS_RUN_ID, descendantRunId: acknowledged.runId, +})); +process.stdout.write("native verifier stdin EOF\\n"); +process.stderr.write("native verifier diagnostics\\n"); +setInterval(() => {}, 1000); +setTimeout(() => process.exit(1), 20000); +`; + +const driverSource = ` +import { existsSync } from "node:fs"; +import { readFile } from "node:fs/promises"; +import { createVerifierCommands } from ${JSON.stringify(moduleUrl("scripts/lib/verify-commands.mjs"))}; +import { systemProcessOwnershipHost } from ${JSON.stringify(moduleUrl("dist/process/ProcessOwnershipObservation.js"))}; +import { liveProcesses } from ${JSON.stringify(moduleUrl("dist/process/ProcessOwnershipProcessTree.js"))}; +let failure; +const commands = await createVerifierCommands(); +try { + await commands.exec(process.execPath, [process.env.NATIVE_LEADER], { + cwd: process.env.NATIVE_LAB, env: process.env, timeout: 1500, maxBuffer: 1024 * 1024, + stdin: "closed-pipe", + }); +} catch (cause) { + failure = { reason: cause.reason, signal: cause.snapshot?.signal, stdout: cause.snapshot?.stdout.text, stderr: cause.snapshot?.stderr.text }; +} finally { + await commands.close(); +} +const ready = JSON.parse(await readFile(process.env.NATIVE_READY, "utf8")); +const liveAtCompletion = liveProcesses(await systemProcessOwnershipHost.listProcesses()).filter(({ pid }) => ready.pids.includes(pid)).map(({ pid }) => pid); +const artifactAtCompletion = (await readFile(process.env.NATIVE_ARTIFACT)).toString("base64"); +process.stdout.write(JSON.stringify({ ready, failure, liveAtCompletion, labExists: existsSync(process.env.NATIVE_LAB), artifactAtCompletion })); +`; + +const reportSchema = z.object({ + ready: z.object({ + pids: z.array(z.number().int().positive()).length(2), + stdin: z.object({ + fifo: z.boolean(), + socket: z.boolean(), + character: z.literal(false), + file: z.literal(false), + }), + stdinEof: z.literal(true), + inputBytes: z.literal(0), + runId: z.string().min(1), + descendantRunId: z.string().min(1), + }), + failure: z.object({ + reason: z.literal("timeout"), + // Canonical cleanup releases verified owned groups with SIGKILL. + signal: z.literal("SIGKILL"), + stdout: z.string(), + stderr: z.string(), + }), + liveAtCompletion: z.array(z.number().int().positive()), + labExists: z.boolean(), + artifactAtCompletion: z.string(), +}); + +const awaitAuthoredExit = async (readyPath: string) => { + const ready = await readFile(readyPath, "utf8").then( + (text) => reportSchema.shape.ready.parse(JSON.parse(text)), + (cause: unknown) => { + if (cause instanceof Error && "code" in cause && cause.code === "ENOENT") + return undefined; + throw cause; + }, + ); + if (ready === undefined) return; + const deadline = Date.now() + 7_500; + while ( + liveProcesses(await systemProcessOwnershipHost.listProcesses()).some( + ({ pid }) => ready.pids.includes(pid), + ) + ) { + if (Date.now() >= deadline) + throw new Error( + "Authored native verifier processes did not finish cleanup", + ); + await delay(25); + } +}; + +// The native client is authored; actual stdin, timeout, descendants and census +// exercise the owned verifier command boundary without onboarding or network access. +it.skipIf(process.platform === "win32")( + "joins a timed-out native verifier descendant after pipe EOF while retaining lab artifacts", + async () => { + const workspace = await createTestWorkspace("rea-native-verifier-owner-"); + const release = workspace.path("release-descendant"); + const readyPath = workspace.path("ready.json"); + const artifact = Buffer.from("retained native verifier lab α\0\n"); + let acquired: + | Awaited> + | undefined; + let supervisor: ProviderProcessSupervisor | undefined; + try { + const lab = await workspace.mkdir("retained lab"); + const artifactPath = await workspace.write( + "retained lab/evidence.bin", + artifact, + ); + const leader = await workspace.write("leader.mjs", leaderSource); + const descendant = await workspace.write( + "descendant.mjs", + descendantSource, + ); + const driver = await workspace.write("driver.mjs", driverSource); + acquired = await spawnOwnedProviderProcess({ + command: process.execPath, + arguments: [driver], + cwd: process.cwd(), + runId: `rea-native-verifier-test-${randomUUID()}`, + sweepTokenOwnedProcesses: true, + hostEnvironment: { + ...process.env, + ...workspace.environment, + NATIVE_LAB: lab, + NATIVE_ARTIFACT: artifactPath, + NATIVE_LEADER: leader, + NATIVE_DESCENDANT: descendant, + NATIVE_READY: readyPath, + NATIVE_RELEASE: release, + NATIVE_EVENTS: workspace.path("events.jsonl"), + }, + }); + const owned = acquired; + supervisor = new ProviderProcessSupervisor( + { + ...owned, + ownsProcessLifetime: true, + cleanup: + owned.cleanup ?? (() => cleanupOwnedProcessGroup(owned.ownership)), + }, + { maxDiagnosticBytes: 64 * 1024 }, + ); + expect(await supervisor.waitForOutputClose(15_000)).toBe(true); + const completed = supervisor.snapshot(); + expect(completed.exitCode, completed.stderr.text).toBe(0); + expect(completed.signal, completed.stderr.text).toBeNull(); + const parsed = reportSchema.safeParse(JSON.parse(completed.stdout.text)); + if (!parsed.success) + throw new Error( + `Native verifier report failed validation: ${completed.stdout.text}`, + { cause: parsed.error }, + ); + const report = parsed.data; + expect(report.ready.stdin.fifo || report.ready.stdin.socket).toBe(true); + expect(report.ready.descendantRunId).toBe(report.ready.runId); + expect(report.failure.stdout).toBe("native verifier stdin EOF\n"); + expect(report.failure.stderr).toBe("native verifier diagnostics\n"); + expect(report.labExists).toBe(true); + expect(report.artifactAtCompletion).toBe(artifact.toString("base64")); + expect(await readFile(artifactPath)).toEqual(artifact); + expect(report.liveAtCompletion).toEqual([]); + } finally { + await writeFile( + release, + "release this authored native verifier descendant", + ); + if (supervisor !== undefined) { + const stopped = await supervisor.stop(); + if (stopped.status === "incomplete") throw new Error(stopped.reason); + } else if (acquired !== undefined) { + const stopped = await (acquired.cleanup === undefined + ? cleanupOwnedProcessGroup(acquired.ownership) + : acquired.cleanup()); + if (!stopped.cleaned) throw new Error(stopped.reason); + } + await awaitAuthoredExit(readyPath); + await removeTestWorkspace(workspace.root); + } + }, + 30_000, +); diff --git a/tests/boundary/process/packageVerifierCommandSemantics.test.ts b/tests/boundary/process/packageVerifierCommandSemantics.test.ts index 2fda609923..2b58f6a829 100644 --- a/tests/boundary/process/packageVerifierCommandSemantics.test.ts +++ b/tests/boundary/process/packageVerifierCommandSemantics.test.ts @@ -132,10 +132,12 @@ it("returns exact stdout for clean exit one and rejects exit two", async () => { expect(second).toEqual({ ok: false, cause: expect.objectContaining({ - name: "PackageCommandExitError", + name: "VerifierCommandExitError", code: 2, - stdout: "exit two stdout", - stderr: "exit two diagnostics", + snapshot: expect.objectContaining({ + stdout: expect.objectContaining({ text: "exit two stdout" }), + stderr: expect.objectContaining({ text: "exit two diagnostics" }), + }), }), }); }); @@ -382,7 +384,7 @@ it("retains the same adopted capability and workspace after failed close, then r expect(outcome).toEqual({ ok: false, cause: expect.objectContaining({ - message: "Package command owner is closed", + message: "Verifier command owner is closed", }), }); }); diff --git a/tests/boundary/process/verifierCommandSemantics.test.ts b/tests/boundary/process/verifierCommandSemantics.test.ts new file mode 100644 index 0000000000..683eb805e4 --- /dev/null +++ b/tests/boundary/process/verifierCommandSemantics.test.ts @@ -0,0 +1,606 @@ +import { existsSync } from "node:fs"; +import { open, readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +import { expect, it } from "vitest"; +import { z } from "zod"; + +import { createVerifierCommands } from "../../../scripts/lib/verify-commands.mjs"; +import { cleanupOwnedProcessGroup } from "../../../src/process/ProcessOwnership.js"; +import { systemProcessOwnershipHost } from "../../../src/process/ProcessOwnershipObservation.js"; +import { liveProcesses } from "../../../src/process/ProcessOwnershipProcessTree.js"; +import { spawnOwnedProviderProcess } from "../../../src/process/ProviderProcess.js"; +import { waitForProviderProcessReady } from "../../fixtures/providerProcess.js"; +import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; + +const observed = (promise: Promise) => + promise.then( + (value) => ({ ok: true as const, value }), + (cause: unknown) => ({ ok: false as const, cause }), + ); + +const deferred = () => { + let resolve!: () => void; + const promise = new Promise((finish) => { + resolve = finish; + }); + return { promise, resolve }; +}; + +it("rejects unsupported stdin before acquiring a launcher", async () => { + const directory = await createTestTempDirectory("rea-verifier-stdin-"); + const forbidden = join(directory, "forbidden launch"); + let acquisitions = 0; + const commands = await createVerifierCommands({ + launcher(input) { + acquisitions += 1; + return spawnOwnedProviderProcess(input); + }, + }); + let outcome; + try { + outcome = await observed( + commands.exec( + process.execPath, + [ + "-e", + `require("node:fs").writeFileSync(${JSON.stringify(forbidden)}, "unexpected launch");`, + ], + { + cwd: directory, + // @ts-expect-error An untyped caller supplied an unsupported stdin mode. + stdin: "ignore", + }, + ), + ); + } finally { + await commands.close(); + } + + expect(outcome).toEqual({ ok: false, cause: expect.any(TypeError) }); + expect(acquisitions).toBe(0); + expect(commands.cleanupResources).toEqual([]); + expect(existsSync(forbidden)).toBe(false); + expect(existsSync(directory)).toBe(true); +}); + +it.skipIf(process.platform === "win32")( + "delivers EOF through a physical stdin pipe while preserving caller inputs and artifact bytes", + async () => { + const directory = await createTestTempDirectory("rea-verifier-command-"); + const artifact = join(directory, "retained evidence.bin"); + const bytes = Buffer.from([0, 255, 10, 13, 65]); + const arguments_ = [ + artifact, + "value with spaces", + "--literal=$HOME", + "quote'and\"double", + ]; + const sentinel = "REA_VERIFIER_AMBIENT_SENTINEL"; + const previous = process.env[sentinel]; + const source = ` +const fs = require("node:fs"); +const input = fs.fstatSync(0); +let received = 0; +process.stdin.on("data", data => { received += data.length; }); +process.stdin.on("end", () => { + fs.writeFileSync(process.argv[1], Buffer.from([0, 255, 10, 13, 65])); + process.stderr.write("closed input diagnostics\\n"); + process.stdout.write(JSON.stringify({ + args: process.argv.slice(1), cwd: process.cwd(), received, eof: true, + pipe: input.isFIFO() || input.isSocket(), character: input.isCharacterDevice(), + regular: input.isFile(), selected: process.env.REA_VERIFIER_SELECTED, + ambient: process.env.${sentinel} ?? null, omitted: process.env.REA_VERIFIER_OMITTED ?? null + })); +}); +process.stdin.resume(); +`; + const commands = await createVerifierCommands(); + let output; + process.env[sentinel] = "ambient-only"; + try { + output = await commands.exec( + process.execPath, + ["-e", source, "--", ...arguments_], + { + cwd: directory, + env: { + REA_VERIFIER_SELECTED: "caller-selected", + REA_VERIFIER_OMITTED: undefined, + }, + stdin: "closed-pipe", + timeout: 2_000, + }, + ); + } finally { + if (previous === undefined) delete process.env[sentinel]; + else process.env[sentinel] = previous; + await commands.close(); + } + + expect(JSON.parse(output!.stdout)).toEqual({ + args: arguments_, + cwd: directory, + received: 0, + eof: true, + pipe: true, + character: false, + regular: false, + selected: "caller-selected", + ambient: null, + omitted: null, + }); + expect(output?.stderr).toBe("closed input diagnostics\n"); + expect(commands.cleanupResources).toEqual([]); + expect(existsSync(directory)).toBe(true); + expect(await readFile(artifact)).toEqual(bytes); + }, +); + +it("reports a numeric nonzero exit with its complete output snapshot after cleanup", async () => { + const directory = await createTestTempDirectory("rea-verifier-exit-"); + const artifact = join(directory, "existing evidence.bin"); + const bytes = Buffer.from([255, 0, 128, 10]); + await writeFile(artifact, bytes); + const commands = await createVerifierCommands(); + const stdout = "stdout α\n "; + const stderr = "stderr β\n"; + const arguments_ = [ + "-e", + `process.stdout.write(${JSON.stringify(stdout)}); process.stderr.write(${JSON.stringify(stderr)}); process.exitCode = 7;`, + ]; + let outcome; + try { + outcome = await observed( + commands.exec(process.execPath, arguments_, { cwd: directory }), + ); + } finally { + await commands.close(); + } + + expect(outcome).toEqual({ + ok: false, + cause: expect.objectContaining({ + name: "VerifierCommandExitError", + command: process.execPath, + arguments_: arguments_, + code: 7, + snapshot: expect.objectContaining({ + exitCode: 7, + signal: null, + stdout: expect.objectContaining({ + text: stdout, + bytes: Buffer.byteLength(stdout), + observedBytes: Buffer.byteLength(stdout), + }), + stderr: expect.objectContaining({ + text: stderr, + bytes: Buffer.byteLength(stderr), + observedBytes: Buffer.byteLength(stderr), + }), + }), + }), + }); + expect(commands.cleanupResources).toEqual([]); + expect(existsSync(directory)).toBe(true); + expect(await readFile(artifact)).toEqual(bytes); +}); + +it("retries only the failed exact file capability after a peer releases, retaining both evidence files", async () => { + const directory = await createTestTempDirectory("rea-verifier-retained-"); + const artifact = join(directory, "retained evidence.bin"); + const peerArtifact = join(directory, "released peer evidence.bin"); + const bytes = Buffer.from([0, 255, 65]); + await writeFile(artifact, bytes); + await writeFile(peerArtifact, bytes); + const handle = await open(artifact, "r"); + const peerHandle = await open(peerArtifact, "r"); + const commands = await createVerifierCommands(); + const denied = new Error("authored file capability release denial"); + const releases: object[] = []; + let peerReleases = 0; + let deny = true; + let released = false; + const capability = { + get cleanupResources() { + return released ? [] : [artifact]; + }, + async close() { + releases.push(this); + if (deny) throw denied; + await handle.close(); + released = true; + }, + }; + commands.adopt(capability, artifact); + commands.adopt(capability, artifact); + commands.adopt( + { + async close() { + peerReleases += 1; + await peerHandle.close(); + }, + }, + peerArtifact, + ); + let first; + let retained; + try { + first = await observed(commands.close()); + retained = { + resources: [...commands.cleanupResources], + size: (await handle.stat()).size, + peerDescriptor: peerHandle.fd, + peerReleases, + directory: existsSync(directory), + }; + } finally { + deny = false; + await commands.close(); + } + const failure = first?.ok === false ? first.cause : undefined; + const cause = z + .object({ cause: z.object({ errors: z.array(z.unknown()) }) }) + .parse(failure).cause; + + expect(cause.errors).toHaveLength(1); + expect(cause.errors[0]).toBe(denied); + expect(retained).toEqual({ + resources: [artifact], + size: bytes.length, + peerDescriptor: -1, + peerReleases: 1, + directory: true, + }); + expect(releases).toHaveLength(2); + for (const owner of releases) expect(owner).toBe(capability); + expect(handle.fd).toBe(-1); + expect(peerReleases).toBe(1); + expect(peerHandle.fd).toBe(-1); + expect(commands.cleanupResources).toEqual([]); + expect(existsSync(directory)).toBe(true); + expect(await readFile(artifact)).toEqual(bytes); + expect(await readFile(peerArtifact)).toEqual(bytes); +}); + +it("refuses new owners and commands after successful close while leaving new resources with their caller", async () => { + const directory = await createTestTempDirectory("rea-verifier-admission-"); + const artifact = join(directory, "caller evidence.bin"); + const forbidden = join(directory, "forbidden launch"); + await writeFile(artifact, "caller-owned evidence"); + const commands = await createVerifierCommands(); + await commands.close(); + const handle = await open(artifact, "r"); + let releases = 0; + const capability = { + async close() { + releases += 1; + await handle.close(); + }, + }; + let rejection: unknown; + let retained; + let execution; + try { + try { + commands.adopt(capability, artifact); + } catch (cause: unknown) { + rejection = cause; + } + retained = { size: (await handle.stat()).size, releases }; + execution = await observed( + commands.exec( + process.execPath, + [ + "-e", + `require("node:fs").writeFileSync(${JSON.stringify(forbidden)}, "unexpected launch");`, + ], + { cwd: directory }, + ), + ); + } finally { + await capability.close(); + await commands.close(); + } + + expect(rejection).toBeInstanceOf(Error); + expect(retained).toEqual({ + size: Buffer.byteLength("caller-owned evidence"), + releases: 0, + }); + expect(execution).toEqual({ + ok: false, + cause: expect.objectContaining({ + message: "Verifier command owner is closed", + }), + }); + expect(releases).toBe(1); + expect(handle.fd).toBe(-1); + expect(commands.cleanupResources).toEqual([]); + expect(existsSync(forbidden)).toBe(false); + expect(existsSync(directory)).toBe(true); + expect(await readFile(artifact, "utf8")).toBe("caller-owned evidence"); +}); + +it.skipIf(process.platform === "win32")( + "keeps a selected physical stdin pipe open until the command deadline", + async () => { + const directory = await createTestTempDirectory("rea-verifier-open-stdin-"); + const source = ` +import { fstatSync } from "node:fs"; +const input = fstatSync(0); +process.stdout.write(JSON.stringify({ + pipe: input.isFIFO() || input.isSocket(), + character: input.isCharacterDevice(), + regular: input.isFile() +}) + "\\n"); +for await (const _chunk of process.stdin) {} +process.stdout.write("unexpected stdin EOF\\n"); +`; + const expectedOutput = + JSON.stringify({ pipe: true, character: false, regular: false }) + "\n"; + const commands = await createVerifierCommands(); + let outcome; + try { + outcome = await observed( + commands.exec(process.execPath, ["--input-type=module", "-e", source], { + cwd: directory, + stdin: "pipe", + timeout: 1_500, + }), + ); + } finally { + await commands.close(); + } + + expect(outcome).toEqual({ + ok: false, + cause: expect.objectContaining({ + reason: "timeout", + cleanupFailure: null, + snapshot: expect.objectContaining({ + stdout: expect.objectContaining({ + text: expectedOutput, + bytes: Buffer.byteLength(expectedOutput), + observedBytes: Buffer.byteLength(expectedOutput), + }), + stderr: expect.objectContaining({ text: "" }), + }), + }), + }); + expect(commands.cleanupResources).toEqual([]); + expect(existsSync(directory)).toBe(true); + }, + 10_000, +); + +it.skipIf(process.platform === "win32")( + "retains a late acquired child's exact cleanup after cancelled close fails and retries it", + async () => { + const directory = await createTestTempDirectory( + "rea-verifier-late-owned-launch-", + ); + const artifact = join(directory, "retained evidence.bin"); + const bytes = Buffer.from([0, 255, 65]); + await writeFile(artifact, bytes); + const entered = deferred(); + const cancelled = deferred(); + const release = deferred(); + const controller = new AbortController(); + const reason = { request: "cancel before owned launch delivery" }; + const denial = "authored exact process cleanup denial"; + let deny = true; + let acquired: + | Awaited> + | undefined; + const cleanupCalls: Array< + Awaited> + > = []; + const commands = await createVerifierCommands({ + launcher: async (input) => { + const owned = await spawnOwnedProviderProcess({ + ...input, + stdin: "pipe", + }); + acquired = owned; + await waitForProviderProcessReady(owned.process); + input.signal?.addEventListener("abort", cancelled.resolve, { + once: true, + }); + if (input.signal?.aborted) cancelled.resolve(); + entered.resolve(); + await release.promise; + return { + ...owned, + cleanup: async () => { + cleanupCalls.push(owned); + if (deny) return { cleaned: false, reason: denial }; + return owned.cleanup === undefined + ? cleanupOwnedProcessGroup(owned.ownership) + : owned.cleanup(); + }, + }; + }, + }); + const running = observed( + commands.exec( + process.execPath, + [ + "-e", + 'process.stdout.write("ready\\n"); process.stderr.write("late child diagnostic\\n"); process.stdin.resume();', + ], + { cwd: directory, signal: controller.signal }, + ), + ); + try { + await Promise.race([ + entered.promise, + running.then(() => { + throw new Error("Command settled before owned launch delivery"); + }), + ]); + const owned = acquired; + if (owned === undefined) + throw new Error("Expected the exact acquired process owner"); + controller.abort(reason); + const closing = commands.close(); + expect(commands.close()).toBe(closing); + let closeSettled = false; + const closeOutcome = observed(closing).then((outcome) => { + closeSettled = true; + return outcome; + }); + await cancelled.promise; + expect(closeSettled).toBe(false); + release.resolve(); + const [execution, close] = await Promise.all([running, closeOutcome]); + expect(execution.ok).toBe(false); + if (execution.ok) throw new Error("Expected cancelled execution"); + const failure = z + .object({ + reason: z.literal("cancelled"), + cancellation: z.object({ reason: z.unknown() }), + cleanupFailure: z.literal(denial), + cleanupOwner: z.object({ + launch: z.object({ process: z.unknown(), ownership: z.unknown() }), + }), + resources: z.array(z.string()), + snapshot: z.object({ + stderr: z.object({ text: z.literal("late child diagnostic\n") }), + }), + }) + .parse(execution.cause); + expect(failure.cancellation.reason).toBe(reason); + expect(failure.cleanupOwner.launch.process).toBe(owned.process); + expect(failure.cleanupOwner.launch.ownership).toBe(owned.ownership); + if (close.ok) throw new Error("Expected retained command cleanup"); + const retryFailure = z + .object({ + cause: z.object({ + errors: z.array(z.object({ cause: z.unknown() })), + }), + }) + .parse(close.cause); + expect(retryFailure.cause.errors).toHaveLength(1); + expect(retryFailure.cause.errors[0]?.cause).toBe(execution.cause); + expect(commands.cleanupResources).toEqual(failure.resources); + expect(cleanupCalls).toHaveLength(2); + expect( + liveProcesses(await systemProcessOwnershipHost.listProcesses()).some( + ({ pid }) => pid === owned.process.pid, + ), + ).toBe(true); + + deny = false; + await commands.close(); + expect(cleanupCalls).toHaveLength(3); + for (const owner of cleanupCalls) expect(owner).toBe(owned); + expect(commands.cleanupResources).toEqual([]); + expect( + liveProcesses(await systemProcessOwnershipHost.listProcesses()).some( + ({ pid }) => pid === owned.process.pid, + ), + ).toBe(false); + expect(existsSync(directory)).toBe(true); + expect(await readFile(artifact)).toEqual(bytes); + } finally { + deny = false; + release.resolve(); + try { + await commands.close(); + } finally { + await running; + if ( + acquired !== undefined && + acquired.process.exitCode === null && + acquired.process.signalCode === null + ) { + const stopped = await (acquired.cleanup === undefined + ? cleanupOwnedProcessGroup(acquired.ownership) + : acquired.cleanup()); + if (!stopped.cleaned) throw new Error(stopped.reason); + } + } + } + }, + 15_000, +); + +it.skipIf(process.platform === "win32")( + "joins a delayed real launcher on close without spawning or removing caller evidence", + async () => { + const directory = await createTestTempDirectory( + "rea-verifier-late-launch-", + ); + const artifact = join(directory, "retained evidence"); + const forbidden = join(directory, "forbidden launch"); + await writeFile(artifact, "retained before launch"); + const entered = deferred(); + const cancelled = deferred(); + const release = deferred(); + const commands = await createVerifierCommands({ + launcher: async (input) => { + input.signal?.addEventListener("abort", cancelled.resolve, { + once: true, + }); + if (input.signal?.aborted) cancelled.resolve(); + entered.resolve(); + await release.promise; + return spawnOwnedProviderProcess(input); + }, + }); + const running = observed( + commands.exec( + process.execPath, + [ + "-e", + `require("node:fs").writeFileSync(${JSON.stringify(forbidden)}, "unexpected launch");`, + ], + { cwd: directory, timeout: 2_000 }, + ), + ); + let outcome; + let sameClose = false; + let closeSettled = false; + let settledBeforeLaunchRelease = false; + try { + await Promise.race([ + entered.promise, + running.then(() => { + throw new Error("Command settled before the authored launch gate"); + }), + ]); + const first = commands.close(); + const second = commands.close(); + sameClose = first === second; + void first.then( + () => { + closeSettled = true; + }, + () => { + closeSettled = true; + }, + ); + await cancelled.promise; + settledBeforeLaunchRelease = closeSettled; + release.resolve(); + await Promise.all([first, second]); + outcome = await running; + } finally { + release.resolve(); + await commands.close(); + await running; + } + + expect(sameClose).toBe(true); + expect(settledBeforeLaunchRelease).toBe(false); + expect(outcome).toEqual({ + ok: false, + cause: expect.objectContaining({ reason: "cancelled", snapshot: null }), + }); + expect(commands.cleanupResources).toEqual([]); + expect(existsSync(forbidden)).toBe(false); + expect(existsSync(directory)).toBe(true); + expect(await readFile(artifact, "utf8")).toBe("retained before launch"); + }, +); diff --git a/tests/boundary/process/verifierTerminalOwner.test.ts b/tests/boundary/process/verifierTerminalOwner.test.ts index c031c902da..e7cf557850 100644 --- a/tests/boundary/process/verifierTerminalOwner.test.ts +++ b/tests/boundary/process/verifierTerminalOwner.test.ts @@ -10,10 +10,8 @@ import { type VerifierTerminalOptions, type VerifierTerminalResult, } from "../../../scripts/lib/owned-verifier-terminal.mjs"; -import { - createPackageVerifierCommands, - type PackageCleanupOwner, -} from "../../../scripts/lib/verify-package-commands.mjs"; +import { createPackageVerifierCommands } from "../../../scripts/lib/verify-package-commands.mjs"; +import type { VerifierCleanupOwner } from "../../../scripts/lib/verifier-command-owner.mjs"; import { cleanupOwnedProcessGroup, type PersistentOwnedProcessGroup, @@ -78,7 +76,7 @@ const terminalFixture = async () => { }, }, }); - const adopted: PackageCleanupOwner[] = []; + const adopted: VerifierCleanupOwner[] = []; const options: Omit = { commands: { adopt(owner, resource) { @@ -121,11 +119,11 @@ const expectReleased = ( expect(fixture.rootsAtRemoval).toEqual([true]); }; -const observeClose = (owner: PackageCleanupOwner) => { +const observeClose = (owner: VerifierCleanupOwner) => { const entered = deferred(); const original = owner.close; const spy = vi.spyOn(owner, "close").mockImplementation(function ( - this: PackageCleanupOwner, + this: VerifierCleanupOwner, ) { const closing = original.call(this); entered.resolve();