mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-23 10:55:31 -06:00
refactor(process): supervise Claude node invocations (#127475)
* refactor(process): supervise Claude node invocations * fix(process): preserve exact node invocation environment * test(telegram): await album reply dispatch deterministically
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
a5e9b3189d
commit
5165fbb4e7
@@ -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
|
||||
|
||||
@@ -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<typeof vi.spyOn>) {
|
||||
return flushTimer;
|
||||
}
|
||||
|
||||
async function waitForBufferedProcessing() {
|
||||
await delay(75);
|
||||
}
|
||||
|
||||
async function flushChannelPostMediaGroup(setTimeoutSpy: ReturnType<typeof vi.spyOn>) {
|
||||
const replyDispatched = new Promise<void>((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");
|
||||
|
||||
@@ -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<T = Record<string, unknown>>() {
|
||||
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<typeof import("node:child_process")>("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<string, string>,
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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<string> {
|
||||
return file;
|
||||
}
|
||||
|
||||
function runCommand(
|
||||
executable: string,
|
||||
request: Parameters<typeof runClaudeCliNodeCommand>[0]["request"],
|
||||
overrides: Partial<Parameters<typeof runClaudeCliNodeCommand>[0]> = {},
|
||||
) {
|
||||
return runClaudeCliNodeCommand({
|
||||
client: client([]),
|
||||
frame: frame(request),
|
||||
request,
|
||||
argv: [executable, ...request.argv],
|
||||
cwd: undefined,
|
||||
env: process.env as Record<string, string>,
|
||||
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<string, string>,
|
||||
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<string, string>,
|
||||
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<T>(): Promise<T> {
|
||||
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<string, string>,
|
||||
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<string, string>,
|
||||
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();
|
||||
});
|
||||
|
||||
@@ -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<RunResult>((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<typeof setTimeout>;
|
||||
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 });
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -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));
|
||||
});
|
||||
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<NodeJS.Signals | null>;
|
||||
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<string | Buffer> = [];
|
||||
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();
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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<RunExit> => {
|
||||
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);
|
||||
},
|
||||
|
||||
@@ -44,6 +44,8 @@ export type ManagedRun = {
|
||||
startedAtMs: number;
|
||||
stdin?: ManagedRunStdin;
|
||||
wait: () => Promise<RunExit>;
|
||||
/** Present only when tree cleanup can outlive the root process result. */
|
||||
waitForExtinction?: () => Promise<void>;
|
||||
cancel: (reason?: TerminationReason) => void;
|
||||
/** Stop delivering output callbacks before owner teardown kills the child. */
|
||||
detachOutput?: () => void;
|
||||
@@ -68,9 +70,10 @@ export type SpawnProcessAdapter<WaitSignal = NodeJS.Signals | number | null> = {
|
||||
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<void>;
|
||||
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 & {
|
||||
|
||||
Reference in New Issue
Block a user