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