From 93f82b6b2ea9242179ced5dbdfe00a9bcd6939c3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 3 Sep 2026 05:43:19 -0500 Subject: [PATCH] fix(runner): admit recovered ACPX provider ownership --- .../acpx/codex-runtime-adapter.test.ts | 55 +++++++++ .../src/drivers/acpx/codex-runtime-adapter.ts | 107 +++++++++++++++--- 2 files changed, 145 insertions(+), 17 deletions(-) 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 index 721cf47547..55964b46cc 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts @@ -1304,6 +1304,61 @@ describe("Codex ACPX runtime adapter", () => { }); }); + it("admits a verified provider that starts with the first recovered turn", async () => { + const runtime = fakeRuntime(); + const child = fakeChild(); + const command = fakeCommand(); + vi.mocked(command.spawn).mockReturnValue(child); + let runtimeOptions: AcpRuntimeOptions | undefined; + let resolvePromptStarted: (() => void) | undefined; + const promptStarted = new Promise((resolve) => { + resolvePromptStarted = resolve; + }); + const rawTurn = { + requestId: "turn-recovered", + promptStarted, + events: { async *[Symbol.asyncIterator]() {} }, + result: new Promise(() => undefined), + cancel: vi.fn(), + closeStream: vi.fn(), + }; + vi.mocked(runtime.startTurn).mockImplementation(() => { + queueMicrotask(() => { + runtimeOptions?.spawnAgent?.({ + command: "ignored", + args: ["--stdio"], + options: {}, + }); + resolvePromptStarted?.(); + }); + return rawTurn; + }); + const port = await openCodexAcpxRuntime(openOptions(command), { + createRegistry: () => registry(), + createStore: () => store(), + awaitProviderOwnership: providerOwnershipEstablished, + awaitProviderExit: providerOwnershipEstablished, + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + }); + + const turn = port.startTurn({ + text: "Resume the task.", + requestId: "turn-recovered", + }); + await expect(turn.promptStarted).resolves.toBeUndefined(); + expect(command.spawn).toHaveBeenCalledTimes(1); + expect(() => + runtimeOptions?.spawnAgent?.({ + command: "ignored", + args: ["--stdio"], + options: {}, + }), + ).toThrow("provider spawned after ownership admission was sealed"); + }); + it("projects only ephemeral MCP bindings and applies fail-closed permissions", async () => { const runtime = fakeRuntime(); let runtimeOptions: AcpRuntimeOptions | undefined; diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts index a3049f2045..13c122bcc2 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts @@ -17,6 +17,7 @@ import type { AcpxRuntimePort, AcpxRuntimePortIdentity, AcpxRuntimePortOpenOptions, + AcpxRuntimeTurn, } from "./runtime-host.js"; import { assertVerifiedAcpxProviderPlatform, @@ -1045,20 +1046,71 @@ function runtimePort( } : {}), startTurn(input) { - return runtime.startTurn({ - handle, - text: input.text, - mode: "prompt", - requestId: input.requestId, - ...(input.signal ? { signal: input.signal } : {}), - ...(input.onElicitation ? { onElicitation: input.onElicitation } : {}), - }); + const finishOwnershipAdmission = + children.beginLifetimeOwnershipAdmission(); + let turn: AcpxRuntimeTurn; + try { + turn = runtime.startTurn({ + handle, + text: input.text, + mode: "prompt", + requestId: input.requestId, + ...(input.signal ? { signal: input.signal } : {}), + ...(input.onElicitation + ? { onElicitation: input.onElicitation } + : {}), + }); + } catch (error) { + void finishOwnershipAdmission().catch(() => undefined); + throw error; + } + return turnWithVerifiedLifetimeOwnership(turn, finishOwnershipAdmission); }, close: closeRuntime, }; return port; } +function turnWithVerifiedLifetimeOwnership( + turn: AcpxRuntimeTurn, + finishOwnershipAdmission: () => Promise, +): AcpxRuntimeTurn { + // A persisted ACPX session can be loaded without starting an agent process. + // Keep the narrowly scoped turn admission open until either the provider has + // accepted the prompt or the turn has already terminalized. The synchronous + // stable-empty seal in SpawnedChildSet then rejects every later spawn. + const reachedAdmissionBoundary = Promise.race([ + turn.promptStarted.then( + () => undefined, + () => undefined, + ), + turn.result.then( + () => undefined, + () => undefined, + ), + ]); + const ownershipVerified = reachedAdmissionBoundary.then(() => + finishOwnershipAdmission(), + ); + void ownershipVerified.catch(() => undefined); + return { + requestId: turn.requestId, + promptStarted: ownershipVerified.then(() => turn.promptStarted), + events: eventsAfterLifetimeOwnership(turn.events, ownershipVerified), + result: ownershipVerified.then(() => turn.result), + cancel: (input) => turn.cancel(input), + closeStream: (input) => turn.closeStream(input), + }; +} + +async function* eventsAfterLifetimeOwnership( + events: AsyncIterable, + ownershipVerified: Promise, +): AsyncIterable { + await ownershipVerified; + yield* events; +} + async function persistedRuntimeStatus( sessionStore: AcpSessionStore, handle: AcpRuntimeHandle, @@ -1304,19 +1356,40 @@ class SpawnedChildSet { } async verifyLifetimeOwnership(): Promise { - for (;;) { - const ownership = this.#lifetimeOwnership.splice(0); - if (ownership.length === 0) { - // This check and seal are synchronous. Any spawn added while an - // earlier batch was pending is observed by the next loop iteration; - // no later provider can race admission after the stable-empty point. - this.#lifetimeOwnershipSealed = true; - return; + try { + for (;;) { + const ownership = this.#lifetimeOwnership.splice(0); + if (ownership.length === 0) { + // This check and seal are synchronous. Any spawn added while an + // earlier batch was pending is observed by the next loop iteration; + // no later provider can race admission after the stable-empty point. + this.#lifetimeOwnershipSealed = true; + return; + } + await Promise.all(ownership); } - await Promise.all(ownership); + } catch (error) { + this.#lifetimeOwnershipSealed = true; + throw error; } } + beginLifetimeOwnershipAdmission(): () => Promise { + if (this.#sealed) { + throw new Error("ACPX provider ownership admission is closed"); + } + if (!this.#lifetimeOwnershipSealed) { + throw new Error("ACPX provider ownership admission is already active"); + } + this.#lifetimeOwnershipSealed = false; + let finished = false; + return async () => { + if (finished) return; + finished = true; + await this.verifyLifetimeOwnership(); + }; + } + #track(child: ChildProcess, providerExit: ProviderExitObservation): void { this.#children.add(child); const onError = (error: unknown) => this.#errors.add(error);