feat(runner): bridge Codex ACPX questions (#12408)
## 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
This commit is contained in:
Dotta
GitHub
parent
96421b0663
commit
4fe3189f02
|
|
@ -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<void>();
|
||||
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<void>;
|
||||
} = {},
|
||||
): {
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -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<string, string>();
|
||||
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<string, unknown>;
|
||||
} | 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<void> {
|
||||
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<unknown> {
|
||||
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<void> {
|
||||
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<AcpElicitationResponse> {
|
||||
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<AcpElicitationResponse>((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<string, unknown>,
|
||||
|
|
@ -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<string, unknown> {
|
||||
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" ||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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" }),
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue