From 4fe3189f0256873a359d2d53c209076919fd1c3b Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 30 Aug 2026 21:48:14 -0500 Subject: [PATCH] feat(runner): bridge Codex ACPX questions (#12408) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The runner package gives provider sessions one normalized execution contract. > - The merged Codex ACPX path can run and recover settled turns. > - It cannot yet send a structured provider question through the existing Paperclip question boundary. > - Provider questions must not expose ACPX-specific data to later server integrations. > - Every pending provider request must also settle on resolution, handoff, cancellation, failure, or close. > - This pull request adds a bounded Codex ACPX form bridge inside the runner package. > - The benefit is a provider-neutral question flow with fail-closed lifecycle handling. ## Linked Issues or Issue Description Refs #12407 This pull request builds on the settled Codex ACPX recovery path merged in #12407. It adds only package-local structured-question support for Codex ACPX sessions. ## What Changed - Enable ACPX form elicitation for the Codex runtime and pass its handler through the runtime host boundary. - Normalize ACPX forms to `paperclip.question_set.v1` and emit `paperclip.runtime_request.v2` events. - Validate `paperclip.question_response.v1` resolutions before conversion to ACP form responses. - Support explicit resolution and durable handoff. Cancel pending requests on provider abort, stream failure, turn settlement, and session close. - Limit each session to 16 pending requests. Reject unsupported input modes, regular expression patterns, session-wide acceptance, stale turns, and late responses. - Fail closed under bounded event-queue pressure: cancel an input whose creation event cannot be retained, and preserve a live request when its durable-handoff event cannot be retained. - Persist pending-request facts in snapshots and reject recovery while a provider request is still pending. - Add focused driver, runtime-adapter, and runtime-host regression tests for round trips, aborts, stream failures, handoff, and handler forwarding. ## Verification - Replay base: `96421b0663d8b740ac5d5d53359aef65c5a158ca` (`master` after #12407 merged). - Exact replay head: `d8184502e1c4570d1003379365850f3419f64d82`. - Stable patch ID for the exact replay delta: `00807ebfcd1b153f759366db023437b681d69017`. - The pull request delta contains exactly these six files: - `packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts` - `packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts` - `packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts` - `packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts` - `packages/paperclip-runner/src/drivers/acpx/runtime-host.ts` - `packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts` - The exact combined delta is 716 additions and 14 deletions. - This change does not add a dependency, lockfile update, migration, workflow, server route, UI change, documentation file, or public package export. - GitHub Actions run `33351323368` passed the full matrix on the exact replay head, including Paperclip Runner verification, build, typecheck/release-registry, general and serialized server suites, canary, and all e2e shards. - Superagent, Socket, Snyk, contributor-trust, policy, and PR-review checks pass on the exact replay head. - Greptile reviewed the exact replay head at 5/5 with no blocking finding and zero review threads. - No local test result is claimed. GitHub Actions is the authoritative verification environment for the replayed revision. ## Risks This change has medium package-local risk because it connects provider-owned input to durable runner state. Unsupported modes and unbounded patterns fail closed. The pending-request limit bounds retained provider state. Provider abort, turn settlement, stream failure, and session close cancel live questions. A durable handoff expires the request before it interrupts the turn. Recovery rejects a snapshot that still contains a pending provider request. Existing direct adapters, server selection, and task-page behavior do not use this route. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used OpenAI Codex with GPT-5.6, extended reasoning, repository tool use, and code execution. ## 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 (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [ ] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --- .../drivers/acpx/codex-acpx-driver.test.ts | 347 ++++++++++++++++- .../src/drivers/acpx/codex-acpx-driver.ts | 357 +++++++++++++++++- .../acpx/codex-runtime-adapter.test.ts | 4 + .../src/drivers/acpx/codex-runtime-adapter.ts | 4 + .../src/drivers/acpx/runtime-host.test.ts | 8 +- .../src/drivers/acpx/runtime-host.ts | 10 +- 6 files changed, 716 insertions(+), 14 deletions(-) diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts index 0069921f40..218b702664 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts @@ -14,6 +14,7 @@ import { type CodexAcpxDriverOptions, } from "./codex-acpx-driver.js"; import type { + AcpxRuntimeTurnInput, AcpxRuntimeTurn, OpenAcpxRuntimeHostOptions, } from "./runtime-host.js"; @@ -140,7 +141,8 @@ describe("Codex ACPX harness driver", () => { resume: true, interruption: true, dynamicTools: true, - runtimeRequestResolution: false, + runtimeRequestResolution: true, + runtimeRequestHandoff: true, }, runtimeContextCapabilities: { instructions: "native", @@ -1298,6 +1300,342 @@ describe("Codex ACPX harness driver", () => { }); }); + it("round-trips a provider-neutral ACP form through the runtime request boundary", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-question", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const createdEvent = collectUntil( + session.events(), + "runtime_request.created", + ); + const { turnId } = await session.startTurn({ + message: { role: "user", text: "Choose a region." }, + }); + const onElicitation = + fixture.host.startTurn.mock.calls[0]![0].onElicitation!; + const controller = new AbortController(); + const providerResponse = onElicitation( + { + mode: "form", + message: "Choose deployment settings.", + requestedSchema: { + type: "object", + title: "Deployment", + required: ["region"], + properties: { + region: { + type: "string", + title: "Region", + enum: ["us-east-1", "eu-west-1"], + }, + }, + }, + }, + { requestId: "rpc-question-1", signal: controller.signal }, + ); + const events = await createdEvent; + const request = session.pendingRuntimeRequests!()[0]!; + + expect(events.at(-1)).toMatchObject({ + eventType: "runtime_request.created", + turnId, + payload: { + request: { + schema: "paperclip.runtime_request.v2", + requestKind: "runtime", + type: "input", + status: "pending", + input: { schema: "paperclip.question_set.v1" }, + origin: { adapter: "acpx-runtime", provider: "codex" }, + }, + }, + }); + const question = request.input!.questions[0]!; + const resolvedEvent = collectUntil( + session.events(), + "runtime_request.resolved", + ); + await session.resolveRuntimeRequest!({ + requestId: request.requestId, + turnId, + resolution: { + action: "submit", + response: { + schema: "paperclip.question_response.v1", + answers: { + [question.id]: { + selectedOptionIds: [question.options![1]!.id], + }, + }, + }, + }, + }); + + await expect(providerResponse).resolves.toEqual({ + action: "accept", + content: { region: "eu-west-1" }, + }); + await expect(resolvedEvent).resolves.toEqual( + expect.arrayContaining([ + expect.objectContaining({ eventType: "runtime_request.resolved" }), + ]), + ); + expect(session.pendingRuntimeRequests!()).toEqual([]); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await collectUntil(session.events(), "turn.completed"); + await session.close({ reason: "question verified" }); + }); + + it("rejects provider input when queue pressure omits its creation event", async () => { + const fixture = driverFixture( + {}, + { maxBufferedEvents: 6, runtimeEvents: [] }, + ); + const session = await fixture.driver.openSession({ + runId: "run-question-created-pressure", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + await session.startTurn({ + message: { role: "user", text: "Fill the event queue." }, + }); + const onElicitation = + fixture.host.startTurn.mock.calls[0]![0].onElicitation!; + + await expect( + onElicitation( + { + mode: "form", + requestedSchema: { + type: "object", + properties: { value: { type: "string" } }, + }, + }, + { + requestId: "rpc-question-created-pressure", + signal: new AbortController().signal, + }, + ), + ).resolves.toEqual({ action: "cancel" }); + expect(session.pendingRuntimeRequests!()).toEqual([]); + + await session.close({ reason: "creation pressure verified" }); + }); + + it("cancels a pending ACP form when its owning provider request aborts", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-question-abort", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const createdEvent = collectUntil( + session.events(), + "runtime_request.created", + ); + await session.startTurn({ + message: { role: "user", text: "Ask and abort." }, + }); + const onElicitation = + fixture.host.startTurn.mock.calls[0]![0].onElicitation!; + const controller = new AbortController(); + const providerResponse = onElicitation( + { + mode: "form", + requestedSchema: { + type: "object", + properties: { value: { type: "string" } }, + }, + }, + { requestId: "rpc-question-abort", signal: controller.signal }, + ); + await createdEvent; + const cancelledEvent = collectUntil( + session.events(), + "runtime_request.cancelled", + ); + + controller.abort(); + + await expect(providerResponse).resolves.toEqual({ action: "cancel" }); + await expect(cancelledEvent).resolves.toEqual( + expect.arrayContaining([ + expect.objectContaining({ eventType: "runtime_request.cancelled" }), + ]), + ); + expect(session.pendingRuntimeRequests!()).toEqual([]); + fixture.finishTurn({ status: "cancelled", stopReason: "aborted" }); + await collectUntil(session.events(), "turn.interrupted"); + await session.close({ reason: "abort verified" }); + }); + + it("cancels a pending ACP form when the provider event stream fails", async () => { + const eventStreamFailure = deferred(); + const fixture = driverFixture( + {}, + { eventStreamFailure: eventStreamFailure.promise }, + ); + const session = await fixture.driver.openSession({ + runId: "run-question-stream-failure", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const failedEvent = collectUntil(session.events(), "turn.failed"); + await session.startTurn({ + message: { role: "user", text: "Ask before the stream fails." }, + }); + const onElicitation = + fixture.host.startTurn.mock.calls[0]![0].onElicitation!; + const providerResponse = onElicitation( + { + mode: "form", + requestedSchema: { + type: "object", + properties: { value: { type: "string" } }, + }, + }, + { + requestId: "rpc-question-stream-failure", + signal: new AbortController().signal, + }, + ); + await vi.waitFor(() => { + expect(session.pendingRuntimeRequests!()).toHaveLength(1); + }); + + eventStreamFailure.resolve(); + + await expect(providerResponse).resolves.toEqual({ action: "cancel" }); + await expect(failedEvent).resolves.toEqual( + expect.arrayContaining([ + expect.objectContaining({ eventType: "runtime_request.cancelled" }), + expect.objectContaining({ eventType: "turn.failed" }), + ]), + ); + expect(session.pendingRuntimeRequests!()).toEqual([]); + await session.close({ reason: "stream failure verified" }); + }); + + it("expires an ACP form before a durable wait without accepting late answers", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-question-handoff", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const createdEvent = collectUntil( + session.events(), + "runtime_request.created", + ); + const { turnId } = await session.startTurn({ + message: { role: "user", text: "Ask for input." }, + }); + const onElicitation = + fixture.host.startTurn.mock.calls[0]![0].onElicitation!; + const providerResponse = onElicitation( + { + mode: "form", + requestedSchema: { + type: "object", + properties: { value: { type: "string" } }, + }, + }, + { + requestId: "rpc-question-handoff", + signal: new AbortController().signal, + }, + ); + await createdEvent; + const [request] = session.pendingRuntimeRequests!(); + + const ownership = new AbortController(); + ownership.abort(); + const abortedHandoff = session.handoffRuntimeRequest!({ + requestId: request!.requestId, + turnId, + reason: "durable_handoff", + signal: ownership.signal, + }); + expect(abortedHandoff.result).toBe("already_settled"); + await expect(abortedHandoff.cleanup).resolves.toBeUndefined(); + expect(session.pendingRuntimeRequests!()).toHaveLength(1); + expect(fixture.host.interruptActiveTurn).not.toHaveBeenCalled(); + + const handoff = session.handoffRuntimeRequest!({ + requestId: request!.requestId, + turnId, + reason: "durable_handoff", + signal: new AbortController().signal, + }); + expect(handoff.result).toBe("handed_off"); + await expect(handoff.cleanup).resolves.toBeUndefined(); + await expect(providerResponse).resolves.toEqual({ action: "cancel" }); + expect(fixture.host.interruptActiveTurn).toHaveBeenCalledWith( + "Paperclip parked the ACPX input on a durable wait.", + ); + await expect( + session.resolveRuntimeRequest!({ + requestId: request!.requestId, + turnId, + resolution: { action: "cancel" }, + }), + ).rejects.toThrow("no longer pending"); + fixture.finishTurn({ status: "cancelled", stopReason: "durable_wait" }); + await collectUntil(session.events(), "turn.interrupted"); + await session.close({ reason: "handoff verified" }); + }); + + it("preserves a pending ACP form when queue pressure blocks durable handoff", async () => { + const fixture = driverFixture( + {}, + { maxBufferedEvents: 7, runtimeEvents: [] }, + ); + const session = await fixture.driver.openSession({ + runId: "run-question-handoff-pressure", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const { turnId } = await session.startTurn({ + message: { role: "user", text: "Fill the handoff lane." }, + }); + const onElicitation = + fixture.host.startTurn.mock.calls[0]![0].onElicitation!; + const providerResponse = onElicitation( + { + mode: "form", + requestedSchema: { + type: "object", + properties: { value: { type: "string" } }, + }, + }, + { + requestId: "rpc-question-handoff-pressure", + signal: new AbortController().signal, + }, + ); + await vi.waitFor(() => { + expect(session.pendingRuntimeRequests!()).toHaveLength(1); + }); + const [request] = session.pendingRuntimeRequests!(); + + expect(() => + session.handoffRuntimeRequest!({ + requestId: request!.requestId, + turnId, + reason: "durable_handoff", + signal: new AbortController().signal, + }), + ).toThrow("event consumer must drain provider events"); + expect(session.pendingRuntimeRequests!()).toEqual([request]); + expect(fixture.host.interruptActiveTurn).not.toHaveBeenCalled(); + + await session.close({ reason: "handoff pressure verified" }); + await expect(providerResponse).resolves.toEqual({ action: "cancel" }); + }); + it("recovers a settled session with the exact persisted identity", async () => { const fixture = driverFixture(); const session = await fixture.driver.openSession({ @@ -2054,6 +2392,7 @@ function driverFixture( readRecoveryWorkspace?: NonNullable< CodexAcpxDriverDependencies["readRecoveryWorkspace"] >; + eventStreamFailure?: Promise; } = {}, ): { driver: CodexAcpxDriver; @@ -2091,6 +2430,10 @@ function driverFixture( if (fixtureOptions.runtimeEventFailure) { await fixtureOptions.runtimeEventFailure; } + if (fixtureOptions.eventStreamFailure) { + await fixtureOptions.eventStreamFailure; + throw new Error("provider event stream failed"); + } }, }, result: activeResult.promise, @@ -2196,7 +2539,7 @@ function fakeHost(createTurn: () => AcpxRuntimeTurn, onClose: () => void) { availableModelIds: ["gpt-5.6-sol"], }, })), - startTurn: vi.fn(createTurn), + startTurn: vi.fn((_input: AcpxRuntimeTurnInput) => createTurn()), interruptActiveTurn: vi.fn(async () => undefined), close: vi.fn(async () => { onClose(); diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts index 8b1b965543..bd5f96248b 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts @@ -1,6 +1,11 @@ import { createHash, randomBytes } from "node:crypto"; -import type { AcpRuntimeEvent } from "acpx/runtime"; +import type { + AcpElicitationContext, + AcpElicitationRequest, + AcpElicitationResponse, + AcpRuntimeEvent, +} from "acpx/runtime"; import { PRP_BLOCK_TOOL_NAME, @@ -8,17 +13,25 @@ import { } from "../../contracts/completion-result.js"; import { HarnessCapabilityUnavailableError, + HarnessRuntimeRequestResolutionError, HarnessStaleTurnError, + harnessRuntimeInputExpiredOutcome, + harnessRuntimeRequestOutcome, + parseHarnessRuntimeRequestResolution, type HarnessDriver, type HarnessDriverConfigValidation, type HarnessDriverDescriptor, type HarnessSession, type HarnessSessionRecoveryOptions, type HarnessSessionRecoveryResult, + type HarnessRuntimeRequest, + type HarnessRuntimeRequestHandoff, + type HarnessRuntimeRequestResolution, type HarnessTranscriptSnapshot, type OpenHarnessSessionInput, type PersistedHarnessSession, } from "../../contracts/harness-driver.js"; +import { PAPERCLIP_RUNTIME_REQUEST_SCHEMA_V2 } from "../../contracts/question-set.js"; import type { NativeAcpxPermissionMode } from "../../contracts/native-execution.js"; import type { NativeUserMessage } from "../../contracts/types.js"; import type { @@ -38,6 +51,10 @@ import { DEFAULT_CODEX_ACPX_RUNTIME_SHUTDOWN_BOUND_MS, openCodexAcpxRuntime, } from "./codex-runtime-adapter.js"; +import { + normalizeAcpFormElicitation, + type NormalizedAcpForm, +} from "./acp-question-adapter.js"; import { acpxDriverDescriptor, validateAcpxDriverConfig, @@ -60,6 +77,7 @@ const MAX_TRANSCRIPT_EVENTS = 1_024; const MAX_TRANSCRIPT_BYTES = 8 * 1024 * 1024; const MAX_RECOVERY_TERMINAL_TURNS = 4_096; const MAX_RECOVERY_TERMINAL_BYTES = 8 * 1024 * 1024; +const MAX_PENDING_RUNTIME_REQUESTS = 16; const CLOSE_TURN_SETTLEMENT_TIMEOUT_MS = 2_000; const MAX_AUTONOMOUS_HOST_CLOSE_RETRIES = 3; const MAX_QUARANTINED_HOST_CLOSE_RETRIES = 3; @@ -83,6 +101,14 @@ const QUARANTINED_HOST_ADMISSION_GRACE_MS = MAX_QUARANTINED_HOST_ATTEMPT_RETRY_DELAY_MS + 1_000; +interface PendingAcpxRuntimeRequest { + request: HarnessRuntimeRequest; + normalized: NormalizedAcpForm; + settle(response: AcpElicitationResponse): void; + cleanup(): void; + settling: boolean; +} + export interface CodexAcpxDynamicToolCall { tool: string; callId: string; @@ -201,15 +227,9 @@ export class CodexAcpxDriver implements HarnessDriver { capabilities: { ...descriptor.capabilities, resume: true, - runtimeRequestResolution: false, - runtimeRequestHandoff: false, - unsupported: [ - "steering", - "runtimeRequestResolution", - "runtimeRequestHandoff", - "goals", - "threadLineage", - ], + runtimeRequestResolution: true, + runtimeRequestHandoff: true, + unsupported: ["steering", "goals", "threadLineage"], }, }; } @@ -617,6 +637,10 @@ class CodexAcpxSession implements HarnessSession { readonly #quarantineCleanup: (host: CodexAcpxHost, reason: string) => void; readonly #transcript: Array<{ event: PrpEvent; bytes: number }> = []; readonly #terminalTurns = new Map(); + readonly #pendingRuntimeRequests = new Map< + string, + PendingAcpxRuntimeRequest + >(); readonly #sourceInstanceId: string; readonly #providerRecoveryPolicy: NonNullable< PersistedHarnessSession["providerRecoveryPolicy"] @@ -653,6 +677,7 @@ class CodexAcpxSession implements HarnessSession { eventType: "turn.completed" | "turn.failed" | "turn.interrupted"; payload: Record; } | null = null; + #runtimeRequestSequence = 0; constructor(input: { host: CodexAcpxHost; @@ -750,6 +775,8 @@ class CodexAcpxSession implements HarnessSession { turn = this.#host.startTurn({ text: input.message.text, requestId: `${safeId(this.#input.runId, "run")}:${turnId}`, + onElicitation: (request, context) => + this.#handleElicitation(turnId, request, context), }); } catch (error) { this.#publishTerminal( @@ -790,6 +817,120 @@ class CodexAcpxSession implements HarnessSession { await this.#host.interruptActiveTurn(input.reason ?? "interrupted"); } + pendingRuntimeRequests(): HarnessRuntimeRequest[] { + return [...this.#pendingRuntimeRequests.values()].map(({ request }) => + structuredClone(request), + ); + } + + async resolveRuntimeRequest(input: { + requestId: string; + turnId: string; + resolution: HarnessRuntimeRequestResolution; + }): Promise { + this.#assertOpen(); + const pending = this.#pendingRuntimeRequests.get(input.requestId); + if (!pending) { + throw new HarnessCapabilityUnavailableError( + "runtime request resolution", + `request ${input.requestId} is no longer pending`, + ); + } + if ( + pending.request.turnId !== input.turnId || + this.#activeTurnId !== input.turnId + ) { + throw new HarnessStaleTurnError(input.turnId); + } + if (pending.settling) { + throw new HarnessCapabilityUnavailableError( + "runtime request resolution", + `request ${input.requestId} is already settling`, + ); + } + pending.settling = true; + try { + const resolution = parseHarnessRuntimeRequestResolution( + pending.request.requestKind, + input.resolution, + pending.request.input, + ); + const providerResponse = acpElicitationResponse( + pending.normalized, + resolution, + ); + if (!this.#pendingRuntimeRequests.delete(input.requestId)) return; + pending.cleanup(); + this.#emit( + "runtime_request.resolved", + harnessRuntimeRequestOutcome(pending.request, { + action: resolution.action, + ...(resolution.action === "submit" && "response" in resolution + ? { response: resolution.response } + : {}), + }), + { turnId: input.turnId, itemId: pending.request.itemId }, + ); + pending.settle(providerResponse); + } catch (error) { + pending.settling = false; + throw error; + } + } + + handoffRuntimeRequest(input: { + requestId: string; + turnId: string; + reason: "durable_handoff"; + signal: AbortSignal; + }): HarnessRuntimeRequestHandoff { + if (input.signal.aborted) { + return { result: "already_settled", cleanup: Promise.resolve() }; + } + this.#assertOpen(); + const pending = this.#pendingRuntimeRequests.get(input.requestId); + if ( + !pending || + pending.request.turnId !== input.turnId || + this.#activeTurnId !== input.turnId || + pending.settling + ) { + return { result: "already_settled", cleanup: Promise.resolve() }; + } + if ( + !this.#emit( + "runtime_request.expired", + harnessRuntimeInputExpiredOutcome(pending.request, input.reason), + { turnId: input.turnId, itemId: pending.request.itemId }, + ) + ) { + throw new HarnessCapabilityUnavailableError( + "runtime request handoff", + "the event consumer must drain provider events before the durable handoff can be retained", + ); + } + if (!this.#pendingRuntimeRequests.delete(input.requestId)) { + throw new Error( + `ACPX runtime request ${input.requestId} changed during its synchronous handoff`, + ); + } + pending.cleanup(); + pending.settle({ action: "cancel" }); + const cleanup = Promise.resolve() + .then(() => this.#host.interruptActiveTurn( + "Paperclip parked the ACPX input on a durable wait.", + )) + .catch((error: unknown) => { + if ( + this.#activeTurnId === input.turnId && + !this.#terminalTurns.has(input.turnId) + ) { + throw error; + } + }); + return { result: "handed_off", cleanup }; + } + async dispatchTool(call: RunnerToolCall): Promise { this.#assertOpen(); if (this.#pendingTerminal) { @@ -948,6 +1089,7 @@ class CodexAcpxSession implements HarnessSession { fingerprint, }), ), + pendingRuntimeRequests: this.pendingRuntimeRequests(), }; } @@ -971,6 +1113,7 @@ class CodexAcpxSession implements HarnessSession { async #finishClose(reason: string): Promise { const closingTurnId = this.#activeTurnId; const pump = this.#activePump; + this.#cancelPendingRuntimeRequests(reason); const hostClose = this.#hostClosePromise ?? this.#startHostClose({ reason }); let hostCloseError: unknown = null; @@ -1099,6 +1242,7 @@ class CodexAcpxSession implements HarnessSession { this.#mapRuntimeEvent(normalizeToolEvent(event), turnId, ++index); } const result = await turn.result; + this.#cancelPendingRuntimeRequests("provider turn settled", turnId); if (this.#terminalTurns.has(turnId)) return; if (result.status === "completed") { const completedSemanticFingerprint = @@ -1183,6 +1327,7 @@ class CodexAcpxSession implements HarnessSession { } catch (error) { if (this.#terminalTurns.has(turnId)) return; if (error instanceof TerminalEventCapacityError) throw error; + this.#cancelPendingRuntimeRequests("provider turn failed", turnId); if (this.#closed || this.#closingStarted) { const reaffirmedSemanticResult = this.#pendingSemanticTransfer?.turnId === turnId @@ -1312,6 +1457,156 @@ class CodexAcpxSession implements HarnessSession { } } + async #handleElicitation( + turnId: string, + request: AcpElicitationRequest, + context: AcpElicitationContext, + ): Promise { + if ( + this.#closed || + this.#activeTurnId !== turnId || + context.signal.aborted + ) { + return { action: "cancel" }; + } + if (this.#pendingRuntimeRequests.size >= MAX_PENDING_RUNTIME_REQUESTS) { + this.#emit( + "harness.diagnostic", + { + code: "runtime_input_limit_reached", + adapter: "acpx-runtime", + reason: "The active ACPX turn has too many pending input requests.", + }, + { turnId }, + ); + return { action: "cancel" }; + } + let normalized: NormalizedAcpForm | null; + try { + normalized = normalizeAcpFormElicitation(request); + } catch (error) { + this.#emit( + "harness.diagnostic", + { + code: "runtime_input_rejected", + adapter: "acpx-runtime", + reason: safeMessage(error), + }, + { turnId }, + ); + return { action: "cancel" }; + } + if (!normalized) { + this.#emit( + "harness.diagnostic", + { + code: "runtime_input_unsupported", + adapter: "acpx-runtime", + reason: "The ACPX provider requested an unsupported input mode.", + }, + { turnId }, + ); + return { action: "cancel" }; + } + if ( + normalized.questionSet.questions.some( + (question) => question.textValidation?.pattern !== undefined, + ) + ) { + this.#emit( + "harness.diagnostic", + { + code: "runtime_input_pattern_unsupported", + adapter: "acpx-runtime", + reason: + "ACPX form patterns require a bounded regular expression dialect.", + }, + { turnId }, + ); + return { action: "cancel" }; + } + const requestId = stableId( + "acpx-request", + `${turnId}:${++this.#runtimeRequestSequence}:${typeof context.requestId}:${String(context.requestId)}`, + ); + const runtimeRequest: HarnessRuntimeRequest = { + requestId, + requestKind: "elicitation", + method: "elicitation/create", + turnId, + itemId: requestId, + status: "pending", + prompt: boundedText( + normalized.questionSet.title ?? + normalized.questionSet.description ?? + "Additional information is required.", + 1_000, + ), + details: { mode: "form" }, + input: structuredClone(normalized.questionSet), + origin: { + adapter: "acpx-runtime", + provider: "codex", + method: "elicitation/create", + }, + }; + if ( + !this.#emit( + "runtime_request.created", + { request: runtimeInputProtocolPayload(runtimeRequest) }, + { turnId, itemId: requestId }, + ) + ) { + return { action: "cancel" }; + } + return await new Promise((settle) => { + const cancel = () => { + const pending = this.#pendingRuntimeRequests.get(requestId); + if (!pending || pending.settling) return; + if (!this.#pendingRuntimeRequests.delete(requestId)) return; + pending.cleanup(); + this.#emit( + "runtime_request.cancelled", + harnessRuntimeRequestOutcome(runtimeRequest, { + action: "cancel", + reason: "provider request aborted", + }), + { turnId, itemId: requestId }, + ); + settle({ action: "cancel" }); + }; + context.signal.addEventListener("abort", cancel, { once: true }); + this.#pendingRuntimeRequests.set(requestId, { + request: runtimeRequest, + normalized, + settle, + cleanup: () => context.signal.removeEventListener("abort", cancel), + settling: false, + }); + if (context.signal.aborted) cancel(); + }); + } + + #cancelPendingRuntimeRequests(reason: string, turnId?: string): void { + for (const [requestId, pending] of this.#pendingRuntimeRequests) { + if (turnId && pending.request.turnId !== turnId) continue; + if (!this.#pendingRuntimeRequests.delete(requestId)) continue; + pending.cleanup(); + this.#emit( + "runtime_request.cancelled", + harnessRuntimeRequestOutcome(pending.request, { + action: "cancel", + reason: boundedText(safeMessage(reason), 1_000), + }), + { + turnId: pending.request.turnId, + itemId: pending.request.itemId, + }, + ); + pending.settle({ action: "cancel" }); + } + } + #emit( eventType: PrpEvent["eventType"], payload: Record, @@ -1417,6 +1712,48 @@ function canonicalJson(value: unknown): string { return JSON.stringify(value) ?? "undefined"; } +function acpElicitationResponse( + normalized: NormalizedAcpForm, + resolution: HarnessRuntimeRequestResolution, +): AcpElicitationResponse { + if (resolution.action === "submit") { + if (!("response" in resolution)) { + throw new HarnessRuntimeRequestResolutionError( + "elicitation", + "ACPX form submissions require a canonical question response", + ); + } + return normalized.accept(resolution.response); + } + if (resolution.action === "accept_for_session") { + throw new HarnessRuntimeRequestResolutionError( + "elicitation", + "ACPX form input does not support session acceptance", + ); + } + return { action: resolution.action }; +} + +function runtimeInputProtocolPayload( + request: HarnessRuntimeRequest, +): Record { + if (!request.input) { + throw new Error("ACPX runtime input request omitted its question set"); + } + return { + schema: PAPERCLIP_RUNTIME_REQUEST_SCHEMA_V2, + requestKind: "runtime", + requestId: request.requestId, + type: "input", + status: request.status, + prompt: request.prompt, + input: structuredClone(request.input), + origin: structuredClone(request.origin), + turnId: request.turnId, + itemId: request.itemId, + }; +} + function validateRecoverySnapshot(snapshot: PersistedHarnessSession): void { if ( snapshot.driverKind !== "acpx_runtime" || 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 ab310259a7..6c900f2951 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 @@ -120,6 +120,7 @@ describe("Codex ACPX runtime adapter", () => { OPENAI_API_KEY: "credential-secret", }); expect(runtimeOptions?.spawnCwd).toBe("/workspace"); + expect(runtimeOptions?.elicitationModes).toEqual(["form"]); expect(await port.identity()).toEqual({ acpxRecordId: "record-1", backendSessionId: "backend-1", @@ -1106,12 +1107,14 @@ describe("Codex ACPX runtime adapter", () => { createRuntime: () => runtime, }); const signal = new AbortController().signal; + const onElicitation = vi.fn(); expect( port.startTurn({ text: "Complete the task.", requestId: "turn-1", signal, + onElicitation, }), ).toBe(turn); expect(runtime.startTurn).toHaveBeenCalledWith({ @@ -1120,6 +1123,7 @@ describe("Codex ACPX runtime adapter", () => { mode: "prompt", requestId: "turn-1", signal, + onElicitation, }); }); 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 e17903aab4..df2aa93337 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts @@ -179,6 +179,7 @@ export async function openCodexAcpxRuntime( overrides: { codex: [VERIFIED_COMMAND_SENTINEL] }, }), permissionMode: options.permissionMode, + elicitationModes: ["form"], nonInteractivePermissions: "fail", permissionPolicy: { ...options.permissionPolicy, @@ -900,6 +901,9 @@ function runtimePort( mode: "prompt", requestId: input.requestId, ...(input.signal ? { signal: input.signal } : {}), + ...(input.onElicitation + ? { onElicitation: input.onElicitation } + : {}), }); }, close: closeRuntime, diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts index a42074920f..17baf4c64d 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts @@ -633,6 +633,7 @@ describe("ACPX runtime host", () => { const fixture = await hostFixture(); const turn = runtimeTurn(); const startTurn = vi.fn(() => turn); + const onElicitation = vi.fn(); const runtime = runtimePort({ startTurn }); const host = await AcpxRuntimeHost.open( { @@ -646,11 +647,16 @@ describe("ACPX runtime host", () => { ); expect( - host.startTurn({ text: "Complete the task.", requestId: "turn-1" }), + host.startTurn({ + text: "Complete the task.", + requestId: "turn-1", + onElicitation, + }), ).toBe(turn); expect(startTurn).toHaveBeenCalledWith({ text: "Complete the task.", requestId: "turn-1", + onElicitation, }); expect(() => host.startTurn({ text: "Concurrent", requestId: "turn-2" }), diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts index 44d38d4249..070e861627 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts @@ -1,4 +1,8 @@ -import type { AcpRuntimeEvent, AcpRuntimeTurnResult } from "acpx/runtime"; +import type { + AcpElicitationHandler, + AcpRuntimeEvent, + AcpRuntimeTurnResult, +} from "acpx/runtime"; import type { NativeAcpxPermissionMode } from "../../contracts/native-execution.js"; import { @@ -54,6 +58,7 @@ export interface AcpxRuntimeTurnInput { text: string; requestId: string; signal?: AbortSignal; + onElicitation?: AcpElicitationHandler; } export interface AcpxRuntimeTurn { @@ -463,6 +468,9 @@ export class AcpxRuntimeHost { text, requestId, ...(input.signal ? { signal: input.signal } : {}), + ...(input.onElicitation + ? { onElicitation: input.onElicitation } + : {}), }); this.#activeTurn = turn; void turn.result