diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index c52b4bc7dfc2..fe6e92d7d411 100644 --- a/config/assertion-safety-baseline.txt +++ b/config/assertion-safety-baseline.txt @@ -3454,7 +3454,7 @@ src/model-picker/apply-session-model-selection.ts 1 src/node-host/config.ts 1 src/node-host/desktop-stream-command.ts 2 src/node-host/invoke-agent-cli-claude-params.ts 4 -src/node-host/invoke-agent-cli-claude.ts 2 +src/node-host/invoke-agent-cli-claude.ts 1 src/node-host/invoke-file-commands.ts 1 src/node-host/invoke-payload.ts 5 src/node-host/invoke.ts 7 @@ -3706,7 +3706,7 @@ src/process/exec-spawn.ts 4 src/process/exec.ts 3 src/process/spawn-secret-input.ts 1 src/process/spawn-utils.ts 1 -src/process/supervisor/adapters/child.ts 3 +src/process/supervisor/adapters/child.ts 2 src/process/supervisor/adapters/pty.ts 1 src/process/terminal-pty.ts 1 src/process/windows-command.ts 1 diff --git a/extensions/telegram/src/bot.create-telegram-bot.media-group-skip-warning.test.ts b/extensions/telegram/src/bot.create-telegram-bot.media-group-skip-warning.test.ts index 91e9701c9144..2ade575e4961 100644 --- a/extensions/telegram/src/bot.create-telegram-bot.media-group-skip-warning.test.ts +++ b/extensions/telegram/src/bot.create-telegram-bot.media-group-skip-warning.test.ts @@ -1,5 +1,4 @@ // Telegram tests cover bot.create telegram bot.media group skip warning plugin behavior. -import { setTimeout as delay } from "node:timers/promises"; import { coerceErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { telegramBotInfoForTest } from "./bot.create-telegram-bot.test-support.js"; @@ -101,15 +100,19 @@ function resolveFlushTimer(setTimeoutSpy: ReturnType) { return flushTimer; } -async function waitForBufferedProcessing() { - await delay(75); -} - async function flushChannelPostMediaGroup(setTimeoutSpy: ReturnType) { + const replyDispatched = new Promise((resolve) => { + const previousReply = replySpy.getMockImplementation(); + replySpy.mockImplementationOnce(async (...args) => { + const result = await previousReply?.(...args); + resolve(); + return result; + }); + }); const flushTimer = resolveFlushTimer(setTimeoutSpy); expect(flushTimer).toBeTypeOf("function"); await flushTimer?.(); - await waitForBufferedProcessing(); + await replyDispatched; } function createChannelPostContext(params: { @@ -206,7 +209,7 @@ describe("createTelegramBot media-group skip warning (#55216)", () => { expect(sendMessageSpy).not.toHaveBeenCalled(); await flushChannelPostMediaGroup(setTimeoutSpy); - await vi.waitFor(() => expect(sendMessageSpy).toHaveBeenCalledTimes(1)); + expect(sendMessageSpy).toHaveBeenCalledTimes(1); expect(sendMessageSpy).toHaveBeenCalledWith( CHANNEL_ID, expect.stringContaining("1 of 2 images"), @@ -219,7 +222,7 @@ describe("createTelegramBot media-group skip warning (#55216)", () => { ); const warningText = String(sendMessageSpy.mock.calls[0]?.[1]); expect(warningText).toContain("1 could not be fetched and was skipped"); - await vi.waitFor(() => expect(replySpy).toHaveBeenCalled()); + expect(replySpy).toHaveBeenCalled(); expect(replySpy.mock.calls[0]?.[0]?.media).toEqual([ expect.objectContaining({ path: "/tmp/p1.jpg", contentType: "image/png" }), expect.objectContaining({ kind: "image" }), @@ -246,7 +249,7 @@ describe("createTelegramBot media-group skip warning (#55216)", () => { }); await flushChannelPostMediaGroup(setTimeoutSpy); - await vi.waitFor(() => expect(sendMessageSpy).toHaveBeenCalledTimes(1)); + expect(sendMessageSpy).toHaveBeenCalledTimes(1); const warningText = String(sendMessageSpy.mock.calls[0]?.[1]); expect(warningText).toContain("0 of 2 images"); expect(warningText).toContain("2 could not be fetched and were skipped"); @@ -281,7 +284,7 @@ describe("createTelegramBot media-group skip warning (#55216)", () => { }); await flushChannelPostMediaGroup(setTimeoutSpy); - await vi.waitFor(() => expect(sendMessageSpy).toHaveBeenCalledTimes(1)); + expect(sendMessageSpy).toHaveBeenCalledTimes(1); const warningText = String(sendMessageSpy.mock.calls[0]?.[1]); expect(warningText).toContain("1 of 3 images"); expect(warningText).toContain("2 could not be fetched and were skipped"); diff --git a/src/node-host/invoke-agent-cli-claude-pipe-error.test.ts b/src/node-host/invoke-agent-cli-claude-pipe-error.test.ts deleted file mode 100644 index f4167ac021ab..000000000000 --- a/src/node-host/invoke-agent-cli-claude-pipe-error.test.ts +++ /dev/null @@ -1,69 +0,0 @@ -import { afterEach, describe, expect, it, vi } from "vitest"; - -const spawnMock = vi.hoisted(() => vi.fn()); - -vi.mock("node:child_process", () => ({ - spawn: spawnMock, -})); - -import type { NodeHostClient } from "./client.js"; -import type { NodeInvokeRequestPayload } from "./invoke.js"; - -function frame(params: unknown): NodeInvokeRequestPayload { - return { - id: "invoke-pipe-error", - nodeId: "node-pipe-error", - command: "agent.cli.claude.run.v1", - paramsJSON: JSON.stringify(params), - }; -} - -function client(): NodeHostClient { - return { - async request>() { - return {} as T; - }, - }; -} - -describe("Claude CLI node command pipe errors", () => { - let realChild: import("node:child_process").ChildProcessWithoutNullStreams | undefined; - - afterEach(() => { - if (realChild && !realChild.killed) { - realChild.kill("SIGKILL"); - } - realChild = undefined; - spawnMock.mockReset(); - vi.resetModules(); - }); - - it("guards real child stdout/stderr pipe error events", async () => { - const childProcess = - await vi.importActual("node:child_process"); - realChild = childProcess.spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { - stdio: ["pipe", "pipe", "pipe"], - }); - spawnMock.mockReturnValueOnce(realChild as never); - const { runClaudeCliNodeCommand } = await import("./invoke-agent-cli-claude.js"); - - const request = { argv: ["-p"], idleTimeoutMs: 100, timeoutMs: 5_000 }; - - const run = runClaudeCliNodeCommand({ - client: client(), - frame: frame(request), - request, - argv: [process.execPath, ...request.argv], - cwd: undefined, - env: process.env as Record, - timeoutMs: request.timeoutMs, - }); - - expect(() => realChild?.stdout.emit("error", new Error("stdout pipe failure"))).not.toThrow(); - expect(() => realChild?.stderr.emit("error", new Error("stderr pipe failure"))).not.toThrow(); - realChild.kill("SIGKILL"); - - await expect(run).resolves.toMatchObject({ success: false }); - expect(spawnMock).toHaveBeenCalledOnce(); - }); -}); diff --git a/src/node-host/invoke-agent-cli-claude.test.ts b/src/node-host/invoke-agent-cli-claude.test.ts index 545e99c23ab3..91ac8d1b2d33 100644 --- a/src/node-host/invoke-agent-cli-claude.test.ts +++ b/src/node-host/invoke-agent-cli-claude.test.ts @@ -2,6 +2,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { withEnvAsync } from "../test-utils/env.js"; import type { NodeHostClient } from "./client.js"; import { decodeClaudeCliNodeRunParams } from "./invoke-agent-cli-claude-params.js"; import { runClaudeCliNodeCommand } from "./invoke-agent-cli-claude.js"; @@ -41,6 +42,23 @@ async function executableScript(source: string): Promise { return file; } +function runCommand( + executable: string, + request: Parameters[0]["request"], + overrides: Partial[0]> = {}, +) { + return runClaudeCliNodeCommand({ + client: client([]), + frame: frame(request), + request, + argv: [executable, ...request.argv], + cwd: undefined, + env: process.env as Record, + timeoutMs: request.timeoutMs, + ...overrides, + }); +} + describe("Claude CLI node command", () => { it.each([ { argv: ["--unknown"], error: "unsupported Claude CLI argument" }, @@ -369,15 +387,7 @@ process.stdin.on("end", () => { idleTimeoutMs: 1_000, timeoutMs: 5_000, }; - const result = await runClaudeCliNodeCommand({ - client: client(calls), - frame: frame(request), - request, - argv: [executable, ...request.argv], - cwd: undefined, - env: process.env as Record, - timeoutMs: request.timeoutMs, - }); + const result = await runCommand(executable, request, { client: client(calls) }); const progress = calls .filter((call) => call.method === "node.invoke.progress") @@ -393,6 +403,47 @@ process.stdin.on("end", () => { await expect(fs.stat(promptPath ?? "")).rejects.toThrow(); }); + it.runIf(process.platform !== "win32")( + "retains the prompt for an authoritative descendant without delaying the root result", + async () => { + const markerDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-node-claude-prompt-")); + tempDirs.push(markerDir); + const marker = path.join(markerDir, "descendant-read"); + const executable = await executableScript(` +const { spawn } = require("node:child_process"); +const prompt = process.argv[process.argv.indexOf("--append-system-prompt-file") + 1]; +const child = spawn(process.execPath, ["-e", + "setTimeout(() => require('node:fs').writeFileSync(" + + JSON.stringify(${JSON.stringify(marker)}) + ", require('node:fs').readFileSync(" + + JSON.stringify(prompt) + ", 'utf8')), 300)" +], { stdio: ["ignore", "ignore", "ignore", 3] }); +child.unref(); +process.stdout.write(JSON.stringify({ type: "result", result: prompt }) + "\\n");`); + await withEnvAsync({ OPENCLAW_SERVICE_MARKER: "openclaw" }, async () => { + const calls: Array<{ method: string; params: unknown }> = []; + const request = { + argv: ["-p"], + systemPrompt: "descendant-owned prompt", + idleTimeoutMs: 2_000, + timeoutMs: 5_000, + }; + const result = await runCommand(executable, request, { client: client(calls) }); + const output = calls + .filter((call) => call.method === "node.invoke.progress") + .map((call) => (call.params as { chunk: string }).chunk) + .join(""); + const promptPath = (JSON.parse(output) as { result: string }).result; + + expect(result).toMatchObject({ exitCode: 0, success: true }); + await expect(fs.readFile(promptPath, "utf8")).resolves.toBe("descendant-owned prompt"); + await vi.waitFor(async () => { + expect(await fs.readFile(marker, "utf8")).toBe("descendant-owned prompt"); + await expect(fs.stat(promptPath)).rejects.toThrow(); + }); + }); + }, + ); + it.each([ { descriptorEnv: "CLAUDE_CODE_OAUTH_TOKEN_FILE_DESCRIPTOR", @@ -417,12 +468,8 @@ process.stdout.write(JSON.stringify({ }) + "\\n");`); const request = { argv: ["-p"], idleTimeoutMs: 1_000, timeoutMs: 5_000 }; const calls: Array<{ method: string; params: unknown }> = []; - const result = await runClaudeCliNodeCommand({ + const result = await runCommand(executable, request, { client: client(calls), - frame: frame(request), - request, - argv: [executable, ...request.argv], - cwd: undefined, env: { ...process.env, [descriptorEnv]: "3", @@ -431,7 +478,6 @@ process.stdout.write(JSON.stringify({ fd: 3, createData: () => Buffer.from("selected-node-secret"), }, - timeoutMs: request.timeoutMs, }); const progress = calls @@ -457,19 +503,13 @@ function writeChunk() { } process.stdout.write("\\n" + JSON.stringify({ type: "result", session_id: "tail-session", result: "done" }) + "\\n", () => process.stderr.write("late failure diagnostic")); } -writeChunk();`, +process.stdout.write(Buffer.concat([ + Buffer.alloc(199_997, 120), Buffer.from([0xe2, 0x82]), Buffer.from("A\\n") +]), writeChunk);`, ); const calls: Array<{ method: string; params: unknown }> = []; const request = { argv: ["-p"], idleTimeoutMs: 1_000, timeoutMs: 5_000 }; - const result = await runClaudeCliNodeCommand({ - client: client(calls), - frame: frame(request), - request, - argv: [executable, ...request.argv], - cwd: undefined, - env: process.env as Record, - timeoutMs: request.timeoutMs, - }); + const result = await runCommand(executable, request, { client: client(calls) }); const progressBytes = calls .filter((call) => call.method === "node.invoke.progress") .reduce((sum, call) => sum + Buffer.byteLength((call.params as { chunk: string }).chunk), 0); @@ -478,6 +518,7 @@ writeChunk();`, .filter((call) => call.method === "node.invoke.progress") .map((call) => (call.params as { chunk: string }).chunk) .join(""); + expect(progress.startsWith(`${"x".repeat(199_997)}�A`)).toBe(true); // OUTPUT_CAP_BYTES + TERMINAL_EVENT_MAX_BYTES from invoke-agent-cli-claude.ts. expect(progressBytes).toBeLessThanOrEqual(200_000 + 1024 * 1024); expect(progress).toContain('"session_id":"tail-session"'); @@ -491,26 +532,88 @@ writeChunk();`, ).toBeLessThanOrEqual(2); }); + it.each([ + { + idleTimeoutMs: 40, + timeoutMs: 400, + noOutputTimedOut: true, + stderr: "Claude CLI produced no output before the idle timeout", + }, + { + idleTimeoutMs: 400, + timeoutMs: 40, + noOutputTimedOut: false, + stderr: "Claude CLI exceeded the hard timeout", + }, + ])("preserves the exact timeout result: $stderr", async (request) => { + const executable = await executableScript("setInterval(() => {}, 1000);"); + await expect(runCommand(executable, { argv: ["-p"], ...request })).resolves.toMatchObject({ + exitCode: 124, + timedOut: true, + noOutputTimedOut: request.noOutputTimedOut, + stderr: request.stderr, + }); + }); + + it.each([ + { name: "spawn", command: "/definitely/not/a/claude-command", error: "ENOENT" }, + { name: "secret input", command: process.execPath, error: "secret delivery failed" }, + { name: "progress", command: process.execPath, error: "progress delivery failed" }, + ])("surfaces $name failures in the invocation result", async ({ name, command, error }) => { + const request = { argv: ["-p"], idleTimeoutMs: 1_000, timeoutMs: 5_000 }; + await expect( + runCommand(command, request, { + argv: [command, "-e", 'process.stdout.write("progress"); setInterval(() => {}, 1000)'], + ...(name === "secret input" + ? { + secretInput: { + fd: 3, + createData: () => { + throw new Error(error); + }, + }, + } + : {}), + ...(name === "progress" + ? { + client: { + async request(): Promise { + throw new Error(error); + }, + } satisfies NodeHostClient, + } + : {}), + }), + ).resolves.toMatchObject({ + exitCode: 1, + success: false, + timedOut: false, + error: expect.stringContaining(error), + stderr: expect.stringContaining(error), + }); + }); + it("terminates an active Claude command when its invoke is cancelled", async () => { - const executable = await executableScript(`setInterval(() => {}, 1000);`); + const executable = await executableScript( + `process.stdout.write("ready"); setInterval(() => {}, 1000);`, + ); const controller = new AbortController(); const request = { argv: ["-p"], idleTimeoutMs: 5_000, timeoutMs: 10_000 }; - const run = runClaudeCliNodeCommand({ - client: client([]), - frame: frame(request), - request, - argv: [executable, ...request.argv], - cwd: undefined, - env: process.env as Record, - timeoutMs: request.timeoutMs, + const calls: Array<{ method: string; params: unknown }> = []; + const run = runCommand(executable, request, { + client: client(calls), signal: controller.signal, }); + await vi.waitFor(() => + expect(calls).toContainEqual(expect.objectContaining({ method: "node.invoke.progress" })), + ); controller.abort(); await expect(run).resolves.toMatchObject({ exitCode: 130, success: false, + timedOut: false, stderr: expect.stringContaining("cancelled"), }); }); @@ -527,16 +630,7 @@ writeChunk();`, const request = { argv: ["-p"], idleTimeoutMs: 5_000, timeoutMs: 10_000 }; await expect( - runClaudeCliNodeCommand({ - client: client([]), - frame: frame(request), - request, - argv: [executable, ...request.argv], - cwd: undefined, - env: process.env as Record, - timeoutMs: request.timeoutMs, - signal: controller.signal, - }), + runCommand(executable, request, { signal: controller.signal }), ).resolves.toMatchObject({ exitCode: 130, success: false }); await expect(fs.stat(marker)).rejects.toThrow(); }); diff --git a/src/node-host/invoke-agent-cli-claude.ts b/src/node-host/invoke-agent-cli-claude.ts index 337bdfbcb15c..1ef07d24ab7e 100644 --- a/src/node-host/invoke-agent-cli-claude.ts +++ b/src/node-host/invoke-agent-cli-claude.ts @@ -1,17 +1,12 @@ /** Validates and streams one approval-gated Claude CLI turn on a headless node. */ -import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { StringDecoder } from "node:string_decoder"; -import { signalProcessTree } from "../process/kill-tree.js"; -import { - addSecretInputStdio, - type SpawnStdioEntry, - writeSecretInputToChild, -} from "../process/spawn-secret-input.js"; -import type { SpawnSecretInput } from "../process/supervisor/types.js"; -import { resolveSafeChildProcessInvocation } from "../process/windows-command.js"; +import { logWarn } from "../logger.js"; +import { getProcessSupervisor } from "../process/supervisor/index.js"; +import type { RunExit, SpawnSecretInput } from "../process/supervisor/types.js"; import { truncateUtf8Suffix } from "../utils/utf8-truncate.js"; import type { NodeHostClient } from "./client.js"; import type { ClaudeCliNodeRunParams } from "./invoke-agent-cli-claude-params.js"; @@ -67,207 +62,160 @@ export async function runClaudeCliNodeCommand(params: { if (params.signal?.aborted) { return cancelledResult(); } - return await new Promise((resolve) => { - let settled = false; - let hardTimedOut = false; - let idleTimedOut = false; - let cancelled = false; - let truncated = false; - let outputBytes = 0; - let stderr = ""; - const decoder = new StringDecoder("utf8"); - const stderrDecoder = new StringDecoder("utf8"); - const terminalDecoder = new StringDecoder("utf8"); - let terminalLineBuffer = ""; - let terminalLineTouchesTruncation = false; - let terminalResultLine: string | undefined; - const invocation = resolveSafeChildProcessInvocation({ + const supervisor = getProcessSupervisor(); + const runId = randomUUID(); + let cancelled = false; + let truncated = false; + let outputBytes = 0; + let stderr = ""; + let terminalLineBuffer = ""; + let terminalLineTouchesTruncation = false; + let terminalResultLine: string | undefined; + const decoder = new StringDecoder("utf8"); + const stderrDecoder = new StringDecoder("utf8"); + const terminalDecoder = new StringDecoder("utf8"); + const progress = createNodeInvokeProgressWriter({ + client: params.client, + frame: params.frame, + idleTimeoutMs: params.request.idleTimeoutMs, + onError: () => supervisor.cancel(runId), + }); + const abortRun = () => { + cancelled = true; + supervisor.cancel(runId); + }; + const retain = (chunk: Buffer) => { + const remaining = Math.max(0, OUTPUT_CAP_BYTES - outputBytes); + const retained = chunk.subarray(0, remaining); + outputBytes += retained.length; + truncated ||= retained.length !== chunk.length; + return retained; + }; + const captureTerminalLines = (raw: Buffer, touchesTruncation: boolean) => { + terminalLineBuffer += terminalDecoder.write(raw); + terminalLineTouchesTruncation ||= touchesTruncation; + for (let newline = terminalLineBuffer.indexOf("\n"); newline >= 0;) { + const line = terminalLineBuffer.slice(0, newline).replace(/\r$/u, ""); + terminalLineBuffer = terminalLineBuffer.slice(newline + 1); + if ( + terminalLineTouchesTruncation && + Buffer.byteLength(line, "utf8") <= TERMINAL_EVENT_MAX_BYTES && + isClaudeResultLine(line) + ) { + terminalResultLine = line; + } + terminalLineTouchesTruncation = touchesTruncation; + newline = terminalLineBuffer.indexOf("\n"); + } + if (Buffer.byteLength(terminalLineBuffer, "utf8") > TERMINAL_EVENT_MAX_BYTES) { + terminalLineBuffer = ""; + terminalLineTouchesTruncation = false; + } + }; + + let exit: RunExit | undefined; + let runError: Error | undefined; + params.signal?.addEventListener("abort", abortRun, { once: true }); + try { + const runPromise = supervisor.spawn({ + runId, + sessionId: params.request.sessionKey ?? params.frame.id, + backendId: "node-host-claude", + mode: "child", argv, - cwd: params.cwd, - env: params.env ?? process.env, - }); - const stdio: SpawnStdioEntry[] = ["pipe", "pipe", "pipe"]; - addSecretInputStdio(stdio, params.secretInput); - const child = spawn(invocation.command, invocation.args, { cwd: params.cwd, env: params.env, - stdio, - ...(process.platform !== "win32" ? { detached: true } : {}), - windowsHide: invocation.windowsHide, - windowsVerbatimArguments: invocation.windowsVerbatimArguments, - }) as ChildProcessWithoutNullStreams; - - const kill = () => { - const pid = child.pid; - if (typeof pid === "number" && pid > 0) { - signalProcessTree(pid, "SIGKILL", { detached: process.platform !== "win32" }); - } - try { - child.kill("SIGKILL"); - } catch { - // Best effort; close/error settles the result. - } - }; - const progress = createNodeInvokeProgressWriter({ - client: params.client, - frame: params.frame, - idleTimeoutMs: params.request.idleTimeoutMs, - onError: kill, + exactEnv: true, + input: params.request.stdin ?? "", + secretInput: params.secretInput, + timeoutMs: params.timeoutMs ?? params.request.timeoutMs, + noOutputTimeoutMs: params.request.idleTimeoutMs, + captureOutput: false, + onStdoutRaw: (raw) => { + const retained = retain(raw); + captureTerminalLines(retained, false); + if (retained.length < raw.length) { + captureTerminalLines(raw.subarray(retained.length), true); + } + if (retained.length === 0) { + progress.queueHeartbeat(); + return; + } + void progress.write(decoder.write(retained)); + }, + onStderrRaw: (raw) => { + retain(raw); + stderr = truncateUtf8Suffix(`${stderr}${stderrDecoder.write(raw)}`, STDERR_TAIL_BYTES); + progress.queueHeartbeat(); + }, }); - const abortRun = () => { - cancelled = true; - kill(); - }; - params.signal?.addEventListener("abort", abortRun, { once: true }); if (params.signal?.aborted) { abortRun(); } - const hardTimer = setTimeout(() => { - hardTimedOut = true; - kill(); - }, params.timeoutMs ?? params.request.timeoutMs); - let idleTimer: ReturnType; - const resetIdleTimer = () => { - clearTimeout(idleTimer); - idleTimer = setTimeout(() => { - idleTimedOut = true; - kill(); - }, params.request.idleTimeoutMs); - }; - resetIdleTimer(); + const run = await runPromise; + if (promptDir && run.waitForExtinction) { + const ownedPromptDir = promptDir; + promptDir = undefined; + // Descendants may still own this file after their root result is already visible. + void run + .waitForExtinction() + .then(() => fs.rm(ownedPromptDir, { recursive: true, force: true })) + .catch((error: unknown) => { + logWarn(`Claude CLI system prompt cleanup failed: ${String(error)}`); + }); + } + exit = await run.wait(); + } catch (error) { + runError = error instanceof Error ? error : new Error(String(error)); + } finally { + params.signal?.removeEventListener("abort", abortRun); + progress.stopHeartbeats(); + } - const retain = (chunk: Buffer): Buffer => { - if (outputBytes >= OUTPUT_CAP_BYTES) { - truncated = true; - return Buffer.alloc(0); - } - const remaining = OUTPUT_CAP_BYTES - outputBytes; - const retained = chunk.length > remaining ? chunk.subarray(0, remaining) : chunk; - outputBytes += retained.length; - if (retained.length !== chunk.length) { - truncated = true; - } - return retained; - }; + void progress.write(decoder.end()); + terminalLineBuffer += terminalDecoder.end(); + stderr = truncateUtf8Suffix(`${stderr}${stderrDecoder.end()}`, STDERR_TAIL_BYTES); + if ( + terminalLineTouchesTruncation && + Buffer.byteLength(terminalLineBuffer, "utf8") <= TERMINAL_EVENT_MAX_BYTES && + isClaudeResultLine(terminalLineBuffer) + ) { + terminalResultLine = terminalLineBuffer; + } + if (truncated && terminalResultLine) { + void progress.write(`\n${terminalResultLine}\n`); + } + await progress.flush(); + progress.stop(); - const captureTerminalLines = (raw: Buffer, touchesTruncation: boolean) => { - terminalLineBuffer += terminalDecoder.write(raw); - terminalLineTouchesTruncation ||= touchesTruncation; - while (true) { - const newline = terminalLineBuffer.indexOf("\n"); - if (newline < 0) { - break; - } - const line = terminalLineBuffer.slice(0, newline).replace(/\r$/u, ""); - terminalLineBuffer = terminalLineBuffer.slice(newline + 1); - if ( - terminalLineTouchesTruncation && - Buffer.byteLength(line, "utf8") <= TERMINAL_EVENT_MAX_BYTES && - isClaudeResultLine(line) - ) { - terminalResultLine = line; - } - terminalLineTouchesTruncation = touchesTruncation; - } - if (Buffer.byteLength(terminalLineBuffer, "utf8") > TERMINAL_EVENT_MAX_BYTES) { - terminalLineBuffer = ""; - terminalLineTouchesTruncation = false; - } - }; - // Output pipes can fail independently; child close/error remains authoritative. - const ignoreOutputStreamError = () => {}; - child.stdout.on("error", ignoreOutputStreamError); - child.stderr.on("error", ignoreOutputStreamError); - child.stdout.on("data", (raw: Buffer) => { - const retained = retain(raw); - if (retained.length > 0) { - captureTerminalLines(retained, false); - } - if (retained.length < raw.length) { - captureTerminalLines(raw.subarray(retained.length), true); - } - // The Gateway's inactivity timer observes stdout progress events only; - // keep the node-local kill timer on the same signal to avoid orphan runs. - resetIdleTimer(); - if (retained.length === 0) { - progress.queueHeartbeat(); - return; - } - const text = decoder.write(retained); - void progress.write(text, child.stdout); - }); - child.stderr.on("data", (raw: Buffer) => { - retain(raw); - stderr = truncateUtf8Suffix(`${stderr}${stderrDecoder.write(raw)}`, STDERR_TAIL_BYTES); - resetIdleTimer(); - progress.queueHeartbeat(); - }); - child.stdin.on("error", () => {}); - child.stdin.end(params.request.stdin ?? ""); - - const finish = async (exitCode: number | null, error?: Error) => { - if (settled) { - return; - } - settled = true; - clearTimeout(hardTimer); - clearTimeout(idleTimer); - progress.stopHeartbeats(); - params.signal?.removeEventListener("abort", abortRun); - const finalText = decoder.end(); - if (finalText) { - void progress.write(finalText); - } - const terminalText = terminalDecoder.end(); - if (terminalText) { - terminalLineBuffer += terminalText; - } - const finalStderr = stderrDecoder.end(); - if (finalStderr) { - stderr = truncateUtf8Suffix(`${stderr}${finalStderr}`, STDERR_TAIL_BYTES); - } - if ( - terminalLineTouchesTruncation && - Buffer.byteLength(terminalLineBuffer, "utf8") <= TERMINAL_EVENT_MAX_BYTES && - isClaudeResultLine(terminalLineBuffer) - ) { - terminalResultLine = terminalLineBuffer; - } - if (truncated && terminalResultLine) { - void progress.write(`\n${terminalResultLine}\n`); - } - await progress.flush(); - progress.stop(); - const timeoutMessage = idleTimedOut - ? "Claude CLI produced no output before the idle timeout" - : hardTimedOut - ? "Claude CLI exceeded the hard timeout" - : ""; - const finalError = progress.error ?? error; - const cancelledMessage = cancelled ? "Claude CLI invocation cancelled" : ""; - resolve({ - exitCode: exitCode ?? (idleTimedOut || hardTimedOut ? 124 : cancelled ? 130 : 1), - timedOut: idleTimedOut || hardTimedOut, - noOutputTimedOut: idleTimedOut, - success: exitCode === 0 && !idleTimedOut && !hardTimedOut && !cancelled && !finalError, - stdout: "", - stderr: truncateUtf8Suffix( - [stderr, timeoutMessage, cancelledMessage, finalError?.message] - .filter(Boolean) - .join("\n"), - STDERR_TAIL_BYTES, - ), - error: finalError?.message ?? null, - truncated, - }); - }; - void writeSecretInputToChild(child, params.secretInput).catch((error: unknown) => { - kill(); - void finish(null, error instanceof Error ? error : new Error(String(error))); - }); - child.once("error", (error) => void finish(null, error)); - child.once("close", (code) => void finish(code)); - }); + const idleTimedOut = !cancelled && exit?.noOutputTimedOut === true; + const timedOut = !cancelled && exit?.timedOut === true; + const timeoutMessage = idleTimedOut + ? "Claude CLI produced no output before the idle timeout" + : timedOut + ? "Claude CLI exceeded the hard timeout" + : ""; + const finalError = progress.error ?? runError; + return { + exitCode: cancelled ? 130 : (exit?.exitCode ?? (timedOut ? 124 : 1)), + timedOut, + noOutputTimedOut: idleTimedOut, + success: exit?.exitCode === 0 && !timedOut && !cancelled && !finalError, + stdout: "", + stderr: truncateUtf8Suffix( + [ + stderr, + timeoutMessage, + cancelled ? "Claude CLI invocation cancelled" : "", + finalError?.message, + ] + .filter(Boolean) + .join("\n"), + STDERR_TAIL_BYTES, + ), + error: finalError?.message ?? null, + truncated, + }; } finally { if (promptDir) { await fs.rm(promptDir, { recursive: true, force: true }); diff --git a/src/process/decoded-output.ts b/src/process/decoded-output.ts index afea7338f988..aca06cfc4e99 100644 --- a/src/process/decoded-output.ts +++ b/src/process/decoded-output.ts @@ -1,7 +1,11 @@ import type { Readable } from "node:stream"; import { createWindowsOutputDecoder } from "../infra/windows-encoding.js"; -export function onDecodedOutput(stream: Readable, listener: (chunk: string) => void): void { +export function onDecodedOutput( + stream: Readable, + listener: (chunk: string) => void, + onRaw?: (chunk: Buffer) => void, +): void { const decoder = createWindowsOutputDecoder(); const emit = (text: string) => { if (text) { @@ -16,7 +20,10 @@ export function onDecodedOutput(stream: Readable, listener: (chunk: string) => v flushed = true; emit(decoder.flush()); }; - stream.on("data", (chunk: Buffer | string) => emit(decoder.decode(chunk))); + stream.on("data", (chunk: Buffer | string) => { + onRaw?.(typeof chunk === "string" ? Buffer.from(chunk) : chunk); + emit(decoder.decode(chunk)); + }); stream.once("end", flush); stream.once("close", flush); } diff --git a/src/process/supervisor/adapters/child-stdin.ts b/src/process/supervisor/adapters/child-stdin.ts new file mode 100644 index 000000000000..c791d4120767 --- /dev/null +++ b/src/process/supervisor/adapters/child-stdin.ts @@ -0,0 +1,65 @@ +import type { Writable } from "node:stream"; +import type { ManagedRunStdin } from "../types.js"; + +/** Keep direct children and service relays on the same observable stdin lifecycle. */ +export function createManagedChildStdin( + stream: Writable | null | undefined, +): ManagedRunStdin | undefined { + if (!stream) { + return undefined; + } + let ended = stream.writableEnded || stream.writableFinished; + let destroyed = stream.destroyed; + stream.once("finish", () => { + ended = true; + }); + stream.once("close", () => { + ended = true; + destroyed = true; + }); + stream.once("error", () => { + destroyed = true; + }); + return { + get destroyed() { + return destroyed || stream.destroyed; + }, + get writable() { + return !destroyed && !ended && stream.writable; + }, + get writableEnded() { + return ended || stream.writableEnded; + }, + get writableFinished() { + return stream.writableFinished; + }, + write(data, callback) { + if (destroyed || ended || !stream.writable) { + callback?.(new Error("stdin is not writable")); + return; + } + try { + stream.write(data, callback); + } catch (error) { + callback?.(error instanceof Error ? error : new Error(String(error))); + } + }, + end() { + ended = true; + try { + stream.end(); + } catch { + // Closing an already-failed child pipe is best effort. + } + }, + destroy() { + ended = true; + destroyed = true; + try { + stream.destroy(); + } catch { + // Destroying an already-failed child pipe is best effort. + } + }, + }; +} diff --git a/src/process/supervisor/adapters/child.service-lifecycle.test.ts b/src/process/supervisor/adapters/child.service-lifecycle.test.ts index 5c771865900e..4bbf5c04e23a 100644 --- a/src/process/supervisor/adapters/child.service-lifecycle.test.ts +++ b/src/process/supervisor/adapters/child.service-lifecycle.test.ts @@ -135,6 +135,7 @@ describe.skipIf(process.platform === "win32")("service-managed child lifecycle", expect(elapsed).toBeLessThan(300); expect(isAlive(descendantPid)).toBe(true); + await adapter.waitForExtinction?.(); await waitFor(() => !isAlive(descendantPid)); }); diff --git a/src/process/supervisor/adapters/child.test.ts b/src/process/supervisor/adapters/child.test.ts index 22c215d4b8cd..9e4ed0f3569b 100644 --- a/src/process/supervisor/adapters/child.test.ts +++ b/src/process/supervisor/adapters/child.test.ts @@ -958,14 +958,16 @@ describe("createChildAdapter", () => { }); const first = vi.fn(); const second = vi.fn(); + const raw = vi.fn(); - adapter.onStdout(first); + adapter.onStdout(first, raw); adapter.onStdout(second); child.stdout?.emit("data", Buffer.from([0xb2])); expect(createWindowsOutputDecoderMock).toHaveBeenCalledTimes(2); expect(first).toHaveBeenCalledWith("first"); expect(second).toHaveBeenCalledWith("second"); + expect(raw).toHaveBeenCalledWith(Buffer.from([0xb2])); }); it("guards stream errors before output listeners are registered", async () => { diff --git a/src/process/supervisor/adapters/child.ts b/src/process/supervisor/adapters/child.ts index e9bfe4b1bb3e..9b493cfed66e 100644 --- a/src/process/supervisor/adapters/child.ts +++ b/src/process/supervisor/adapters/child.ts @@ -21,7 +21,8 @@ import { resolveWindowsCommandShim, } from "../../windows-command.js"; import { createServiceChildRelayAdapter } from "../service-child-relay-host.js"; -import type { ManagedRunStdin, SpawnProcessAdapter, SpawnSecretInput } from "../types.js"; +import type { SpawnProcessAdapter, SpawnSecretInput } from "../types.js"; +import { createManagedChildStdin } from "./child-stdin.js"; import { toStringEnv } from "./env.js"; const FORCE_KILL_WAIT_FALLBACK_MS = 4000; @@ -193,77 +194,19 @@ export async function createChildAdapter(params: { child.stdout.on("error", ignoreOutputStreamError); child.stderr.on("error", ignoreOutputStreamError); const childStdin = spawned.child.stdin; - let stdinDestroyed = childStdin?.destroyed ?? false; - let stdinEnded = childStdin?.writableEnded === true || childStdin?.writableFinished === true; - if (childStdin) { - childStdin.once("finish", () => { - stdinEnded = true; - }); - childStdin.once("close", () => { - stdinEnded = true; - stdinDestroyed = true; - }); - childStdin.once("error", () => { - stdinDestroyed = true; - }); - if (params.input !== undefined) { - childStdin.write(params.input); - stdinEnded = true; - childStdin.end(); - } else if (stdinMode === "pipe-closed") { - stdinEnded = true; - childStdin.end(); - } + const stdin = createManagedChildStdin(childStdin); + if (params.input !== undefined) { + childStdin?.write(params.input); + stdin?.end(); + } else if (stdinMode === "pipe-closed") { + stdin?.end(); } - const stdin: ManagedRunStdin | undefined = childStdin - ? { - get destroyed() { - return stdinDestroyed || childStdin.destroyed; - }, - get writable() { - return !stdinDestroyed && !stdinEnded && childStdin.writable; - }, - get writableEnded() { - return stdinEnded || childStdin.writableEnded; - }, - get writableFinished() { - return childStdin.writableFinished; - }, - write: (data: string, cb?: (err?: Error | null) => void) => { - if (stdinDestroyed || stdinEnded || !childStdin.writable) { - cb?.(new Error("stdin is not writable")); - return; - } - try { - childStdin.write(data, cb); - } catch (err) { - cb?.(err as Error); - } - }, - end: () => { - try { - stdinEnded = true; - childStdin.end(); - } catch { - // ignore close errors - } - }, - destroy: () => { - try { - stdinDestroyed = true; - stdinEnded = true; - childStdin.destroy(); - } catch { - // ignore destroy errors - } - }, - } - : undefined; + const onStdout: ChildAdapter["onStdout"] = (listener, onRaw) => + onDecodedOutput(child.stdout, listener, onRaw); - const onStdout = (listener: (chunk: string) => void) => onDecodedOutput(child.stdout, listener); - - const onStderr = (listener: (chunk: string) => void) => onDecodedOutput(child.stderr, listener); + const onStderr: ChildAdapter["onStderr"] = (listener, onRaw) => + onDecodedOutput(child.stderr, listener, onRaw); let waitResult: { code: number | null; signal: NodeJS.Signals | null } | null = null; let waitError: unknown; diff --git a/src/process/supervisor/service-child-relay-host.ts b/src/process/supervisor/service-child-relay-host.ts index 47ffb63d30b2..21028f2c0f60 100644 --- a/src/process/supervisor/service-child-relay-host.ts +++ b/src/process/supervisor/service-child-relay-host.ts @@ -1,11 +1,12 @@ import { spawn, type ChildProcess } from "node:child_process"; import { randomUUID } from "node:crypto"; -import type { Duplex, Readable, Writable } from "node:stream"; +import type { Duplex, Readable } from "node:stream"; import { fileURLToPath } from "node:url"; -import { toErrorObject } from "../../infra/errors.js"; import { resolveRuntimeWorkerUrl } from "../../infra/runtime-worker-url.js"; +import { createDeferredCore } from "../../shared/deferred.js"; import { onDecodedOutput } from "../decoded-output.js"; import { addSecretInputStdio, writeSecretInputToChild } from "../spawn-secret-input.js"; +import { createManagedChildStdin } from "./adapters/child-stdin.js"; import { toStringEnv } from "./adapters/env.js"; import { encodeServiceChildMessage, @@ -13,7 +14,7 @@ import { type ServiceChildRelayMessage, type ServiceChildStart, } from "./service-child-protocol.js"; -import type { ManagedRunStdin, SpawnProcessAdapter, SpawnSecretInput } from "./types.js"; +import type { SpawnProcessAdapter, SpawnSecretInput } from "./types.js"; type ServiceAdapter = SpawnProcessAdapter; type AuthorityState = "starting" | "active" | "closing" | "closed" | "identity-lost"; @@ -49,101 +50,57 @@ function reserveStdioEntry(stdio: StdioEntry[], value: StdioEntry): number { return fd; } -function createManagedStdin(stream: Writable | null): ManagedRunStdin | undefined { - if (!stream) { - return undefined; - } - let ended = stream.writableEnded || stream.writableFinished; - let destroyed = stream.destroyed; - stream.once("finish", () => { - ended = true; - }); - stream.once("close", () => { - ended = true; - destroyed = true; - }); - stream.once("error", () => { - destroyed = true; - }); - return { - get destroyed() { - return destroyed || stream.destroyed; - }, - get writable() { - return !destroyed && !ended && stream.writable; - }, - get writableEnded() { - return ended || stream.writableEnded; - }, - get writableFinished() { - return stream.writableFinished; - }, - write(data, callback) { - if (destroyed || ended || !stream.writable) { - callback?.(new Error("stdin is not writable")); - return; - } - try { - stream.write(data, callback); - } catch (error) { - callback?.(toErrorObject(error, "stdin write failed")); - } - }, - end() { - ended = true; - stream.end(); - }, - destroy() { - ended = true; - destroyed = true; - stream.destroy(); - }, - }; -} - function createOutputRelay(stream: Readable) { const listeners = new Set<(chunk: string) => void>(); - const pending: string[] = []; + const rawListeners = new Set<(chunk: Buffer) => void>(); + const pending: Array = []; let pendingBytes = 0; let active = false; - const activate = (deliver: boolean) => { + const deliver = (chunk: string | Buffer) => { + if (typeof chunk === "string") { + listeners.forEach((listener) => listener(chunk)); + } else { + rawListeners.forEach((listener) => listener(chunk)); + } + }; + const activate = (keepOutput: boolean) => { if (active) { return; } active = true; - if (deliver) { - for (const text of pending) { - for (const listener of listeners) { - listener(text); - } - } + if (keepOutput) { + pending.forEach(deliver); } pending.length = 0; pendingBytes = 0; stream.resume(); }; - onDecodedOutput(stream, (text) => { + const enqueue = (chunk: string | Buffer) => { if (active) { - for (const listener of listeners) { - listener(text); - } + deliver(chunk); return; } - pending.push(text); - pendingBytes += Buffer.byteLength(text); - // Bound host memory while retaining the rest in the native pipe until subscription. - if (pendingBytes >= stream.readableHighWaterMark) { - stream.pause(); + pending.push(chunk); + if (Buffer.isBuffer(chunk)) { + pendingBytes += chunk.length; + if (pendingBytes >= stream.readableHighWaterMark) { + stream.pause(); + } } - }); + }; + onDecodedOutput(stream, enqueue, enqueue); return { - subscribe: (listener: (chunk: string) => void) => { + subscribe: (listener: (chunk: string) => void, onRaw?: (chunk: Buffer) => void) => { listeners.add(listener); + if (onRaw) { + rawListeners.add(onRaw); + } activate(true); }, drain: () => activate(false), clear: () => { listeners.clear(); + rawListeners.clear(); pending.length = 0; pendingBytes = 0; }, @@ -215,6 +172,8 @@ export async function createServiceChildRelayAdapter(params: { let rejectWait: ((error: Error) => void) | undefined; let waitPromise: Promise<{ code: number | null; signal: NodeJS.Signals | null }> | undefined; let waitSettled = false; + const extinction = createDeferredCore(); + void extinction.promise.catch(() => undefined); const settleWait = () => { if (waitSettled) { @@ -256,6 +215,7 @@ export async function createServiceChildRelayAdapter(params: { rejectStartup(waitError); } settleWait(); + extinction.reject(waitError); }; let pending = ""; @@ -314,6 +274,7 @@ export async function createServiceChildRelayAdapter(params: { state = "closed"; rootResult ??= { code: null, signal: requestedSignal ?? null }; settleWait(); + extinction.resolve(); }); control.on("error", (error) => { loseIdentity(error.message); @@ -368,7 +329,7 @@ export async function createServiceChildRelayAdapter(params: { throw startupError ?? secretDeliveryError; } - const stdin = createManagedStdin(relay.stdin); + const stdin = createManagedChildStdin(relay.stdin); if (params.input !== undefined) { stdin?.write(params.input); stdin?.end(); @@ -417,6 +378,7 @@ export async function createServiceChildRelayAdapter(params: { }); return await waitPromise; }, + waitForExtinction: () => extinction.promise, kill, dispose: () => { stdoutRelay.clear(); diff --git a/src/process/supervisor/supervisor.test.ts b/src/process/supervisor/supervisor.test.ts index b790f5511d72..3b3f3be3bc75 100644 --- a/src/process/supervisor/supervisor.test.ts +++ b/src/process/supervisor/supervisor.test.ts @@ -143,24 +143,66 @@ describe("process supervisor", () => { expect(adapter.disposeMock).toHaveBeenCalledTimes(1); }); - it("passes private secret input to the child adapter", async () => { + it.each([ + { outcome: "process-tree extinction", failure: false }, + { outcome: "cleanup identity loss", failure: true }, + ])("retains root-result cancellation ownership until $outcome", async ({ failure }) => { + const extinction = createDeferred(); + const adapter = Object.assign(createStubChildAdapter(), { + waitForExtinction: () => extinction.promise, + }); + createChildAdapterMock.mockResolvedValue(adapter); + const supervisor = createProcessSupervisor(); + const run = await spawnChild(supervisor, { + sessionId: "root-result-before-extinction", + scopeKey: "scope:root-result-before-extinction", + argv: createSilentIdleArgv(), + }); + adapter.emitStdout("authentic root output"); + adapter.settle(23); + const root = await run.wait(); + + expect(root).toMatchObject({ reason: "exit", exitCode: 23, stdout: "authentic root output" }); + expect(adapter.disposeMock).not.toHaveBeenCalled(); + supervisor.cancelScope("scope:root-result-before-extinction"); + expect(adapter.killMock).toHaveBeenCalledWith("SIGTERM"); + expect(supervisor.getRecord(run.runId)).toMatchObject({ + state: "exited", + terminationReason: "exit", + exitCode: 23, + }); + + if (failure) { + extinction.reject(new Error("cleanup identity lost")); + await expect(run.waitForExtinction?.()).rejects.toThrow("cleanup identity lost"); + } else { + extinction.resolve(); + await expect(run.waitForExtinction?.()).resolves.toBeUndefined(); + } + expect(adapter.disposeMock).toHaveBeenCalledOnce(); + await expect(run.wait()).resolves.toBe(root); + supervisor.cancel(run.runId); + expect(adapter.killMock).toHaveBeenCalledOnce(); + }); + + it("passes private secret input and exact environment to the child adapter", async () => { const adapter = createStubChildAdapter(); createChildAdapterMock.mockResolvedValue(adapter); - const secretInput = { - fd: 3, - createData: () => Buffer.from("secret"), - }; + const secretInput = { fd: 3, createData: () => Buffer.from("secret") }; const supervisor = createProcessSupervisor(); const run = await spawnChild(supervisor, { sessionId: "s1", argv: createWriteStdoutArgv("ok"), + exactEnv: true, secretInput, }); adapter.settle(0); await run.wait(); - expect(createChildAdapterMock).toHaveBeenCalledWith(expect.objectContaining({ secretInput })); + expect(createChildAdapterMock).toHaveBeenCalledWith( + expect.objectContaining({ exactEnv: true, secretInput }), + ); }); it("enforces no-output timeout for silent processes", async () => { diff --git a/src/process/supervisor/supervisor.ts b/src/process/supervisor/supervisor.ts index 1d92f5468503..9084a72b4d05 100644 --- a/src/process/supervisor/supervisor.ts +++ b/src/process/supervisor/supervisor.ts @@ -195,6 +195,7 @@ export function createProcessSupervisor(): ProcessSupervisor { let forcedReason: TerminationReason | null = startingRun.terminationReason ?? null; let settled = false; + let extinguished = false; let stdout = ""; let stderr = ""; let stdoutListener = input.onStdout; @@ -222,7 +223,9 @@ export function createProcessSupervisor(): ProcessSupervisor { let cancelAdapter: ((reason: TerminationReason) => void) | null = null; const requestCancel = (reason: TerminationReason) => { - setForcedReason(reason); + if (!settled) { + setForcedReason(reason); + } cancelAdapter?.(reason); }; startingRun.cancel = requestCancel; @@ -292,6 +295,7 @@ export function createProcessSupervisor(): ProcessSupervisor { argv: input.argv, cwd: input.cwd, env: input.env, + exactEnv: input.exactEnv, windowsVerbatimArguments: input.windowsVerbatimArguments, input: input.input, stdinMode: input.stdinMode, @@ -303,7 +307,7 @@ export function createProcessSupervisor(): ProcessSupervisor { ...(forcedReason ? { terminationReason: forcedReason } : {}), }); - const clearTimers = () => { + const clearTimers = (includeForceKill = true) => { if (timeoutTimer) { clearTimeout(timeoutTimer); timeoutTimer = null; @@ -312,14 +316,24 @@ export function createProcessSupervisor(): ProcessSupervisor { clearTimeout(noOutputTimer); noOutputTimer = null; } - if (forceKillTimer) { + if (includeForceKill && forceKillTimer) { clearTimeout(forceKillTimer); forceKillTimer = null; } }; + const releaseAuthority = () => { + if (extinguished) { + return; + } + extinguished = true; + clearTimers(); + adapter.dispose(); + active.delete(runId); + }; + cancelAdapter = (reason: TerminationReason) => { - if (settled || cancelRequested) { + if (extinguished || cancelRequested) { return; } cancelRequested = true; @@ -335,7 +349,7 @@ export function createProcessSupervisor(): ProcessSupervisor { } adapter.kill("SIGTERM"); forceKillTimer = setTimeout(() => { - if (!settled) { + if (!extinguished) { adapter.kill("SIGKILL"); } }, GRACEFUL_CANCEL_TIMEOUT_MS); @@ -359,20 +373,29 @@ export function createProcessSupervisor(): ProcessSupervisor { ); } + const onRawOutput = (listener?: (chunk: Buffer) => void) => + listener && + ((chunk: Buffer) => { + listener(chunk); + touchOutput(); + }); + const rawInput = input.mode === "child" ? input : undefined; adapter.onStdout((chunk) => { if (captureOutput) { stdout = appendCapturedOutput(stdout, chunk, "stdout", maxCapturedOutputChars); } stdoutListener?.(chunk); touchOutput(); - }); + }, onRawOutput(rawInput?.onStdoutRaw)); adapter.onStderr((chunk) => { if (captureOutput) { stderr = appendCapturedOutput(stderr, chunk, "stderr", maxCapturedOutputChars); } stderrListener?.(chunk); touchOutput(); - }); + }, onRawOutput(rawInput?.onStderrRaw)); + + const adapterExtinction = adapter.waitForExtinction?.(); const waitPromise = (async (): Promise => { const result = await adapter.wait(); @@ -383,9 +406,7 @@ export function createProcessSupervisor(): ProcessSupervisor { }); const terminalReason = forcedReason ?? deadlineReason; settled = true; - clearTimers(); - adapter.dispose(); - active.delete(runId); + clearTimers(false); const reason: TerminationReason = terminalReason ?? (result.signal != null ? ("signal" as const) : ("exit" as const)); @@ -405,13 +426,19 @@ export function createProcessSupervisor(): ProcessSupervisor { exitCode: exit.exitCode, exitSignal: exit.exitSignal, }); + if (!adapterExtinction) { + releaseAuthority(); + } return exit; })().catch((err: unknown) => { if (!settled) { settled = true; - clearTimers(); - active.delete(runId); - adapter.dispose(); + clearTimers(false); + if (adapterExtinction) { + adapter.kill("SIGKILL"); + } else { + releaseAuthority(); + } registry.finalize(runId, { reason: "spawn-error", exitCode: null, @@ -421,12 +448,23 @@ export function createProcessSupervisor(): ProcessSupervisor { throw err; }); + const extinctionPromise = adapterExtinction + ? Promise.allSettled([waitPromise, adapterExtinction]).then(([, extinction]) => { + releaseAuthority(); + if (extinction.status === "rejected") { + throw extinction.reason; + } + }) + : undefined; + void extinctionPromise?.catch(() => undefined); + const managedRun: ManagedRun = { runId, pid: adapter.pid, startedAtMs, stdin: adapter.stdin, wait: async () => await waitPromise, + ...(extinctionPromise ? { waitForExtinction: async () => await extinctionPromise } : {}), cancel: (reason = "manual-cancel") => { requestCancel(reason); }, diff --git a/src/process/supervisor/types.ts b/src/process/supervisor/types.ts index d1459c8f11da..0a7c44a6c740 100644 --- a/src/process/supervisor/types.ts +++ b/src/process/supervisor/types.ts @@ -44,6 +44,8 @@ export type ManagedRun = { startedAtMs: number; stdin?: ManagedRunStdin; wait: () => Promise; + /** Present only when tree cleanup can outlive the root process result. */ + waitForExtinction?: () => Promise; cancel: (reason?: TerminationReason) => void; /** Stop delivering output callbacks before owner teardown kills the child. */ detachOutput?: () => void; @@ -68,9 +70,10 @@ export type SpawnProcessAdapter = { pid?: number; stdin?: ManagedRunStdin; oomScoreWrapperSelected?: boolean; - onStdout: (listener: (chunk: string) => void) => void; - onStderr: (listener: (chunk: string) => void) => void; + onStdout: (listener: (chunk: string) => void, onRaw?: (chunk: Buffer) => void) => void; + onStderr: (listener: (chunk: string) => void, onRaw?: (chunk: Buffer) => void) => void; wait: () => Promise<{ code: number | null; signal: WaitSignal }>; + waitForExtinction?: () => Promise; kill: (signal?: NodeJS.Signals) => void; dispose: () => void; }; @@ -101,10 +104,14 @@ type SpawnBaseInput = { type SpawnChildInput = SpawnBaseInput & { mode: "child"; argv: string[]; + /** Preserve a caller-prepared environment without environment-mutating spawn wrappers. */ + exactEnv?: true; windowsVerbatimArguments?: boolean; input?: string; stdinMode?: "inherit" | "pipe-open" | "pipe-closed"; secretInput?: SpawnSecretInput; + onStdoutRaw?: (chunk: Buffer) => void; + onStderrRaw?: (chunk: Buffer) => void; }; type SpawnPtyInput = SpawnBaseInput & {