diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts
index 4a0ca8354a8..3fc4438b0b2 100644
--- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts
+++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts
@@ -1653,6 +1653,8 @@ class DurableCoreStore {
   #writeRevision = 0;
   #externallyPersisted = false;
   revision = 0;
+  /** Called after the stored state is replaced or saved. */
+  onChange: (() => void) | null = null;
   #pendingEvent: {
     buildCandidate: (state: StoredCoreState) => StoredCoreState | null;
     isCurrentOwner: () => boolean;
@@ -1714,6 +1716,7 @@ class DurableCoreStore {
     this.#state.checkpointEventPolicy = "control-v1";
     this.#externallyPersisted = true;
     this.revision += 1;
+    this.onChange?.();
   }
 
   save(): void {
@@ -1732,6 +1735,7 @@ class DurableCoreStore {
       this.#state = candidate;
       if (pending && withEvent) pending.committed = true;
       this.revision += 1;
+      this.onChange?.();
     } catch (error) {
       this.#writeIndeterminate = true;
       throw error;
@@ -1757,6 +1761,7 @@ class DurableCoreStore {
         this.#state = candidate;
         ++this.#writeRevision;
         if (checkpointRequired) this.revision += 1;
+        this.onChange?.();
         return true;
       }
       await withStagedPrivateWrite(this.path, `${JSON.stringify(candidate)}\n`, (temporary) => {
@@ -1768,6 +1773,7 @@ class DurableCoreStore {
         ++this.#writeRevision;
         pending.committed = true;
         if (checkpointRequired) this.revision += 1;
+        this.onChange?.();
       });
       // Explicit authority commits do not absorb pending events. On conflict,
       // leave the event unacknowledged for replay instead of rewriting in a loop.
@@ -1801,6 +1807,7 @@ class DurableCoreStore {
       }
       this.#state = candidate;
       if (checkpointRequired) this.revision += 1;
+      this.onChange?.();
     } catch (error) {
       // Rename may already have succeeded before directory fsync failed.
       // Never overwrite that possibly durable receipt using stale memory.
@@ -2091,6 +2098,7 @@ export class DurablePrpControlPlane {
   #persistenceTail = Promise.resolve();
   #sendQueues = new Map<AuthorityConnection, Promise<void>>();
   #recoveredCursor: { runId: string; sourceSeq: number } | null = null;
+  #readableStateListeners = new Set<() => void>();
 
   constructor(options: DurablePrpControlPlaneOptions) {
     if (
@@ -2111,6 +2119,7 @@ export class DurablePrpControlPlane {
       options.stateDirectory,
       options.identity,
     );
+    this.#store.onChange = () => this.#readableStateChanged();
     if (this.#store.state.checkpointEventPolicy === "control-v1") {
       this.#recoveredCursor = { runId: options.identity.runId, sourceSeq: this.#store.state.ackedSourceSeq };
     }
@@ -2143,6 +2152,31 @@ export class DurablePrpControlPlane {
     return this.#publishedStore ?? this.#store;
   }
 
+  /**
+   * Tell a reader of `store` that what it can see may have changed: an event
+   * or command was committed, or a saved checkpoint released the events that
+   * were waiting for it. This is a hint, and it can arrive partway through the
+   * change, so the reader must look at the state in a later task. Returns the
+   * function that removes the listener.
+   */
+  onReadableStateChange(listener: () => void): () => void {
+    this.#readableStateListeners.add(listener);
+    return () => {
+      this.#readableStateListeners.delete(listener);
+    };
+  }
+
+  #readableStateChanged(): void {
+    for (const listener of this.#readableStateListeners) {
+      // A reader can never fail the commit that woke it.
+      try {
+        listener();
+      } catch {
+        // The reader owns its own failure handling.
+      }
+    }
+  }
+
   #capturePublishedState(snapshot: DurablePrpCheckpoint): StoredCoreState {
     const header = JSON.parse(snapshot.records.get("header")!) as CheckpointHeader;
     const cursor = JSON.parse(snapshot.records.get("cursor")!) as { ackedSourceSeq: number; replayDeliveries: number };
@@ -2227,6 +2261,7 @@ export class DurablePrpControlPlane {
         this.#publishedState = published;
         this.#persistedRevision = revision;
         this.#checkpointPersistenceFailed = false;
+        this.#readableStateChanged();
       } catch (error) {
         this.#checkpointPersistenceFailed = true;
         throw error;
@@ -3795,6 +3830,7 @@ export class DurablePrpControlPlane {
       const cursor = Math.max(this.#store.state.ackedSourceSeq, pending.acknowledgedSourceSeq);
       this.#store.state.ackedSourceSeq = cursor;
       this.#recoveredCursor = { runId: this.#identity.runId, sourceSeq: cursor };
+      this.#readableStateChanged();
     }
     connection.pendingChallenge = null;
     connection.lease = lease;
diff --git a/packages/paperclip-runner/src/live/runnerd-acpx-permission-forwarding.test.ts b/packages/paperclip-runner/src/live/runnerd-acpx-permission-forwarding.test.ts
index 2a0ff6779d7..21dc9f76f50 100644
--- a/packages/paperclip-runner/src/live/runnerd-acpx-permission-forwarding.test.ts
+++ b/packages/paperclip-runner/src/live/runnerd-acpx-permission-forwarding.test.ts
@@ -23,6 +23,11 @@ vi.mock("../control-plane/durable-prp-control-plane.js", async (importOriginal)
       readonly recoveredSourceSeq = 0;
       store: TestPeer["store"];
       connectUrl = "ws://fixture.invalid";
+      readers = new Set<() => void>();
+      onReadableStateChange(reader: () => void) {
+        this.readers.add(reader);
+        return () => { this.readers.delete(reader); };
+      }
       constructor(options: { identity: DurableRecoveryIdentity }) {
         this.store = { state: { identity: options.identity, committedEvents: [], commands: [] } };
         peer.current = this;
@@ -54,6 +59,7 @@ vi.mock("../control-plane/durable-prp-control-plane.js", async (importOriginal)
           priority: 0, deliveryCount: 1, logicalEffectCount: 1,
           envelope: { turnId: this.store.state.identity.turnId, payload: { payload } },
         });
+        for (const reader of this.readers) reader();
       }
     },
   };
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
index 4800a50f5e4..6946b67c192 100644
--- a/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts
+++ b/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts
@@ -21,6 +21,11 @@ vi.mock("../control-plane/durable-prp-control-plane.js", async (importOriginal)
       readonly recoveredSourceSeq = 0;
       store: TestPeer["store"];
       connectUrl = "ws://fixture.invalid";
+      readers = new Set<() => void>();
+      onReadableStateChange(reader: () => void) {
+        this.readers.add(reader);
+        return () => { this.readers.delete(reader); };
+      }
       constructor(options: { identity: DurableRecoveryIdentity }) {
         this.store = { state: { identity: options.identity, committedEvents: [], commands: [] } };
         peer.current = this;
@@ -51,6 +56,7 @@ vi.mock("../control-plane/durable-prp-control-plane.js", async (importOriginal)
           priority: 0, deliveryCount: 1, logicalEffectCount: 1,
           envelope: { turnId: this.store.state.identity.turnId, payload: { payload } },
         });
+        for (const reader of this.readers) reader();
       }
     },
   };
diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-pump.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-pump.test.ts
new file mode 100644
index 00000000000..8a38927582d
--- /dev/null
+++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-pump.test.ts
@@ -0,0 +1,135 @@
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join, resolve } from "node:path";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { probeTransportPump, type ReadableEvent, type TransportPumpProbe } from "../../test/support/transport-pump-probe.js";
+import {
+  createCapabilityRunnerdCodexTransport,
+  defaultCapabilityRunnerdBinary,
+} from "./runnerd-codex-transport.js";
+
+// A live transport used to look for committed events every 5 ms for as long as
+// it was open, so a dozen open sessions kept the server's main thread busy
+// with nothing to deliver, and every event waited for the next tick.
+const fixture = resolve(import.meta.dirname, "../../test/fixtures/fake-paced-codex-app-server.mjs");
+const DELTAS = 40;
+
+async function openTransport(
+  persistence: "file" | "external",
+  onReadable?: (event: ReadableEvent) => void,
+) {
+  const root = await mkdtemp(join(tmpdir(), "paperclip-transport-pump-"));
+  let probe: TransportPumpProbe | null = null;
+  const bundle = createCapabilityRunnerdCodexTransport({
+    runnerBinary: defaultCapabilityRunnerdBinary(),
+    codexCommand: process.execPath,
+    codexArgs: [fixture, String(DELTAS), "15"],
+    environment: {},
+    stateDirectory: join(root, "session"),
+    lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
+    controlPlaneRegistration: async (core) => {
+      probe = probeTransportPump(core, persistence, onReadable);
+      return probe.registration();
+    },
+  });
+  try {
+    await bundle.transport.request("thread/start", { cwd: root, model: "fixture-no-model", dynamicTools: [] });
+  } catch (error) {
+    await bundle.transport.close().catch(() => undefined);
+    await rm(root, { recursive: true, force: true });
+    throw error;
+  }
+  return {
+    transport: bundle.transport,
+    probe: probe!,
+    async close() {
+      await bundle.transport.close().catch(() => undefined);
+      await rm(root, { recursive: true, force: true });
+    },
+  };
+}
+
+describe.each(["file", "external"] as const)("runnerd transport event pump with %s persistence", (persistence) => {
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("stops looking for events while an open session has none to deliver", async () => {
+    const session = await openTransport(persistence);
+    try {
+      const before = session.probe.stateReads();
+      await new Promise((elapsed) => setTimeout(elapsed, 2_000));
+      // The 5 ms poll took the state several hundred times a second. Without
+      // it, only the tail of session startup reads the state in this window.
+      expect(session.probe.stateReads() - before).toBeLessThan(100);
+
+      // The quiet session still delivers the first event that arrives.
+      const notifications = session.transport.notifications()[Symbol.asyncIterator]();
+      await session.transport.request("turn/start", { input: [{ type: "text", text: "Stream the paced fixture." }] });
+      for (;;) {
+        const next = await notifications.next();
+        expect(next.done).not.toBe(true);
+        if (next.value.method === "turn/completed") break;
+      }
+    } finally {
+      await session.close();
+    }
+  }, 30_000);
+
+  it("delivers each committed event in the next task, without a timer tick", async () => {
+    // No interval ever fires in this test, so nothing here can be delivered by
+    // a periodic poll.
+    vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
+    // The control plane wakes the transport before this probe sees the event,
+    // so a task queued here runs after the transport's delivery task. That
+    // order does not depend on the host's speed.
+    const overtaken = new Set<number>();
+    const pending = new Map<number, NodeJS.Immediate>();
+    const session = await openTransport(persistence, (event) => {
+      if (event.eventType !== "item.delta") return;
+      pending.set(event.sourceSeq, setImmediate(() => {
+        pending.delete(event.sourceSeq);
+        overtaken.add(event.sourceSeq);
+      }));
+    });
+    try {
+      const notifications = session.transport.notifications()[Symbol.asyncIterator]();
+      await session.transport.request("turn/start", { input: [{ type: "text", text: "Stream the paced fixture." }] });
+      // Events committed before this point can wait for the start response.
+      const startAnswered = performance.now();
+      const deliveredInOrder: boolean[] = [];
+      let completed = false;
+      let stalled: ReturnType<typeof setTimeout> | undefined;
+      await Promise.race([
+        (async () => {
+          for (;;) {
+            const next = await notifications.next();
+            if (next.done) return;
+            if (next.value.method === "item/agentMessage/delta") {
+              const event = session.probe.readable.filter((candidate) => candidate.eventType === "item.delta")[deliveredInOrder.length]!;
+              clearImmediate(pending.get(event.sourceSeq));
+              pending.delete(event.sourceSeq);
+              deliveredInOrder.push(event.at < startAnswered || !overtaken.has(event.sourceSeq));
+            }
+            if (next.value.method === "turn/completed") {
+              completed = true;
+              return;
+            }
+          }
+        })(),
+        new Promise<void>((elapsed) => { stalled = setTimeout(elapsed, 10_000); }),
+      ]);
+      clearTimeout(stalled);
+
+      expect(completed, "the turn's events were delivered without a periodic poll").toBe(true);
+      expect(deliveredInOrder).toHaveLength(DELTAS);
+      expect(deliveredInOrder.every(Boolean), "every event was delivered in the task after it became readable").toBe(true);
+      // The assertion above covers the events streamed after the start response.
+      expect(session.probe.readable.some((event) => event.eventType === "item.delta" && event.at >= startAnswered)).toBe(true);
+    } finally {
+      for (const task of pending.values()) clearImmediate(task);
+      vi.useRealTimers();
+      await session.close();
+    }
+  }, 30_000);
+});
diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts
index 130cbc2878c..0275f959d0d 100644
--- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts
+++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts
@@ -115,6 +115,8 @@ const packageRoot = resolve(fileURLToPath(new URL("../..", import.meta.url)));
 const executableSuffix = process.platform === "win32" ? ".exe" : "";
 const MAX_NOTIFICATION_COUNT = 2_048;
 const MAX_NOTIFICATION_BYTES = 4 * 1024 * 1024;
+// How often the event pump retries events that it is holding back itself.
+const PUMP_HELD_RETRY_MS = 5;
 const RUNNER_CLIENT_VERSION = "0.3.0";
 const RUNNER_BOOTSTRAP_TICKET_TTL_MS = 60_000;
 const RUNNER_RESTORE_STABLE_MS = 60_000;
@@ -3950,7 +3952,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
   #lostRunnerHandle: RunnerProcessHandle | null = null;
   #adoptedRunnerMonitor: NodeJS.Timeout | null = null;
   #adoptedRunnerAuthenticated = false;
-  #pump: NodeJS.Timeout | null = null;
+  /** Events are delivered only between session start and release, retirement, failure or close. */
+  #pumpActive = false;
+  #pumpWake: NodeJS.Immediate | null = null;
+  #pumpRetry: NodeJS.Timeout | null = null;
+  #stopFollowingCore: (() => void) | null = null;
   #eventSourceSeq = 0;
   #eventIdentity: DurableRecoveryIdentity | null = null;
   #sourceRecoveryIdentity: DurableRecoveryIdentity | null = null;
@@ -4640,8 +4646,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
   }
 
   #quietReleasedRunner(): void {
-    if (this.#pump !== null) clearInterval(this.#pump);
-    this.#pump = null;
+    this.#stopPump();
     if (this.#adoptedRunnerMonitor !== null) clearInterval(this.#adoptedRunnerMonitor);
     this.#adoptedRunnerMonitor = null;
   }
@@ -4768,10 +4773,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       await this.#awaitRegistrationReady(registration?.ready);
       this.#assertStartupOpen();
     }
-    if (this.#pump === null) {
-      this.#pump = setInterval(() => this.#pumpEventsSafely(), 5);
-      this.#pump.unref?.();
-    }
+    if (!this.#pumpActive) this.#startPump();
     if (this.options.adoptExistingRunner && this.#adoptedRunnerMonitor === null) this.#watchAdoptedRunner(this.options.adoptExistingRunner);
   }
 
@@ -5017,6 +5019,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       this.#eventSourceSeq = 0;
       this.#deferredTurnStartEvents = [];
       this.#durableTurnId = desired.turnId;
+      this.#wakePump();
       if (!previousReleased) await previousRelease?.();
       previousReleased = true;
       this.#controlPlaneRelease = registration?.release ?? null;
@@ -5101,6 +5104,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       this.#pendingDriverTraceInterpretations.length < 4_096
     ) {
       this.#pendingDriverTraceInterpretations.push(structuredClone(input));
+      this.#retryPumpWhile(true);
     } else if (traceResult === "retry") {
       this.#traceRehydrationSpoolOverflow = true;
     }
@@ -5378,8 +5382,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
   }
 
   async #retireControllerForRecovery(deadline: number): Promise<void> {
-    if (this.#pump !== null) clearInterval(this.#pump);
-    this.#pump = null;
+    this.#stopPump();
     if (this.#adoptedRunnerMonitor !== null)
       clearInterval(this.#adoptedRunnerMonitor);
     this.#adoptedRunnerMonitor = null;
@@ -5614,8 +5617,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       this.#pendingDriverTraceInterpretations = [];
       this.#traceFrameIndex.clear();
     }
-    if (this.#pump !== null) clearInterval(this.#pump);
-    this.#pump = null;
+    this.#stopPump();
+    this.#stopFollowingCore?.();
+    this.#stopFollowingCore = null;
     if (this.#adoptedRunnerMonitor !== null)
       clearInterval(this.#adoptedRunnerMonitor);
     this.#adoptedRunnerMonitor = null;
@@ -6089,6 +6093,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       connectionLeaseTtlMs: 60 * 60 * 1_000,
     });
     this.#core = core;
+    this.#followCommittedEvents(core);
     this.#carriedCommandCount = core.store.state.commands.length;
     // Externally launched runners own their state directory (for example in a
     // Daytona sandbox). Do not create an empty controller-side placeholder:
@@ -6180,7 +6185,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     this.#evidence.runnerPid = handle.child.pid ?? null;
     this.#evidence.runnerProcessGroupId = handle.processGroupId ?? null;
     this.#publish();
-    this.#pump = setInterval(() => this.#pumpEventsSafely(), 5);
+    this.#startPump();
     await this.#waitCommand("run.prepare", undefined, undefined, true);
     this.#startupSteps?.advance("runner.process.launch", "provider.session.load");
     await this.#waitCommand("session.open", undefined, admissionDeadline, true);
@@ -6665,6 +6670,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       connectionLeaseTtlMs: 60 * 60 * 1_000,
     });
     this.#core = core;
+    this.#followCommittedEvents(core);
     this.#carriedCommandCount = core.store.state.commands.length;
     // Reusing resources still requires current environment authorization. The
     // optional hook avoids re-preparation after controller storage is lost.
@@ -7133,7 +7139,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       }
     }
     this.#assertStartupOpen();
