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, + ), + ); +}