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 <noreply@paperclip.ing>

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<Record<string, unknown>> } };
+  emit(type: string, payload: Record<string, unknown>): void;
+}
+vi.mock("../control-plane/durable-prp-control-plane.js", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("../control-plane/durable-prp-control-plane.js")>();
+  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<string, unknown>, 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<string, unknown>) {
+        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<ReturnType<typeof driver.openSession>> | undefined;
+    let drain: Promise<void> | 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(); }
+});

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<CodexRpcNotification> {
-  #values: Array<{ value: CodexRpcNotification; bytes: number }> = [];
-  #waiters: Array<{
-    resolve: (value: IteratorResult<CodexRpcNotification>) => 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<CodexRpcNotification> {
     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<CodexRpcNotification> {
     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<void>(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;