-    this.#pump = setInterval(() => this.#pumpEventsSafely(), 5);
+    this.#startPump();
     if (adoptedRunner) {
       await this.#awaitAdoptedRunnerConnection(
         adoptedRunner,
@@ -7807,7 +7813,28 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     }
 
 
+  /**
+   * Deliver every committed event that is ready. The control plane wakes this
+   * when a reader can see a change, so an event never waits for a timer and an
+   * idle session runs none. Only what this transport holds back itself, behind
+   * a pending turn start, a recovery binding or a trace write, is retried on a
+   * timer, and only for as long as it is held.
+   */
   #pumpEvents(): void {
+    const remaining = this.#deliverCommittedEvents();
+    if (remaining === "next_task") this.#wakePump();
+    this.#retryPumpWhile(
+      remaining === "held" ||
+        this.#pendingTraceRehydrations.length > 0 ||
+        this.#pendingDriverTraceInterpretations.length > 0,
+    );
+  }
+
+  /**
+   * Returns what a later pass still has to deliver: nothing, events held
+   * behind a gate, or events that only had to wait for the next task.
+   */
+  #deliverCommittedEvents(): "none" | "held" | "next_task" {
     const core = this.#core;
     // The controller may activate its new epoch before attachRun observes the
     // confirmed handoff. Never consume either epoch with the other's cursor.
