From 1a941e85e57a9c01569a848b12d7bdffaf6a3bc1 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 30 Sep 2026 21:55:19 -0500 Subject: [PATCH] fix(runner): preserve event order before native input dispatch Queue bounded internal request callbacks alongside notifications so the Codex driver maps earlier tool activity before exposing permission cards. Dispatch without awaiting human answers and discard stale callbacks after closure, replacement, or durable settlement. Co-Authored-By: Paperclip --- .../runnerd-acpx-permission-ordering.test.ts | 208 ++++++++++++++++++ .../src/live/runnerd-codex-transport.ts | 95 ++++---- 2 files changed, 266 insertions(+), 37 deletions(-) create mode 100644 packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts diff --git a/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts b/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts new file mode 100644 index 00000000000..3d48678b09f --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts @@ -0,0 +1,208 @@ +import { mkdtempSync, writeFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, it, vi } from "vitest"; +import type { DurableRecoveryCommittedEvent, DurableRecoveryIdentity } from "../contracts/durable-recovery.js"; + +// Replace only the PRP peer and process boundary. The production transport, +// event pump, registered handler, and request emission below all run unchanged. +const peer = vi.hoisted(() => ({ current: null as TestPeer | null })); +interface TestPeer { + store: { state: { identity: DurableRecoveryIdentity; committedEvents: DurableRecoveryCommittedEvent[]; commands: Array> } }; + emit(type: string, payload: Record): void; +} +vi.mock("../control-plane/durable-prp-control-plane.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + spawnRunner: vi.fn(() => ({ child: { exitCode: null }, completion: new Promise(() => {}) })), + DurablePrpControlPlane: class implements TestPeer { + store: TestPeer["store"]; + connectUrl = "ws://fixture.invalid"; + constructor(options: { identity: DurableRecoveryIdentity }) { + this.store = { state: { identity: options.identity, committedEvents: [], commands: [] } }; + peer.current = this; + } + async start() {} + async stop() {} + disconnectActiveRunner() {} + persistRunAttachTemplate() {} + issueBootstrapTicket() { return {}; } + getCommand(id: string) { return this.store.state.commands.find(c => c.commandId === id); } + queueCommand(type: string, payload: Record, id = type) { + this.store.state.commands.push({ type, commandId: id, status: "completed", result: { result: { providerTurnId: payload.turnId } } }); + if (type === "session.open") this.emit("session.started", { + threadId: "thread-1", sessionId: "provider-session-1", + runtimeIdentity: { executionKind: "remote_service" }, + }); + if (type === "turn.start") this.emit("turn.started", { providerTurnId: payload.turnId }); + } + emit(eventType: string, payload: Record) { + const sourceSeq = this.store.state.committedEvents.length + 1; + this.store.state.committedEvents.push({ + eventType, sourceSeq, sourceEventId: `fixture:${sourceSeq}`, + priority: 0, deliveryCount: 1, logicalEffectCount: 1, + envelope: { turnId: this.store.state.identity.turnId, payload: { payload } }, + }); + } + }, + }; +}); +import { createCapabilityRunnerdCodexTransport } from "./runnerd-codex-transport.js"; +import { makeDriver, WORKSPACE } from "../drivers/codex/codex-app-server-driver.test-support.js"; +import type { PrpEvent } from "../protocol/replay-contract.js"; + + +describe("permission and tool order through the actual transport and harness", () => { + it.each(["tool-first", "permission-first"])("preserves committed batch order with an unanswered permission: %s", async order => { + const root = mkdtempSync(join(tmpdir(), "permission-ordering-")); + const binary = join(root, "unexecuted-runner"); + writeFileSync(binary, "not executable: process boundary is mocked"); + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: binary, stateDirectory: root, + sourceCodexHome: join(root, "absent-source-home"), + environment: { HOME: root, PATH: "/usr/bin:/bin" }, + }); + const driver = makeDriver([], { transportFactory: () => bundle.transport }); + const seen: PrpEvent[] = []; + let session: Awaited> | undefined; + let drain: Promise | undefined; + try { + session = await driver.openSession({ runId: "ordering-run", normalizedSessionId: "ordering-session", workingDirectory: WORKSPACE }); + const { turnId } = await session.startTurn({ message: { role: "user", text: "Fixture" } }); + drain = (async () => { for await (const event of session!.events()) { + seen.push(event); + if (seen.some(e => e.eventType === "runtime_request.created") && seen.some(e => e.eventType === "tool.execution.started")) break; + } })(); + const tool = () => peer.current!.emit("tool.execution.started", { + schema: "paperclip.tool.execution.v1", executionId: "native-tool-7", transport: "builtin", namespace: null, + name: "Run command", operation: "execute", target: null, status: "running", readOnly: false, + inputUpdated: true, output: null, outputBytes: 0, outputTruncated: false, outputDigest: null, + durationMs: null, exitCode: null, progress: [], + }); + const permission = () => peer.current!.emit("runtime_request.created", { request: { + requestId: "permission-7", type: "permission", requestKind: "permission_approval", + itemId: "opaque-item-7", prompt: "Run command", details: { toolCallId: "native-tool-7" }, + choices: [{ key: "decline", label: "Deny" }, { key: "cancel", label: "Cancel" }], + origin: { adapter: "acpx-runtime-sidecar", provider: "cursor", method: "session/request_permission" }, + } }); + // Both durable events exist before the production pump's next tick. + // Keep the permission unanswered while the independent tool pump drains. + if (order === "tool-first") { tool(); permission(); } else { permission(); tool(); } + await vi.waitFor(() => expect(seen.filter(e => ["tool.execution.started", "runtime_request.created"].includes(e.eventType))).toHaveLength(2), { timeout: 1_000, interval: 5 }); + const nativeOrder = peer.current!.store.state.committedEvents.filter(e => ["tool.execution.started", "runtime_request.created"].includes(e.eventType)).map(e => e.eventType); + const projectedOrder = seen.filter(e => ["tool.execution.started", "runtime_request.created"].includes(e.eventType)).map(e => e.eventType); + expect(session.pendingRuntimeRequests!()).toMatchObject([{ requestId: "permission-7", turnId, status: "pending" }]); + expect(projectedOrder).toEqual(nativeOrder); + } finally { + // This test owns no actual process or database. Retire the mocked peer + // and its controller timer without inventing a provider terminal event. + await bundle.detachControllerForRestart(); + await drain; + rmSync(root, { recursive: true, force: true }); + } + }); +}); + +async function bareTransport() { + const root = mkdtempSync(join(tmpdir(), "permission-queue-lifecycle-")); + const binary = join(root, "unexecuted-runner"); + writeFileSync(binary, "not executable: process boundary is mocked"); + const bundle = createCapabilityRunnerdCodexTransport({ runnerBinary: binary, stateDirectory: root, + sourceCodexHome: join(root, "absent-source-home"), environment: { HOME: root, PATH: "/usr/bin:/bin" }, + }); + await bundle.transport.request("initialize", {}); + await bundle.transport.request("thread/start", { cwd: root }); + await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Fixture" }] }); + const iterator = bundle.transport.notifications()[Symbol.asyncIterator](); + expect((await iterator.next()).value?.method).toBe("turn/started"); + const emitPermission = (requestId = "permission-7", prompt = "Run command") => peer.current!.emit("runtime_request.created", { request: { + requestId, type: "permission", requestKind: "permission_approval", itemId: requestId, prompt, + choices: [{ key: "cancel", label: "Cancel" }], details: { toolCallId: "tool-7" }, + origin: { adapter: "acpx-runtime-sidecar", provider: "cursor", method: "session/request_permission" }, + } }); + const emitMarker = () => peer.current!.emit("item.delta", { kind: "agentMessage", text: "marker" }); + return { bundle, iterator, emitPermission, emitMarker, async close() { await bundle.detachControllerForRestart(); rmSync(root, { recursive: true, force: true }); } }; +} + +it("discards a queued permission on close before the notification consumer asks for next", async () => { + const fixture = await bareTransport(); const handler = vi.fn(async () => ({})); + fixture.bundle.transport.setServerRequestHandler(handler); + try { + fixture.emitMarker(); fixture.emitPermission(); + expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta"); + expect(handler).not.toHaveBeenCalled(); + await fixture.bundle.detachControllerForRestart(); + await expect(fixture.iterator.next()).resolves.toMatchObject({ done: true }); + expect(handler).not.toHaveBeenCalled(); + } finally { await fixture.close(); } +}); + +it("does not dispatch an expired durable request when its consumer resumes", async () => { + const fixture = await bareTransport(); const handler = vi.fn(async () => ({})); + fixture.bundle.transport.setServerRequestHandler(handler); + try { + fixture.emitPermission(); + peer.current!.emit("runtime_request.expired", { requestId: "permission-7" }); + fixture.emitMarker(); + expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta"); + expect(handler).not.toHaveBeenCalled(); + } finally { await fixture.close(); } +}); + +it("does not dispatch queued approval callbacks into a replacement turn", async () => { + const fixture = await bareTransport(); const handler = vi.fn(async () => ({})); + fixture.bundle.transport.setServerRequestHandler(handler); + try { + fixture.emitMarker(); fixture.emitPermission(); + expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta"); + await fixture.bundle.transport.request("turn/start", { input: [{ type: "text", text: "Next turn" }] }); + expect((await fixture.iterator.next()).value?.method).toBe("turn/started"); + expect(handler).not.toHaveBeenCalled(); + } finally { await fixture.close(); } +}); + +it.each(["throw", "reject"])("closes the queue and wakes its consumer on handler %s", async failure => { + const fixture = await bareTransport(); + const handler = vi.fn(() => { if (failure === "throw") throw new Error("fixture handler failed"); return Promise.reject(new Error("fixture handler failed")); }); + fixture.bundle.transport.setServerRequestHandler(handler); + try { + fixture.emitPermission(); + await expect(fixture.iterator.next()).rejects.toThrow("fixture handler failed"); + expect(handler).toHaveBeenCalledTimes(1); + await expect(fixture.iterator.next()).rejects.toThrow("fixture handler failed"); + } finally { await fixture.close(); } +}); + +it.each(["count", "bytes"])("bounds queued permission callback %s without a consumer", async bound => { + const fixture = await bareTransport(); const handler = vi.fn(async () => ({})); + fixture.bundle.transport.setServerRequestHandler(handler); + try { + if (bound === "count") for (let i = 0; i < 2049; i++) fixture.emitPermission(`permission-${i}`); + else fixture.emitPermission("oversized", "x".repeat(4 * 1024 * 1024)); + // next waits until the actual event pump reaches the closed queue limit. + await expect(fixture.iterator.next()).rejects.toThrow("notification queue bound exceeded"); + expect(handler).not.toHaveBeenCalled(); + } finally { await fixture.close(); } +}); + +it("wakes an idle notification consumer during controller shutdown", async () => { + const fixture = await bareTransport(); + try { + const pending = fixture.iterator.next(); + await fixture.bundle.detachControllerForRestart(); + await expect(pending).resolves.toMatchObject({ done: true }); + } finally { await fixture.close(); } +}); + +it("does not transfer a queued request to a newly registered handler", async () => { + const fixture = await bareTransport(); const original = vi.fn(async () => ({})), replacement = vi.fn(async () => ({})); + fixture.bundle.transport.setServerRequestHandler(original); + try { + fixture.emitMarker(); fixture.emitPermission(); fixture.emitMarker(); + expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta"); + fixture.bundle.transport.setServerRequestHandler(replacement); + expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta"); + expect(original).not.toHaveBeenCalled(); expect(replacement).not.toHaveBeenCalled(); + } finally { await fixture.close(); } +}); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index cfd656c38f9..b81798e5571 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -1488,32 +1488,37 @@ export function resolveRunnerdSessionIdentity(input: unknown): { }; } +type NotificationQueueEntry = + | { kind: "notification"; value: CodexRpcNotification; bytes: number } + | { kind: "control"; dispatch(): void; bytes: number }; + class NotificationQueue implements AsyncIterable { - #values: Array<{ value: CodexRpcNotification; bytes: number }> = []; - #waiters: Array<{ - resolve: (value: IteratorResult) => void; - reject: (error: Error) => void; - }> = []; + #values: NotificationQueueEntry[] = []; + #waiters: Array<() => void> = []; #bytes = 0; #closed = false; #error: Error | null = null; push(value: CodexRpcNotification): void { if (this.#closed) return; - const bytes = Buffer.byteLength(JSON.stringify(value)); - const waiter = this.#waiters.shift(); - if (waiter !== undefined) { - waiter.resolve({ value, done: false }); - return; - } - if ( - this.#values.length >= MAX_NOTIFICATION_COUNT || - this.#bytes + bytes > MAX_NOTIFICATION_BYTES - ) { + this.#push({ kind: "notification", value, bytes: Buffer.byteLength(JSON.stringify(value)) }); + } + + pushControl(value: CodexRpcServerRequest, dispatch: () => void): void { + if (this.#closed) return; + // Count the retained request just like notification payloads. The callback + // is internal; a provider notification cannot create a control entry. + this.#push({ kind: "control", dispatch, bytes: Buffer.byteLength(JSON.stringify(value)) }); + } + + #push(entry: NotificationQueueEntry): void { + if (this.#closed) return; + if (this.#values.length >= MAX_NOTIFICATION_COUNT || this.#bytes + entry.bytes > MAX_NOTIFICATION_BYTES) { throw new Error("PRP provider notification queue bound exceeded"); } - this.#values.push({ value, bytes }); - this.#bytes += bytes; + this.#values.push(entry); + this.#bytes += entry.bytes; + this.#waiters.shift()?.(); } close(error?: Error): void { @@ -1522,25 +1527,29 @@ class NotificationQueue implements AsyncIterable { this.#error = error ?? null; this.#values = []; this.#bytes = 0; - for (const waiter of this.#waiters.splice(0)) { - if (this.#error !== null) waiter.reject(this.#error); - else waiter.resolve({ value: undefined, done: true }); - } + for (const wake of this.#waiters.splice(0)) wake(); } [Symbol.asyncIterator](): AsyncIterator { return { next: async () => { - const queued = this.#values.shift(); - if (queued !== undefined) { - this.#bytes -= queued.bytes; - return { value: queued.value, done: false }; + for (;;) { + if (this.#error !== null) throw this.#error; + if (this.#closed) return { value: undefined, done: true }; + const queued = this.#values.shift(); + if (queued !== undefined) { + this.#bytes -= queued.bytes; + if (queued.kind === "control") { + // The consumer asks for next only after mapping the previous + // notification. Start the request here without awaiting the + // human response, so subsequent provider activity keeps flowing. + queued.dispatch(); + continue; + } + return { value: queued.value, done: false }; + } + await new Promise(resolve => this.#waiters.push(resolve)); } - if (this.#error !== null) throw this.#error; - if (this.#closed) return { value: undefined, done: true }; - return new Promise((resolveValue, reject) => - this.#waiters.push({ resolve: resolveValue, reject }), - ); }, }; } @@ -6087,14 +6096,15 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { isAcpxCanonicalInputMethod(method) || permission) && !this.#bridgedRuntimeInputs.has(requestId) ) { - this.#bridgedRuntimeInputs.set(requestId, { + const binding = { permission, durableTurnId: typeof event.envelope.turnId === "string" ? event.envelope.turnId : this.#durableTurnId, - }); - void this.#handler({ + }; + this.#bridgedRuntimeInputs.set(requestId, binding); + const request: CodexRpcServerRequest = { id: requestId, method, params, @@ -6102,10 +6112,21 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { sourceEventId: event.sourceEventId, sourceEventType: event.eventType, }, - }).catch((error) => { - this.#failTransport( - error instanceof Error ? error : new Error(String(error)), - ); + }; + const threadId = this.#threadId, turnId = this.#turnId, handler = this.#handler; + this.#queue.pushControl(request, () => { + // A detached controller, replaced turn or settled durable request + // cannot acquire approval authority when a slow consumer resumes. + if (this.#closed || this.#failure || this.#core !== core || this.#handler !== handler + || this.#threadId !== threadId || this.#turnId !== turnId + || this.#bridgedRuntimeInputs.get(requestId) !== binding) return; + try { + void handler(request).catch((error) => { + this.#failTransport(error instanceof Error ? error : new Error(String(error))); + }); + } catch (error) { + this.#failTransport(error instanceof Error ? error : new Error(String(error))); + } }); } continue;