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;