diff --git a/extensions/codex/src/app-server/sandbox-exec-server-registry.ts b/extensions/codex/src/app-server/sandbox-exec-server-registry.ts index e79596db6665..025fb016157c 100644 --- a/extensions/codex/src/app-server/sandbox-exec-server-registry.ts +++ b/extensions/codex/src/app-server/sandbox-exec-server-registry.ts @@ -2,6 +2,28 @@ import type { OpenClawExecServer } from "./sandbox-exec-server/types.js"; export const sandboxExecServerRegistry = { servers: new Map>(), + async close(server: OpenClawExecServer): Promise { + if (server.closed) { + return; + } + server.closed = true; + for (const client of server.server.clients) { + client.close(1001, "shutdown"); + } + await new Promise((resolve) => { + server.server.close(() => resolve()); + }); + const cleanup = await Promise.allSettled([ + ...server.cleanupTasks, + ...[...server.children].map(async (child) => await child.terminate()), + ]); + const failures = cleanup.flatMap((result) => + result.status === "rejected" ? [result.reason] : [], + ); + if (failures.length > 0) { + throw new AggregateError(failures, "Codex sandbox exec-server child cleanup failed"); + } + }, async closeAll(): Promise { const servers = await Promise.allSettled(this.servers.values()); this.servers.clear(); @@ -12,16 +34,7 @@ export const sandboxExecServerRegistry = { } const server = entry.value; server.refCount = 0; - if (server.closed) { - return; - } - server.closed = true; - for (const client of server.server.clients) { - client.close(1001, "shutdown"); - } - await new Promise((resolve) => { - server.server.close(() => resolve()); - }); + await this.close(server); }), ); }, diff --git a/extensions/codex/src/app-server/sandbox-exec-server.lifecycle.test.ts b/extensions/codex/src/app-server/sandbox-exec-server.lifecycle.test.ts index e1882f7b5e5a..ebd764837b55 100644 --- a/extensions/codex/src/app-server/sandbox-exec-server.lifecycle.test.ts +++ b/extensions/codex/src/app-server/sandbox-exec-server.lifecycle.test.ts @@ -7,6 +7,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import type { WebSocket } from "ws"; const spawnMock = vi.hoisted(() => vi.fn()); +const killProcessTreeMock = vi.hoisted(() => vi.fn()); vi.mock("node:child_process", async (importOriginal) => { const actual = await importOriginal(); return { @@ -14,10 +15,17 @@ vi.mock("node:child_process", async (importOriginal) => { spawn: (...args: Parameters) => spawnMock(...args), }; }); +vi.mock("openclaw/plugin-sdk/process-runtime", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + killProcessTree: (...args: unknown[]) => killProcessTreeMock(...args), + }; +}); import { createSandboxContext } from "./sandbox-exec-server.test-helpers.js"; import { httpRequest } from "./sandbox-exec-server/http.js"; -import { startProcess } from "./sandbox-exec-server/processes.js"; +import { startProcess, terminateProcess } from "./sandbox-exec-server/processes.js"; import type { ManagedProcess, OpenClawExecServer } from "./sandbox-exec-server/types.js"; type FakeSocket = WebSocket & { send: ReturnType }; @@ -40,7 +48,7 @@ function createFakeSocket(): FakeSocket { } function createExecServer(sandbox: SandboxContext): OpenClawExecServer { - return { sandbox } as OpenClawExecServer; + return { sandbox, children: new Set(), cleanupTasks: new Set() } as OpenClawExecServer; } function processStartParams(processId: string) { @@ -65,10 +73,223 @@ function streamingHttpParams(requestId: string) { } afterEach(() => { + vi.useRealTimers(); spawnMock.mockReset(); + killProcessTreeMock.mockReset(); }); describe("Codex sandbox exec-server lifecycle", () => { + it("reaps and finalizes a TERM-resistant child before acknowledging termination", async () => { + vi.useFakeTimers(); + const child = createFakeChild(); + spawnMock.mockReturnValue(child); + let finishFinalize: (() => void) | undefined; + const finalizeExec = vi.fn( + async () => + await new Promise((resolve) => { + finishFinalize = resolve; + }), + ); + const sandbox = createSandboxContext({ + buildExecSpec: async () => ({ + argv: ["sandbox-child"], + env: {}, + finalizeToken: "terminate-token", + stdinMode: "pipe-closed", + }), + finalizeExec, + }); + const processes = new Map(); + await startProcess( + createExecServer(sandbox), + processes, + createFakeSocket(), + processStartParams("process-resistant"), + ); + killProcessTreeMock.mockImplementation(() => { + setTimeout(() => child.emit("close", null, "SIGKILL"), 1_000); + }); + + let settled = false; + const termination = Promise.resolve( + terminateProcess(processes, { processId: "process-resistant" }), + ).then((result) => { + settled = true; + return result; + }); + await vi.advanceTimersByTimeAsync(0); + + expect(settled).toBe(false); + expect(killProcessTreeMock).toHaveBeenCalledWith(child.pid, { + detached: process.platform !== "win32", + graceMs: 1_000, + }); + + await vi.runOnlyPendingTimersAsync(); + expect(finalizeExec).toHaveBeenCalledOnce(); + expect(settled).toBe(false); + + finishFinalize?.(); + await expect(termination).resolves.toEqual({ running: true }); + expect(finalizeExec).toHaveBeenCalledWith({ + status: "completed", + exitCode: 1, + timedOut: false, + token: "terminate-token", + }); + }); + + it("preserves cooperative TERM exit without force killing", async () => { + const child = createFakeChild(); + spawnMock.mockReturnValue(child); + killProcessTreeMock.mockImplementation(() => child.emit("close", 143, "SIGTERM")); + const finalizeExec = vi.fn(async () => undefined); + const processes = new Map(); + await startProcess( + createExecServer( + createSandboxContext({ + buildExecSpec: async () => ({ + argv: ["sandbox-child"], + env: {}, + finalizeToken: "cooperative-token", + stdinMode: "pipe-closed", + }), + finalizeExec, + }), + ), + processes, + createFakeSocket(), + processStartParams("process-cooperative"), + ); + + await expect( + terminateProcess(processes, { processId: "process-cooperative" }), + ).resolves.toEqual({ running: true }); + + expect(killProcessTreeMock).toHaveBeenCalledOnce(); + expect(finalizeExec).toHaveBeenCalledWith({ + status: "completed", + exitCode: 143, + timedOut: false, + token: "cooperative-token", + }); + }); + + it("shares termination and finalization across concurrent cleanup", async () => { + vi.useFakeTimers(); + const child = createFakeChild(); + spawnMock.mockReturnValue(child); + killProcessTreeMock.mockImplementation(() => { + setTimeout(() => child.emit("close", null, "SIGKILL"), 1_000); + }); + const finalizeExec = vi.fn(async () => undefined); + const processes = new Map(); + await startProcess( + createExecServer( + createSandboxContext({ + buildExecSpec: async () => ({ + argv: ["sandbox-child"], + env: {}, + finalizeToken: "race-token", + stdinMode: "pipe-closed", + }), + finalizeExec, + }), + ), + processes, + createFakeSocket(), + processStartParams("process-race"), + ); + + const first = terminateProcess(processes, { processId: "process-race" }); + const second = terminateProcess(processes, { processId: "process-race" }); + await vi.runOnlyPendingTimersAsync(); + + await expect(Promise.all([first, second])).resolves.toEqual([ + { running: true }, + { running: true }, + ]); + expect(killProcessTreeMock).toHaveBeenCalledOnce(); + expect(finalizeExec).toHaveBeenCalledOnce(); + }); + + it("reports a surviving tree instead of acknowledging termination", async () => { + vi.useFakeTimers(); + const child = createFakeChild(); + spawnMock.mockReturnValue(child); + killProcessTreeMock.mockImplementation(() => undefined); + const finalizeExec = vi.fn(async () => undefined); + const processes = new Map(); + await startProcess( + createExecServer( + createSandboxContext({ + buildExecSpec: async () => ({ + argv: ["sandbox-child"], + env: {}, + finalizeToken: "survivor-token", + stdinMode: "pipe-closed", + }), + finalizeExec, + }), + ), + processes, + createFakeSocket(), + processStartParams("process-survivor"), + ); + + const termination = terminateProcess(processes, { processId: "process-survivor" }); + const rejection = expect(termination).rejects.toThrow( + `Sandbox child process tree ${child.pid} survived SIGKILL; tear down the sandbox environment and inspect the surviving process tree before retrying.`, + ); + await vi.advanceTimersByTimeAsync(4_500); + + await rejection; + expect(finalizeExec).not.toHaveBeenCalled(); + }); + + it("reaps a TERM-resistant streaming HTTP child on socket close", async () => { + vi.useFakeTimers(); + const child = createFakeChild(); + spawnMock.mockReturnValue(child); + killProcessTreeMock.mockImplementation(() => { + setTimeout(() => child.emit("close", null, "SIGKILL"), 1_000); + }); + const finalizeExec = vi.fn(async () => undefined); + const socket = createFakeSocket(); + const request = httpRequest( + createExecServer( + createSandboxContext({ + buildExecSpec: async () => ({ + argv: ["sandbox-http-child"], + env: {}, + finalizeToken: "http-terminate-token", + stdinMode: "pipe-closed", + }), + finalizeExec, + }), + ), + socket, + streamingHttpParams("http-resistant"), + ); + await vi.waitFor(() => expect(spawnMock).toHaveBeenCalledOnce()); + (child.stdout as PassThrough).write( + `${JSON.stringify({ type: "headers", status: 200, headers: [] })}\n`, + ); + await expect(request).resolves.toEqual({ status: 200, headers: [], bodyBase64: "" }); + + socket.emit("close"); + await vi.runOnlyPendingTimersAsync(); + + expect(killProcessTreeMock).toHaveBeenCalledOnce(); + expect(finalizeExec).toHaveBeenCalledOnce(); + expect(finalizeExec).toHaveBeenCalledWith({ + status: "failed", + exitCode: 1, + timedOut: false, + token: "http-terminate-token", + }); + }); + it("retains the process backend lease after child error until close", async () => { const child = createFakeChild(); spawnMock.mockReturnValue(child); diff --git a/extensions/codex/src/app-server/sandbox-exec-server.test.ts b/extensions/codex/src/app-server/sandbox-exec-server.test.ts index 874b3c84c7f9..a251336a91b8 100644 --- a/extensions/codex/src/app-server/sandbox-exec-server.test.ts +++ b/extensions/codex/src/app-server/sandbox-exec-server.test.ts @@ -44,6 +44,27 @@ function echoFirstInputLineScript(prefix: string): string { ].join(" "); } +async function readStartedPid( + socket: Awaited>, + processId: string, +): Promise { + for (let attempt = 0; attempt < 20; attempt += 1) { + const read = (await rpc(socket, "process/read", { + processId, + afterSeq: 0, + waitMs: 100, + })) as { chunks?: Array<{ chunk: string }> }; + const output = (read.chunks ?? []) + .map((chunk) => Buffer.from(chunk.chunk, "base64").toString("utf8")) + .join(""); + const pid = /PID=(\d+)/u.exec(output)?.[1]; + if (pid) { + return Number(pid); + } + } + throw new Error(`process ${processId} did not report its PID`); +} + describe("OpenClaw Codex sandbox exec-server", () => { it("reports unavailable app-server remote environment support without exposing an environment", async () => { const sandbox = createSandboxContext({}); @@ -213,7 +234,12 @@ describe("OpenClaw Codex sandbox exec-server", () => { expect(buildExecSpec).toHaveBeenCalledWith( expect.objectContaining({ command: "'/bin/sh' '-lc' 'printf ok'", - env: { POLICY_ONLY: "1", POLICY_SET: "env-wins", TEST_FLAG: "1" }, + env: expect.objectContaining({ + CODEX_SANDBOX_EXEC_ID: expect.any(String), + POLICY_ONLY: "1", + POLICY_SET: "env-wins", + TEST_FLAG: "1", + }), usePty: false, workdir: "/workspace", }), @@ -463,11 +489,10 @@ describe("OpenClaw Codex sandbox exec-server", () => { arg0: null, }); - expect(buildExecSpec).toHaveBeenCalledWith( - expect.objectContaining({ - env: {}, - }), - ); + const [{ env: execEnv }] = buildExecSpec.mock.calls[0] as unknown as [ + { env: Record }, + ]; + expect(execEnv).toEqual({ CODEX_SANDBOX_EXEC_ID: expect.any(String) }); socket.close(); }); @@ -626,6 +651,79 @@ describe("OpenClaw Codex sandbox exec-server", () => { await expect(openSocket(execServerUrl)).rejects.toThrow(); }); + it.runIf(process.platform !== "win32")( + "reaps TERM-resistant process groups on socket loss and turn environment release", + async () => { + for (const cleanup of ["socket", "environment"] as const) { + const finalizeExec = vi.fn(async () => undefined); + const sandbox = createSandboxContext({ + buildExecSpec: async () => ({ + argv: [ + "/bin/sh", + "-c", + 'echo "PID=$$"; trap -- "" TERM; while :; do echo heartbeat; sleep 0.1; done', + ], + env: testExecEnv(), + finalizeToken: `${cleanup}-lease`, + stdinMode: "pipe-closed", + }), + finalizeExec, + }); + sandbox.runtimeId = `openclaw-test-runtime-${cleanup}`; + const client = createClient(); + await ensureCodexSandboxExecServerEnvironment({ + client: client as never, + sandbox, + }); + const socket = await openSocket(execServerUrlFromClient(client)); + let pid: number | undefined; + try { + await rpc(socket, "initialize", { clientName: "test" }); + socket.send(JSON.stringify({ method: "initialized" })); + await rpc(socket, "process/start", { + processId: `process-${cleanup}`, + argv: ["ignored"], + cwd: "file:///workspace", + env: {}, + tty: false, + pipeStdin: false, + arg0: null, + }); + pid = await readStartedPid(socket, `process-${cleanup}`); + + if (cleanup === "socket") { + const closed = waitForSocketClose(socket); + socket.terminate(); + await closed; + await vi.waitFor(() => expect(finalizeExec).toHaveBeenCalledOnce(), { + timeout: 3_000, + }); + } else { + await releaseCodexSandboxExecServerEnvironment(sandbox); + } + + expect(() => process.kill(pid!, 0)).toThrow(); + expect(finalizeExec).toHaveBeenCalledOnce(); + expect(finalizeExec).toHaveBeenCalledWith({ + status: "completed", + exitCode: 1, + timedOut: false, + token: `${cleanup}-lease`, + }); + } finally { + if (pid) { + try { + process.kill(-pid, "SIGKILL"); + } catch { + // The owner already reaped the process group. + } + } + await releaseCodexSandboxExecServerEnvironment(sandbox); + } + } + }, + ); + it("keeps a shared exec-server open when another turn reacquires during release", async () => { const sandbox = createSandboxContext({}); const client = createClient(); diff --git a/extensions/codex/src/app-server/sandbox-exec-server.ts b/extensions/codex/src/app-server/sandbox-exec-server.ts index 9753aaafed69..127af7aaa3fa 100644 --- a/extensions/codex/src/app-server/sandbox-exec-server.ts +++ b/extensions/codex/src/app-server/sandbox-exec-server.ts @@ -193,6 +193,8 @@ async function startOpenClawExecServer(sandbox: SandboxContext): Promise { // ws emits error for maxPayload rejections before auth or JSON-RPC sees the frame. @@ -228,20 +230,7 @@ async function releaseOpenClawExecServer(execServer: OpenClawExecServer): Promis if (current === execServer) { sandboxExecServerRegistry.servers.delete(execServer.sandbox.runtimeId); } - await closeOpenClawExecServer(execServer); -} - -async function closeOpenClawExecServer(execServer: OpenClawExecServer): Promise { - if (execServer.closed) { - return; - } - execServer.closed = true; - for (const client of execServer.server.clients) { - client.close(1001, "shutdown"); - } - await new Promise((resolve) => { - execServer.server.close(() => resolve()); - }); + await sandboxExecServerRegistry.close(execServer); } function buildEnvironmentId(sandbox: SandboxContext): string { @@ -267,9 +256,21 @@ function handleConnection(execServer: OpenClawExecServer, socket: WebSocket): vo }); socket.on("close", () => { closeAllFileReads(fileReads); - for (const process of processes.values()) { - process.abortController.abort(); - } + const cleanup = Promise.all( + [...processes].map(async ([processId]) => { + await terminateProcess(processes, { processId }); + }), + ).then(() => undefined); + execServer.cleanupTasks.add(cleanup); + void cleanup.then( + () => execServer.cleanupTasks.delete(cleanup), + (error: unknown) => { + execServer.cleanupTasks.delete(cleanup); + embeddedAgentLog.warn("codex sandbox exec-server socket cleanup failed", { + error: error instanceof Error ? error.message : String(error), + }); + }, + ); }); } @@ -341,7 +342,7 @@ async function dispatchRequest( case "process/write": return writeProcess(processes, request.params); case "process/terminate": - return terminateProcess(processes, request.params); + return await terminateProcess(processes, request.params); case "fs/open": return await openFile(execServer, fileReads, request.params); case "fs/readBlock": diff --git a/extensions/codex/src/app-server/sandbox-exec-server/http.ts b/extensions/codex/src/app-server/sandbox-exec-server/http.ts index 93665ed466bd..edbbf39d1464 100644 --- a/extensions/codex/src/app-server/sandbox-exec-server/http.ts +++ b/extensions/codex/src/app-server/sandbox-exec-server/http.ts @@ -2,15 +2,18 @@ * Implements sandboxed HTTP requests for Codex native tools by routing network * access through the active OpenClaw sandbox backend. */ -import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; import { embeddedAgentLog } from "openclaw/plugin-sdk/agent-harness-runtime"; -import type { SandboxContext } from "openclaw/plugin-sdk/sandbox"; import { SsrFBlockedError, isBlockedHostnameOrIp } from "openclaw/plugin-sdk/ssrf-runtime"; import { sliceUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; import type { WebSocket } from "ws"; import type { JsonObject, JsonValue } from "../protocol.js"; import { readHttpHeaders, requireNumber, requireObject, requireString } from "./json-rpc.js"; import { requireBackend } from "./runtime.js"; +import { + prepareSandboxChildExec, + spawnSandboxChild, + type SandboxChildOwner, +} from "./sandbox-child.js"; import type { HttpHeader, OpenClawExecServer } from "./types.js"; /** Maximum JSON-line size accepted from the streaming HTTP helper process. */ @@ -111,38 +114,34 @@ async function runStreamingSandboxHttpRequest( params: SandboxHttpRequest, ): Promise { const backend = requireBackend(execServer); + const remoteExec = prepareSandboxChildExec(backend, {}); const execSpec = await backend.buildExecSpec({ command: SANDBOX_HTTP_REQUEST_SCRIPT, workdir: execServer.sandbox.containerWorkdir, - env: {}, + env: remoteExec.env, usePty: false, }); - let child: ChildProcessWithoutNullStreams; - try { - const [command, ...args] = execSpec.argv; - if (!command) { - throw new Error("OpenClaw sandbox HTTP exec spec did not provide a command."); - } - child = spawn(command, args, { - env: execSpec.env, - stdio: ["pipe", "pipe", "pipe"], + const lifecycle = { failed: false }; + const owner = await spawnSandboxChild({ + argv: execSpec.argv, + env: execSpec.env, + finalizeExec: backend.finalizeExec, + finalizeToken: execSpec.finalizeToken, + finalizeStatus: (outcome) => + lifecycle.failed || outcome.exitCode !== 0 ? "failed" : "completed", + onFinalizeError: (error) => { + embeddedAgentLog.warn("codex sandbox http/request finalize failed", { error }); + }, + owners: execServer.children, + terminateRemote: remoteExec.terminate, + }); + const child = owner.process; + const abortOnSocketClose = () => { + lifecycle.failed = true; + void owner.terminate().catch((error: unknown) => { + embeddedAgentLog.warn("codex sandbox http/request cleanup failed", { error }); }); - } catch (error) { - try { - await backend.finalizeExec?.({ - status: "failed", - exitCode: null, - timedOut: false, - token: execSpec.finalizeToken, - }); - } catch (finalizeError) { - embeddedAgentLog.warn("codex sandbox http/request finalize after start failure failed", { - error: finalizeError, - }); - } - throw error; - } - const abortOnSocketClose = () => child.kill("SIGTERM"); + }; socket.once("close", abortOnSocketClose); child.once("close", () => { socket.off("close", abortOnSocketClose); @@ -156,17 +155,17 @@ async function runStreamingSandboxHttpRequest( child.stdin.end(JSON.stringify(params)); return await readStreamingSandboxHttpResponse({ child, - execSpec, - finalizeExec: backend.finalizeExec, + lifecycle, + owner, requestId, socket, }); } function readStreamingSandboxHttpResponse(params: { - child: ChildProcessWithoutNullStreams; - execSpec: { finalizeToken?: unknown }; - finalizeExec?: NonNullable["finalizeExec"]; + child: SandboxChildOwner["process"]; + lifecycle: { failed: boolean }; + owner: SandboxChildOwner; requestId: string; socket: WebSocket; }): Promise { @@ -177,21 +176,14 @@ function readStreamingSandboxHttpResponse(params: { let lastBodySeq = 0; let stdoutBuffer = ""; let stderr = ""; - const finalize = async (status: "completed" | "failed", exitCode: number | null) => { - await params.finalizeExec?.({ - status, - exitCode, - timedOut: false, - token: params.execSpec.finalizeToken, - }); - }; - const fail = (message: string, exitCode: number | null) => { + const fail = (message: string, _exitCode: number | null) => { if (failed) { return; } failed = true; - void finalize("failed", exitCode).catch((error: unknown) => { - embeddedAgentLog.warn("codex sandbox http/request finalize failed", { error }); + params.lifecycle.failed = true; + void params.owner.terminate().catch((error: unknown) => { + embeddedAgentLog.warn("codex sandbox http/request cleanup failed", { error }); }); if (headerResolved) { sendHttpBodyDelta(params.socket, { @@ -241,7 +233,6 @@ function readStreamingSandboxHttpResponse(params: { newline = stdoutBuffer.indexOf("\n"); } if (stdoutBuffer.length > SANDBOX_HTTP_STREAM_LINE_MAX_CHARS) { - params.child.kill("SIGKILL"); fail( `sandbox http/request produced an unterminated stdout line longer than ${SANDBOX_HTTP_STREAM_LINE_MAX_CHARS} characters`, null, @@ -256,6 +247,7 @@ function readStreamingSandboxHttpResponse(params: { // ChildProcess error can precede close while the helper is still alive. // Keep its backend lease until close provides the terminal exit state. childFailure ??= error.message; + params.lifecycle.failed = true; }); params.child.once("close", (code) => { const exitCode = code ?? 1; @@ -267,10 +259,8 @@ function readStreamingSandboxHttpResponse(params: { return; } if (exitCode === 0) { - void finalize("completed", exitCode).catch((error: unknown) => { - embeddedAgentLog.warn("codex sandbox http/request finalize failed", { error }); - }); if (!headerResolved) { + params.lifecycle.failed = true; reject(new Error("sandbox http/request exited before returning headers")); } return; diff --git a/extensions/codex/src/app-server/sandbox-exec-server/processes.ts b/extensions/codex/src/app-server/sandbox-exec-server/processes.ts index 67552e3437d9..5e3ebfe0dbce 100644 --- a/extensions/codex/src/app-server/sandbox-exec-server/processes.ts +++ b/extensions/codex/src/app-server/sandbox-exec-server/processes.ts @@ -2,7 +2,6 @@ * Manages subprocess lifecycle, streaming output buffers, stdin writes, and * termination for Codex sandbox exec-server process RPCs. */ -import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; import { embeddedAgentLog } from "openclaw/plugin-sdk/agent-harness-runtime"; import { coerceErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { sanitizeEnvVars } from "openclaw/plugin-sdk/sandbox"; @@ -10,6 +9,7 @@ import type { WebSocket } from "ws"; import type { JsonObject, JsonValue } from "../protocol.js"; import { requireObject, requireString, requireStringArray } from "./json-rpc.js"; import { resolveExecServerPath } from "./path-uri.js"; +import { prepareSandboxChildExec, spawnSandboxChild } from "./sandbox-child.js"; import type { ManagedProcess, OpenClawExecServer, ProcessChunk } from "./types.js"; const ENV_KEY_RE = /^[A-Za-z_][A-Za-z0-9_]*$/; @@ -45,9 +45,8 @@ export async function startProcess( failure: null, tty, pipeStdin, - abortController: new AbortController(), + terminationRequested: false, child: null, - finalized: false, waiters: [], emitNotification: (method, notificationParams) => { if (socket.readyState === 1) { @@ -67,8 +66,10 @@ export async function startProcess( }, }; processes.set(processId, managed); + const startPromise = runProcess(execServer, managed, { argv, cwd, env }); + managed.startPromise = startPromise; try { - await runProcess(execServer, managed, { argv, cwd, env }); + await startPromise; } catch (error) { processes.delete(processId); managed.failure = coerceErrorMessage(error); @@ -77,6 +78,10 @@ export async function startProcess( managed.closed = true; notifyProcessWaiters(managed); throw error; + } finally { + if (managed.startPromise === startPromise) { + managed.startPromise = undefined; + } } return { processId }; } @@ -91,43 +96,44 @@ async function runProcess( throw new Error("OpenClaw sandbox backend is unavailable."); } throwIfProcessStartCancelled(managed); + const remoteExec = prepareSandboxChildExec(backend, params.env); const execSpec = await backend.buildExecSpec({ command: shellCommandFromArgv(params.argv), workdir: params.cwd, - env: params.env, + env: remoteExec.env, // This bridge currently owns only pipe-backed child processes. Asking the // backend for a PTY can produce commands such as `docker exec -t`, which // require this process itself to own a real TTY. usePty: false, }); - managed.finalizeToken = execSpec.finalizeToken; - managed.finalizeExec = backend.finalizeExec; - let child: ChildProcessWithoutNullStreams; - try { - if (managed.abortController.signal.aborted) { - throw new Error("process start cancelled"); - } - const [command, ...args] = execSpec.argv; - if (!command) { - throw new Error("OpenClaw sandbox exec spec did not provide a command."); - } - child = spawn(command, args, { - env: execSpec.env, - stdio: ["pipe", "pipe", "pipe"], + if (managed.terminationRequested) { + await backend.finalizeExec?.({ + status: "failed", + exitCode: null, + timedOut: false, + token: execSpec.finalizeToken, }); - } catch (error) { - managed.failure = coerceErrorMessage(error); - await finalizeProcess(managed).catch((finalizeError: unknown) => { - embeddedAgentLog.warn("codex sandbox exec-server finalize after start failure failed", { - processId: managed.processId, - error: coerceErrorMessage(finalizeError), - }); - }); - throw error; + throw new Error("process start cancelled"); } - managed.child = child; - const abortListener = () => child.kill("SIGTERM"); - managed.abortController.signal.addEventListener("abort", abortListener, { once: true }); + const owner = await spawnSandboxChild({ + argv: execSpec.argv, + env: execSpec.env, + finalizeExec: backend.finalizeExec, + finalizeToken: execSpec.finalizeToken, + finalizeStatus: () => (managed.failure ? "failed" : "completed"), + onFinalizeError: (error) => { + const message = coerceErrorMessage(error); + managed.failure ??= message; + embeddedAgentLog.warn("codex sandbox exec-server finalize failed", { + processId: managed.processId, + error: message, + }); + }, + owners: execServer.children, + terminateRemote: remoteExec.terminate, + }); + managed.child = owner; + const child = owner.process; child.stdout.on("data", (chunk: Buffer) => appendProcessChunk(managed, managed.tty ? "pty" : "stdout", chunk), ); @@ -139,7 +145,6 @@ async function runProcess( notifyProcessWaiters(managed); }); child.once("close", (code) => { - managed.abortController.signal.removeEventListener("abort", abortListener); emitProcessClosed(managed, code ?? 1); }); if (!managed.tty && !managed.pipeStdin) { @@ -148,7 +153,7 @@ async function runProcess( } function throwIfProcessStartCancelled(managed: ManagedProcess): void { - if (managed.abortController.signal.aborted) { + if (managed.terminationRequested) { throw new Error("process start cancelled"); } } @@ -210,34 +215,12 @@ function emitProcessClosed(managed: ManagedProcess, exitCode: number | null): vo seq: closeSeq, }); } - void finalizeProcess(managed).catch((error: unknown) => { - const message = coerceErrorMessage(error); - managed.failure ??= message; - embeddedAgentLog.warn("codex sandbox exec-server finalize failed", { - processId: managed.processId, - error: message, - }); - }); // Closed processes stay briefly readable so clients that observe close before // their final poll can still drain exit/output state. managed.evictProcess(); notifyProcessWaiters(managed); } -async function finalizeProcess(managed: ManagedProcess): Promise { - if (managed.finalized) { - return; - } - managed.finalized = true; - managed.child?.stdin.destroy(); - await managed.finalizeExec?.({ - status: managed.failure ? "failed" : "completed", - exitCode: managed.exitCode, - timedOut: false, - token: managed.finalizeToken, - }); -} - function limitProcessChunks(chunks: ProcessChunk[], maxBytes: number | undefined): ProcessChunk[] { if (!maxBytes) { return chunks; @@ -298,18 +281,22 @@ export function writeProcess( return { status: "unknownProcess" }; } const chunk = Buffer.from(requireString(record.chunk, "chunk"), "base64"); - if ((!managed.tty && !managed.pipeStdin) || managed.closed || !managed.child?.stdin.writable) { + if ( + (!managed.tty && !managed.pipeStdin) || + managed.closed || + !managed.child?.process.stdin.writable + ) { return { status: "stdinClosed" }; } - managed.child.stdin.write(chunk); + managed.child.process.stdin.write(chunk); return { status: "accepted" }; } /** Requests process termination and reports whether it was running at call time. */ -export function terminateProcess( +export async function terminateProcess( processes: Map, params: JsonValue | undefined, -): JsonObject { +): Promise { const record = requireObject(params, "process/terminate params"); const processId = requireString(record.processId, "processId"); const managed = processes.get(processId); @@ -317,9 +304,11 @@ export function terminateProcess( return { running: false }; } const running = !managed.exited; - managed.abortController.abort(); - managed.child?.kill("SIGTERM"); - if (running && !managed.child) { + managed.terminationRequested = true; + await managed.startPromise?.catch(() => undefined); + if (managed.child) { + await managed.child.terminate(); + } else if (running && !managed.closed) { emitProcessClosed(managed, null); } return { running }; diff --git a/extensions/codex/src/app-server/sandbox-exec-server/sandbox-child.ts b/extensions/codex/src/app-server/sandbox-exec-server/sandbox-child.ts new file mode 100644 index 000000000000..b5e1b77e899a --- /dev/null +++ b/extensions/codex/src/app-server/sandbox-exec-server/sandbox-child.ts @@ -0,0 +1,163 @@ +/** Owns one sandbox subprocess tree through close, reaping, and backend finalization. */ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { killProcessTree } from "openclaw/plugin-sdk/process-runtime"; +import type { SandboxContext } from "openclaw/plugin-sdk/sandbox"; + +const SANDBOX_CHILD_TERM_GRACE_MS = 1_000; +// Covers the post-TERM tree kill plus Windows taskkill completion before failure is reported. +const SANDBOX_CHILD_REAP_TIMEOUT_MS = 4_500; +const SANDBOX_EXEC_MARKER = "CODEX_SANDBOX_EXEC_ID"; + +type SandboxChildOutcome = { exitCode: number; signal: NodeJS.Signals | null }; + +export type SandboxChildOwner = { + process: ChildProcessWithoutNullStreams; + settled: Promise; + terminate: () => Promise; +}; + +export async function spawnSandboxChild(params: { + argv: string[]; + env: NodeJS.ProcessEnv; + finalizeExec?: NonNullable["finalizeExec"]; + finalizeToken?: unknown; + finalizeStatus: (outcome: SandboxChildOutcome) => "completed" | "failed"; + onFinalizeError: (error: unknown) => void; + owners: Set; + terminateRemote?: () => Promise; +}): Promise { + const [command, ...args] = params.argv; + const finalize = async (status: "completed" | "failed", exitCode: number | null) => + await params.finalizeExec?.({ + status, + exitCode, + timedOut: false, + token: params.finalizeToken, + }); + if (!command) { + await finalize("failed", null).catch(params.onFinalizeError); + throw new Error("OpenClaw sandbox exec spec did not provide a command."); + } + let child: ChildProcessWithoutNullStreams; + try { + child = spawn(command, args, { + detached: process.platform !== "win32", + env: params.env, + stdio: ["pipe", "pipe", "pipe"], + }); + } catch (error) { + await finalize("failed", null).catch(params.onFinalizeError); + throw error; + } + + let outcome: SandboxChildOutcome | undefined; + const closed = new Promise((resolve) => { + child.once("close", (code, signal) => resolve((outcome = { exitCode: code ?? 1, signal }))); + }); + let finalizePromise: Promise | undefined; + let terminationCleanup: Promise | undefined; + let terminationError: Error | undefined; + const settled = closed.then(async (result) => { + await terminationCleanup; + child.stdin.destroy(); + await (finalizePromise ??= finalize(params.finalizeStatus(result), result.exitCode)); + return result; + }); + void settled.catch(params.onFinalizeError); + + let terminationPromise: Promise | undefined; + const owner: SandboxChildOwner = { + process: child, + settled, + terminate: () => + (terminationPromise ??= (async () => { + child.stdin.destroy(); + terminationCleanup = params.terminateRemote?.().catch((error: unknown) => { + terminationError = error instanceof Error ? error : new Error(String(error)); + }); + await terminationCleanup; + if (!outcome) { + if (child.pid) { + killProcessTree(child.pid, { + detached: process.platform !== "win32", + graceMs: SANDBOX_CHILD_TERM_GRACE_MS, + }); + } else { + child.kill("SIGTERM"); + } + const reaped = await Promise.race([ + closed.then(() => true), + delay(SANDBOX_CHILD_REAP_TIMEOUT_MS).then(() => false), + ]); + if (!reaped) { + throw new Error( + `Sandbox child process tree ${child.pid ?? "unknown"} survived SIGKILL; tear down the sandbox environment and inspect the surviving process tree before retrying.`, + ); + } + } + const result = await settled; + if (terminationError) { + throw terminationError; + } + return result; + })()), + }; + params.owners.add(owner); + void settled.then( + () => params.owners.delete(owner), + () => params.owners.delete(owner), + ); + return owner; +} + +export function prepareSandboxChildExec( + backend: NonNullable, + env: Record, +): { env: Record; terminate: () => Promise } { + const marker = randomUUID(); + return { + env: { ...env, [SANDBOX_EXEC_MARKER]: marker }, + terminate: async () => { + const result = await backend.runShellCommand({ + script: SANDBOX_REMOTE_TERMINATE_SCRIPT, + args: [`${SANDBOX_EXEC_MARKER}=${marker}`], + allowFailure: true, + signal: AbortSignal.timeout(SANDBOX_CHILD_REAP_TIMEOUT_MS), + }); + if (result.code !== 0) { + const detail = + result.stderr.toString("utf8").trim() || result.stdout.toString("utf8").trim(); + throw new Error( + detail || + `Sandbox process tree cleanup failed with code ${result.code}; tear down the sandbox environment and inspect surviving processes before retrying.`, + ); + } + }, + }; +} + +const SANDBOX_REMOTE_TERMINATE_SCRIPT = String.raw` +find_owned_pids() { + for env_file in /proc/[0-9]*/environ; do + if [ -r "$env_file" ] && tr '\0' '\n' < "$env_file" 2>/dev/null | grep -Fqx "$1"; then + basename "$(dirname "$env_file")" + fi + done +} +owned="$(find_owned_pids "$1")" +[ -z "$owned" ] || kill -TERM $owned 2>/dev/null || true +sleep 1 +owned="$(find_owned_pids "$1")" +[ -z "$owned" ] || kill -KILL $owned 2>/dev/null || true +sleep 1 +owned="$(find_owned_pids "$1")" +[ -z "$owned" ] || { echo "Sandbox process IDs survived SIGKILL: $owned" >&2; exit 1; } +`.trim(); + +function delay(ms: number): Promise { + return new Promise((resolve) => { + const timer = setTimeout(resolve, ms); + timer.unref?.(); + }); +} diff --git a/extensions/codex/src/app-server/sandbox-exec-server/types.ts b/extensions/codex/src/app-server/sandbox-exec-server/types.ts index 78ad0047ca20..346289cd643e 100644 --- a/extensions/codex/src/app-server/sandbox-exec-server/types.ts +++ b/extensions/codex/src/app-server/sandbox-exec-server/types.ts @@ -2,10 +2,10 @@ * Shared protocol and runtime state types for the Codex sandbox exec-server * WebSocket bridge. */ -import type { ChildProcessWithoutNullStreams } from "node:child_process"; import type { SandboxContext } from "openclaw/plugin-sdk/sandbox"; import type { WebSocketServer } from "ws"; import type { JsonObject, JsonValue } from "../protocol.js"; +import type { SandboxChildOwner } from "./sandbox-child.js"; /** Minimal JSON-RPC request shape accepted by the sandbox exec-server. */ export type JsonRpcRequest = { @@ -70,11 +70,9 @@ export type ManagedProcess = { failure: string | null; tty: boolean; pipeStdin: boolean; - abortController: AbortController; - child: ChildProcessWithoutNullStreams | null; - finalizeToken?: unknown; - finalizeExec?: NonNullable["finalizeExec"]; - finalized: boolean; + terminationRequested: boolean; + child: SandboxChildOwner | null; + startPromise?: Promise; evictionTimer?: ReturnType; waiters: Array<() => void>; emitNotification: (method: string, params: JsonObject) => void; @@ -90,4 +88,6 @@ export type OpenClawExecServer = { url: string; sandbox: SandboxContext; server: WebSocketServer; + children: Set; + cleanupTasks: Set>; }; diff --git a/src/plugin-sdk/process-runtime.ts b/src/plugin-sdk/process-runtime.ts index 2f8302a2638d..87673fd38e2a 100644 --- a/src/plugin-sdk/process-runtime.ts +++ b/src/plugin-sdk/process-runtime.ts @@ -12,4 +12,5 @@ export { } from "../process/exec.js"; export { prepareOomScoreAdjustedSpawn } from "../process/linux-oom-score.js"; export type { OomScoreAdjustedSpawn, OomWrapOptions } from "../process/linux-oom-score.js"; +export { killProcessTree } from "../process/kill-tree.js"; export { isPidAlive } from "../shared/pid-alive.js";