@@ -7816,7 +7843,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       !this.#eventIdentity ||
       !recoveryIdentityMatches(core.store.state.identity, this.#eventIdentity)
     )
-      return;
+      return "none";
     this.#flushPendingTraceRehydrations();
     const events = core.store.state.committedEvents;
     for (;;) {
@@ -7830,7 +7857,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
         events.find((candidate) => candidate.sourceSeq > this.#eventSourceSeq);
       if (event === undefined) {
         this.#eventSourceSeq = Math.max(this.#eventSourceSeq, core.recoveredSourceSeq);
-        return;
+        return this.#deferredTurnStartEvents.length > 0 ? "held" : "none";
       }
       const fromDeferredQueue = deferredEvent !== undefined;
       if (!fromDeferredQueue && event.sourceSeq > this.#eventSourceSeq + 1) {
@@ -7906,7 +7933,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
           this.#eventSourceSeq = event.sourceSeq;
           this.#deferredTurnStartEvents.push(structuredClone(event));
         }
-        return;
+        return "held";
       }
       // The control plane retains a sliding committed-event window. Track its
       // durable protocol cursor rather than an array index: once that array is
@@ -8207,15 +8234,16 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       // The strict Codex driver must bind the provider turn from the response
       // before it observes terminal notifications. A fast provider can commit
       // start and terminal events in one durable batch, so stop after exposing
-      // turn/started while the request is pending. The regular pump drains the
-      // remaining events after the promise resolves.
+      // turn/started while the request is pending. The next task drains the
+      // remaining events; a terminal among them stays held until the response
+      // is released.
       if (
         this.#turnStartResponsePending &&
         notifications.some(
           (notification) => notification.method === "turn/started",
         )
       ) {
-        return;
+        return "next_task";
       }
     }
   }
