diff --git a/docs/pi-workflow/README.md b/docs/pi-workflow/README.md index 265ee80..c782a3d 100644 --- a/docs/pi-workflow/README.md +++ b/docs/pi-workflow/README.md @@ -30,7 +30,7 @@ The managed flow is: 4. Choose a fresh plan-seeded session, compacted current session, or unchanged current context. 5. Implement with guarded file ownership. 6. Remove prototype-only behavior. -7. Run plan-approved verification commands and focused tests. +7. Run plan-approved verification commands and focused tests. Each host command is terminated after three minutes so an interactive or stalled command cannot block the run indefinitely. 8. Complete independent specification and standards review. The assistant must inspect the current run before each state-changing call and obtain the user's explicit Gate Decision. It must not infer approval. diff --git a/docs/pi-workflow/design.md b/docs/pi-workflow/design.md index 0b5bfbe..82cbe47 100644 --- a/docs/pi-workflow/design.md +++ b/docs/pi-workflow/design.md @@ -403,6 +403,7 @@ Mutating Assignments run in the Workflow Run's worktree. - Mutating child sessions do not receive `bash` or other process-execution tools. Assignment-scoped file operations reject symlink traversal and require canonical containment beneath both the worktree root and the Assignment's owned path roots. - Plan-approved verification and generation commands run separately through a host executor using fixed executable/argument arrays, never agent-authored shell strings. The only accepted wrapper form is `env PI_OFFLINE=1 ...`; the executor unwraps it, sets that single variable, and applies the normal executable safety checks. - The host executor sets `PYTHONDONTWRITEBYTECODE=1` so Python-backed checks do not create out-of-scope `__pycache__` artifacts. +- Every host command has a three-minute execution timeout and reports exit code `124` when terminated, preventing interactive or stalled commands from blocking a run indefinitely. - The engine records branch, HEAD, worktree identity, and Git diff state before and after every mutating Assignment and host command. - Out-of-scope changes, branch movement, commits, or worktree metadata changes invalidate the Assignment and open a Recovery Gate before more mutation starts. - The scheduler reserves capacity synchronously before launching children. diff --git a/extensions/pi-workflow/workspace.test.ts b/extensions/pi-workflow/workspace.test.ts index 4878407..86f66a3 100644 --- a/extensions/pi-workflow/workspace.test.ts +++ b/extensions/pi-workflow/workspace.test.ts @@ -16,6 +16,7 @@ import { promisify } from "node:util"; import { deflateSync } from "node:zlib"; import { ApprovedHostCommandExecutor, + ExecFileCommandRunner, FilesystemPrototypeValidator, GitStateProbe, GitWorkspaceManager, @@ -283,6 +284,27 @@ test("Git probe and approved host commands enforce identity and ownership", asyn 'process.exit(process.env.PI_OFFLINE === "1" ? 0 : 1)', "utf8", ); + await writeFile( + path.join(directory, "spawn-hanging-child.cjs"), + [ + 'const { spawn } = require("node:child_process");', + 'const { writeFileSync } = require("node:fs");', + 'const child = spawn(process.execPath, ["delayed-write.cjs"], { stdio: "ignore" });', + 'writeFileSync("src/timeout-pids.json", JSON.stringify([process.pid, child.pid]));', + "setInterval(() => {}, 1_000);", + ].join("\n"), + "utf8", + ); + await writeFile( + path.join(directory, "delayed-write.cjs"), + [ + 'const { writeFileSync } = require("node:fs");', + 'process.on("SIGTERM", () => {});', + 'setTimeout(() => writeFileSync("src/delayed.txt", "late\\n"), 1_200);', + "setInterval(() => {}, 1_000);", + ].join("\n"), + "utf8", + ); await execFileAsync("git", ["-C", directory, "add", "."]); await execFileAsync("git", ["-C", directory, "commit", "-qm", "initial"]); const probe = new GitStateProbe(); @@ -315,6 +337,35 @@ test("Git probe and approved host commands enforce identity and ownership", asyn args: ["PI_OFFLINE=1", process.execPath, "check-offline.cjs"], }); assert.equal(offline.result.code, 0); + const boundedExecutor = new ApprovedHostCommandExecutor( + new ExecFileCommandRunner(), + undefined, + 250, + ); + const startedAt = Date.now(); + const timedOut = await boundedExecutor.run({ + worktreeRoot: directory, + ownedPaths: ["src"], + executable: process.execPath, + args: ["spawn-hanging-child.cjs"], + }); + assert.equal(timedOut.result.code, 124); + assert.match(timedOut.result.stderr, /timed out after 250ms/); + assert.ok(Date.now() - startedAt < 2_500); + const timedOutProcessIds = JSON.parse( + await readFile(path.join(directory, "src", "timeout-pids.json"), "utf8"), + ) as number[]; + for (const processId of timedOutProcessIds) { + assert.throws( + () => process.kill(processId, 0), + (error: unknown) => (error as NodeJS.ErrnoException).code === "ESRCH", + ); + } + await new Promise((resolve) => setTimeout(resolve, 1_000)); + await assert.rejects( + access(path.join(directory, "src", "delayed.txt")), + /ENOENT/, + ); await assert.rejects( executor.run({ worktreeRoot: directory, diff --git a/extensions/pi-workflow/workspace.ts b/extensions/pi-workflow/workspace.ts index 5ce6eff..2bcd187 100644 --- a/extensions/pi-workflow/workspace.ts +++ b/extensions/pi-workflow/workspace.ts @@ -1,4 +1,4 @@ -import { execFile } from "node:child_process"; +import { execFile, spawn } from "node:child_process"; import { createHash } from "node:crypto"; import { constants } from "node:fs"; import { @@ -25,6 +25,7 @@ export interface CommandResult { export interface CommandRunOptions { env?: NodeJS.ProcessEnv; + timeoutMs?: number; } export interface CommandRunner { @@ -504,16 +505,21 @@ const PACKAGE_EXEC_SUBCOMMANDS = new Map>([ ["bun", new Set(["x"])], ]); +const DEFAULT_HOST_COMMAND_TIMEOUT_MS = 3 * 60 * 1_000; + export class ApprovedHostCommandExecutor { private readonly runner: CommandRunner; private readonly probe: GitStateProbe; + private readonly timeoutMs: number; constructor( runner: CommandRunner = new ExecFileCommandRunner(), probe: GitStateProbe = new GitStateProbe(runner), + timeoutMs = DEFAULT_HOST_COMMAND_TIMEOUT_MS, ) { this.runner = runner; this.probe = probe; + this.timeoutMs = timeoutMs; } async run(options: { @@ -570,6 +576,7 @@ export class ApprovedHostCommandExecutor { const before = beforeRepository.get(root) ?? (await this.probe.capture(root)); const result = await this.runner.run(root, executable, args, { env: commandEnv, + timeoutMs: this.timeoutMs, }); try { const afterRepository = await this.probe.captureRepository(root); @@ -813,6 +820,50 @@ export class WorkspaceValidationError extends Error { } } +const PROCESS_TERMINATION_GRACE_MS = 100; +const PROCESS_TERMINATION_POLL_MS = 10; +const PROCESS_TERMINATION_TIMEOUT_MS = 1_000; + +function wait(milliseconds: number) { + return new Promise((resolve) => setTimeout(resolve, milliseconds)); +} + +function processGroupExists(processId: number) { + try { + process.kill(-processId, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code !== "ESRCH"; + } +} + +function signalProcessGroup(processId: number, signal: NodeJS.Signals) { + try { + process.kill(-processId, signal); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; + } +} + +async function terminateProcessTree(processId: number) { + if (process.platform === "win32") { + await execFileAsync("taskkill", ["/pid", String(processId), "/T", "/F"]); + return; + } + signalProcessGroup(processId, "SIGTERM"); + await wait(PROCESS_TERMINATION_GRACE_MS); + signalProcessGroup(processId, "SIGKILL"); + const deadline = Date.now() + PROCESS_TERMINATION_TIMEOUT_MS; + while (processGroupExists(processId)) { + if (Date.now() >= deadline) { + throw new Error(`process group ${processId} remained alive after SIGKILL`); + } + await wait(PROCESS_TERMINATION_POLL_MS); + } +} + +const COMMAND_OUTPUT_LIMIT_BYTES = 4 * 1024 * 1024; + export class ExecFileCommandRunner implements CommandRunner { async run( cwd: string, @@ -820,26 +871,83 @@ export class ExecFileCommandRunner implements CommandRunner { args: string[], options: CommandRunOptions = {}, ) { - try { - const result = await execFileAsync(executable, args, { + return new Promise((resolve) => { + const stdoutChunks: Buffer[] = []; + const stderrChunks: Buffer[] = []; + let stdoutBytes = 0; + let stderrBytes = 0; + let spawnError: Error | undefined; + let timedOut = false; + let outputExceeded = false; + let termination: Promise | undefined; + let timeout: NodeJS.Timeout | undefined; + const child = spawn(executable, args, { cwd, - encoding: "utf8", - maxBuffer: 4 * 1024 * 1024, + detached: process.platform !== "win32", env: options.env, + stdio: ["ignore", "pipe", "pipe"], }); - return { code: 0, stdout: result.stdout, stderr: result.stderr }; - } catch (error) { - const failed = error as Error & { - code?: number | string; - stdout?: string; - stderr?: string; + const terminate = () => { + if (!termination && child.pid !== undefined) { + termination = terminateProcessTree(child.pid); + } }; - return { - code: typeof failed.code === "number" ? failed.code : 1, - stdout: failed.stdout ?? "", - stderr: failed.stderr ?? failed.message, + const capture = (chunks: Buffer[], chunk: Buffer, stream: "stdout" | "stderr") => { + const capturedBytes = stream === "stdout" ? stdoutBytes : stderrBytes; + const remaining = COMMAND_OUTPUT_LIMIT_BYTES - capturedBytes; + if (remaining > 0) chunks.push(chunk.subarray(0, remaining)); + if (stream === "stdout") stdoutBytes += chunk.length; + else stderrBytes += chunk.length; + if (chunk.length > remaining && !outputExceeded) { + outputExceeded = true; + terminate(); + } }; - } + child.stdout.on("data", (chunk: Buffer) => capture(stdoutChunks, chunk, "stdout")); + child.stderr.on("data", (chunk: Buffer) => capture(stderrChunks, chunk, "stderr")); + child.on("error", (error) => { + spawnError = error; + }); + child.on("close", async (code) => { + if (timeout !== undefined) clearTimeout(timeout); + let terminationError: string | undefined; + try { + await termination; + } catch (failure) { + terminationError = failure instanceof Error ? failure.message : String(failure); + } + const stdout = Buffer.concat(stdoutChunks).toString("utf8"); + const stderr = Buffer.concat(stderrChunks).toString("utf8"); + if (timedOut) { + const detail = [stderr, terminationError].filter(Boolean).join("\n"); + resolve({ + code: terminationError ? 125 : 124, + stdout, + stderr: `Host command timed out after ${options.timeoutMs}ms${detail ? `\n${detail}` : ""}`, + }); + return; + } + if (outputExceeded) { + resolve({ + code: terminationError ? 125 : 1, + stdout, + stderr: `Host command output exceeded ${COMMAND_OUTPUT_LIMIT_BYTES} bytes${terminationError ? `\n${terminationError}` : ""}`, + }); + return; + } + resolve({ + code: code ?? 1, + stdout, + stderr: stderr || spawnError?.message || "", + }); + }); + if (options.timeoutMs !== undefined) { + timeout = setTimeout(() => { + timedOut = true; + terminate(); + }, options.timeoutMs); + } + }); } }