From db52ec0ca085e988067c4653c99711353936a263 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 30 Aug 2026 18:47:14 -0500 Subject: [PATCH] feat(runner): adapt the pinned Codex ACPX runtime (#12401) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The runner admits a verified Codex ACPX profile before any provider process can start. > - The pinned ACPX library needs a narrow adapter to the admitted runtime host. > - That adapter must keep credentials and launch controls out of durable session records. > - It must preserve exact recovery identity, model controls, and ownership of the complete provider process tree. > - This pull request adds the Codex-only package adapter without registering production execution. ## Linked Issues or Issue Description **Agent or provider** Codex through the exact ACPX and Codex ACP packages landed in #12400. **Why this adapter is useful** The package-local runtime host has an injected port, but no production implementation. This implementation uses the verified executable lease and private runtime sandbox without persisting managed credentials or other launch-only state in ACPX recovery records. **How the agent is invoked** The adapter creates one persistent ACPX Codex session. ACPX receives a placeholder registry command, while its patched spawn callback launches through Paperclip's verified command lease. The private launch environment is supplied only at spawn time. Durable session state receives only the session key, workspace, model, and bounded system instructions. **Additional context** #12400 is merged. This PR does not register an adapter, start runnerd, expose a server route, or change any direct adapter. It supports Codex only, rejects non-Codex profiles, and fails closed on Windows until provider descendants can be contained with an owned Job Object or equivalent. ## What Changed - Add a Codex-only adapter from the pinned ACPX library to the admitted runtime port. - Create the ACPX store inside the private runtime state directory. - Open one persistent session with the qualified model and bounded system instructions. - Route provider launches through the verified executable lease and a dedicated POSIX process group. - Retain cleanup ownership through asynchronous errors and late termination. - Supply the private launch environment at spawn time without persisting it. - Require all ACPX recovery identity fields before returning the runtime port. - Map status, exact model selection, and state-preserving close operations. - Add regression coverage for secret isolation, verified spawning, process-tree cleanup, lifecycle mapping, identity failure, and the Codex-only boundary. ## Verification - Exact verified head: `dc89439d0b2e3dee46d212715caeefc8ae0c0959`. - Full GitHub PR workflow passed in [run 33341468207, attempt 3](https://github.com/paperclipai/paperclip/actions/runs/33341468207/attempts/3), including runner verification/build, typecheck, all test shards, canary, and e2e. - Greptile is 5/5 on the exact head with zero unresolved review threads. - Superagent Security, Snyk, contributor trust, and commitperclip passed on the exact head. - Storybook skipped by path as expected. - The diff contains 2 files and does not change `pnpm-lock.yaml`, workflows, migrations, server selection, or UI behavior. - No additional local suite was run during the final restack; GitHub Actions is the authoritative verification environment. ## Risks The primary risk is leaking launch credentials into durable ACPX state. Session options are constructed explicitly and regression-tested; the launch environment remains behind the spawn-time callback. Another risk is orphaning credential-bearing descendants. Supported launches use a retained POSIX process-group identity with bounded TERM-to-KILL cleanup. Windows fails closed before runtime construction until equivalent process-tree containment exists. ## Model Used OpenAI Codex with GPT-5 and repository tool use. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used with version and capability details - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either linked an existing public item or described the issue in this PR - [x] I have not referenced internal or instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal task identifier - [x] I have added or updated tests where applicable - [x] I have documented the process, credential, recovery, and rollout risks - [x] All applicable GitHub Actions are green - [x] Greptile is 5/5 with every actionable comment resolved - [x] I have addressed all review findings before merge --- .../acpx/codex-runtime-adapter.test.ts | 1216 +++++++++++++++++ .../src/drivers/acpx/codex-runtime-adapter.ts | 827 +++++++++++ 2 files changed, 2043 insertions(+) create mode 100644 packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts create mode 100644 packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts new file mode 100644 index 0000000000..9bf7faf1b2 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts @@ -0,0 +1,1216 @@ +import type { ChildProcess } from "node:child_process"; +import { EventEmitter } from "node:events"; + +import type { + AcpAgentRegistry, + AcpRuntime, + AcpRuntimeHandle, + AcpRuntimeOptions, + AcpSessionStore, +} from "acpx/runtime"; +import { decodeAcpxRuntimeHandleState } from "acpx/runtime"; +import { describe, expect, it, vi } from "vitest"; + +import type { VerifiedAcpxCommandLease } from "./installation-integrity.js"; +import { openCodexAcpxRuntime } from "./codex-runtime-adapter.js"; +import type { AcpxRuntimePortOpenOptions } from "./runtime-host.js"; + +const HANDLE: AcpRuntimeHandle = { + sessionKey: "session-key", + backend: "acpx", + runtimeSessionName: "runtime-name", + cwd: "/workspace", + acpxRecordId: "record-1", + backendSessionId: "backend-1", + agentSessionId: "agent-1", +}; + +describe("Codex ACPX runtime adapter", () => { + it("rejects a pre-aborted admission before constructing or spawning ACPX", async () => { + const cancellation = new Error("runtime admission cancelled"); + const controller = new AbortController(); + controller.abort(cancellation); + const command = fakeCommand(); + const createRuntime = vi.fn(); + + await expect( + openCodexAcpxRuntime( + { ...openOptions(command), signal: controller.signal }, + { createRuntime }, + ), + ).rejects.toBe(cancellation); + + expect(createRuntime).not.toHaveBeenCalled(); + expect(command.spawn).not.toHaveBeenCalled(); + }); + + it("rejects Windows before constructing or spawning ACPX", async () => { + const command = fakeCommand(); + const createRuntime = vi.fn(); + const platformDescriptor = Object.getOwnPropertyDescriptor( + process, + "platform", + ); + if (platformDescriptor === undefined) { + throw new Error("Node process.platform descriptor is unavailable"); + } + Object.defineProperty(process, "platform", { + ...platformDescriptor, + value: "win32", + }); + try { + await expect( + openCodexAcpxRuntime(openOptions(command), { createRuntime }), + ).rejects.toThrow( + "provider process-tree containment unavailable on Windows", + ); + } finally { + Object.defineProperty(process, "platform", platformDescriptor); + } + + expect(createRuntime).not.toHaveBeenCalled(); + expect(command.spawn).not.toHaveBeenCalled(); + }); + + it("opens a persistent Codex session without persisting launch secrets", async () => { + const runtime = fakeRuntime(); + let runtimeOptions: AcpRuntimeOptions | undefined; + const command = fakeCommand(); + const port = await openCodexAcpxRuntime(openOptions(command), { + createRegistry: ({ overrides }) => { + expect(overrides).toEqual({ + codex: ["paperclip-verified-acpx-command"], + }); + return registry(); + }, + createStore: ({ stateDir }) => { + expect(stateDir).toBe("/runtime/state"); + return store(); + }, + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + }); + + expect(runtime.ensureSession).toHaveBeenCalledWith({ + sessionKey: "provider-key", + agent: "codex", + mode: "persistent", + cwd: "/workspace", + sessionOptions: { + model: "gpt-5.6-sol", + systemPrompt: { append: "Use Paperclip tools." }, + }, + }); + expect( + JSON.stringify(vi.mocked(runtime.ensureSession).mock.calls[0]?.[0]), + ).not.toContain("credential-secret"); + expect(runtimeOptions?.spawnEnvironment?.()).toEqual({ + CODEX_HOME: "/runtime/agent-home", + OPENAI_API_KEY: "credential-secret", + }); + expect(runtimeOptions?.spawnCwd).toBe("/workspace"); + expect(await port.identity()).toEqual({ + acpxRecordId: "record-1", + backendSessionId: "backend-1", + agentSessionId: "agent-1", + }); + }); + + it("launches only through the verified command lease", async () => { + const runtime = fakeRuntime(); + let runtimeOptions: AcpRuntimeOptions | undefined; + const command = fakeCommand(); + await openCodexAcpxRuntime(openOptions(command), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + }); + const child = fakeChild(); + vi.mocked(command.spawn).mockReturnValue(child); + const spawnOptions = { cwd: "/runtime/spawn" }; + + expect( + runtimeOptions?.spawnAgent?.({ + command: "/attacker/replacement", + args: ["--stdio"], + options: spawnOptions, + }), + ).toBe(child); + expect(command.spawn).toHaveBeenCalledWith(["--stdio"], { + ...spawnOptions, + detached: true, + }); + }); + + it.runIf(process.platform !== "win32")( + "terminates the provider process group after its leader exits", + async () => { + const child = fakeProcessGroupChild(54_321); + const command = fakeCommand(); + vi.mocked(command.spawn).mockReturnValue(child); + const handshakeFailure = new Error("ACP handshake rejected"); + const runtime = fakeRuntime(); + let groupRunning = true; + const processKill = vi + .spyOn(process, "kill") + .mockImplementation((pid, signal) => { + expect(pid).toBe(-54_321); + if (signal === 0) { + if (!groupRunning) { + throw Object.assign(new Error("process group exited"), { + code: "ESRCH", + }); + } + return true; + } + if (signal === "SIGTERM") { + child.signalCode = "SIGTERM"; + queueMicrotask(() => child.emit("exit", null, "SIGTERM")); + return true; + } + if (signal === "SIGKILL") { + groupRunning = false; + return true; + } + return true; + }); + vi.useFakeTimers(); + try { + const openingError = openCodexAcpxRuntime(openOptions(command), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (runtimeOptions) => { + vi.mocked(runtime.ensureSession).mockImplementation(async () => { + runtimeOptions.spawnAgent?.({ + command: "ignored", + args: ["--stdio"], + options: {}, + }); + throw handshakeFailure; + }); + return runtime; + }, + }).then( + () => undefined, + (error: unknown) => error, + ); + for (let turn = 0; turn < 5; turn += 1) await Promise.resolve(); + expect(command.spawn).toHaveBeenCalledOnce(); + + await vi.advanceTimersByTimeAsync(2_001); + await expect(openingError).resolves.toBe(handshakeFailure); + expect(processKill).toHaveBeenCalledWith(-54_321, "SIGTERM"); + expect(processKill).toHaveBeenCalledWith(-54_321, "SIGKILL"); + expect(child.kill).not.toHaveBeenCalled(); + } finally { + processKill.mockRestore(); + vi.useRealTimers(); + } + }, + ); + + it("maps status, model selection, and state-preserving close", async () => { + const runtime = fakeRuntime(); + vi.mocked(runtime.getStatus!).mockResolvedValue({ + models: { + currentModelId: "gpt-5.6-sol", + availableModelIds: ["gpt-5.6-sol"], + }, + }); + const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: () => runtime, + }); + + expect(await port.getStatus()).toEqual({ + models: { + currentModelId: "gpt-5.6-sol", + availableModelIds: ["gpt-5.6-sol"], + }, + }); + await port.setModel?.("gpt-5.6-sol"); + expect(runtime.setConfigOption).toHaveBeenCalledWith({ + handle: HANDLE, + key: "model", + value: "gpt-5.6-sol", + }); + await port.close({ reason: "test complete" }); + expect(runtime.close).toHaveBeenCalledWith({ + handle: HANDLE, + reason: "test complete", + discardPersistentState: false, + }); + }); + + it("fails closed and closes the session when ACPX omits recovery identity", async () => { + const runtime = fakeRuntime({ ...HANDLE, agentSessionId: undefined }); + await expect( + openCodexAcpxRuntime(openOptions(fakeCommand()), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: () => runtime, + }), + ).rejects.toThrow("ACPX runtime omitted agentSessionId"); + expect(runtime.close).toHaveBeenCalledWith({ + handle: { ...HANDLE, agentSessionId: undefined }, + reason: "ACPX runtime identity validation failed", + discardPersistentState: false, + }); + }); + + it("retains failed ordinary admission cleanup until a retry succeeds", async () => { + const runtime = fakeRuntime({ ...HANDLE, agentSessionId: undefined }); + const firstCloseFailure = new Error("runtime close failed"); + vi.mocked(runtime.close) + .mockRejectedValueOnce(firstCloseFailure) + .mockResolvedValueOnce(undefined); + const retainedCleanups: Promise[] = []; + + await expect( + openCodexAcpxRuntime(openOptions(fakeCommand()), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: () => runtime, + retainCleanup: (cleanup) => retainedCleanups.push(cleanup), + }), + ).rejects.toMatchObject({ + errors: [ + expect.objectContaining({ + message: "ACPX runtime omitted agentSessionId", + }), + firstCloseFailure, + ], + }); + + expect(retainedCleanups).toHaveLength(1); + await expect(retainedCleanups[0]).resolves.toBeUndefined(); + expect(runtime.close).toHaveBeenCalledTimes(2); + }); + + it("terminates a provider spawned before the session handshake rejects", async () => { + const child = fakeChild(); + const command = fakeCommand(); + vi.mocked(command.spawn).mockReturnValue(child); + const failure = new Error("ACP handshake rejected"); + const runtime = fakeRuntime(); + + await expect( + openCodexAcpxRuntime(openOptions(command), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + vi.mocked(runtime.ensureSession).mockImplementation(async () => { + options.spawnAgent?.({ + command: "ignored", + args: ["--stdio"], + options: {}, + }); + throw failure; + }); + return runtime; + }, + }), + ).rejects.toBe(failure); + expect(runtime.close).not.toHaveBeenCalled(); + expect(child.kill).toHaveBeenCalledWith("SIGTERM"); + }); + + it("aborts a blocked handshake, reaps its provider, and closes a late session", async () => { + let resolveHandshake: ((handle: AcpRuntimeHandle) => void) | undefined; + const blockedHandshake = new Promise((resolve) => { + resolveHandshake = resolve; + }); + const child = fakeChild(); + const command = fakeCommand(); + vi.mocked(command.spawn).mockReturnValue(child); + const runtime = fakeRuntime(); + const controller = new AbortController(); + const cancellation = new Error("runtime admission cancelled"); + + const opening = openCodexAcpxRuntime( + { ...openOptions(command), signal: controller.signal }, + { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (runtimeOptions) => { + vi.mocked(runtime.ensureSession).mockImplementation(async () => { + runtimeOptions.spawnAgent?.({ + command: "ignored", + args: ["--stdio"], + options: {}, + }); + return await blockedHandshake; + }); + return runtime; + }, + }, + ); + await vi.waitFor(() => expect(command.spawn).toHaveBeenCalledOnce()); + + controller.abort(cancellation); + await expect(opening).rejects.toBe(cancellation); + expect(child.kill).toHaveBeenCalledWith("SIGTERM"); + + resolveHandshake?.(HANDLE); + await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledOnce()); + expect(runtime.close).toHaveBeenCalledWith({ + handle: HANDLE, + reason: "ACPX runtime admission aborted", + discardPersistentState: false, + }); + }); + + it("retries retained admission cleanup after the first close times out", async () => { + let rejectHandshake: ((error: Error) => void) | undefined; + let firstCloseSettled = false; + let overlappingClose = false; + const blockedHandshake = new Promise( + (_resolve, reject) => { + rejectHandshake = reject; + }, + ); + const runtime = fakeRuntime(); + vi.mocked(runtime.ensureSession).mockReturnValue(blockedHandshake); + vi.mocked(runtime.close) + .mockImplementationOnce( + () => + new Promise((_resolve, reject) => { + setTimeout(() => { + firstCloseSettled = true; + reject(new Error("late close rejected after its timeout")); + }, 8); + }), + ) + .mockImplementationOnce(async () => { + overlappingClose = !firstCloseSettled; + }); + const controller = new AbortController(); + const cancellation = new Error("runtime admission cancelled"); + let runtimeOptions: AcpRuntimeOptions | undefined; + + const opening = openCodexAcpxRuntime( + { + ...openOptions(fakeCommand()), + signal: controller.signal, + }, + { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + runtimeCloseTimeoutMs: 5, + }, + ); + await vi.waitFor(() => + expect(runtime.ensureSession).toHaveBeenCalledOnce(), + ); + + controller.abort(cancellation); + const openingFailure = opening.catch((error: unknown) => error); + await runtimeOptions!.sessionStore.save({ + acpxRecordId: "late-record", + acpSessionId: "late-backend-session", + agentSessionId: "late-agent-session", + name: "late-runtime-name", + cwd: "/workspace", + } as never); + + await expect(openingFailure).resolves.toBeInstanceOf(Error); + await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(2)); + expect(firstCloseSettled).toBe(true); + expect(overlappingClose).toBe(false); + expect(runtime.close).toHaveBeenCalledTimes(2); + expect(vi.mocked(runtime.close).mock.calls[1]?.[0]).toEqual( + vi.mocked(runtime.close).mock.calls[0]?.[0], + ); + rejectHandshake?.(new Error("test handshake stopped")); + }); + + it("retries a rejected cleanup for a session returned after admission aborts", async () => { + let resolveHandshake: ((handle: AcpRuntimeHandle) => void) | undefined; + const blockedHandshake = new Promise((resolve) => { + resolveHandshake = resolve; + }); + const firstCloseFailure = new Error("late runtime close failed"); + const runtime = fakeRuntime(); + vi.mocked(runtime.ensureSession).mockReturnValue(blockedHandshake); + vi.mocked(runtime.close) + .mockRejectedValueOnce(firstCloseFailure) + .mockResolvedValueOnce(undefined); + const controller = new AbortController(); + const cancellation = new Error("runtime admission cancelled"); + + const opening = openCodexAcpxRuntime( + { + ...openOptions(fakeCommand()), + signal: controller.signal, + }, + { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: () => runtime, + }, + ); + await vi.waitFor(() => + expect(runtime.ensureSession).toHaveBeenCalledOnce(), + ); + + controller.abort(cancellation); + await expect(opening).rejects.toBe(cancellation); + resolveHandshake?.(HANDLE); + + await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(2)); + expect(vi.mocked(runtime.close).mock.calls[0]?.[0]).toEqual({ + handle: HANDLE, + reason: "ACPX runtime admission aborted", + discardPersistentState: false, + }); + expect(vi.mocked(runtime.close).mock.calls[1]?.[0]).toEqual( + vi.mocked(runtime.close).mock.calls[0]?.[0], + ); + }); + + it("retains ownership of a late close until its exact attempt settles", async () => { + let resolveHandshake: ((handle: AcpRuntimeHandle) => void) | undefined; + const blockedHandshake = new Promise((resolve) => { + resolveHandshake = resolve; + }); + const runtime = fakeRuntime(); + vi.mocked(runtime.ensureSession).mockReturnValue(blockedHandshake); + let resolveClose: (() => void) | undefined; + vi.mocked(runtime.close).mockImplementation( + () => + new Promise((resolve) => { + resolveClose = resolve; + }), + ); + const controller = new AbortController(); + const cancellation = new Error("runtime admission cancelled"); + const retainedCleanups: Promise[] = []; + + const opening = openCodexAcpxRuntime( + { ...openOptions(fakeCommand()), signal: controller.signal }, + { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: () => runtime, + runtimeCloseTimeoutMs: 5, + retainCleanup: (cleanup) => retainedCleanups.push(cleanup), + }, + ); + await vi.waitFor(() => + expect(runtime.ensureSession).toHaveBeenCalledOnce(), + ); + + controller.abort(cancellation); + await expect(opening).rejects.toBe(cancellation); + expect(retainedCleanups).toHaveLength(1); + resolveHandshake?.(HANDLE); + + await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledOnce()); + let cleanupSettled = false; + void retainedCleanups[0]!.then( + () => { + cleanupSettled = true; + }, + () => { + cleanupSettled = true; + }, + ); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(cleanupSettled).toBe(false); + expect(runtime.close).toHaveBeenCalledOnce(); + + resolveClose?.(); + await retainedCleanups[0]; + expect(cleanupSettled).toBe(true); + expect(runtime.close).toHaveBeenCalledOnce(); + }); + + it("coalesces store and late-handshake cleanup while a timed-out close is active", async () => { + let resolveHandshake: ((handle: AcpRuntimeHandle) => void) | undefined; + let firstCloseSettled = false; + let overlappingClose = false; + const blockedHandshake = new Promise((resolve) => { + resolveHandshake = resolve; + }); + const runtime = fakeRuntime(); + vi.mocked(runtime.ensureSession).mockReturnValue(blockedHandshake); + vi.mocked(runtime.close) + .mockImplementationOnce( + () => + new Promise((_resolve, reject) => { + setTimeout(() => { + firstCloseSettled = true; + reject(new Error("timed-out close eventually rejected")); + }, 8); + }), + ) + .mockImplementationOnce(async () => { + overlappingClose = !firstCloseSettled; + }); + const controller = new AbortController(); + const cancellation = new Error("runtime admission cancelled"); + let runtimeOptions: AcpRuntimeOptions | undefined; + + const opening = openCodexAcpxRuntime( + { + ...openOptions(fakeCommand()), + signal: controller.signal, + }, + { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + runtimeCloseTimeoutMs: 5, + }, + ); + await vi.waitFor(() => + expect(runtime.ensureSession).toHaveBeenCalledOnce(), + ); + + controller.abort(cancellation); + await expect(opening).rejects.toBe(cancellation); + await runtimeOptions!.sessionStore.save({ + acpxRecordId: "record-1", + acpSessionId: "backend-1", + agentSessionId: "agent-1", + name: "stored-runtime-name", + cwd: "/workspace", + } as never); + // The returned handle uses a different runtimeSessionName representation, + // but its durable record identity must join the store-published cleanup. + resolveHandshake?.(HANDLE); + await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(2)); + expect(overlappingClose).toBe(false); + }); + + it("promotes a session fallback into the later durable record lifecycle", async () => { + let resolveHandshake: ((handle: AcpRuntimeHandle) => void) | undefined; + let firstCloseSettled = false; + let overlappingClose = false; + const blockedHandshake = new Promise((resolve) => { + resolveHandshake = resolve; + }); + const runtime = fakeRuntime(); + vi.mocked(runtime.ensureSession).mockReturnValue(blockedHandshake); + vi.mocked(runtime.close) + .mockImplementationOnce( + () => + new Promise((_resolve, reject) => { + setTimeout(() => { + firstCloseSettled = true; + reject(new Error("fallback close rejected after timeout")); + }, 15); + }), + ) + .mockImplementationOnce(async () => { + overlappingClose = !firstCloseSettled; + }); + const controller = new AbortController(); + const cancellation = new Error("runtime admission cancelled"); + const retainedCleanups: Promise[] = []; + let runtimeOptions: AcpRuntimeOptions | undefined; + + const opening = openCodexAcpxRuntime( + { + ...openOptions(fakeCommand()), + signal: controller.signal, + }, + { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + runtimeCloseTimeoutMs: 10, + retainCleanup: (cleanup) => retainedCleanups.push(cleanup), + }, + ); + await vi.waitFor(() => + expect(runtime.ensureSession).toHaveBeenCalledOnce(), + ); + + controller.abort(cancellation); + await expect(opening).rejects.toBe(cancellation); + resolveHandshake?.({ + ...HANDLE, + sessionKey: "provider-key", + acpxRecordId: undefined, + agentSessionId: "fallback-agent-session", + } as never); + for (let turn = 0; turn < 5; turn += 1) await Promise.resolve(); + + await runtimeOptions!.sessionStore.save({ + acpxRecordId: "promoted-record", + acpSessionId: "promoted-backend-session", + name: "promoted-runtime-name", + cwd: "/workspace", + } as never); + await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(2)); + expect(overlappingClose).toBe(false); + expect(vi.mocked(runtime.close).mock.calls[1]?.[0].handle).toMatchObject({ + acpxRecordId: "promoted-record", + backendSessionId: "promoted-backend-session", + agentSessionId: "fallback-agent-session", + }); + expect( + decodeAcpxRuntimeHandleState( + vi.mocked(runtime.close).mock.calls[1]![0].handle.runtimeSessionName, + ), + ).toMatchObject({ name: "promoted-runtime-name" }); + await Promise.all(retainedCleanups); + }); + + it("keeps exact record ids distinct when whitespace differs", async () => { + let rejectHandshake: ((error: Error) => void) | undefined; + const blockedHandshake = new Promise( + (_resolve, reject) => { + rejectHandshake = reject; + }, + ); + const runtime = fakeRuntime(); + vi.mocked(runtime.ensureSession).mockReturnValue(blockedHandshake); + const controller = new AbortController(); + const cancellation = new Error("runtime admission cancelled"); + let runtimeOptions: AcpRuntimeOptions | undefined; + + const opening = openCodexAcpxRuntime( + { ...openOptions(fakeCommand()), signal: controller.signal }, + { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + }, + ); + await vi.waitFor(() => + expect(runtime.ensureSession).toHaveBeenCalledOnce(), + ); + + controller.abort(cancellation); + await expect(opening).rejects.toBe(cancellation); + for (const acpxRecordId of ["record-id", " record-id "]) { + await runtimeOptions!.sessionStore.save({ + acpxRecordId, + acpSessionId: `${acpxRecordId}-backend`, + agentSessionId: `${acpxRecordId}-agent`, + name: `${acpxRecordId}-runtime`, + cwd: "/workspace", + } as never); + } + + await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(2)); + expect( + vi + .mocked(runtime.close) + .mock.calls.map(([input]) => input.handle.acpxRecordId), + ).toEqual(["record-id", " record-id "]); + rejectHandshake?.(new Error("test handshake stopped")); + }); + + it("retains cleanup beyond the former admission close retry budget", async () => { + let rejectHandshake: ((error: Error) => void) | undefined; + const blockedHandshake = new Promise( + (_resolve, reject) => { + rejectHandshake = reject; + }, + ); + const runtime = fakeRuntime(); + const closeFailure = new Error("runtime close failed"); + const retainedCleanups: Promise[] = []; + vi.mocked(runtime.ensureSession).mockReturnValue(blockedHandshake); + vi.mocked(runtime.close) + .mockRejectedValueOnce(closeFailure) + .mockRejectedValueOnce(closeFailure) + .mockRejectedValueOnce(closeFailure) + .mockRejectedValueOnce(closeFailure) + .mockResolvedValueOnce(undefined); + const controller = new AbortController(); + const cancellation = new Error("runtime admission cancelled"); + let runtimeOptions: AcpRuntimeOptions | undefined; + + const opening = openCodexAcpxRuntime( + { + ...openOptions(fakeCommand()), + signal: controller.signal, + }, + { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + retainCleanup: (cleanup) => retainedCleanups.push(cleanup), + }, + ); + await vi.waitFor(() => + expect(runtime.ensureSession).toHaveBeenCalledOnce(), + ); + + controller.abort(cancellation); + await runtimeOptions!.sessionStore.save({ + acpxRecordId: "late-record", + acpSessionId: "late-backend-session", + agentSessionId: "late-agent-session", + name: "late-runtime-name", + cwd: "/workspace", + } as never); + + await expect(opening).rejects.toBeInstanceOf(Error); + expect(retainedCleanups).toHaveLength(2); + await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(5)); + await expect(retainedCleanups[0]).resolves.toBeUndefined(); + expect(runtime.close).toHaveBeenCalledTimes(5); + + rejectHandshake?.(new Error("test handshake stopped")); + await expect(retainedCleanups[1]).rejects.toThrow("test handshake stopped"); + }); + + it("aggregates asynchronous provider signal errors after a failed handshake", async () => { + const child = failingSignalChild(); + const command = fakeCommand(); + vi.mocked(command.spawn).mockReturnValue(child.child); + const handshakeError = new Error("ACP handshake rejected"); + const runtime = fakeRuntime(); + + const result = openCodexAcpxRuntime(openOptions(command), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + vi.mocked(runtime.ensureSession).mockImplementation(async () => { + options.spawnAgent?.({ + command: "ignored", + args: ["--stdio"], + options: {}, + }); + throw handshakeError; + }); + return runtime; + }, + }); + + await expect(result).rejects.toMatchObject({ + errors: [ + handshakeError, + ...child.errors, + expect.objectContaining({ + message: "ACPX provider did not exit after SIGKILL", + }), + ], + }); + expect(child.child.kill).toHaveBeenNthCalledWith(1, "SIGTERM"); + expect(child.child.kill).toHaveBeenNthCalledWith(2, "SIGKILL"); + child.child.signalCode = "SIGKILL"; + child.child.emit("exit", null, "SIGKILL"); + }); + + it("retains provider cleanup beyond the former retry budget", async () => { + const child = new EventEmitter() as ChildProcess; + Object.defineProperties(child, { + exitCode: { value: null, writable: true }, + signalCode: { value: null, writable: true }, + }); + let killAttempts = 0; + child.kill = vi.fn((signal) => { + if (signal === "SIGKILL") { + killAttempts += 1; + if (killAttempts === 5) { + child.signalCode = "SIGKILL"; + queueMicrotask(() => child.emit("exit", null, "SIGKILL")); + return true; + } + } + queueMicrotask(() => + child.emit("error", new Error(`${String(signal)} still pending`)), + ); + return true; + }); + const command = fakeCommand(); + vi.mocked(command.spawn).mockReturnValue(child); + const handshakeError = new Error("ACP handshake rejected"); + const runtime = fakeRuntime(); + const retainedCleanups: Promise[] = []; + + await expect( + openCodexAcpxRuntime(openOptions(command), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (runtimeOptions) => { + vi.mocked(runtime.ensureSession).mockImplementation(async () => { + runtimeOptions.spawnAgent?.({ + command: "ignored", + args: ["--stdio"], + options: {}, + }); + throw handshakeError; + }); + return runtime; + }, + retainCleanup: (cleanup) => retainedCleanups.push(cleanup), + }), + ).rejects.toBeInstanceOf(AggregateError); + + expect(retainedCleanups).toHaveLength(1); + await expect(retainedCleanups[0]).resolves.toBeUndefined(); + expect(killAttempts).toBe(5); + }); + + it("closes a recovered session when its handshake rejects before another save", async () => { + const runtime = fakeRuntime(); + const recoveredStore = store(); + vi.mocked(recoveredStore.load).mockResolvedValue({ + acpxRecordId: "recovered-record", + acpSessionId: "recovered-backend-session", + agentSessionId: "recovered-agent-session", + name: "recovered-runtime-name", + cwd: "/workspace", + } as never); + const failure = new Error("recovered ACP handshake rejected"); + + await expect( + openCodexAcpxRuntime(openOptions(fakeCommand()), { + createRegistry: () => registry(), + createStore: () => recoveredStore, + createRuntime: (runtimeOptions) => { + vi.mocked(runtime.ensureSession).mockImplementation(async () => { + await runtimeOptions.sessionStore.load("provider-key"); + throw failure; + }); + return runtime; + }, + }), + ).rejects.toBe(failure); + expect(runtime.close).toHaveBeenCalledOnce(); + const recoveredClose = vi.mocked(runtime.close).mock.calls[0]![0]; + expect(recoveredClose).toMatchObject({ + handle: { + sessionKey: "provider-key", + backend: "acpx", + cwd: "/workspace", + acpxRecordId: "recovered-record", + backendSessionId: "recovered-backend-session", + agentSessionId: "recovered-agent-session", + }, + reason: "ACPX session handshake failed", + discardPersistentState: false, + }); + expect( + decodeAcpxRuntimeHandleState(recoveredClose.handle.runtimeSessionName), + ).toEqual({ + name: "recovered-runtime-name", + agent: "codex", + cwd: "/workspace", + mode: "persistent", + acpxRecordId: "recovered-record", + backendSessionId: "recovered-backend-session", + agentSessionId: "recovered-agent-session", + }); + expect(recoveredStore.save).not.toHaveBeenCalled(); + }); + + it("closes a newly created session when its record save rejects", async () => { + const runtime = fakeRuntime(); + const failingStore = store(); + const failure = new Error("session store unavailable"); + vi.mocked(failingStore.save).mockRejectedValue(failure); + + await expect( + openCodexAcpxRuntime(openOptions(fakeCommand()), { + createRegistry: () => registry(), + createStore: () => failingStore, + createRuntime: (runtimeOptions) => { + vi.mocked(runtime.ensureSession).mockImplementation(async () => { + await runtimeOptions.sessionStore.save({ + acpxRecordId: "new-record", + acpSessionId: "new-backend-session", + agentSessionId: "new-agent-session", + name: "new-runtime-name", + cwd: "/workspace", + } as never); + return HANDLE; + }); + return runtime; + }, + }), + ).rejects.toBe(failure); + expect(runtime.close).toHaveBeenCalledOnce(); + const failedSaveClose = vi.mocked(runtime.close).mock.calls[0]![0]; + expect(failedSaveClose).toMatchObject({ + handle: { + sessionKey: "provider-key", + backend: "acpx", + cwd: "/workspace", + acpxRecordId: "new-record", + backendSessionId: "new-backend-session", + agentSessionId: "new-agent-session", + }, + reason: "ACPX session handshake failed", + discardPersistentState: false, + }); + expect( + decodeAcpxRuntimeHandleState(failedSaveClose.handle.runtimeSessionName), + ).toMatchObject({ name: "new-runtime-name", agent: "codex" }); + }); + + it("bounds a stalled runtime close before terminating a failed-handshake provider", async () => { + const child = fakeChild(); + const command = fakeCommand(); + vi.mocked(command.spawn).mockReturnValue(child); + const runtime = fakeRuntime(); + vi.mocked(runtime.close).mockImplementation( + () => new Promise(() => undefined), + ); + + await expect( + openCodexAcpxRuntime(openOptions(command), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (runtimeOptions) => { + vi.mocked(runtime.ensureSession).mockImplementation(async () => { + await runtimeOptions.sessionStore.save({ + acpxRecordId: "actual-record", + acpSessionId: "backend-session", + agentSessionId: "agent-session", + name: "actual-runtime-name", + cwd: "/workspace", + } as never); + runtimeOptions.spawnAgent?.({ + command: "ignored", + args: ["--stdio"], + options: {}, + }); + throw new Error("ACP handshake rejected"); + }); + return runtime; + }, + runtimeCloseTimeoutMs: 5, + }), + ).rejects.toThrow("ACPX session handshake and runtime cleanup failed"); + expect(runtime.close).toHaveBeenCalledOnce(); + const stalledClose = vi.mocked(runtime.close).mock.calls[0]![0]; + expect(stalledClose).toMatchObject({ + handle: { + sessionKey: "provider-key", + backend: "acpx", + cwd: "/workspace", + acpxRecordId: "actual-record", + backendSessionId: "backend-session", + agentSessionId: "agent-session", + }, + reason: "ACPX session handshake failed", + discardPersistentState: false, + }); + expect( + decodeAcpxRuntimeHandleState(stalledClose.handle.runtimeSessionName), + ).toMatchObject({ name: "actual-runtime-name", agent: "codex" }); + expect(child.kill).toHaveBeenCalledWith("SIGTERM"); + }); + + it("rejects close with asynchronous provider signal errors", async () => { + const runtime = fakeRuntime(); + const command = fakeCommand(); + const child = failingSignalChild(); + let runtimeOptions: AcpRuntimeOptions | undefined; + vi.mocked(command.spawn).mockReturnValue(child.child); + const port = await openCodexAcpxRuntime(openOptions(command), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + }); + runtimeOptions?.spawnAgent?.({ + command: "ignored", + args: ["--stdio"], + options: {}, + }); + + await expect(port.close({ reason: "test complete" })).rejects.toMatchObject( + { + errors: [ + ...child.errors, + expect.objectContaining({ + message: "ACPX provider did not exit after SIGKILL", + }), + ], + }, + ); + expect(child.child.kill).toHaveBeenNthCalledWith(1, "SIGTERM"); + expect(child.child.kill).toHaveBeenNthCalledWith(2, "SIGKILL"); + child.child.signalCode = "SIGKILL"; + child.child.emit("exit", null, "SIGKILL"); + }); + + it("retains a provider error after close removes the child", async () => { + const runtime = fakeRuntime(); + const command = fakeCommand(); + const child = fakeChild(); + const providerError = new Error("provider spawn failed"); + let runtimeOptions: AcpRuntimeOptions | undefined; + vi.mocked(command.spawn).mockReturnValue(child); + const port = await openCodexAcpxRuntime(openOptions(command), { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + }); + runtimeOptions?.spawnAgent?.({ + command: "ignored", + args: ["--stdio"], + options: {}, + }); + + child.emit("error", providerError); + // A real ChildProcess has committed its terminal status before `close`. + // Model that ordering so the process tracker can prove this child no + // longer needs a cleanup signal while retaining its earlier error. + child.exitCode = 1; + child.emit("close", 1, null); + + await expect(port.close({ reason: "test complete" })).rejects.toMatchObject( + { + errors: [providerError], + }, + ); + expect(child.kill).not.toHaveBeenCalled(); + }); + + it("rejects non-Codex profiles before constructing ACPX", async () => { + const createRuntime = vi.fn(); + await expect( + openCodexAcpxRuntime( + { + ...openOptions(fakeCommand()), + profile: { + ...openOptions(fakeCommand()).profile, + agent: "claude", + }, + }, + { createRuntime }, + ), + ).rejects.toThrow("currently supports Codex only"); + expect(createRuntime).not.toHaveBeenCalled(); + }); +}); + +function openOptions( + command: VerifiedAcpxCommandLease, +): AcpxRuntimePortOpenOptions { + return { + command, + profile: { + driverKind: "acpx_runtime", + protocolVersion: 1, + acpxVersion: "0.13.1", + agent: "codex", + agentProfileVersion: 1, + agentServerPackage: "@agentclientprotocol/codex-acp", + agentServerVersion: "1.6.2", + agentRuntimePackage: null, + agentRuntimeVersion: null, + commandDigest: "sha256:test", + qualificationModel: "gpt-5.6-sol", + reportedModelId: "gpt-5.6-sol", + permissionPolicy: "interactive", + }, + cwd: "/workspace", + stateDirectory: "/runtime/state", + providerSessionKey: "provider-key", + permissionMode: "approve-reads", + permissionPolicy: { + autoApprove: ["read"], + escalate: ["write"], + defaultAction: "escalate", + }, + launchEnvironment: { + CODEX_HOME: "/runtime/agent-home", + OPENAI_API_KEY: "credential-secret", + OMITTED: undefined, + }, + systemInstructions: "Use Paperclip tools.", + }; +} + +function fakeRuntime(handle: AcpRuntimeHandle = HANDLE): AcpRuntime { + return { + ensureSession: vi.fn().mockResolvedValue(handle), + startTurn: vi.fn(), + runTurn: vi.fn(), + getStatus: vi.fn(), + setConfigOption: vi.fn(), + cancel: vi.fn(), + close: vi.fn(), + }; +} + +function fakeCommand(): VerifiedAcpxCommandLease { + return { spawn: vi.fn(), close: vi.fn() }; +} + +function fakeChild(): ChildProcess { + const child = new EventEmitter() as ChildProcess; + Object.defineProperties(child, { + exitCode: { value: null, writable: true }, + signalCode: { value: null, writable: true }, + }); + child.kill = vi.fn(() => { + child.signalCode = "SIGTERM"; + queueMicrotask(() => child.emit("exit", null, "SIGTERM")); + return true; + }); + return child; +} + +function fakeProcessGroupChild(pid: number): ChildProcess { + const child = fakeChild(); + Object.defineProperty(child, "pid", { value: pid }); + return child; +} + +function failingSignalChild(): { + child: ChildProcess; + errors: [Error, Error]; +} { + const child = new EventEmitter() as ChildProcess; + const errors: [Error, Error] = [ + new Error("SIGTERM delivery failed"), + new Error("SIGKILL delivery failed"), + ]; + Object.defineProperties(child, { + exitCode: { value: null, writable: true }, + signalCode: { value: null, writable: true }, + }); + child.kill = vi.fn((signal) => { + const error = signal === "SIGTERM" ? errors[0] : errors[1]; + queueMicrotask(() => child.emit("error", error)); + return true; + }); + return { child, errors }; +} + +function registry(): AcpAgentRegistry { + return { resolve: vi.fn(), list: vi.fn() }; +} + +function store(): AcpSessionStore { + return { load: vi.fn(), save: vi.fn() }; +} diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts new file mode 100644 index 0000000000..ade6610540 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts @@ -0,0 +1,827 @@ +import type { ChildProcess } from "node:child_process"; + +import { + createAcpRuntime, + createAgentRegistry, + createRuntimeStore, + encodeAcpxRuntimeHandleState, + type AcpAgentRegistry, + type AcpRuntime, + type AcpRuntimeHandle, + type AcpRuntimeOptions, + type AcpSessionRecord, + type AcpSessionStore, +} from "acpx/runtime"; + +import type { + AcpxRuntimePort, + AcpxRuntimePortIdentity, + AcpxRuntimePortOpenOptions, +} from "./runtime-host.js"; + +const VERIFIED_COMMAND_SENTINEL = "paperclip-verified-acpx-command"; +const DEFAULT_RUNTIME_CLOSE_TIMEOUT_MS = 2_000; +const RETAINED_ADMISSION_CLEANUP_RETRY_MIN_MS = 10; +const RETAINED_ADMISSION_CLEANUP_RETRY_MAX_MS = 30_000; +const activeCodexRuntimeCleanupOwners = new Set>(); + +class AcpxRuntimeCloseTimeoutError extends Error { + constructor() { + super("ACPX runtime close timed out"); + this.name = "AcpxRuntimeCloseTimeoutError"; + } +} + +export interface CodexAcpxRuntimeDependencies { + createRuntime?: (options: AcpRuntimeOptions) => AcpRuntime; + createRegistry?: (input: { + overrides: Record; + }) => AcpAgentRegistry; + createStore?: (input: { stateDir: string }) => AcpSessionStore; + runtimeCloseTimeoutMs?: number; + /** Internal test seam for autonomous failed-admission cleanup ownership. */ + retainCleanup?: (cleanup: Promise) => void; +} + +/** + * Adapt the pinned ACPX library to Paperclip's admitted runtime port. The + * executable, launch environment, and spawn cwd stay host-owned and are never + * persisted in ACPX's session options. + */ +export async function openCodexAcpxRuntime( + options: AcpxRuntimePortOpenOptions, + dependencies: CodexAcpxRuntimeDependencies = {}, +): Promise { + if (options.profile.agent !== "codex") { + throw new Error( + "The production ACPX runtime currently supports Codex only", + ); + } + options.signal?.throwIfAborted(); + // The verified-command boundary already refuses to mint a Windows command + // lease, because Node cannot pin its executable there. Repeat the platform + // gate at this lower boundary so alternate host wiring cannot launch a + // credential-bearing provider without a killable tree. `child.kill()` only + // terminates the direct Windows process, and taskkill cannot reliably find + // descendants after their original parent has exited; Windows support must + // therefore wait for an owned Job Object or equivalent containment. + if (process.platform === "win32") { + throw new Error( + "The production ACPX runtime requires provider process-tree containment unavailable on Windows", + ); + } + + const createRegistry = dependencies.createRegistry ?? createAgentRegistry; + const createStore = dependencies.createStore ?? createRuntimeStore; + const createRuntime = dependencies.createRuntime ?? createAcpRuntime; + const runtimeCloseTimeoutMs = + dependencies.runtimeCloseTimeoutMs ?? DEFAULT_RUNTIME_CLOSE_TIMEOUT_MS; + const children = new SpawnedChildSet(); + const baseStore = createStore({ stateDir: options.stateDirectory }); + let failedHandshakeHandle: AcpRuntimeHandle | null = null; + let admissionCleanup: RuntimeAdmissionCleanup | null = null; + const retainedCleanupOwners = new WeakSet>(); + const retainCleanup = (cleanup: Promise): void => { + if (retainedCleanupOwners.has(cleanup)) { + return; + } + retainedCleanupOwners.add(cleanup); + dependencies.retainCleanup?.(cleanup); + retainCodexRuntimeCleanup(cleanup); + }; + const rememberHandshakeHandle = (record: AcpSessionRecord): void => { + const runtimeSessionName = record.name?.trim(); + if ( + typeof record.acpxRecordId !== "string" || + record.acpxRecordId.length === 0 || + !runtimeSessionName || + record.cwd !== options.cwd + ) { + return; + } + const rememberedHandle: AcpRuntimeHandle = { + sessionKey: options.providerSessionKey, + backend: "acpx", + runtimeSessionName: encodeAcpxRuntimeHandleState({ + name: runtimeSessionName, + agent: "codex", + cwd: record.cwd, + mode: "persistent", + acpxRecordId: record.acpxRecordId, + backendSessionId: record.acpSessionId, + agentSessionId: record.agentSessionId, + }), + cwd: record.cwd, + acpxRecordId: record.acpxRecordId, + backendSessionId: record.acpSessionId, + ...(record.agentSessionId + ? { agentSessionId: record.agentSessionId } + : {}), + }; + failedHandshakeHandle = rememberedHandle; + if (options.signal?.aborted && admissionCleanup !== null) { + retainCleanup( + admissionCleanup.runRetained( + rememberedHandle, + "ACPX runtime admission aborted", + ), + ); + } + }; + const sessionStore: AcpSessionStore = { + async load(sessionId) { + const record = await baseStore.load(sessionId); + if (record !== undefined) rememberHandshakeHandle(record); + return record; + }, + async save(record) { + // ACPX has already created this runtime-owned identity before it asks + // the store to persist it. Capture cleanup authority first so a storage + // rejection cannot orphan the live session created by the handshake. + rememberHandshakeHandle(record); + await baseStore.save(record); + }, + }; + const runtime = createRuntime({ + cwd: options.cwd, + sessionStore, + agentRegistry: createRegistry({ + overrides: { codex: [VERIFIED_COMMAND_SENTINEL] }, + }), + permissionMode: options.permissionMode, + nonInteractivePermissions: "fail", + permissionPolicy: { + ...options.permissionPolicy, + autoApprove: options.permissionPolicy.autoApprove + ? [...options.permissionPolicy.autoApprove] + : undefined, + escalate: options.permissionPolicy.escalate + ? [...options.permissionPolicy.escalate] + : undefined, + }, + spawnEnvironment: () => definedEnvironment(options.launchEnvironment), + spawnCwd: options.cwd, + spawnAgent: (input) => { + // ACPX can invoke this callback after its handshake caller has already + // been cancelled. Check at the last host-owned boundary so a late + // handshake cannot create a provider process after authority is gone. + options.signal?.throwIfAborted(); + // A verified provider can create descendants that inherit its launch + // credential. Give the provider a dedicated POSIX process group so + // cleanup authority covers that complete credential-bearing tree. + return children.add( + options.command.spawn(input.args, { + ...input.options, + detached: true, + }) as ChildProcess, + true, + ); + }, + }); + admissionCleanup = new RuntimeAdmissionCleanup( + runtime, + children, + runtimeCloseTimeoutMs, + retainCleanup, + ); + + let handle: AcpRuntimeHandle | null = null; + try { + const handshake = Promise.resolve().then(() => + runtime.ensureSession({ + sessionKey: options.providerSessionKey, + agent: "codex", + mode: "persistent", + cwd: options.cwd, + sessionOptions: { + model: options.profile.qualificationModel, + ...(options.systemInstructions + ? { systemPrompt: { append: options.systemInstructions } } + : {}), + }, + }), + ); + if (options.signal === undefined) { + handle = await handshake; + } else { + try { + handle = await raceRuntimeHandshakeWithAbort(handshake, options.signal); + } catch (error) { + if (options.signal.aborted) { + retainCleanup( + handshake.then((lateHandle) => + admissionCleanup!.runRetained( + lateHandle, + "ACPX runtime admission aborted", + ), + ), + ); + } + throw error; + } + // The promise and abort notification can settle in the same turn. Do + // not admit a handle if cancellation won immediately afterward. + options.signal.throwIfAborted(); + } + } catch (error) { + const cleanupErrors = await admissionCleanup.run( + handle ?? failedHandshakeHandle, + options.signal?.aborted + ? "ACPX runtime admission aborted" + : "ACPX session handshake failed", + ); + if (cleanupErrors.length > 0) { + throw new AggregateError( + [error, ...cleanupErrors], + "ACPX session handshake and runtime cleanup failed", + ); + } + throw error; + } + + // Assigned by the successful handshake above. Keeping this assertion at the + // boundary makes it impossible to construct a port from a cancelled or + // otherwise absent ACPX session. + if (handle === null) + throw new Error("ACPX runtime omitted its session handle"); + try { + return runtimePort( + runtime, + handle, + requireIdentity(handle), + admissionCleanup, + ); + } catch (error) { + const cleanupErrors = await admissionCleanup.run( + handle, + "ACPX runtime identity validation failed", + ); + if (cleanupErrors.length > 0) { + throw new AggregateError( + [error, ...cleanupErrors], + "ACPX runtime identity validation and cleanup failed", + ); + } + throw error; + } +} + +function raceRuntimeHandshakeWithAbort( + handshake: Promise, + signal: AbortSignal, +): Promise { + signal.throwIfAborted(); + return new Promise((resolve, reject) => { + let settled = false; + const settle = (operation: () => void): void => { + if (settled) return; + settled = true; + signal.removeEventListener("abort", onAbort); + operation(); + }; + const onAbort = (): void => settle(() => reject(signal.reason)); + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) { + onAbort(); + return; + } + void handshake.then( + (value) => settle(() => resolve(value)), + (error: unknown) => settle(() => reject(error)), + ); + }); +} + +function retainCodexRuntimeCleanup(cleanup: Promise): void { + activeCodexRuntimeCleanupOwners.add(cleanup); + void cleanup + .finally(() => activeCodexRuntimeCleanupOwners.delete(cleanup)) + .catch(() => undefined); +} + +type RuntimeAdmissionCleanupTarget = { + handle: AcpRuntimeHandle | null; + reason: string; + cleanup: Promise | null; +}; + +class RuntimeAdmissionCleanup { + readonly #closedHandles = new Set(); + readonly #activeHandleAttempts = new Map< + string, + Promise + >(); + readonly #registeredTargets = new Map< + string, + RuntimeAdmissionCleanupTarget + >(); + readonly #targetAliases = new Map(); + #tail: Promise = Promise.resolve(); + + constructor( + private readonly runtime: AcpRuntime, + private readonly children: SpawnedChildSet, + private readonly runtimeCloseTimeoutMs: number, + private readonly retainCleanup: (cleanup: Promise) => void, + ) {} + + run(handle: AcpRuntimeHandle | null, reason: string): Promise { + const targetKey = this.#resolveTargetKey( + runtimeAdmissionCleanupTargetKey(handle), + handle, + ); + return this.#runAttempt(targetKey, handle, reason).then(({ errors }) => { + if (errors.length > 0) { + this.retainCleanup(this.runRetained(handle, reason)); + } + return errors; + }); + } + + runRetained(handle: AcpRuntimeHandle | null, reason: string): Promise { + const rawTargetKey = runtimeAdmissionCleanupTargetKey(handle); + const targetKey = this.#resolveTargetKey(rawTargetKey, handle); + const existing = this.#registeredTargets.get(targetKey); + if (existing !== undefined) { + if (handle !== null) { + existing.handle = + existing.handle === null + ? handle + : preferRuntimeAdmissionCleanupHandle(existing.handle, handle); + } + this.#targetAliases.set(rawTargetKey, targetKey); + return existing.cleanup!; + } + const target: RuntimeAdmissionCleanupTarget = { + handle, + reason, + cleanup: null, + }; + this.#registeredTargets.set(targetKey, target); + this.#targetAliases.set(rawTargetKey, targetKey); + const cleanup = this.#retryRetained(targetKey, target); + target.cleanup = cleanup; + return cleanup; + } + + #resolveTargetKey( + rawTargetKey: string, + handle: AcpRuntimeHandle | null, + ): string { + const aliasedTargetKey = this.#targetAliases.get(rawTargetKey); + if (aliasedTargetKey !== undefined) return aliasedTargetKey; + if (handle === null) return rawTargetKey; + const recordId = nonEmptyRuntimeIdentity(handle.acpxRecordId); + const sessionTargetKey = runtimeAdmissionCleanupSessionTargetKey(handle); + if (recordId !== undefined) { + const fallbackTargetKey = + this.#targetAliases.get(sessionTargetKey) ?? sessionTargetKey; + const fallbackHandle = + this.#registeredTargets.get(fallbackTargetKey)?.handle; + if ( + fallbackHandle !== undefined && + fallbackHandle !== null && + nonEmptyRuntimeIdentity(fallbackHandle.acpxRecordId) === undefined && + sameRuntimeAdmissionCleanupOwner(fallbackHandle, handle) + ) { + this.#targetAliases.set(rawTargetKey, fallbackTargetKey); + return fallbackTargetKey; + } + return rawTargetKey; + } + const compatibleRecordTargets = [ + ...this.#registeredTargets.entries(), + ].filter( + ([, target]) => + target.handle !== null && + nonEmptyRuntimeIdentity(target.handle.acpxRecordId) !== undefined && + sameRuntimeAdmissionCleanupOwner(target.handle, handle), + ); + return compatibleRecordTargets.length === 1 + ? compatibleRecordTargets[0]![0] + : rawTargetKey; + } + + async #retryRetained( + targetKey: string, + target: RuntimeAdmissionCleanupTarget, + ): Promise { + let retryDelayMs = RETAINED_ADMISSION_CLEANUP_RETRY_MIN_MS; + // Retained cleanup is the continuing owner. Keep one runtime-close attempt + // in flight at a time and retry process-tree termination until both are + // confirmed complete; a finite budget would recreate an orphan boundary. + for (;;) { + const attempt = await this.#runAttempt( + targetKey, + target.handle, + target.reason, + ); + const runtimeNeedsRetry = + target.handle !== null && + attempt.runtimeError !== undefined && + !this.#closedHandles.has(targetKey); + const processNeedsRetry = attempt.processErrors.length > 0; + if (!runtimeNeedsRetry && !processNeedsRetry) { + return; + } + await delay(retryDelayMs); + retryDelayMs = Math.min( + retryDelayMs * 2, + RETAINED_ADMISSION_CLEANUP_RETRY_MAX_MS, + ); + } + } + + #runAttempt( + targetKey: string, + handle: AcpRuntimeHandle | null, + reason: string, + ): Promise<{ + errors: unknown[]; + runtimeError: unknown | undefined; + processErrors: unknown[]; + }> { + const cleanup = this.#tail.then(async () => { + const errors: unknown[] = []; + let runtimeError: unknown | undefined; + if (handle !== null && !this.#closedHandles.has(targetKey)) { + runtimeError = await this.#closeHandleWithin(targetKey, handle, reason); + if (runtimeError !== undefined) errors.push(runtimeError); + } + const processErrors = await this.children.terminate(); + errors.push(...processErrors); + return { + errors, + runtimeError, + processErrors, + }; + }); + this.#tail = cleanup.then( + () => undefined, + () => undefined, + ); + return cleanup; + } + + async #closeHandleWithin( + targetKey: string, + handle: AcpRuntimeHandle, + reason: string, + ): Promise { + let attempt = this.#activeHandleAttempts.get(targetKey); + if (attempt === undefined) { + attempt = runtimeCloseOutcome(this.runtime, { + handle, + reason, + discardPersistentState: false, + }); + this.#activeHandleAttempts.set(targetKey, attempt); + void attempt.then((error) => { + if (this.#activeHandleAttempts.get(targetKey) === attempt) { + this.#activeHandleAttempts.delete(targetKey); + } + if (error === undefined) this.#closedHandles.add(targetKey); + }); + } + return await closeOutcomeWithin(attempt, this.runtimeCloseTimeoutMs); + } +} + +function runtimeAdmissionCleanupTargetKey( + handle: AcpRuntimeHandle | null, +): string { + if (handle === null) return JSON.stringify(["children"]); + const recordId = nonEmptyRuntimeIdentity(handle.acpxRecordId); + return recordId === undefined + ? runtimeAdmissionCleanupSessionTargetKey(handle) + : JSON.stringify(["record", recordId]); +} + +function runtimeAdmissionCleanupSessionTargetKey( + handle: AcpRuntimeHandle, +): string { + return JSON.stringify(["session", handle.sessionKey]); +} + +function preferRuntimeAdmissionCleanupHandle( + current: AcpRuntimeHandle, + incoming: AcpRuntimeHandle, +): AcpRuntimeHandle { + const currentRecordId = nonEmptyRuntimeIdentity(current.acpxRecordId); + const incomingRecordId = nonEmptyRuntimeIdentity(incoming.acpxRecordId); + if ( + !sameRuntimeAdmissionCleanupOwner(current, incoming) || + (currentRecordId !== undefined && incomingRecordId !== currentRecordId) + ) { + return current; + } + const currentAgentSessionId = nonEmptyRuntimeIdentity(current.agentSessionId); + const incomingAgentSessionId = nonEmptyRuntimeIdentity( + incoming.agentSessionId, + ); + if ( + currentAgentSessionId !== undefined && + incomingAgentSessionId !== undefined && + incomingAgentSessionId !== currentAgentSessionId + ) { + return current; + } + const backendSessionId = + nonEmptyRuntimeIdentity(incoming.backendSessionId) ?? + nonEmptyRuntimeIdentity(current.backendSessionId); + const agentSessionId = incomingAgentSessionId ?? currentAgentSessionId; + return { + ...current, + ...incoming, + ...((incomingRecordId ?? currentRecordId) === undefined + ? {} + : { acpxRecordId: incomingRecordId ?? currentRecordId }), + ...(backendSessionId === undefined ? {} : { backendSessionId }), + ...(agentSessionId === undefined ? {} : { agentSessionId }), + }; +} + +function sameRuntimeAdmissionCleanupOwner( + current: AcpRuntimeHandle, + incoming: AcpRuntimeHandle, +): boolean { + return ( + current.sessionKey === incoming.sessionKey && + current.backend === incoming.backend && + current.cwd === incoming.cwd + ); +} + +function nonEmptyRuntimeIdentity( + value: string | undefined, +): string | undefined { + return typeof value === "string" && value.length > 0 ? value : undefined; +} + +function runtimePort( + runtime: AcpRuntime, + handle: AcpRuntimeHandle, + identity: AcpxRuntimePortIdentity, + admissionCleanup: RuntimeAdmissionCleanup, +): AcpxRuntimePort { + return { + async identity() { + return structuredClone(identity); + }, + async getStatus() { + if (!runtime.getStatus) { + throw new Error("The pinned ACPX runtime cannot report session status"); + } + return structuredClone(await runtime.getStatus({ handle })); + }, + ...(runtime.setConfigOption + ? { + async setModel(model: string) { + await runtime.setConfigOption?.({ + handle, + key: "model", + value: model, + }); + }, + } + : {}), + async close(input) { + const errors = await admissionCleanup.run(handle, input.reason); + if (errors.length > 0) { + throw new AggregateError( + errors, + "ACPX runtime and provider cleanup failed", + ); + } + }, + }; +} + +function runtimeCloseOutcome( + runtime: AcpRuntime, + input: Parameters[0], +): Promise { + return Promise.resolve() + .then(() => runtime.close(input)) + .then( + () => undefined, + (error: unknown) => error, + ); +} + +async function closeOutcomeWithin( + closeOutcome: Promise, + timeoutMs: number, +): Promise { + const boundedTimeoutMs = Math.max(1, timeoutMs); + let timer: ReturnType | undefined; + const timeoutOutcome = new Promise((resolve) => { + timer = setTimeout( + () => resolve(new AcpxRuntimeCloseTimeoutError()), + boundedTimeoutMs, + ); + }); + const outcome = await Promise.race([closeOutcome, timeoutOutcome]); + if (timer !== undefined) clearTimeout(timer); + return outcome; +} + +function delay(timeoutMs: number): Promise { + return new Promise((resolve) => { + const timer = setTimeout(resolve, timeoutMs); + timer.unref?.(); + }); +} + +class SpawnedChildSet { + readonly #children = new Map(); + readonly #errors = new Set(); + + add(child: ChildProcess, processGroup: boolean): ChildProcess { + const onError = (error: unknown) => this.#errors.add(error); + const tracked: SpawnedProviderProcess = { + child, + processGroupId: processGroup ? (child.pid ?? null) : null, + onError, + }; + this.#children.set(child, tracked); + const forgetExitedTree = () => { + if (!providerTreeRunning(tracked)) this.#forget(tracked); + }; + // ChildProcess reports some spawn and signal-delivery failures through an + // asynchronous `error` event. Observe those for the child's whole tracked + // lifetime so cleanup can report them instead of crashing runnerd. + child.on("error", onError); + child.once("exit", forgetExitedTree); + child.once("close", forgetExitedTree); + return child; + } + + async terminate(): Promise { + const errors: unknown[] = []; + const children = [...this.#children.values()]; + await Promise.all( + children.map(async (tracked) => { + if (providerTreeRunning(tracked)) { + const terminateOutcome = await signalAndWaitForExit( + tracked, + "SIGTERM", + 2_000, + ); + if (terminateOutcome.error !== undefined) { + pushUnique(errors, terminateOutcome.error); + } + if (!terminateOutcome.exited && providerTreeRunning(tracked)) { + const killOutcome = await signalAndWaitForExit( + tracked, + "SIGKILL", + 2_000, + ); + if (killOutcome.error !== undefined) { + pushUnique(errors, killOutcome.error); + } + if (!killOutcome.exited && providerTreeRunning(tracked)) { + errors.push( + new Error("ACPX provider did not exit after SIGKILL"), + ); + } + } + } + if (!providerTreeRunning(tracked)) this.#forget(tracked); + }), + ); + // A failed spawn or signal can emit `error` and then `close` before this + // method snapshots the live children. Keep those errors independently of + // child membership, report each object once, and drain them only after all + // in-flight termination attempts have had a chance to emit. + for (const error of this.#errors) pushUnique(errors, error); + this.#errors.clear(); + return errors; + } + + #forget(tracked: SpawnedProviderProcess): void { + if (this.#children.get(tracked.child) !== tracked) return; + this.#children.delete(tracked.child); + tracked.child.off("error", tracked.onError); + } +} + +interface SpawnedProviderProcess { + child: ChildProcess; + processGroupId: number | null; + onError: (error: unknown) => void; +} + +function running(child: ChildProcess): boolean { + return child.exitCode === null && child.signalCode === null; +} + +function providerTreeRunning(tracked: SpawnedProviderProcess): boolean { + if (tracked.processGroupId === null) return running(tracked.child); + try { + process.kill(-tracked.processGroupId, 0); + return true; + } catch (error) { + return errorCode(error) !== "ESRCH"; + } +} + +async function signalAndWaitForExit( + tracked: SpawnedProviderProcess, + signal: NodeJS.Signals, + timeoutMs: number, +): Promise<{ exited: boolean; error?: unknown }> { + if (!providerTreeRunning(tracked)) return { exited: true }; + const { child } = tracked; + return await new Promise<{ exited: boolean; error?: unknown }>((resolve) => { + let settled = false; + const finish = (outcome: { exited: boolean; error?: unknown }) => { + if (settled) return; + settled = true; + clearTimeout(timer); + if (poll !== undefined) clearInterval(poll); + child.off("exit", onExit); + child.off("close", onExit); + child.off("error", onError); + resolve(outcome); + }; + const onExit = () => { + if (!providerTreeRunning(tracked)) finish({ exited: true }); + }; + const onError = (error: unknown) => finish({ exited: false, error }); + const timer = setTimeout( + () => finish({ exited: !providerTreeRunning(tracked) }), + timeoutMs, + ); + timer.unref(); + const poll = + tracked.processGroupId === null + ? undefined + : setInterval(() => { + if (!providerTreeRunning(tracked)) finish({ exited: true }); + }, 25); + poll?.unref(); + child.once("exit", onExit); + child.once("close", onExit); + child.once("error", onError); + if (!providerTreeRunning(tracked)) { + finish({ exited: true }); + return; + } + try { + if (tracked.processGroupId === null) { + if (!child.kill(signal) && providerTreeRunning(tracked)) { + finish({ + exited: false, + error: new Error(`ACPX provider rejected ${signal}`), + }); + return; + } + } else { + process.kill(-tracked.processGroupId, signal); + } + if (!providerTreeRunning(tracked)) finish({ exited: true }); + } catch (error) { + if (errorCode(error) === "ESRCH" && !providerTreeRunning(tracked)) { + finish({ exited: true }); + } else { + finish({ exited: false, error }); + } + } + }); +} + +function errorCode(error: unknown): string | undefined { + if (typeof error !== "object" || error === null || !("code" in error)) { + return undefined; + } + return typeof error.code === "string" ? error.code : undefined; +} + +function pushUnique(errors: unknown[], error: unknown): void { + if (!errors.includes(error)) errors.push(error); +} + +function requireIdentity(handle: AcpRuntimeHandle): AcpxRuntimePortIdentity { + const identity = { + acpxRecordId: handle.acpxRecordId, + backendSessionId: handle.backendSessionId, + agentSessionId: handle.agentSessionId, + }; + for (const [name, value] of Object.entries(identity)) { + if (typeof value !== "string" || value.length === 0) { + throw new Error(`ACPX runtime omitted ${name}`); + } + } + return identity as AcpxRuntimePortIdentity; +} + +function definedEnvironment( + environment: Readonly, +): Record { + return Object.fromEntries( + Object.entries(environment).filter( + (entry): entry is [string, string] => entry[1] !== undefined, + ), + ); +}