@@ -8394,6 +8422,47 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     }
   }
 
+  #followCommittedEvents(core: DurablePrpControlPlane): void {
+    this.#stopFollowingCore?.();
+    this.#stopFollowingCore = core.onReadableStateChange(() => {
+      if (this.#core === core) this.#wakePump();
+    });
+  }
+
+  #startPump(): void {
+    this.#pumpActive = true;
+    this.#wakePump();
+  }
+
+  #stopPump(): void {
+    this.#pumpActive = false;
+    if (this.#pumpWake !== null) clearImmediate(this.#pumpWake);
+    this.#pumpWake = null;
+    this.#retryPumpWhile(false);
+  }
+
+  /**
+   * Deliver in the next task. The control plane can report a change partway
+   * through its own commit, so the pump never runs inside that call; several
+   * changes in one task share a single pass.
+   */
+  #wakePump(): void {
+    if (!this.#pumpActive || this.#pumpWake !== null) return;
+    this.#pumpWake = setImmediate(() => {
+      this.#pumpWake = null;
+      if (this.#pumpActive) this.#pumpEventsSafely();
+    });
+  }
+
+  #retryPumpWhile(held: boolean): void {
+    if (held && this.#pumpActive) {
+      this.#pumpRetry ??= setInterval(() => this.#pumpEventsSafely(), PUMP_HELD_RETRY_MS);
+    } else if (this.#pumpRetry !== null) {
+      clearInterval(this.#pumpRetry);
+      this.#pumpRetry = null;
+    }
+  }
+
   /** True only when a runnerd this transport spawned on this host has exited
    * and its journalled provider and tool processes are proven stopped. */
   async #reapLostLocalRunnerLineage(closeDeadline: number): Promise<boolean> {
@@ -8778,8 +8847,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     if (this.#failure !== null || this.#closed) return;
     this.#failure = error;
     this.#rejectFailureSignal(error);
-    if (this.#pump !== null) clearInterval(this.#pump);
-    this.#pump = null;
+    this.#stopPump();
     try {
       this.#diagnostic(error.message);
     } finally {
diff --git a/packages/paperclip-runner/src/live/runnerd-transport-pump.benchmark.test.ts b/packages/paperclip-runner/src/live/runnerd-transport-pump.benchmark.test.ts
new file mode 100644
index 00000000000..20e3b81c394
--- /dev/null
+++ b/packages/paperclip-runner/src/live/runnerd-transport-pump.benchmark.test.ts
@@ -0,0 +1,147 @@
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join, resolve } from "node:path";
+import { expect, it } from "vitest";
+import { probeTransportPump, type TransportPumpProbe } from "../../test/support/transport-pump-probe.js";
+import {
+  createCapabilityRunnerdCodexTransport,
+  defaultCapabilityRunnerdBinary,
+} from "./runnerd-codex-transport.js";
+
+// Opt in explicitly. Run from packages/paperclip-runner:
+// PAPERCLIP_TRANSPORT_PUMP_BENCHMARK=1 pnpm exec vitest run src/live/runnerd-transport-pump.benchmark.test.ts
+// PAPERCLIP_TRANSPORT_PUMP_SESSIONS (default 12) sets how many transports stay open.
+// PAPERCLIP_TRANSPORT_PUMP_DELTAS (default 40) sets how many events each turn
+// streams, and so how many committed events each session retains afterwards.
+//
+// It opens that many real transports against runnerd and a credential-free
+// provider fixture, then reports two things for each persistence mode:
+// - what the open transports cost the main thread while nothing arrives, and
+// - how long a committed event waits before the transport delivers it.
+// This is a local measurement that shares the host's background load, not a
+// CPU-isolated assertion; only delivery itself is asserted.
+const enabled = process.env.PAPERCLIP_TRANSPORT_PUMP_BENCHMARK === "1";
+const sessions = Number(process.env.PAPERCLIP_TRANSPORT_PUMP_SESSIONS ?? "12");
+const deltas = Number(process.env.PAPERCLIP_TRANSPORT_PUMP_DELTAS ?? "40");
+if (enabled && (!Number.isInteger(sessions) || sessions < 1 || sessions > 32)) {
+  throw new Error("transport_pump_sessions_must_be_between_1_and_32");
+}
+if (enabled && (!Number.isInteger(deltas) || deltas < 1 || deltas > 512)) {
+  throw new Error("transport_pump_deltas_must_be_between_1_and_512");
+}
+const fixture = resolve(import.meta.dirname, "../../test/fixtures/fake-paced-codex-app-server.mjs");
+const DELTA_INTERVAL_MS = 15;
+const IDLE_SETTLE_MS = 1_500;
+const IDLE_WINDOW_MS = 3_000;
+
+const percentile = (sorted: number[], fraction: number) =>
+  sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * fraction))]!;
+const round = (value: number) => Math.round(value * 1_000) / 1_000;
+
+it.skipIf(!enabled).each(["file", "external"] as const)(
+  "measures idle pump cost and delivery wait with %s persistence",
+  async (persistence) => {
+    const root = await mkdtemp(join(tmpdir(), "paperclip-transport-pump-benchmark-"));
+    const open: Array<{
+      bundle: ReturnType<typeof createCapabilityRunnerdCodexTransport>;
+      probe: () => TransportPumpProbe;
+    }> = [];
+    const measureIdle = async () => {
+      await new Promise((settled) => setTimeout(settled, IDLE_SETTLE_MS));
+      const reads = () => open.reduce((total, session) => total + session.probe().stateReads(), 0);
+      const readsBefore = reads();
+      const cpuBefore = process.cpuUsage();
+      const loopBefore = performance.eventLoopUtilization();
+      const started = performance.now();
+      await new Promise((elapsed) => setTimeout(elapsed, IDLE_WINDOW_MS));
+      const seconds = (performance.now() - started) / 1_000;
+      const loop = performance.eventLoopUtilization(loopBefore);
+      const cpu = process.cpuUsage(cpuBefore);
+      return {
+        seconds: round(seconds),
+        stateReadsPerSessionPerSecond: round((reads() - readsBefore) / seconds / sessions),
+        eventLoopUtilization: round(loop.utilization),
+        cpuMsPerSecond: round((cpu.user + cpu.system) / 1_000 / seconds),
+      };
+    };
+    try {
+      for (let index = 0; index < sessions; index += 1) {
+        let probe: TransportPumpProbe | null = null;
+        const bundle = createCapabilityRunnerdCodexTransport({
+          runnerBinary: defaultCapabilityRunnerdBinary(),
+          codexCommand: process.execPath,
+          codexArgs: [fixture, String(deltas), String(DELTA_INTERVAL_MS)],
+          environment: {},
+          stateDirectory: join(root, `session-${index}`),
+          lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
+          controlPlaneRegistration: async (core) => {
+            probe = probeTransportPump(core, persistence);
+            return probe.registration();
+          },
+        });
+        open.push({ bundle, probe: () => probe! });
+        await bundle.transport.request("thread/start", { cwd: root, model: "fixture-no-model", dynamicTools: [] });
+      }
+
+      // Idle: every session is open and connected, and no event arrives.
+      const idleFresh = await measureIdle();
+
+      // Streaming: every session runs one paced turn at the same time.
+      const waits: number[] = [];
+      const terminalWaits: number[] = [];
+      await Promise.all(open.map(async ({ bundle, probe }) => {
+        const delivered: number[] = [];
+        let terminalAt: number | null = null;
+        const notifications = bundle.transport.notifications()[Symbol.asyncIterator]();
+        await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Stream the paced fixture." }] });
+        for (;;) {
+          const next = await notifications.next();
+          if (next.done) throw new Error("transport_pump_stream_ended_without_terminal");
+          if (next.value.method === "item/agentMessage/delta") delivered.push(performance.now());
+          if (next.value.method === "turn/completed") {
+            terminalAt = performance.now();
+            break;
+          }
+        }
+        const readable = probe().readable;
+        const streamed = readable.filter((event) => event.eventType === "item.delta");
+        const terminal = readable.find((event) => event.eventType === "turn.completed");
+        expect(delivered).toHaveLength(deltas);
+        expect(streamed).toHaveLength(deltas);
+        expect(terminal).toBeDefined();
+        streamed.forEach((event, index) => waits.push(delivered[index]! - event.at));
+        terminalWaits.push(terminalAt! - terminal!.at);
+      }));
+      waits.sort((left, right) => left - right);
+      terminalWaits.sort((left, right) => left - right);
+
+      // Idle again: each session now retains the turn's committed events.
+      const idleAfterTurn = await measureIdle();
+
+      process.stdout.write(`TRANSPORT_PUMP_BENCHMARK ${JSON.stringify({
+        schema: "paperclip.transport_pump_benchmark.v1",
+        persistence,
+        sessions,
+        retainedEventsPerSession: open[0]!.probe().readable.length,
+        idleFresh,
+        idleAfterTurn,
+        deliveryWaitMs: {
+          samples: waits.length,
+          mean: round(waits.reduce((total, wait) => total + wait, 0) / waits.length),
+          p50: round(percentile(waits, 0.5)),
+          p95: round(percentile(waits, 0.95)),
+          max: round(waits.at(-1)!),
+        },
+        terminalDeliveryWaitMs: {
+          samples: terminalWaits.length,
+          p50: round(percentile(terminalWaits, 0.5)),
+          max: round(terminalWaits.at(-1)!),
+        },
+      })}\n`);
+    } finally {
+      await Promise.all(open.map(({ bundle }) => bundle.transport.close().catch(() => undefined)));
+      await rm(root, { recursive: true, force: true });
+    }
+  },
+  120_000,
+);
diff --git a/packages/paperclip-runner/test/fixtures/fake-paced-codex-app-server.mjs b/packages/paperclip-runner/test/fixtures/fake-paced-codex-app-server.mjs
new file mode 100644
index 00000000000..dddfaf05af5
--- /dev/null
+++ b/packages/paperclip-runner/test/fixtures/fake-paced-codex-app-server.mjs
@@ -0,0 +1,110 @@
+// Credential-free, deterministic provider that streams one delta at a time.
+// No network, model, tool execution, or user files are used by this fixture.
+// Usage: node fake-paced-codex-app-server.mjs <deltaCount> <intervalMs>
+import { createInterface } from "node:readline";
+
+const [countArg, intervalArg] = process.argv.slice(2);
+const deltaCount = Number(countArg);
+const intervalMs = Number(intervalArg);
+if (
+  !Number.isInteger(deltaCount) || deltaCount < 1 || deltaCount > 512 ||
+  !Number.isInteger(intervalMs) || intervalMs < 1 || intervalMs > 1_000
+) {
+  throw new Error("paced_fixture_invalid_arguments");
+}
+const threadId = "paced-thread";
+const turns = new Map();
+let nextTurn = 0;
+const send = (value) => process.stdout.write(`${JSON.stringify(value)}\n`);
+
+function stream(turnId) {
+  let index = 0;
+  const timer = setInterval(() => {
+    if (turns.get(turnId) !== "inProgress") {
+      clearInterval(timer);
+      return;
+    }
+    if (index < deltaCount) {
+      send({
+        method: "item/agentMessage/delta",
+        params: {
+          threadId,
+          turnId,
+          itemId: `message-${turnId}`,
+          delta: `${index.toString().padStart(4, "0")} `,
+        },
+      });
+      index += 1;
+      return;
+    }
+    clearInterval(timer);
+    send({
+      method: "item/completed",
+      params: {
+        threadId,
+        turnId,
+        item: { id: `message-${turnId}`, type: "agentMessage", text: "Fixture complete." },
+      },
+    });
+    turns.set(turnId, "completed");
+    send({
+      method: "turn/completed",
+      params: { threadId, turn: { id: turnId, status: "completed" } },
+    });
+  }, intervalMs);
+}
+
+createInterface({ input: process.stdin }).on("line", (line) => {
+  const { id, method, params = {} } = JSON.parse(line);
+  if (!method || id === undefined) return;
+  if (method === "initialize") {
+    send({ id, result: { user: { sessionId: "paced-fixture" } } });
+  } else if (["thread/start", "thread/resume", "thread/read"].includes(method)) {
+    send({
+      id,
+      result: {
+        model: "fixture-no-model",
+        modelProvider: "fixture-no-provider",
+        thread: {
+          id: threadId,
+          sessionId: "paced-fixture",
+          status: { type: [...turns.values()].includes("inProgress") ? "active" : "idle" },
+          ...(params.includeTurns
+            ? { turns: [...turns].map(([turnId, status]) => ({ id: turnId, status })) }
+            : {}),
+        },
+      },
+    });
+  } else if (method === "thread/turns/list") {
+    const listed = [...turns].map(([turnId, status]) => ({
+      id: turnId, status, items: [], itemsView: "notLoaded",
+    }));
+    if (params.sortDirection === "desc") listed.reverse();
+    const offset = Number(params.cursor ?? 0);
+    const limit = params.limit ?? 100;
+    send({ id, result: {
+      data: listed.slice(offset, offset + limit),
+      nextCursor: offset + limit < listed.length ? String(offset + limit) : null,
+    } });
+  } else if (method === "turn/start") {
+    const turnId = `paced-turn-${++nextTurn}`;
+    turns.set(turnId, "inProgress");
+    send({ id, result: { turn: { id: turnId, status: "inProgress" } } });
+    send({
+      method: "turn/started",
+      params: { threadId, turn: { id: turnId, status: "inProgress" } },
+    });
+    stream(turnId);
+  } else if (method === "turn/interrupt") {
+    send({ id, result: {} });
+    if (turns.get(params.turnId) === "inProgress") {
+      turns.set(params.turnId, "interrupted");
+      send({
+        method: "turn/completed",
+        params: { threadId, turn: { id: params.turnId, status: "interrupted" } },
+      });
+    }
+  } else {
+    send({ id, error: { code: -32601, message: "paced_fixture_unsupported_method" } });
+  }
+});
diff --git a/packages/paperclip-runner/test/support/transport-pump-probe.ts b/packages/paperclip-runner/test/support/transport-pump-probe.ts
new file mode 100644
index 00000000000..91518eff25f
--- /dev/null
+++ b/packages/paperclip-runner/test/support/transport-pump-probe.ts
@@ -0,0 +1,103 @@
+import type { DurablePrpControlPlane } from "../../src/control-plane/durable-prp-control-plane.js";
+
+export interface ReadableEvent {
+  readonly sourceSeq: number;
+  readonly eventType: string;
+  /** `performance.now()` when a reader could first see this committed event. */
+  readonly at: number;
+}
+
+export interface TransportPumpProbe {
+  /** How often anything took the control plane's readable state. A transport takes it on every pump pass. */
+  stateReads(): number;
+  /** Committed events in the order readers could first see them. */
+  readonly readable: ReadableEvent[];
+  registration(): Promise<{ connectUrl: string; release: () => void }>;
+}
+
+/**
+ * Test-only observation of one control plane: every wrapped method delegates
+ * unchanged. It records when each committed event becomes readable and counts
+ * state reads, so a test can tell an event-driven reader from a polling one.
+ *
+ * `external` selects the production persistence mode, where a non-streaming
+ * event stays hidden from readers until its checkpoint is saved.
+ */
+export function probeTransportPump(
+  core: DurablePrpControlPlane,
+  persistence: "file" | "external",
+  onReadable?: (event: ReadableEvent) => void,
+): TransportPumpProbe {
+  const readable: ReadableEvent[] = [];
+  let lastSeq = 0;
+  let reads = 0;
+  let observing = false;
+  const observe = () => {
+    observing = true;
+    try {
+      for (const event of core.store.state.committedEvents) {
+        if (event.sourceSeq <= lastSeq) continue;
+        lastSeq = event.sourceSeq;
+        const seen = { sourceSeq: event.sourceSeq, eventType: event.eventType, at: performance.now() };
+        readable.push(seen);
+        onReadable?.(seen);
+      }
+    } finally {
+      observing = false;
+    }
+  };
+
+  // The working store exists before external persistence replaces the public one.
+  const working = core.store as unknown as Record<"commit" | "commitEvent" | "save", (...args: unknown[]) => unknown>;
+  for (const method of ["commit", "save"] as const) {
+    const original = working[method].bind(working);
+    working[method] = (...args: unknown[]) => {
+      const result = original(...args);
+      observe();
+      return result;
+    };
+  }
+  // A staged event write publishes its state partway through, right after it
+  // asks whether the sender still owns the session. Look once that synchronous
+  // step ends, which is before any timer or I/O callback can read the event.
+  const commitEvent = working.commitEvent.bind(working);
+  working.commitEvent = (build: unknown, isCurrentOwner: unknown, ...rest: unknown[]) =>
+    (commitEvent(build, () => {
+      queueMicrotask(observe);
+      return (isCurrentOwner as () => boolean)();
+    }, ...rest) as Promise<unknown>).finally(observe);
+  if (persistence === "external") {
+    core.setCheckpointPersistence(async () => undefined);
+    const flush = core.flushCheckpoint.bind(core);
+    core.flushCheckpoint = () => {
+      const saved = flush();
+      void saved.then(observe, () => undefined);
+      return saved;
+    };
+  }
+
+  const store = core.store;
+  let owner: object | null = store;
+  let descriptor: PropertyDescriptor | undefined;
+  while (owner !== null && descriptor === undefined) {
+    descriptor = Object.getOwnPropertyDescriptor(owner, "state");
+    owner = Object.getPrototypeOf(owner);
+  }
+  const readState = descriptor!.get!;
+  Object.defineProperty(store, "state", {
+    configurable: true,
+    get() {
+      if (!observing) reads += 1;
+      return readState.call(store);
+    },
+  });
+
+  return {
+    stateReads: () => reads,
+    readable,
+    async registration() {
+      await core.start();
+      return { connectUrl: core.connectUrl, release: () => undefined };
+    },
+  };
+}