diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc
index e171124eed1..f6ff1d595e8 100644
--- a/config/reliability-gates.jsonc
+++ b/config/reliability-gates.jsonc
@@ -3278,7 +3278,7 @@
           "platform": "linux",
           "command": "pnpm exec playwright test tests/e2e/daemon-generation-legacy-close-safety.spec.ts --config tests/playwright.config.ts --project electron-headless --workers=1",
           "result": "passed",
-          "durationSeconds": 5.1,
+          "durationSeconds": 6.6,
           "summary": "Current-head capability-gated oracle passed in E2E run https://github.com/stablyai/orca/actions/runs/30250731941/job/89928386794: capable v25/v26 roots and descendants survived with snapshot republish and post-close I/O, byte-identical legacy closes retained current-main shutdown behavior, and the unrelated fifth-worktree control survived."
         },
         {
@@ -8804,18 +8804,21 @@
       "providers": ["local", "daemon", "ssh", "wsl", "remote-runtime"],
       "coveredPlatforms": ["macos"],
       "coveredProviders": ["local"],
-      "coverageNotes": "A deterministic in-process runtime/SQLite/RPC harness covers two remint-equivalent panes across Run and Dispatch ownership changes, captures mocked PTY-controller writes, executes the production check, reopens a durable production-shape mismatch under a fresh runtime, and verifies Run Delivery acknowledgment plus worker-check consumption. A 151-row backlog proves reconciliation stays within one 50-row page and one coalesced mailbox wake per event-loop turn. The authority decision adds no wire field, renderer state, polling, or provider call. Daemon, paired remote-runtime, SSH, WSL, Linux, Windows, and headed restart evidence remain gaps.",
+      "coverageNotes": "A deterministic in-process runtime/SQLite/RPC harness covers two remint-equivalent panes across Run and Dispatch ownership changes, captures mocked PTY-controller writes, executes the production check, reopens durable production-shape state under a fresh runtime, and verifies Run Delivery acknowledgment plus worker-check consumption. A 151-row backlog proves live-idle reconciliation stays within one 50-row page and one coalesced continuation per event-loop turn; explicit checks cooperatively drain their start-of-check sequence snapshot before filtering or waiting. Controlled setImmediate races prove Run/Dispatch consumers are fenced after ownership changes, split current/stored handles both drain, cancellation stops paging, and a proven-inactive Dispatch transfers unread rows to its reachable Run for supported consumption and acknowledgment both after page routing and filtered wait cancellation. The authority decision adds no wire field, renderer state, polling, or provider call. Daemon, paired remote-runtime, live SSH, WSL, Linux, Windows, and headed restart evidence remain gaps.",
       "motivatingLinks": ["internal incident evidence: improve-vps-setup, 2026-08-10"],
-      "invariant": "Every injected orchestration pointer must select the same mailbox as the exact check from that pane and expose at least the notified unread messages through that mailbox's acknowledgment semantics. Rebinding cannot let foreign direct mail wake a new owner, newer Run mail cannot be announced behind an outstanding fixed Delivery, and restart cannot replay a successfully submitted pointer. Notification records only a durable pointer cursor; it never marks mail read or globally clears it.",
-      "oracle": "Create Run A with unread current-delivery mail addressed directly to its coordinator handle, bind the same pane to Run B, drive a controlled working-to-idle event, and require zero PTY pointers while Run B check returns zero and Run A mail stays unread. Reopen a database already containing that mismatch and require live-idle reconciliation to make the row actionable to Run A without pointing Run B. For same-Run mail, require one Run-pinned pointer, a bound Delivery containing that row, explicit acknowledgment, and no replay after a fresh runtime. Require reminted active-worker mail to retain Dispatch ownership, suppress newer Run pointers behind an outstanding Delivery, preserve waiter reservations, reject absent local PTYs, and reconcile a 151-row mismatch in 50-row pages with one mailbox wake per turn.",
+      "invariant": "Every injected orchestration pointer must select the same mailbox as the exact check from that pane and expose at least the notified unread messages through that mailbox's acknowledgment semantics. Rebinding cannot let foreign direct mail wake a new owner, newer Run mail cannot be announced behind an outstanding fixed Delivery, and restart cannot replay pointer text already accepted by the PTY even if delayed Enter has not fired. Notification records only a durable pointer cursor; it never marks mail read or globally clears it. Direct and Dispatch mail without a mixed-version-compatible pinned consuming action must not synthesize a terminal turn.",
+      "oracle": "Create Run A with unread current-delivery mail addressed directly to its coordinator handle, bind the same pane to Run B, drive a controlled working-to-idle event, and require zero PTY pointers while Run B check returns zero and Run A mail stays unread. Reopen a database already containing that mismatch and require live-idle reconciliation to make the row actionable to Run A without pointing Run B. For same-Run mail, require one Run-pinned pointer, a bound Delivery containing that row, explicit acknowledgment, and no replay after a fresh runtime, including restart between accepted pointer payload and delayed Enter. Require direct and Dispatch mail to remain explicitly checkable without an unpinned pointer, reminted active-worker mail to retain Dispatch ownership, newer Run mail to stay suppressed behind an outstanding fixed Delivery, waiter reservations to remain single-consumer across routing, absent local PTYs to retain mail, and a 151-row mismatch to reconcile in 50-row pages with one continuation per turn. During explicit paging, deterministically rebind the Run or complete the Dispatch and require a fenced response; require the inactive Dispatch's coordinator to consume and acknowledge the preserved rows through two fixed Run Deliveries; split current and stored handles and require one check to route both; add mail beyond the captured sequence and require finite snapshot completion; abort between pages and require paging to stop.",
       "commands": [
-        "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts",
+        "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-mailbox-routing-races.test.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts src/main/runtime/orchestration/formatter.test.ts src/main/providers/ssh-pty-provider.test.ts",
         "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.test.ts src/main/runtime/rpc/methods/orchestration.test.ts"
       ],
       "testFiles": [
         "src/main/runtime/orchestration-mailbox-notification-consistency.test.ts",
+        "src/main/runtime/orchestration-mailbox-routing-races.test.ts",
         "src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts",
         "src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts",
+        "src/main/runtime/orchestration/formatter.test.ts",
+        "src/main/providers/ssh-pty-provider.test.ts",
         "src/main/runtime/orca-runtime.test.ts",
         "src/main/runtime/terminal-send-stale-leaf-liveness.test.ts",
         "src/main/runtime/rpc/methods/orchestration.test.ts"
@@ -8829,9 +8832,11 @@
             "a fresh runtime reconciles a database that already contains the production mismatch",
             "large persisted mismatches route in 50-row pages with coalesced per-mailbox wakes",
             "Run pointers pin the existing mixed-version-compatible --run identity",
+            "direct and Dispatch mail remain durable without unpinned synthetic terminal turns",
             "a delayed pointer submit cannot press Enter after the agent becomes working",
             "an explicit check releases its staged pointer reservation without redrive",
-            "a provider-refused Enter remains undelivered and retryable",
+            "accepted pointer text is durably deduplicated before delayed Enter and provider Enter refusal",
+            "a known PTY exit releases staged rows for the replacement process",
             "a staged pointer follows its Run to a new live pane without cross-PTY duplication",
             "equivalent panes serialize one mailbox while newer mail arrives during Enter delay",
             "same-Run direct mail is delivered and acknowledged through the bound Run",
@@ -8840,6 +8845,8 @@
             "active worker direct mail is exposed by its Dispatch check after handle remint",
             "late direct mail is normalized when its displaced Run rebinds",
             "a bound waiter wakes without a duplicate PTY pointer",
+            "filtered waiters remain single-consumer when direct rows route into Run or Dispatch mailboxes",
+            "same-owner filtered checks drain later 50-row pages before waiting",
             "working-agent arrival uses bounded ownership lookup without a full mailbox scan"
           ]
         },
@@ -8848,26 +8855,38 @@
           "assertions": [
             "remote Dispatch attachment lookup remains bounded across pane remints and malformed suffix collisions"
           ]
+        },
+        {
+          "file": "src/main/runtime/orchestration-mailbox-routing-races.test.ts",
+          "assertions": [
+            "Run and Dispatch consumers are fenced after ownership changes during a routing-page yield",
+            "current and stored reminted handles both drain in one explicit check",
+            "explicit paging terminates at its indexed start-of-check sequence snapshot and stops on cancellation",
+            "a fenced inactive Dispatch transfers unread rows to its Run for supported consumption and acknowledgment",
+            "filtered Dispatch wait cancellation reconciles unrelated unread types before fencing",
+            "unfiltered and filtered parked reservations union in either arrival order",
+            "reminted Dispatch pane lookup uses a bounded partial expression index while exact legacy keys remain supported"
+          ]
         }
       ],
       "evidenceRuns": [
         {
-          "date": "2026-08-10",
+          "date": "2026-08-11",
           "runner": "local",
           "platform": "macos",
-          "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts",
+          "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-mailbox-routing-races.test.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts src/main/runtime/orchestration/formatter.test.ts src/main/providers/ssh-pty-provider.test.ts",
           "result": "passed",
           "durationSeconds": 5.1,
-          "summary": "37 tests passed across mailbox identity, persisted-upgrade reconciliation, 50-row paging and coalesced wakes, cross-PTY serialization, live ownership transfer, durable restart, submit races, outstanding Deliveries, reminted Dispatch ownership, acknowledgment, waiters, Run Delivery, and remote pane remints. The original 10-scenario mailbox oracle failed 10/10 on origin/main plus the test commit and again with the implementation commit explicitly reverted, including pointerCount > 0 with bound check count 0."
+          "summary": "127 tests passed across mailbox identity, direct and Dispatch pointer suppression, persisted reconciliation, 50-row paging and filtered waits, cross-PTY serialization, live ownership transfer, restart before delayed Enter, submit races, outstanding Deliveries, reminted Dispatch ownership, acknowledgment, remote pane remints, split-handle drains, ownership-change fencing, inactive-mailbox reachability after paging and filtered waits, snapshot termination, cancellation, reservation ordering, and bounded pane lookup. The original 10-scenario mailbox oracle failed 10/10 on origin/main plus the test commit and again with the implementation commit explicitly reverted, including pointerCount > 0 with bound check count 0."
         },
         {
-          "date": "2026-08-10",
+          "date": "2026-08-11",
           "runner": "local",
           "platform": "macos",
           "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.test.ts src/main/runtime/rpc/methods/orchestration.test.ts",
           "result": "passed",
-          "durationSeconds": 10.1,
-          "summary": "1,250 tests passed and 1 was skipped across runtime pointer delivery, PTY retirement and respawn, stale-leaf liveness, and orchestration RPC behavior."
+          "durationSeconds": 12.5,
+          "summary": "1,251 tests passed and 1 was skipped across Run-bound pointer delivery, PTY retirement and respawn, stale-leaf liveness, direct-mail routing, filtered waiter ownership, and orchestration RPC behavior."
         }
       ],
       "runtimeBudget": {
@@ -8876,7 +8895,7 @@
       },
       "flakeHistory": {
         "status": "unknown",
-        "evidence": "First deterministic local macOS run only; no sleeps or wall-clock delivery assertions are used."
+        "evidence": "Repeated deterministic local macOS runs passed during four review rounds; controlled fake timers and lifecycle events replace arbitrary sleeps and wall-clock delivery assertions."
       },
       "redGreenEvidence": {
         "status": "complete",
@@ -8884,7 +8903,7 @@
       },
       "performanceBudget": {
         "required": true,
-        "evidence": "Arrival ownership uses indexed identity and SELECT 1 LIMIT 1 probes. Pointer selection and direct-mail reconciliation are capped to 50 rows inside SQLite; a single setImmediate continuation yields between pages and coalesces each mailbox/type wake. Submit-time row revalidation uses the unique message-ID index, durable cursor writes update only the submitted batch in 500-ID chunks, and outstanding-Delivery suppression is a bounded existence query. No polling, subprocesses, provider calls, persistent timers, listeners, or renderer work were added."
+        "evidence": "Arrival ownership uses indexed identity and SELECT 1 LIMIT 1 probes, including exact and remint-stable Dispatch pane lookups. Live-idle direct-mail reconciliation is capped to 50 rows per event-loop turn with one coalesced setImmediate continuation; explicit checks capture an indexed maximum sequence per distinct owner handle, route only that finite snapshot with one cooperative setImmediate yield per page, and stop on request cancellation. Submit-time row revalidation uses the unique message-ID index, durable cursor writes update only the staged batch in 500-ID chunks, and outstanding-Delivery suppression is a bounded existence query. A proven-inactive Dispatch uses one transactional indexed mailbox UPDATE to preserve reachability without polling or repeated scans. No subprocesses, provider fanout, persistent timers, listeners, or renderer work were added."
       },
       "promotionCriteria": [
         "Add isolated live restart and SSH or paired-runtime evidence.",
@@ -8894,6 +8913,7 @@
         "No headed Electron restart journey was required to prove the runtime/PTY/check boundary.",
         "No daemon, paired remote-runtime, live SSH, WSL, Linux, or Windows journey has been collected.",
         "Daemon and SSH PTY writes remain transport-accepted rather than owner-acknowledged; capability-negotiated owner acknowledgments are separate remote-protocol work.",
+        "Direct and Dispatch mail stay durable and explicitly checkable but do not synthesize agent turns until a backward-compatible pinned action or passive UI exists.",
         "Passive renderer-native unread UI and opt-in coordinator wake policy remain separate product work."
       ],
       "demotionRule": "Demote or quarantine if the deterministic oracle flakes without a product or harness bug owned by orchestration."
diff --git a/src/main/providers/ssh-pty-provider.ts b/src/main/providers/ssh-pty-provider.ts
index 2bc52fe491d..9e95f445a50 100644
--- a/src/main/providers/ssh-pty-provider.ts
+++ b/src/main/providers/ssh-pty-provider.ts
@@ -23,6 +23,7 @@ import { SshPtySpawnExitRaceTracker } from './ssh-pty-spawn-exit-race'
 import { SshAgentSessionCapabilities } from './ssh-agent-session-capabilities'
 import type { PtyProcessInspection } from './pty-process-inspection'
 import { SSH_SESSION_EXPIRED_ERROR } from './ssh-pty-errors'
+import { writeToSshPty } from './ssh-pty-write'
 
 // Why: sequential relay teardown calls share one absolute budget; convert to the mux-relative timeout only at dispatch.
 function relayTimeoutOptions(deadlineMs: number | undefined): { timeoutMs: number } | undefined {
@@ -204,8 +205,8 @@ export class SshPtyProvider implements IPtyProvider {
     })
   }
 
-  write(id: string, data: string): void {
-    this.mux.notify('pty.data', { id: this.toRelayPtyId(id), data })
+  write(id: string, data: string): boolean {
+    return writeToSshPty(this.mux, this.toRelayPtyId(id), data)
   }
 
   resize(id: string, cols: number, rows: number): void {
diff --git a/src/main/providers/ssh-pty-write.test.ts b/src/main/providers/ssh-pty-write.test.ts
new file mode 100644
index 00000000000..e1a4f374695
--- /dev/null
+++ b/src/main/providers/ssh-pty-write.test.ts
@@ -0,0 +1,16 @@
+import { describe, expect, it, vi } from 'vitest'
+import { SshPtyProvider } from './ssh-pty-provider'
+
+describe('SSH PTY writes', () => {
+  it('rejects writes synchronously after the transport is disposed', () => {
+    const mux = {
+      isDisposed: vi.fn().mockReturnValue(true),
+      notify: vi.fn(),
+      onNotification: vi.fn()
+    }
+    const provider = new SshPtyProvider('conn-1', mux as never)
+
+    expect(provider.write('ssh:conn-1@@pty-1', 'pointer')).toBe(false)
+    expect(mux.notify).not.toHaveBeenCalled()
+  })
+})
diff --git a/src/main/providers/ssh-pty-write.ts b/src/main/providers/ssh-pty-write.ts
new file mode 100644
index 00000000000..d21522479c6
--- /dev/null
+++ b/src/main/providers/ssh-pty-write.ts
@@ -0,0 +1,13 @@
+import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer'
+
+export function writeToSshPty(
+  mux: SshChannelMultiplexer,
+  relayPtyId: string,
+  data: string
+): boolean {
+  if (mux.isDisposed()) {
+    return false
+  }
+  mux.notify('pty.data', { id: relayPtyId, data })
+  return !mux.isDisposed()
+}
diff --git a/src/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts
index 893d9326ac2..50ca3dc6e30 100644
--- a/src/main/runtime/orca-runtime.test.ts
+++ b/src/main/runtime/orca-runtime.test.ts
@@ -1154,6 +1154,42 @@ class InMemoryOrchestrationMessages {
     return [...this.runs.values()].find((run) => run.coordinator_pane_key === paneKey)
   }
 
+  hasUndeliveredDirectMessageForRun(runId: string, directHandle: string): boolean {
+    return this.messages.some(
+      (message) =>
+        message.run_id === runId &&
+        message.to_handle === directHandle &&
+        message.read === 0 &&
+        !message.delivered_at
+    )
+  }
+
+  routeUnreadDirectMessagesToRunMailbox(
+    runId: string,
+    directHandle: string
+  ): { routedCount: number; hasMore: boolean; types: MessageType[] } {
+    const routed = this.messages.filter(
+      (message) =>
+        message.run_id === runId && message.to_handle === directHandle && message.read === 0
+    )
+    for (const message of routed) {
+      message.to_handle = `run:${runId}`
+    }
+    return {
+      routedCount: routed.length,
+      hasMore: false,
+      types: [...new Set(routed.map((message) => message.type))]
+    }
+  }
+
+  areUnreadMessages(toHandle: string, ids: string[]): boolean {
+    return ids.every((id) =>
+      this.messages.some(
+        (message) => message.id === id && message.to_handle === toHandle && message.read === 0
+      )
+    )
+  }
+
   markAsDelivered(ids: string[]): void {
     const deliveredIds = new Set(ids)
     for (const message of this.messages) {
@@ -1163,6 +1199,15 @@ class InMemoryOrchestrationMessages {
     }
   }
 
+  markAsUndelivered(ids: string[]): void {
+    const releasedIds = new Set(ids)
+    for (const message of this.messages) {
+      if (releasedIds.has(message.id) && message.read === 0) {
+        message.delivered_at = null
+      }
+    }
+  }
+
   close(): void {}
 }
 
@@ -1173,6 +1218,15 @@ function setInMemoryOrchestrationMessages(
   runtime.setOrchestrationDb(db as unknown as OrchestrationDb)
 }
 
+function bindSinglePtyRun(db: InMemoryOrchestrationMessages, terminalHandle: string): string {
+  db.setRun({
+    id: 'run_test',
+    coordinator_handle: terminalHandle,
+    coordinator_pane_key: 'tab-1:pane:1'
+  })
+  return 'run:run_test'
+}
+
 function expectStablePaneKeyEnv(env: Record<string, string>): string {
   expect(env.ORCA_TAB_ID).toMatch(UUID_RE)
   const leafId = env.ORCA_PANE_KEY?.slice(`${env.ORCA_TAB_ID}:`.length)
@@ -19572,6 +19626,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      const mailbox = bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       db.setActiveCoordinatorRun({ coordinator_handle: 'term_other' })
@@ -19586,7 +19641,7 @@ describe('OrcaRuntimeService', () => {
       await vi.advanceTimersByTimeAsync(500)
       expect(write).toHaveBeenCalledWith('pty-1', '\r')
 
-      const unread = db.getUnreadMessages(terminal.handle)
+      const unread = db.getUnreadMessages(mailbox)
       expect(unread).toHaveLength(1)
       expect(unread[0].read).toBe(0)
       expect(unread[0].delivered_at).toEqual(expect.any(String))
@@ -19611,6 +19666,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      const mailbox = bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       db.setActiveCoordinatorRun({ coordinator_handle: terminal.handle })
@@ -19632,7 +19688,7 @@ describe('OrcaRuntimeService', () => {
       )
       expect(submitWrites).toHaveLength(1)
 
-      const unread = db.getUnreadMessages(terminal.handle)
+      const unread = db.getUnreadMessages(mailbox)
       expect(unread).toHaveLength(1)
       expect(unread[0].read).toBe(0)
       expect(unread[0].delivered_at).toEqual(expect.any(String))
@@ -19657,6 +19713,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      const mailbox = bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;\u280b Cursor Agent\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Cursor ready\x07', 101)
       db.insertMessage({ from: 'term_sender', to: terminal.handle, subject: 'hello cursor' })
@@ -19673,7 +19730,7 @@ describe('OrcaRuntimeService', () => {
       )
       expect(submitWrites).toHaveLength(0)
 
-      const unread = db.getUnreadMessages(terminal.handle)
+      const unread = db.getUnreadMessages(mailbox)
       expect(unread).toHaveLength(1)
       expect(unread[0].read).toBe(0)
       expect(unread[0].delivered_at).toEqual(expect.any(String))
@@ -19698,6 +19755,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime, 'pty-1', { tabTitle: 'cursor-repro-branch' })
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;. Investigate Cursor Agent\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;* Investigate Cursor Agent\x07', 101)
       db.insertMessage({ from: 'term_sender', to: terminal.handle, subject: 'hello claude' })
@@ -19731,6 +19789,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       db.insertMessage({ from: 'term_sender', to: terminal.handle, subject: 'hello' })
@@ -34577,6 +34636,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35009,6 +35069,7 @@ describe('OrcaRuntimeService', () => {
     syncSinglePty(runtime)
 
     const [terminal] = (await runtime.listTerminals()).terminals
+    bindSinglePtyRun(db, terminal.handle)
     runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
     const message = db.insertMessage({
       from: 'sender',
@@ -35042,6 +35103,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      bindSinglePtyRun(db, terminal.handle)
       runtime.seedTerminalRestoreTail('pty-1', { lastTitle: 'Codex done' })
       const message = db.insertMessage({
         from: 'sender',
@@ -35084,6 +35146,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      bindSinglePtyRun(db, terminal.handle)
       // Why: the persisted title is historical — the agent may have gone busy
       // across the relaunch, so a seeded 'idle' must not authorize a PTY write.
       runtime.seedTerminalRestoreTail('pty-1', { lastTitle: 'Codex done' })
@@ -35131,6 +35194,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      const mailbox = bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35141,8 +35205,8 @@ describe('OrcaRuntimeService', () => {
       // shared in-flight promise put a no-waiter notify inside that window, so the
       // push must not inject rows the resolved check is about to return.
       const consumed = runtime
-        .waitForMessage(terminal.handle, { timeoutMs: 5_000 })
-        .then(() => db.getUnreadMessages(terminal.handle).map((row) => (row.read = 1)))
+        .waitForMessage(mailbox, { timeoutMs: 5_000 })
+        .then(() => db.getUnreadMessages(mailbox).map((row) => (row.read = 1)))
       const first = db.insertMessage({ from: 'sender', to: terminal.handle, subject: 'pulled' })
       runtime.notifyMessageArrived(terminal.handle, 'status')
       const second = db.insertMessage({
@@ -35178,6 +35242,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      const mailbox = bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35187,7 +35252,7 @@ describe('OrcaRuntimeService', () => {
       // `status` notify is unclaimed and pushes, but the worker_done row landing
       // in the same drain belongs to this waiter's check — injecting it too would
       // deliver that completion twice (pane + check return).
-      const waitPromise = runtime.waitForMessage(terminal.handle, {
+      const waitPromise = runtime.waitForMessage(mailbox, {
         typeFilter: ['worker_done'],
         timeoutMs: 5_000
       })
@@ -35212,7 +35277,7 @@ describe('OrcaRuntimeService', () => {
         .map(([, data]) => data)
         .filter((data): data is string => typeof data === 'string')
       expect(payloads).toContain(
-        '\nYou have 1 orchestration message. Run `orca orchestration check`.\n'
+        '\nYou have 1 orchestration message. Run `orca orchestration check --run run_test`.\n'
       )
       expect(payloads.some((data) => data.includes('reserved completion'))).toBe(false)
       expect(status.delivered_at).toEqual(expect.any(String))
@@ -35238,6 +35303,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      const mailbox = bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35253,7 +35319,7 @@ describe('OrcaRuntimeService', () => {
       // blocks before the deferred push runs still owns this row, so only the
       // push-time read of live waiters can keep it out of the pane.
       runtime.notifyMessageArrived(terminal.handle, 'status')
-      const waitPromise = runtime.waitForMessage(terminal.handle, {
+      const waitPromise = runtime.waitForMessage(mailbox, {
         typeFilter: ['status'],
         timeoutMs: 5_000
       })
@@ -35340,6 +35406,7 @@ describe('OrcaRuntimeService', () => {
 
       setInMemoryOrchestrationMessages(runtime, db)
       const [republished] = (await runtime.listTerminals()).terminals
+      bindSinglePtyRun(db, republished.handle)
       const message = db.insertMessage({
         from: 'sender',
         to: republished.handle,
@@ -35372,6 +35439,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35429,6 +35497,7 @@ describe('OrcaRuntimeService', () => {
       runtime.onPtyExit('pty-1', 0)
       runtime.onPtySpawned('pty-1', undefined, { awaitsRegistration: false })
       setInMemoryOrchestrationMessages(runtime, db)
+      bindSinglePtyRun(db, terminal.handle)
       const message = db.insertMessage({
         from: 'sender',
         to: terminal.handle,
@@ -35473,6 +35542,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      const mailbox = bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35480,7 +35550,7 @@ describe('OrcaRuntimeService', () => {
       // Why: a `check --wait --types worker_done` waiter never returns a status
       // row — check re-reads under the same filter — so it is not the consumer
       // and treating it as one would strand the message (#12536).
-      const waitPromise = runtime.waitForMessage(terminal.handle, {
+      const waitPromise = runtime.waitForMessage(mailbox, {
         typeFilter: ['worker_done'],
         timeoutMs: 5_000
       })
@@ -35526,6 +35596,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      const mailbox = bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35538,7 +35609,7 @@ describe('OrcaRuntimeService', () => {
 
       // Why: blocked orchestration.check --wait is an explicit pull; push must
       // not stamp delivered_at or type into the pane (double delivery, #12584).
-      const waitPromise = runtime.waitForMessage(terminal.handle, { timeoutMs: 5_000 })
+      const waitPromise = runtime.waitForMessage(mailbox, { timeoutMs: 5_000 })
       runtime.notifyMessageArrived(terminal.handle, 'status')
 
       await expect(waitPromise).resolves.toBe('notified')
@@ -35565,6 +35636,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35612,6 +35684,7 @@ describe('OrcaRuntimeService', () => {
       syncSinglePty(runtime)
 
       const [terminal] = (await runtime.listTerminals()).terminals
+      bindSinglePtyRun(db, terminal.handle)
       runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
       runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
       await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
@@ -35670,6 +35743,7 @@ describe('OrcaRuntimeService', () => {
     syncSinglePty(runtime)
 
     const [terminal] = (await runtime.listTerminals()).terminals
+    bindSinglePtyRun(db, terminal.handle)
     runtime.onPtyData('pty-1', '\x1b]0;Codex working\x07', 100)
     runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101)
     await runtime.waitForTerminal(terminal.handle, { condition: 'tui-idle' })
diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts
index c857a0d22c6..63a737d4f2a 100644
--- a/src/main/runtime/orca-runtime.ts
+++ b/src/main/runtime/orca-runtime.ts
@@ -2899,13 +2899,10 @@ export class OrcaRuntimeService {
   private handles = new Map<string, TerminalHandleRecord>()
   private handleByLeafKey = new Map<string, string>()
   private handleByPtyId = new Map<string, string>()
-  // Why: provisional pointer state serializes PTY injection before the durable delivered_at commit.
-  private readonly lastPointedMessageSequenceByHandle = new Map<string, number>()
-  // Why: a waiter can reserve an older row while a newer row advances the sequence watermark.
-  private readonly pointedMessageIdsByHandle = new Map<string, Set<string>>()
   private readonly mailPointerRepointScheduler = new MailPointerRepointScheduler((handle) =>
     this.repointPendingMessagesForHandle(handle)
   )
+  // Why: in-process reservations serialize delayed Enter after durable pointer staging.
   private readonly pointedMessageWatermarkOwnerByHandle = new Map<
     string,
     { ptyId: string; sequence: number; leafKey: string; active: boolean }
@@ -2915,7 +2912,7 @@ export class OrcaRuntimeService {
     string,
     ReadonlySet<string> | null
   >()
-  private readonly pendingForeignDirectReconciliationsByLeafKey = new Set<string>()
+  private readonly pendingDirectMailboxReconciliationsByLeafKey = new Set<string>()
   private syntheticTerminalHandles = new Set<string>()
   private detachedPreAllocatedLeaves = new Map<string, RuntimeLeafRecord>()
   private graphSyncCallbacks: (() => void)[] = []
@@ -3953,8 +3950,6 @@ export class OrcaRuntimeService {
   setOrchestrationDb(db: OrchestrationDb): void {
     this.stopOrchestrationFederationRelay()
     this.mailPointerRepointScheduler.clear()
-    this.lastPointedMessageSequenceByHandle.clear()
-    this.pointedMessageIdsByHandle.clear()
     this._orchestrationDb = db
     this.ensureOrchestrationFederationRelay()
     this.scheduleRestoredMessageRepoints()
@@ -32428,14 +32423,14 @@ export class OrcaRuntimeService {
     }
   }
 
-  private scheduleForeignDirectReconciliation(leaf: RuntimeLeafRecord): void {
+  private scheduleDirectMailboxReconciliation(leaf: RuntimeLeafRecord): void {
     const leafKey = this.getLeafKey(leaf.tabId, leaf.leafId)
-    if (this.pendingForeignDirectReconciliationsByLeafKey.has(leafKey)) {
+    if (this.pendingDirectMailboxReconciliationsByLeafKey.has(leafKey)) {
       return
     }
-    this.pendingForeignDirectReconciliationsByLeafKey.add(leafKey)
+    this.pendingDirectMailboxReconciliationsByLeafKey.add(leafKey)
     setImmediate(() => {
-      this.pendingForeignDirectReconciliationsByLeafKey.delete(leafKey)
+      this.pendingDirectMailboxReconciliationsByLeafKey.delete(leafKey)
       const currentLeaf = this.leaves.get(leafKey)
       if (!currentLeaf || currentLeaf.ptyId !== leaf.ptyId) {
         return
@@ -32443,6 +32438,7 @@ export class OrcaRuntimeService {
       this.notifyForwardedOrchestrationMailboxes(
         this.routeForeignDirectMessagesForLeaf(currentLeaf)
       )
+      this.resolveActionableMailboxForLeaf(currentLeaf)
     })
   }
 
@@ -32476,7 +32472,7 @@ export class OrcaRuntimeService {
       paneKey
     )
     if (routed?.hasMore) {
-      this.scheduleForeignDirectReconciliation(leaf)
+      this.scheduleDirectMailboxReconciliation(leaf)
     }
     return routed?.mailboxes ?? []
   }
@@ -32491,6 +32487,25 @@ export class OrcaRuntimeService {
     }
   }
 
+  private wakeRoutedMessageWaiters(mailboxHandle: string, types: readonly string[]): void {
+    const waiters = [...(this.messageWaitersByHandle.get(mailboxHandle) ?? [])]
+    if (waiters.length === 0 || types.length === 0) {
+      return
+    }
+    // Why: the arrival hook must snapshot waiter ownership before routed rows wake those waiters.
+    queueMicrotask(() => {
+      const liveWaiters = this.messageWaitersByHandle.get(mailboxHandle)
+      for (const waiter of waiters) {
+        if (
+          liveWaiters?.has(waiter) &&
+          (!waiter.typeFilter || types.some((type) => waiter.typeFilter?.includes(type)))
+        ) {
+          this.resolveMessageWaiter(waiter, 'notified')
+        }
+      }
+    })
+  }
+
   private resolveActionableMailboxForLeaf(
     leaf: RuntimeLeafRecord,
     requestedMailbox?: string,
@@ -32515,7 +32530,10 @@ export class OrcaRuntimeService {
         return null
       }
       const hasDirectRunMail =
-        this._orchestrationDb.hasUndeliveredDirectMessageForRun?.(run.id, terminalHandle) ?? false
+        options.routeDirectMail === false
+          ? false
+          : (this._orchestrationDb.hasUndeliveredDirectMessageForRun?.(run.id, terminalHandle) ??
+            false)
       if (
         options.requireRequestedMail &&
         requestedMailbox === terminalHandle &&
@@ -32524,7 +32542,14 @@ export class OrcaRuntimeService {
         return null
       }
       if (hasDirectRunMail && options.routeDirectMail !== false) {
-        this._orchestrationDb.routeUnreadDirectMessagesToRunMailbox?.(run.id, terminalHandle)
+        const routed = this._orchestrationDb.routeUnreadDirectMessagesToRunMailbox?.(
+          run.id,
+          terminalHandle
+        )
+        this.wakeRoutedMessageWaiters(runMailbox, routed?.types ?? [])
+        if (routed?.hasMore) {
+          this.scheduleDirectMailboxReconciliation(leaf)
+        }
       }
       return runMailbox
     }
@@ -32540,10 +32565,12 @@ export class OrcaRuntimeService {
         return null
       }
       const hasDirectDispatchMail =
-        this._orchestrationDb.hasUndeliveredDirectMessageForRun?.(
-          dispatch.run_id,
-          terminalHandle
-        ) ?? false
+        options.routeDirectMail === false
+          ? false
+          : (this._orchestrationDb.hasUndeliveredDirectMessageForRun?.(
+              dispatch.run_id,
+              terminalHandle
+            ) ?? false)
       if (
         options.requireRequestedMail &&
         requestedMailbox === terminalHandle &&
@@ -32552,11 +32579,15 @@ export class OrcaRuntimeService {
         return null
       }
       if (hasDirectDispatchMail && options.routeDirectMail !== false) {
-        this._orchestrationDb.routeUnreadDirectMessagesToDispatchMailbox?.(
+        const routed = this._orchestrationDb.routeUnreadDirectMessagesToDispatchMailbox?.(
           dispatch.id,
           dispatch.run_id,
           terminalHandle
         )
+        this.wakeRoutedMessageWaiters(dispatchMailbox, routed?.types ?? [])
+        if (routed?.hasMore) {
+          this.scheduleDirectMailboxReconciliation(leaf)
+        }
       }
       return dispatchMailbox
     }
@@ -33237,7 +33268,7 @@ export class OrcaRuntimeService {
   // retire must not clear a newer same-id flight or flush its parked trigger.
   private readonly messageDeliveryFlightsByPtyId = new Map<
     string,
-    { enterTimer: ReturnType<typeof setTimeout> | null }
+    { enterTimer: ReturnType<typeof setTimeout> | null; stagedMessageIds: string[] }
   >()
 
   private readonly parkedMessageRedeliveriesByPtyId = new Map<
@@ -33247,7 +33278,7 @@ export class OrcaRuntimeService {
 
   private settlePendingMessageDelivery(
     ptyId: string,
-    flight: { enterTimer: ReturnType<typeof setTimeout> | null }
+    flight: { enterTimer: ReturnType<typeof setTimeout> | null; stagedMessageIds: string[] }
   ): void {
     if (this.messageDeliveryFlightsByPtyId.get(ptyId) !== flight) {
       return
@@ -33338,7 +33369,7 @@ export class OrcaRuntimeService {
     reservedTypes: ReadonlySet<string> | undefined
   ): void {
     const prior = this.parkedMessageRedeliveryTypesByMailboxHandle.get(mailboxHandle)
-    if (prior === null || (prior === undefined && reservedTypes === undefined)) {
+    if (!this.parkedMessageRedeliveryTypesByMailboxHandle.has(mailboxHandle) && !reservedTypes) {
       this.parkedMessageRedeliveryTypesByMailboxHandle.set(mailboxHandle, null)
       return
     }
@@ -33369,19 +33400,18 @@ export class OrcaRuntimeService {
     if (flight?.enterTimer != null) {
       clearTimeout(flight.enterTimer)
     }
+    if (flight?.stagedMessageIds.length) {
+      this._orchestrationDb?.markAsUndelivered(flight.stagedMessageIds)
+    }
     this.messageDeliveryFlightsByPtyId.delete(ptyId)
     this.parkedMessageRedeliveriesByPtyId.delete(ptyId)
     for (const leaf of this.getLeavesForPty(ptyId)) {
       const handle = this.handleByLeafKey.get(this.getLeafKey(leaf.tabId, leaf.leafId))
       if (handle) {
-        this.lastPointedMessageSequenceByHandle.delete(handle)
-        this.pointedMessageIdsByHandle.delete(handle)
         this.mailPointerRepointScheduler.schedule(handle)
       }
       const run = this._orchestrationDb?.getCurrentRunForPane?.(`${leaf.tabId}:${leaf.leafId}`)
       if (run) {
-        this.lastPointedMessageSequenceByHandle.delete(`run:${run.id}`)
-        this.pointedMessageIdsByHandle.delete(`run:${run.id}`)
         this.mailPointerRepointScheduler.schedule(`run:${run.id}`)
       }
     }
@@ -33416,7 +33446,7 @@ export class OrcaRuntimeService {
     if (messages.some((message) => messageTypeHasLiveWaiter(waiters, message.type))) {
       return 'release'
     }
-    return (this._orchestrationDb?.areUndeliveredUnreadMessages?.(
+    return (this._orchestrationDb?.areUnreadMessages?.(
       mailboxHandle,
       messages.map((message) => message.id)
     ) ?? true)
@@ -33442,11 +33472,11 @@ export class OrcaRuntimeService {
       return
     }
     const mailboxHandle = options.mailboxHandle ?? handle
+    if (!mailboxHandle.startsWith('run:')) {
+      return
+    }
 
-    if (
-      mailboxHandle.startsWith('run:') &&
-      this._orchestrationDb.hasOutstandingRunDelivery?.(mailboxHandle.slice('run:'.length))
-    ) {
+    if (this._orchestrationDb.hasOutstandingRunDelivery?.(mailboxHandle.slice('run:'.length))) {
       return
     }
 
@@ -33489,18 +33519,6 @@ export class OrcaRuntimeService {
       excludeTypes: [...excludedTypes],
       limit: ORCHESTRATION_DELIVERY_BATCH_LIMIT
     })
-    const pendingIds = new Set(pending.map((message) => message.id))
-    const pointedIds = this.pointedMessageIdsByHandle.get(mailboxHandle)
-    if (pointedIds) {
-      for (const id of pointedIds) {
-        if (!pendingIds.has(id)) {
-          pointedIds.delete(id)
-        }
-      }
-      if (pointedIds.size === 0) {
-        this.pointedMessageIdsByHandle.delete(mailboxHandle)
-      }
-    }
     const unread = pending
       .filter(
         (message) =>
@@ -33512,16 +33530,6 @@ export class OrcaRuntimeService {
       return
     }
 
-    const watermark = this.lastPointedMessageSequenceByHandle.get(mailboxHandle) ?? -1
-    const priorPointedIds = this.pointedMessageIdsByHandle.get(mailboxHandle)
-    if (
-      !unread.some(
-        (message) => message.sequence > watermark || priorPointedIds?.has(message.id) !== true
-      )
-    ) {
-      return
-    }
-
     if (!leaf.writable || !leaf.ptyId) {
       return
     }
@@ -33599,7 +33607,10 @@ export class OrcaRuntimeService {
     }
 
     const deliveryPtyId = leaf.ptyId
-    const flight: { enterTimer: ReturnType<typeof setTimeout> | null } = { enterTimer: null }
+    const flight: {
+      enterTimer: ReturnType<typeof setTimeout> | null
+      stagedMessageIds: string[]
+    } = { enterTimer: null, stagedMessageIds: [] }
     this.messageDeliveryFlightsByPtyId.set(deliveryPtyId, flight)
     let settlesInEnterCallback = false
     try {
@@ -33608,16 +33619,8 @@ export class OrcaRuntimeService {
       if (!wrote) {
         return
       }
-      this.lastPointedMessageSequenceByHandle.set(
-        mailboxHandle,
-        Math.max(watermark, newestSequence)
-      )
-      const pointedIdsAfterWrite =
-        this.pointedMessageIdsByHandle.get(mailboxHandle) ?? new Set<string>()
-      for (const message of unread) {
-        pointedIdsAfterWrite.add(message.id)
-      }
-      this.pointedMessageIdsByHandle.set(mailboxHandle, pointedIdsAfterWrite)
+      flight.stagedMessageIds = unread.map((message) => message.id)
+      this._orchestrationDb.markAsDelivered(flight.stagedMessageIds)
       this.setPointedMessageWatermark(
         mailboxHandle,
         newestSequence,
@@ -33627,17 +33630,7 @@ export class OrcaRuntimeService {
 
       const tabTitle = this.tabs.get(leaf.tabId)?.title
       if (isCursorAgentOrchestrationTarget(leaf, tabTitle)) {
-        let delivered = false
-        try {
-          this._orchestrationDb.markAsDelivered(unread.map((message) => message.id))
-          delivered = true
-        } finally {
-          if (delivered) {
-            this.clearPointedMessageWatermark(mailboxHandle, newestSequence, deliveryPtyId)
-          } else {
-            this.deactivatePointedMessageWatermark(mailboxHandle, newestSequence, deliveryPtyId)
-          }
-        }
+        this.clearPointedMessageWatermark(mailboxHandle, newestSequence, deliveryPtyId)
         this.redrivePendingMessageMailbox(mailboxHandle)
         return
       }
@@ -33645,7 +33638,7 @@ export class OrcaRuntimeService {
       // Why: agent TUIs can swallow a \r in the same PTY write; submit separately after a delay.
       flight.enterTimer = setTimeout(() => {
         let clearAndRedrive = false
-        let delivered = false
+        let submitted = false
         let releaseWithoutRedrive = false
         let finalizeReservation = true
         void this.isLeafPtyProvenAbsent(deliveryPtyId)
@@ -33679,11 +33672,7 @@ export class OrcaRuntimeService {
               releaseWithoutRedrive = true
               return
             }
-            const submitted = this.ptyController?.write(deliveryPtyId, '\r') ?? false
-            if (submitted) {
-              this._orchestrationDb?.markAsDelivered(unread.map((message) => message.id))
-              delivered = true
-            }
+            submitted = this.ptyController?.write(deliveryPtyId, '\r') ?? false
           })
           .catch(() => {
             // Terminal liveness is uncertain; mail remains queued for explicit check.
@@ -33691,8 +33680,11 @@ export class OrcaRuntimeService {
           .finally(() => {
             let released = false
             if (finalizeReservation) {
+              if (clearAndRedrive) {
+                this._orchestrationDb?.markAsUndelivered(unread.map((message) => message.id))
+              }
               released =
-                delivered || clearAndRedrive || releaseWithoutRedrive
+                submitted || clearAndRedrive || releaseWithoutRedrive
                   ? this.clearPointedMessageWatermark(mailboxHandle, newestSequence, deliveryPtyId)
                   : this.deactivatePointedMessageWatermark(
                       mailboxHandle,
diff --git a/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts b/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts
index f4902ce06d0..eafee93b331 100644
--- a/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts
+++ b/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts
@@ -3,9 +3,29 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { afterEach, describe, expect, it, vi } from 'vitest'
 import { ORCHESTRATION_CONTRACT_VERSION } from '../../shared/protocol-version'
-import type Database from '../sqlite/sync-database'
-import { OrcaRuntimeService } from './orca-runtime'
 import { OrchestrationDb } from './orchestration/db'
+import {
+  checkBoundMailbox,
+  createBoundRun,
+  createDatabase,
+  createRuntime,
+  driveToLiveIdle,
+  insertDirectRunMessage,
+  LAUNCH_TOKEN,
+  LEAF_ID,
+  PANE_KEY,
+  pointerCount,
+  PTY_ID,
+  registerSecondPane,
+  SECOND_LAUNCH_TOKEN,
+  SECOND_LEAF_ID,
+  SECOND_PANE_KEY,
+  SECOND_PTY_ID,
+  SECOND_TERMINAL_HANDLE,
+  sqliteFor,
+  temporaryDirectories,
+  TERMINAL_HANDLE
+} from './orchestration-mailbox-notification-test-harness'
 import { RpcDispatcher } from './rpc/dispatcher'
 import { ORCHESTRATION_METHODS } from './rpc/methods/orchestration'
 
@@ -16,212 +36,6 @@ vi.mock('electron', () => ({
   webContents: { fromId: vi.fn(() => null) }
 }))
 
-const TAB_ID = '11111111-1111-4111-8111-111111111111'
-const LEAF_ID = '22222222-2222-4222-8222-222222222222'
-const PANE_KEY = `${TAB_ID}:${LEAF_ID}`
-const PTY_ID = 'pty-mailbox'
-const TERMINAL_HANDLE = 'term_mailbox_consistency'
-const SECOND_TAB_ID = '33333333-3333-4333-8333-333333333333'
-const SECOND_LEAF_ID = '44444444-4444-4444-8444-444444444444'
-const SECOND_PANE_KEY = `${SECOND_TAB_ID}:${SECOND_LEAF_ID}`
-const SECOND_PTY_ID = 'pty-mailbox-second'
-const SECOND_TERMINAL_HANDLE = 'term_mailbox_consistency_second'
-const SECOND_LAUNCH_TOKEN = 'mailbox-consistency-second-launch'
-const WORKTREE_ID = 'repo-mailbox::/tmp/mailbox'
-const LAUNCH_TOKEN = 'mailbox-consistency-launch'
-const temporaryDirectories: string[] = []
-
-function createDatabase(prefix: string): OrchestrationDb {
-  const directory = mkdtempSync(join(tmpdir(), prefix))
-  temporaryDirectories.push(directory)
-  return new OrchestrationDb(join(directory, 'orchestration.db'))
-}
-
-function sqliteFor(db: OrchestrationDb): Database.Database {
-  return (db as unknown as { db: Database.Database }).db
-}
-
-function createBoundRun(db: OrchestrationDb, objective: string) {
-  return db.createRun({
-    objective,
-    coordinatorHandle: TERMINAL_HANDLE,
-    coordinatorPaneKey: PANE_KEY
-  })
-}
-
-function insertDirectRunMessage(db: OrchestrationDb, runId: string, subject: string) {
-  return db.insertMessage({
-    from: 'term_worker',
-    to: TERMINAL_HANDLE,
-    subject,
-    type: 'status',
-    runId,
-    deliveryContract: 'current_delivery'
-  })
-}
-
-type Harness = {
-  runtime: OrcaRuntimeService
-  write: ReturnType<typeof vi.fn>
-}
-
-type CheckResult = {
-  runId: string
-  dispatchId?: string
-  deliveryId: string | null
-  count: number
-  messages: unknown[]
-  acknowledged?: string | null
-}
-
-function createRuntime(db: OrchestrationDb): Harness {
-  const runtime = new OrcaRuntimeService(null, undefined, {
-    attestAgentHookCompatibilityAuthority: ({ paneKey }) =>
-      paneKey === PANE_KEY || paneKey.startsWith(`${SECOND_TAB_ID}:`)
-        ? { paneKey, source: 'current_hook' }
-        : null
-  })
-  const write = vi.fn(() => true)
-  runtime.setOrchestrationDb(db)
-  runtime.setPtyController({
-    write,
-    kill: vi.fn(),
-    getForegroundProcess: async () => null
-  })
-  runtime.registerPty(PTY_ID, WORKTREE_ID, null, {
-    tabId: TAB_ID,
-    leafId: LEAF_ID,
-    incarnationId: 'mailbox-incarnation',
-    agentLaunchAuthority: { launchToken: LAUNCH_TOKEN, launchAgent: 'codex' }
-  })
-  runtime.registerPreAllocatedHandleForPty(PTY_ID, TERMINAL_HANDLE)
-  runtime.attachWindow(1)
-  runtime.syncWindowGraph(1, {
-    tabs: [
-      {
-        tabId: TAB_ID,
-        worktreeId: WORKTREE_ID,
-        title: 'Codex',
-        activeLeafId: LEAF_ID,
-        layout: null
-      }
-    ],
-    leaves: [
-      {
-        tabId: TAB_ID,
-        worktreeId: WORKTREE_ID,
-        leafId: LEAF_ID,
-        paneRuntimeId: 1,
-        ptyId: PTY_ID
-      }
-    ]
-  })
-  return { runtime, write }
-}
-
-function registerSecondPane(
-  runtime: OrcaRuntimeService,
-  leafId = SECOND_LEAF_ID,
-  includeFirstPane = true
-): void {
-  runtime.registerPty(SECOND_PTY_ID, WORKTREE_ID, null, {
-    tabId: SECOND_TAB_ID,
-    leafId,
-    incarnationId: 'mailbox-second-incarnation',
-    agentLaunchAuthority: { launchToken: SECOND_LAUNCH_TOKEN, launchAgent: 'codex' }
-  })
-  runtime.registerPreAllocatedHandleForPty(SECOND_PTY_ID, SECOND_TERMINAL_HANDLE)
-  runtime.syncWindowGraph(1, {
-    tabs: [
-      ...(includeFirstPane
-        ? [
-            {
-              tabId: TAB_ID,
-              worktreeId: WORKTREE_ID,
-              title: 'Codex',
-              activeLeafId: LEAF_ID,
-              layout: null
-            }
-          ]
-        : []),
-      {
-        tabId: SECOND_TAB_ID,
-        worktreeId: WORKTREE_ID,
-        title: 'Codex',
-        activeLeafId: leafId,
-        layout: null
-      }
-    ],
-    leaves: [
-      ...(includeFirstPane
-        ? [
-            {
-              tabId: TAB_ID,
-              worktreeId: WORKTREE_ID,
-              leafId: LEAF_ID,
-              paneRuntimeId: 1,
-              ptyId: PTY_ID
-            }
-          ]
-        : []),
-      {
-        tabId: SECOND_TAB_ID,
-        worktreeId: WORKTREE_ID,
-        leafId,
-        paneRuntimeId: 2,
-        ptyId: SECOND_PTY_ID
-      }
-    ]
-  })
-}
-
-async function driveToLiveIdle(runtime: OrcaRuntimeService): Promise<void> {
-  await runtime.listTerminals()
-  runtime.onPtyData(PTY_ID, '\x1b]0;Codex working\x07', 1)
-  runtime.onPtyData(PTY_ID, '\x1b]0;Codex done\x07', 2)
-  await Promise.resolve()
-}
-
-function pointerCount(write: ReturnType<typeof vi.fn>): number {
-  return write.mock.calls.filter(([, payload]) =>
-    String(payload).includes('orca orchestration check')
-  ).length
-}
-
-async function checkBoundMailbox(
-  runtime: OrcaRuntimeService,
-  options: {
-    ack?: string
-    wait?: boolean
-    terminal?: string
-    paneKey?: string
-    launchToken?: string
-  } = {}
-): Promise<CheckResult> {
-  const terminal = options.terminal ?? TERMINAL_HANDLE
-  const response = await new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }).dispatch({
-    id: 'req-mailbox-consistency',
-    authToken: 'test-auth-token',
-    method: 'orchestration.check',
-    orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION,
-    orchestrationCompatibilityEvidence: {
-      terminalHandle: terminal,
-      paneKey: options.paneKey ?? PANE_KEY,
-      launchToken: options.launchToken ?? LAUNCH_TOKEN
-    },
-    params: {
-      terminal,
-      ...(options.ack ? { ack: options.ack } : {}),
-      ...(options.wait ? { wait: true, timeoutMs: 5_000 } : {})
-    }
-  })
-  expect(response.ok).toBe(true)
-  if (!response.ok) {
-    throw new Error(response.error.message)
-  }
-  return response.result as CheckResult
-}
-
 describe('orchestration notification mailbox consistency', () => {
   afterEach(() => {
     vi.useRealTimers()
@@ -338,6 +152,46 @@ describe('orchestration notification mailbox consistency', () => {
     db.close()
   })
 
+  it('uses composite indexes for direct ownership reconciliation', () => {
+    const db = createDatabase('orca-mailbox-routing-indexes-')
+    const sqlite = sqliteFor(db)
+    const plan = (sql: string, ...params: string[]): string =>
+      (sqlite.prepare(`EXPLAIN QUERY PLAN ${sql}`).all(...params) as { detail: string }[])
+        .map((row) => row.detail)
+        .join(' ')
+
+    expect(
+      plan(
+        `SELECT * FROM dispatch_contexts
+         WHERE run_id = ? AND assignee_handle = ? AND status IN ('pending', 'dispatched')
+         ORDER BY rowid DESC LIMIT 1`,
+        'run_indexed',
+        TERMINAL_HANDLE
+      )
+    ).toContain('idx_dispatch_active_run_assignee_handle')
+    expect(
+      plan(
+        `SELECT * FROM dispatch_contexts
+         WHERE run_id = ? AND assignee_pane_key IS NOT NULL
+           AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1
+           AND substr(assignee_pane_key, instr(assignee_pane_key, ':') + 1) = ?
+         ORDER BY rowid DESC LIMIT 1`,
+        'run_indexed',
+        LEAF_ID
+      )
+    ).toContain('idx_dispatch_active_run_pane_leaf')
+    expect(
+      plan(
+        `SELECT 1 FROM messages INDEXED BY idx_messages_undelivered_direct_run
+         WHERE run_id = ? AND to_handle = ? AND read = 0 AND delivered_at IS NULL
+           AND delivery_contract = 'current_delivery' LIMIT 1`,
+        'run_indexed',
+        TERMINAL_HANDLE
+      )
+    ).toContain('idx_messages_undelivered_direct_run')
+    db.close()
+  })
+
   it('does not submit or replay a pointer after ownership changes during Enter delay', async () => {
     vi.useFakeTimers()
     const directory = mkdtempSync(join(tmpdir(), 'orca-mailbox-restart-'))
@@ -371,6 +225,33 @@ describe('orchestration notification mailbox consistency', () => {
     restartedDb.close()
   })
 
+  it('does not replay a staged same-Run pointer when the runtime restarts before Enter', async () => {
+    vi.useFakeTimers()
+    const directory = mkdtempSync(join(tmpdir(), 'orca-mailbox-staged-restart-'))
+    temporaryDirectories.push(directory)
+    const dbPath = join(directory, 'orchestration.db')
+    const firstDb = new OrchestrationDb(dbPath)
+    const first = createRuntime(firstDb)
+    const run = createBoundRun(firstDb, 'Staged restart Run')
+    const message = insertDirectRunMessage(firstDb, run.id, 'Visible before restart')
+
+    await driveToLiveIdle(first.runtime)
+    expect(pointerCount(first.write)).toBe(1)
+    expect(first.write.mock.calls.filter(([, payload]) => payload === '\r')).toHaveLength(0)
+    expect(firstDb.getMessageById(message.id)?.delivered_at).toEqual(expect.any(String))
+    firstDb.close()
+
+    const restartedDb = new OrchestrationDb(dbPath)
+    const restarted = createRuntime(restartedDb)
+    await driveToLiveIdle(restarted.runtime)
+    const checked = await checkBoundMailbox(restarted.runtime)
+
+    expect(pointerCount(restarted.write)).toBe(0)
+    expect(checked).toMatchObject({ runId: run.id, count: 1 })
+    expect(checked.messages).toEqual([expect.objectContaining({ id: message.id })])
+    restartedDb.close()
+  })
+
   it('fences the pointed mailbox instead of checking a rebound empty Run', async () => {
     vi.useFakeTimers()
     const db = createDatabase('orca-mailbox-post-submit-rebind-')
@@ -514,7 +395,7 @@ describe('orchestration notification mailbox consistency', () => {
     await vi.advanceTimersByTimeAsync(500)
 
     expect(harness.write.mock.calls.filter(([, payload]) => payload === '\r')).toHaveLength(0)
-    expect(db.getMessageById(message.id)?.delivered_at).toBeNull()
+    expect(db.getMessageById(message.id)?.delivered_at).toEqual(expect.any(String))
     harness.runtime.onPtyData(PTY_ID, '\x1b]0;Codex done\x07', 4)
     await Promise.resolve()
     expect(pointerCount(harness.write)).toBe(1)
@@ -564,7 +445,7 @@ describe('orchestration notification mailbox consistency', () => {
       })
     )
     const notificationQuery = vi.spyOn(db, 'getUndeliveredUnreadMessages')
-    const submitRevalidation = vi.spyOn(db, 'areUndeliveredUnreadMessages')
+    const submitRevalidation = vi.spyOn(db, 'areUnreadMessages')
     const prepare = vi.spyOn(sqliteFor(db), 'prepare')
 
     await driveToLiveIdle(harness.runtime)
@@ -642,7 +523,7 @@ describe('orchestration notification mailbox consistency', () => {
     restartedDb.close()
   })
 
-  it('keeps mail retryable when the provider rejects Enter', async () => {
+  it('does not replay a visible pointer when the provider rejects Enter', async () => {
     vi.useFakeTimers()
     const db = createDatabase('orca-mailbox-provider-refusal-')
     const first = createRuntime(db)
@@ -662,11 +543,11 @@ describe('orchestration notification mailbox consistency', () => {
     await driveToLiveIdle(first.runtime)
     await vi.advanceTimersByTimeAsync(500)
     expect(pointerCount(first.write)).toBe(1)
-    expect(db.getMessageById(message.id)?.delivered_at).toBeNull()
+    expect(db.getMessageById(message.id)?.delivered_at).toEqual(expect.any(String))
 
     const restarted = createRuntime(db)
     await driveToLiveIdle(restarted.runtime)
-    expect(pointerCount(restarted.write)).toBe(1)
+    expect(pointerCount(restarted.write)).toBe(0)
     db.close()
   })
 
@@ -702,7 +583,7 @@ describe('orchestration notification mailbox consistency', () => {
     db.close()
   })
 
-  it('routes active worker direct mail through the Dispatch check', async () => {
+  it('routes active worker direct mail without injecting an unpinned Dispatch pointer', async () => {
     vi.useFakeTimers()
     const db = createDatabase('orca-mailbox-dispatch-')
     const harness = createRuntime(db)
@@ -727,13 +608,13 @@ describe('orchestration notification mailbox consistency', () => {
     await vi.advanceTimersByTimeAsync(500)
     const checked = await checkBoundMailbox(harness.runtime)
 
-    expect(pointerCount(harness.write)).toBe(1)
+    expect(pointerCount(harness.write)).toBe(0)
     expect(checked).toMatchObject({ runId: run.id, dispatchId: dispatch.id, count: 1 })
     expect(checked.messages).toEqual([expect.objectContaining({ id: message.id })])
     expect(db.getMessageById(message.id)).toMatchObject({
       to_handle: `dispatch:${dispatch.id}`,
       read: 1,
-      delivered_at: expect.any(String)
+      delivered_at: null
     })
     const internals = harness.runtime as unknown as {
       pointedMessageWatermarkOwnerByHandle: Map<string, unknown>
@@ -877,4 +758,109 @@ describe('orchestration notification mailbox consistency', () => {
     expect(pointerCount(harness.write)).toBe(0)
     db.close()
   })
+
+  it('drains persisted Run pages before installing a filtered waiter', async () => {
+    const db = createDatabase('orca-mailbox-filtered-run-backlog-')
+    const harness = createRuntime(db)
+    const run = createBoundRun(db, 'Filtered Run backlog')
+    for (let index = 0; index < 50; index += 1) {
+      insertDirectRunMessage(db, run.id, `Status ${index}`)
+    }
+    const question = db.insertMessage({
+      from: 'term_worker',
+      to: TERMINAL_HANDLE,
+      subject: 'Question behind first page',
+      type: 'question',
+      runId: run.id
+    })
+
+    const checked = await checkBoundMailbox(harness.runtime, {
+      wait: true,
+      types: 'question'
+    })
+
+    expect(checked).toMatchObject({ runId: run.id, count: 50 })
+    expect(checked.messages).not.toContainEqual(expect.objectContaining({ id: question.id }))
+    expect(db.getMessageById(question.id)?.to_handle).toBe(`run:${run.id}`)
+    const next = await checkBoundMailbox(harness.runtime, {
+      ack: checked.deliveryId!,
+      types: 'question'
+    })
+    expect(next.messages).toEqual(
+      expect.arrayContaining([expect.objectContaining({ id: question.id })])
+    )
+    db.close()
+  })
+
+  it('wakes a filtered waiter when reconciliation moves its type on a later page', async () => {
+    const db = createDatabase('orca-mailbox-filtered-reconciliation-wake-')
+    const harness = createRuntime(db)
+    const run = createBoundRun(db, 'Filtered reconciliation wake')
+    const waiting = checkBoundMailbox(harness.runtime, { wait: true, types: 'question' })
+    const internals = harness.runtime as unknown as {
+      messageWaitersByHandle: Map<string, Set<unknown>>
+    }
+    await vi.waitFor(() => {
+      expect(internals.messageWaitersByHandle.has(`run:${run.id}`)).toBe(true)
+    })
+    for (let index = 0; index < 50; index += 1) {
+      insertDirectRunMessage(db, run.id, `Status before question ${index}`)
+    }
+    const question = db.insertMessage({
+      from: 'term_worker',
+      to: TERMINAL_HANDLE,
+      subject: 'Question moved by continuation',
+      type: 'question',
+      runId: run.id
+    })
+    const arrivingStatus = insertDirectRunMessage(db, run.id, 'Status arrival trigger')
+
+    harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, arrivingStatus.type)
+    const checked = await waiting
+
+    expect(checked).toMatchObject({ runId: run.id, count: 50 })
+    expect(checked.messages).not.toContainEqual(expect.objectContaining({ id: question.id }))
+    expect(db.getMessageById(question.id)?.to_handle).toBe(`run:${run.id}`)
+    const next = await checkBoundMailbox(harness.runtime, {
+      ack: checked.deliveryId!,
+      types: 'question'
+    })
+    expect(next.messages).toEqual(
+      expect.arrayContaining([expect.objectContaining({ id: question.id })])
+    )
+    db.close()
+  })
+
+  it('drains persisted Dispatch pages before installing a filtered waiter', async () => {
+    const db = createDatabase('orca-mailbox-filtered-dispatch-backlog-')
+    const harness = createRuntime(db)
+    const run = db.createRun({
+      objective: 'Filtered Dispatch backlog',
+      coordinatorHandle: 'term_coordinator',
+      coordinatorPaneKey:
+        '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666'
+    })
+    const task = db.createTask({ spec: 'Worker task', runId: run.id })
+    const dispatch = db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY)
+    for (let index = 0; index < 50; index += 1) {
+      insertDirectRunMessage(db, run.id, `Worker status ${index}`)
+    }
+    const question = db.insertMessage({
+      from: 'term_coordinator',
+      to: TERMINAL_HANDLE,
+      subject: 'Worker question behind first page',
+      type: 'question',
+      runId: run.id
+    })
+
+    const checked = await checkBoundMailbox(harness.runtime, {
+      wait: true,
+      types: 'question'
+    })
+
+    expect(checked).toMatchObject({ runId: run.id, dispatchId: dispatch.id, count: 1 })
+    expect(checked.messages).toEqual([expect.objectContaining({ id: question.id })])
+    expect(db.getMessageById(question.id)?.to_handle).toBe(`dispatch:${dispatch.id}`)
+    db.close()
+  })
 })
diff --git a/src/main/runtime/orchestration-mailbox-notification-test-harness.ts b/src/main/runtime/orchestration-mailbox-notification-test-harness.ts
new file mode 100644
index 00000000000..cc476f1f278
--- /dev/null
+++ b/src/main/runtime/orchestration-mailbox-notification-test-harness.ts
@@ -0,0 +1,231 @@
+import { mkdtempSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { expect, vi } from 'vitest'
+import { ORCHESTRATION_CONTRACT_VERSION } from '../../shared/protocol-version'
+import type Database from '../sqlite/sync-database'
+import { OrcaRuntimeService } from './orca-runtime'
+import { OrchestrationDb } from './orchestration/db'
+import { RpcDispatcher } from './rpc/dispatcher'
+import { ORCHESTRATION_METHODS } from './rpc/methods/orchestration'
+
+export const TAB_ID = '11111111-1111-4111-8111-111111111111'
+export const LEAF_ID = '22222222-2222-4222-8222-222222222222'
+export const PANE_KEY = `${TAB_ID}:${LEAF_ID}`
+export const PTY_ID = 'pty-mailbox'
+export const TERMINAL_HANDLE = 'term_mailbox_consistency'
+export const SECOND_TAB_ID = '33333333-3333-4333-8333-333333333333'
+export const SECOND_LEAF_ID = '44444444-4444-4444-8444-444444444444'
+export const SECOND_PANE_KEY = `${SECOND_TAB_ID}:${SECOND_LEAF_ID}`
+export const SECOND_PTY_ID = 'pty-mailbox-second'
+export const SECOND_TERMINAL_HANDLE = 'term_mailbox_consistency_second'
+export const SECOND_LAUNCH_TOKEN = 'mailbox-consistency-second-launch'
+export const WORKTREE_ID = 'repo-mailbox::/tmp/mailbox'
+export const LAUNCH_TOKEN = 'mailbox-consistency-launch'
+export const temporaryDirectories: string[] = []
+
+export function createDatabase(prefix: string): OrchestrationDb {
+  const directory = mkdtempSync(join(tmpdir(), prefix))
+  temporaryDirectories.push(directory)
+  return new OrchestrationDb(join(directory, 'orchestration.db'))
+}
+
+export function sqliteFor(db: OrchestrationDb): Database.Database {
+  return (db as unknown as { db: Database.Database }).db
+}
+
+export function createBoundRun(db: OrchestrationDb, objective: string) {
+  return db.createRun({
+    objective,
+    coordinatorHandle: TERMINAL_HANDLE,
+    coordinatorPaneKey: PANE_KEY
+  })
+}
+
+export function insertDirectRunMessage(db: OrchestrationDb, runId: string, subject: string) {
+  return db.insertMessage({
+    from: 'term_worker',
+    to: TERMINAL_HANDLE,
+    subject,
+    type: 'status',
+    runId,
+    deliveryContract: 'current_delivery'
+  })
+}
+
+export type MailboxNotificationHarness = {
+  runtime: OrcaRuntimeService
+  write: ReturnType<typeof vi.fn>
+}
+
+export type MailboxCheckResult = {
+  runId: string
+  dispatchId?: string
+  deliveryId: string | null
+  count: number
+  messages: unknown[]
+  acknowledged?: string | null
+}
+
+export type MailboxCheckOptions = {
+  ack?: string
+  wait?: boolean
+  terminal?: string
+  paneKey?: string
+  launchToken?: string
+  types?: string
+  signal?: AbortSignal
+}
+
+export function createRuntime(db: OrchestrationDb): MailboxNotificationHarness {
+  const runtime = new OrcaRuntimeService(null, undefined, {
+    attestAgentHookCompatibilityAuthority: ({ paneKey }) =>
+      paneKey === PANE_KEY || paneKey.startsWith(`${SECOND_TAB_ID}:`)
+        ? { paneKey, source: 'current_hook' }
+        : null
+  })
+  const write = vi.fn(() => true)
+  runtime.setOrchestrationDb(db)
+  runtime.setPtyController({
+    write,
+    kill: vi.fn(),
+    getForegroundProcess: async () => null
+  })
+  runtime.registerPty(PTY_ID, WORKTREE_ID, null, {
+    tabId: TAB_ID,
+    leafId: LEAF_ID,
+    incarnationId: 'mailbox-incarnation',
+    agentLaunchAuthority: { launchToken: LAUNCH_TOKEN, launchAgent: 'codex' }
+  })
+  runtime.registerPreAllocatedHandleForPty(PTY_ID, TERMINAL_HANDLE)
+  runtime.attachWindow(1)
+  runtime.syncWindowGraph(1, {
+    tabs: [
+      {
+        tabId: TAB_ID,
+        worktreeId: WORKTREE_ID,
+        title: 'Codex',
+        activeLeafId: LEAF_ID,
+        layout: null
+      }
+    ],
+    leaves: [
+      {
+        tabId: TAB_ID,
+        worktreeId: WORKTREE_ID,
+        leafId: LEAF_ID,
+        paneRuntimeId: 1,
+        ptyId: PTY_ID
+      }
+    ]
+  })
+  return { runtime, write }
+}
+
+export function registerSecondPane(
+  runtime: OrcaRuntimeService,
+  leafId = SECOND_LEAF_ID,
+  includeFirstPane = true
+): void {
+  runtime.registerPty(SECOND_PTY_ID, WORKTREE_ID, null, {
+    tabId: SECOND_TAB_ID,
+    leafId,
+    incarnationId: 'mailbox-second-incarnation',
+    agentLaunchAuthority: { launchToken: SECOND_LAUNCH_TOKEN, launchAgent: 'codex' }
+  })
+  runtime.registerPreAllocatedHandleForPty(SECOND_PTY_ID, SECOND_TERMINAL_HANDLE)
+  runtime.syncWindowGraph(1, {
+    tabs: [
+      ...(includeFirstPane
+        ? [
+            {
+              tabId: TAB_ID,
+              worktreeId: WORKTREE_ID,
+              title: 'Codex',
+              activeLeafId: LEAF_ID,
+              layout: null
+            }
+          ]
+        : []),
+      {
+        tabId: SECOND_TAB_ID,
+        worktreeId: WORKTREE_ID,
+        title: 'Codex',
+        activeLeafId: leafId,
+        layout: null
+      }
+    ],
+    leaves: [
+      ...(includeFirstPane
+        ? [
+            {
+              tabId: TAB_ID,
+              worktreeId: WORKTREE_ID,
+              leafId: LEAF_ID,
+              paneRuntimeId: 1,
+              ptyId: PTY_ID
+            }
+          ]
+        : []),
+      {
+        tabId: SECOND_TAB_ID,
+        worktreeId: WORKTREE_ID,
+        leafId,
+        paneRuntimeId: 2,
+        ptyId: SECOND_PTY_ID
+      }
+    ]
+  })
+}
+
+export async function driveToLiveIdle(runtime: OrcaRuntimeService): Promise<void> {
+  await runtime.listTerminals()
+  runtime.onPtyData(PTY_ID, '\x1b]0;Codex working\x07', 1)
+  runtime.onPtyData(PTY_ID, '\x1b]0;Codex done\x07', 2)
+  await Promise.resolve()
+}
+
+export function pointerCount(write: ReturnType<typeof vi.fn>): number {
+  return write.mock.calls.filter(([, payload]) =>
+    String(payload).includes('orca orchestration check')
+  ).length
+}
+
+export async function checkBoundMailbox(
+  runtime: OrcaRuntimeService,
+  options: MailboxCheckOptions = {}
+): Promise<MailboxCheckResult> {
+  const response = await dispatchMailboxCheck(runtime, options)
+  expect(response.ok).toBe(true)
+  if (!response.ok) {
+    throw new Error(response.error.message)
+  }
+  return response.result as MailboxCheckResult
+}
+
+export async function dispatchMailboxCheck(
+  runtime: OrcaRuntimeService,
+  options: MailboxCheckOptions = {}
+) {
+  const terminal = options.terminal ?? TERMINAL_HANDLE
+  return new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }).dispatch(
+    {
+      id: 'req-mailbox-consistency',
+      authToken: 'test-auth-token',
+      method: 'orchestration.check',
+      orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION,
+      orchestrationCompatibilityEvidence: {
+        terminalHandle: terminal,
+        paneKey: options.paneKey ?? PANE_KEY,
+        launchToken: options.launchToken ?? LAUNCH_TOKEN
+      },
+      params: {
+        terminal,
+        ...(options.types ? { types: options.types } : {}),
+        ...(options.ack ? { ack: options.ack } : {}),
+        ...(options.wait ? { wait: true, timeoutMs: 5_000 } : {})
+      }
+    },
+    { signal: options.signal }
+  )
+}
diff --git a/src/main/runtime/orchestration-mailbox-routing-races.test.ts b/src/main/runtime/orchestration-mailbox-routing-races.test.ts
new file mode 100644
index 00000000000..f92cf8ec0c1
--- /dev/null
+++ b/src/main/runtime/orchestration-mailbox-routing-races.test.ts
@@ -0,0 +1,334 @@
+import { rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import {
+  checkBoundMailbox,
+  createBoundRun,
+  createDatabase,
+  createRuntime,
+  dispatchMailboxCheck,
+  insertDirectRunMessage,
+  LEAF_ID,
+  PANE_KEY,
+  registerSecondPane,
+  SECOND_LAUNCH_TOKEN,
+  SECOND_PANE_KEY,
+  SECOND_TERMINAL_HANDLE,
+  sqliteFor,
+  temporaryDirectories,
+  TERMINAL_HANDLE
+} from './orchestration-mailbox-notification-test-harness'
+
+vi.mock('electron', () => ({
+  app: { getPath: vi.fn(() => tmpdir()), isPackaged: false },
+  BrowserWindow: { fromId: vi.fn(() => null) },
+  ipcMain: { on: vi.fn(), removeListener: vi.fn() },
+  webContents: { fromId: vi.fn(() => null) }
+}))
+
+describe('orchestration mailbox routing races', () => {
+  afterEach(() => {
+    for (const directory of temporaryDirectories.splice(0)) {
+      rmSync(directory, { recursive: true, force: true })
+    }
+  })
+
+  it('fences a Run consumer rebound during a routing-page yield', async () => {
+    const db = createDatabase('orca-mailbox-run-routing-fence-')
+    const harness = createRuntime(db)
+    const run = createBoundRun(db, 'Run routing fence')
+    for (let index = 0; index < 51; index += 1) {
+      insertDirectRunMessage(db, run.id, `Before rebind ${index}`)
+    }
+    const route = db.routeUnreadDirectMessagesToRunMailbox.bind(db)
+    let scheduled = false
+    vi.spyOn(db, 'routeUnreadDirectMessagesToRunMailbox').mockImplementation((...args) => {
+      const page = route(...args)
+      if (page.hasMore && !scheduled) {
+        scheduled = true
+        setImmediate(() => createBoundRun(db, 'Replacement Run'))
+      }
+      return page
+    })
+
+    const response = await dispatchMailboxCheck(harness.runtime)
+
+    expect(response).toMatchObject({ ok: false, error: { code: 'consumer_fenced' } })
+    expect(db.getUnreadMessages(`run:${run.id}`)).toHaveLength(51)
+    db.close()
+  })
+
+  it('fences a Dispatch completed during a routing-page yield', async () => {
+    const db = createDatabase('orca-mailbox-dispatch-routing-fence-')
+    const harness = createRuntime(db)
+    registerSecondPane(harness.runtime)
+    const run = db.createRun({
+      objective: 'Dispatch routing fence',
+      coordinatorHandle: SECOND_TERMINAL_HANDLE,
+      coordinatorPaneKey: SECOND_PANE_KEY
+    })
+    const task = db.createTask({ spec: 'Worker task', runId: run.id })
+    const dispatch = db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY)
+    for (let index = 0; index < 51; index += 1) {
+      insertDirectRunMessage(db, run.id, `Before completion ${index}`)
+    }
+    const route = db.routeUnreadDirectMessagesToDispatchMailbox.bind(db)
+    let scheduled = false
+    vi.spyOn(db, 'routeUnreadDirectMessagesToDispatchMailbox').mockImplementation((...args) => {
+      const page = route(...args)
+      if (page.hasMore && !scheduled) {
+        scheduled = true
+        setImmediate(() => db.completeDispatch(dispatch.id))
+      }
+      return page
+    })
+
+    const response = await dispatchMailboxCheck(harness.runtime)
+
+    expect(response).toMatchObject({ ok: false, error: { code: 'dispatch_inactive' } })
+    expect(db.getUnreadMessages(`dispatch:${dispatch.id}`)).toHaveLength(0)
+    const first = await checkBoundMailbox(harness.runtime, {
+      terminal: SECOND_TERMINAL_HANDLE,
+      paneKey: SECOND_PANE_KEY,
+      launchToken: SECOND_LAUNCH_TOKEN
+    })
+    const second = await checkBoundMailbox(harness.runtime, {
+      ack: first.deliveryId!,
+      terminal: SECOND_TERMINAL_HANDLE,
+      paneKey: SECOND_PANE_KEY,
+      launchToken: SECOND_LAUNCH_TOKEN
+    })
+    const acknowledged = await checkBoundMailbox(harness.runtime, {
+      ack: second.deliveryId!,
+      terminal: SECOND_TERMINAL_HANDLE,
+      paneKey: SECOND_PANE_KEY,
+      launchToken: SECOND_LAUNCH_TOKEN
+    })
+    expect(first.count).toBe(50)
+    expect(second.count).toBe(1)
+    expect(acknowledged).toMatchObject({ count: 0, acknowledged: second.deliveryId })
+    db.close()
+  })
+
+  it('reconciles filtered-out mail when a Dispatch wait loses authority', async () => {
+    const db = createDatabase('orca-mailbox-dispatch-wait-fence-')
+    const harness = createRuntime(db)
+    registerSecondPane(harness.runtime)
+    const run = db.createRun({
+      objective: 'Dispatch wait fence',
+      coordinatorHandle: SECOND_TERMINAL_HANDLE,
+      coordinatorPaneKey: SECOND_PANE_KEY
+    })
+    const task = db.createTask({ spec: 'Waiting worker', runId: run.id })
+    const dispatch = db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY)
+    const status = insertDirectRunMessage(db, run.id, 'Filtered-out status')
+    const controller = new AbortController()
+    const waiting = dispatchMailboxCheck(harness.runtime, {
+      wait: true,
+      types: 'question',
+      signal: controller.signal
+    })
+    const internals = harness.runtime as unknown as {
+      messageWaitersByHandle: Map<string, Set<unknown>>
+    }
+    await vi.waitFor(() => {
+      expect(internals.messageWaitersByHandle.has(`dispatch:${dispatch.id}`)).toBe(true)
+    })
+
+    db.completeDispatch(dispatch.id)
+    controller.abort()
+    const response = await waiting
+
+    expect(response).toMatchObject({ ok: false, error: { code: 'dispatch_inactive' } })
+    const checked = await checkBoundMailbox(harness.runtime, {
+      terminal: SECOND_TERMINAL_HANDLE,
+      paneKey: SECOND_PANE_KEY,
+      launchToken: SECOND_LAUNCH_TOKEN
+    })
+    expect(checked.messages).toEqual([expect.objectContaining({ id: status.id })])
+    const acknowledged = await checkBoundMailbox(harness.runtime, {
+      ack: checked.deliveryId!,
+      terminal: SECOND_TERMINAL_HANDLE,
+      paneKey: SECOND_PANE_KEY,
+      launchToken: SECOND_LAUNCH_TOKEN
+    })
+    expect(acknowledged).toMatchObject({ count: 0, acknowledged: checked.deliveryId })
+    db.close()
+  })
+
+  it('drains current and stored Run handles in one check', async () => {
+    const db = createDatabase('orca-mailbox-split-run-handles-')
+    const harness = createRuntime(db)
+    const run = db.createRun({
+      objective: 'Split Run handles',
+      coordinatorHandle: 'term_previous_coordinator',
+      coordinatorPaneKey: PANE_KEY
+    })
+    const current = insertDirectRunMessage(db, run.id, 'Current handle')
+    const previous = db.insertMessage({
+      from: 'term_worker',
+      to: 'term_previous_coordinator',
+      subject: 'Previous handle',
+      type: 'status',
+      runId: run.id
+    })
+
+    const checked = await checkBoundMailbox(harness.runtime)
+
+    expect(checked.messages).toEqual(
+      expect.arrayContaining([
+        expect.objectContaining({ id: current.id }),
+        expect.objectContaining({ id: previous.id })
+      ])
+    )
+    db.close()
+  })
+
+  it('drains current and stored Dispatch handles in one check', async () => {
+    const db = createDatabase('orca-mailbox-split-dispatch-handles-')
+    const harness = createRuntime(db)
+    const run = db.createRun({
+      objective: 'Split Dispatch handles',
+      coordinatorHandle: 'term_coordinator',
+      coordinatorPaneKey:
+        '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666'
+    })
+    const task = db.createTask({ spec: 'Worker task', runId: run.id })
+    db.createDispatchContext(task.id, 'term_previous_worker', PANE_KEY)
+    const current = insertDirectRunMessage(db, run.id, 'Current worker handle')
+    const previous = db.insertMessage({
+      from: 'term_coordinator',
+      to: 'term_previous_worker',
+      subject: 'Previous worker handle',
+      type: 'status',
+      runId: run.id
+    })
+
+    const checked = await checkBoundMailbox(harness.runtime)
+
+    expect(checked.messages).toEqual(
+      expect.arrayContaining([
+        expect.objectContaining({ id: current.id }),
+        expect.objectContaining({ id: previous.id })
+      ])
+    )
+    db.close()
+  })
+
+  it('routes only the direct-mail snapshot captured when the check starts', async () => {
+    const db = createDatabase('orca-mailbox-routing-snapshot-')
+    const harness = createRuntime(db)
+    const run = createBoundRun(db, 'Routing snapshot')
+    for (let index = 0; index < 51; index += 1) {
+      insertDirectRunMessage(db, run.id, `Initial ${index}`)
+    }
+    const route = db.routeUnreadDirectMessagesToRunMailbox.bind(db)
+    let inserted = false
+    const routeSpy = vi
+      .spyOn(db, 'routeUnreadDirectMessagesToRunMailbox')
+      .mockImplementation((...args) => {
+        const page = route(...args)
+        if (page.hasMore && !inserted) {
+          inserted = true
+          for (let index = 0; index < 60; index += 1) {
+            insertDirectRunMessage(db, run.id, `Concurrent ${index}`)
+          }
+        }
+        return page
+      })
+
+    await checkBoundMailbox(harness.runtime)
+
+    expect(routeSpy).toHaveBeenCalledTimes(2)
+    expect(db.getUnreadMessages(TERMINAL_HANDLE)).toHaveLength(60)
+    db.close()
+  })
+
+  it('stops paged routing when the check is cancelled', async () => {
+    const db = createDatabase('orca-mailbox-routing-cancel-')
+    const harness = createRuntime(db)
+    const run = createBoundRun(db, 'Routing cancellation')
+    for (let index = 0; index < 101; index += 1) {
+      insertDirectRunMessage(db, run.id, `Before cancellation ${index}`)
+    }
+    const controller = new AbortController()
+    const route = db.routeUnreadDirectMessagesToRunMailbox.bind(db)
+    let scheduled = false
+    const routeSpy = vi
+      .spyOn(db, 'routeUnreadDirectMessagesToRunMailbox')
+      .mockImplementation((...args) => {
+        const page = route(...args)
+        if (!scheduled) {
+          scheduled = true
+          setImmediate(() => controller.abort())
+        }
+        return page
+      })
+
+    const response = await dispatchMailboxCheck(harness.runtime, {
+      signal: controller.signal
+    })
+
+    expect(response).toMatchObject({ ok: false, error: { code: 'runtime_error' } })
+    expect(routeSpy).toHaveBeenCalledOnce()
+    db.close()
+  })
+
+  it('preserves a later filtered reservation behind an unfiltered redrive', () => {
+    const db = createDatabase('orca-mailbox-reservation-union-')
+    const harness = createRuntime(db)
+    const internals = harness.runtime as unknown as {
+      parkMessageRedeliveryForMailbox: (
+        mailboxHandle: string,
+        reservedTypes: ReadonlySet<string> | undefined
+      ) => void
+      parkedMessageRedeliveryTypesByMailboxHandle: Map<string, ReadonlySet<string> | null>
+    }
+
+    internals.parkMessageRedeliveryForMailbox('run:run_test', undefined)
+    internals.parkMessageRedeliveryForMailbox('run:run_test', new Set(['worker_done']))
+
+    expect(internals.parkedMessageRedeliveryTypesByMailboxHandle.get('run:run_test')).toEqual(
+      new Set(['worker_done'])
+    )
+    db.close()
+  })
+
+  it('uses a bounded indexed pane lookup for reminted Dispatch identity', () => {
+    const db = createDatabase('orca-mailbox-dispatch-pane-index-')
+    const run = db.createRun({
+      objective: 'Dispatch pane index',
+      coordinatorHandle: 'term_coordinator',
+      coordinatorPaneKey:
+        '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666'
+    })
+    const task = db.createTask({ spec: 'Valid worker', runId: run.id })
+    const valid = db.createDispatchContext(task.id, 'term_old', PANE_KEY)
+    const collisionTask = db.createTask({ spec: 'Malformed collision', runId: run.id })
+    db.createDispatchContext(collisionTask.id, 'term_collision', `:${LEAF_ID}`)
+
+    expect(db.getActiveDispatchForIdentity('term_reminted', PANE_KEY)?.id).toBe(valid.id)
+    const plan = sqliteFor(db)
+      .prepare(
+        `EXPLAIN QUERY PLAN SELECT * FROM dispatch_contexts
+         WHERE assignee_pane_key IS NOT NULL
+           AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1
+           AND substr(assignee_pane_key, instr(assignee_pane_key, ':') + 1) = ?
+         ORDER BY rowid DESC LIMIT 1`
+      )
+      .all(LEAF_ID) as { detail: string }[]
+    expect(plan.map((row) => row.detail).join(' ')).toContain('idx_dispatch_assignee_pane_leaf')
+    const snapshotPlan = sqliteFor(db)
+      .prepare(
+        `EXPLAIN QUERY PLAN SELECT sequence FROM messages
+         WHERE run_id = ? AND to_handle = ? AND read = 0
+           AND delivery_contract = 'current_delivery'
+         ORDER BY sequence DESC LIMIT 1`
+      )
+      .all(run.id, TERMINAL_HANDLE) as { detail: string }[]
+    expect(snapshotPlan.map((row) => row.detail).join(' ')).toContain(
+      'idx_messages_delivery_contract'
+    )
+    db.close()
+  })
+})
diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts
index 0d4724dc6b1..382a5d4d52b 100644
--- a/src/main/runtime/orchestration/db.ts
+++ b/src/main/runtime/orchestration/db.ts
@@ -93,6 +93,7 @@ export const ORCHESTRATION_DELIVERY_BATCH_LIMIT = 50
 export type DirectMailboxRoutingPage = {
   routedCount: number
   hasMore: boolean
+  types: MessageType[]
 }
 
 export type ForeignDirectMailboxRoutingPage = DirectMailboxRoutingPage & {
@@ -1034,6 +1035,15 @@ export class OrchestrationDb {
         CREATE INDEX IF NOT EXISTS idx_dispatch_assignee_pane_leaf
           ON dispatch_contexts(${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL})
           WHERE assignee_pane_key IS NOT NULL AND status IN ('pending', 'dispatched');
+        CREATE INDEX IF NOT EXISTS idx_dispatch_active_run_assignee_handle
+          ON dispatch_contexts(run_id, assignee_handle)
+          WHERE assignee_handle IS NOT NULL AND status IN ('pending', 'dispatched');
+        CREATE INDEX IF NOT EXISTS idx_dispatch_active_run_pane_leaf
+          ON dispatch_contexts(run_id, ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL})
+          WHERE assignee_pane_key IS NOT NULL AND status IN ('pending', 'dispatched');
+        CREATE INDEX IF NOT EXISTS idx_dispatch_active_assignee_pane_key
+          ON dispatch_contexts(assignee_pane_key)
+          WHERE assignee_pane_key IS NOT NULL AND status IN ('pending', 'dispatched');
       `)
       this.createUndeliveredInboxIndexIfPossible()
 
@@ -1424,7 +1434,11 @@ export class OrchestrationDb {
     }
     this.db.exec(`
       CREATE INDEX IF NOT EXISTS idx_messages_undelivered_inbox
-        ON messages(to_handle, read, delivered_at, sequence)
+        ON messages(to_handle, read, delivered_at, sequence);
+      CREATE INDEX IF NOT EXISTS idx_messages_undelivered_direct_run
+        ON messages(run_id, to_handle, sequence)
+        WHERE read = 0 AND delivered_at IS NULL
+          AND delivery_contract = 'current_delivery'
     `)
   }
 
@@ -3579,7 +3593,7 @@ export class OrchestrationDb {
     return Boolean(
       this.db
         .prepare(
-          `SELECT 1 FROM messages
+          `SELECT 1 FROM messages INDEXED BY idx_messages_undelivered_direct_run
            WHERE run_id = ? AND to_handle = ? AND read = 0 AND delivered_at IS NULL
              AND delivery_contract = 'current_delivery'
            LIMIT 1`
@@ -3596,23 +3610,48 @@ export class OrchestrationDb {
     )
   }
 
+  getLatestUnreadDirectMessageSequenceForRun(
+    runId: string,
+    directHandle: string
+  ): number | undefined {
+    const row = this.db
+      .prepare(
+        `SELECT sequence FROM messages
+         WHERE run_id = ? AND to_handle = ? AND read = 0
+           AND delivery_contract = 'current_delivery'
+         ORDER BY sequence DESC LIMIT 1`
+      )
+      .get(runId, directHandle) as { sequence: number } | undefined
+    return row?.sequence
+  }
+
   // Why: change mailbox ownership without changing unread or acknowledgment state.
   private routeDirectMessagePage(
     mailboxHandle: string,
     runId: string,
-    directHandle: string
+    directHandle: string,
+    throughSequence?: number
   ): DirectMailboxRoutingPage {
+    const throughClause = throughSequence === undefined ? '' : ' AND sequence <= ?'
+    const params: (string | number)[] = [runId, directHandle]
+    if (throughSequence !== undefined) {
+      params.push(throughSequence)
+    }
+    params.push(ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1)
     const rows = this.db
       .prepare(
-        `SELECT id FROM messages
+        `SELECT id, type FROM messages
          WHERE run_id = ? AND to_handle = ? AND read = 0
-           AND delivery_contract = 'current_delivery'
+           AND delivery_contract = 'current_delivery'${throughClause}
          ORDER BY sequence LIMIT ?`
       )
-      .all(runId, directHandle, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1) as { id: string }[]
+      .all(...params) as {
+      id: string
+      type: MessageType
+    }[]
     const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT)
     if (page.length === 0) {
-      return { routedCount: 0, hasMore: false }
+      return { routedCount: 0, hasMore: false, types: [] }
     }
     const placeholders = page.map(() => '?').join(',')
     const result = this.db
@@ -3620,23 +3659,58 @@ export class OrchestrationDb {
       .run(mailboxHandle, ...page.map((row) => row.id))
     return {
       routedCount: Number(result.changes),
-      hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT
+      hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT,
+      types: [...new Set(page.map((row) => row.type))]
     }
   }
 
   routeUnreadDirectMessagesToRunMailbox(
     runId: string,
-    directHandle: string
+    directHandle: string,
+    throughSequence?: number
   ): DirectMailboxRoutingPage {
-    return this.routeDirectMessagePage(`run:${runId}`, runId, directHandle)
+    return this.routeDirectMessagePage(`run:${runId}`, runId, directHandle, throughSequence)
   }
 
   routeUnreadDirectMessagesToDispatchMailbox(
     dispatchId: string,
     runId: string,
-    directHandle: string
+    directHandle: string,
+    throughSequence?: number
   ): DirectMailboxRoutingPage {
-    return this.routeDirectMessagePage(`dispatch:${dispatchId}`, runId, directHandle)
+    return this.routeDirectMessagePage(
+      `dispatch:${dispatchId}`,
+      runId,
+      directHandle,
+      throughSequence
+    )
+  }
+
+  routeUnreadDispatchMailboxToRunMailbox(
+    dispatchId: string,
+    runId: string
+  ): { routedCount: number; types: MessageType[] } {
+    const dispatchMailbox = `dispatch:${dispatchId}`
+    this.db.exec('BEGIN IMMEDIATE')
+    try {
+      const rows = this.db
+        .prepare(
+          `SELECT DISTINCT type FROM messages
+           WHERE to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery'`
+        )
+        .all(dispatchMailbox) as { type: MessageType }[]
+      const result = this.db
+        .prepare(
+          `UPDATE messages SET to_handle = ?
+           WHERE to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery'`
+        )
+        .run(`run:${runId}`, dispatchMailbox)
+      this.db.exec('COMMIT')
+      return { routedCount: Number(result.changes), types: rows.map((row) => row.type) }
+    } catch (error) {
+      this.db.exec('ROLLBACK')
+      throw error
+    }
   }
 
   private findActiveDispatchForDirectMessageOwner(
@@ -3685,7 +3759,7 @@ export class OrchestrationDb {
     }[]
     const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT)
     if (page.length === 0) {
-      return { routedCount: 0, hasMore: false, mailboxes: [] }
+      return { routedCount: 0, hasMore: false, types: [], mailboxes: [] }
     }
     const runIds = [...new Set(page.map((row) => row.run_id))]
     const dispatchByRun = new Map<string, DispatchContextRow>()
@@ -3723,6 +3797,7 @@ export class OrchestrationDb {
     return {
       routedCount: page.length,
       hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT,
+      types: [...new Set(page.map((row) => row.type))],
       mailboxes: [...byMailbox].map(([mailboxHandle, types]) => ({
         mailboxHandle,
         types: [...types]
@@ -3747,6 +3822,23 @@ export class OrchestrationDb {
     return matched === ids.length
   }
 
+  areUnreadMessages(toHandle: string, ids: string[]): boolean {
+    let matched = 0
+    for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) {
+      const batch = ids.slice(offset, offset + MESSAGE_ID_UPDATE_BATCH_SIZE)
+      const placeholders = batch.map(() => '?').join(',')
+      const row = this.db
+        .prepare(
+          `SELECT COUNT(*) AS count FROM messages INDEXED BY idx_messages_id
+           WHERE to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery'
+             AND id IN (${placeholders})`
+        )
+        .get(toHandle, ...batch) as { count: number }
+      matched += row.count
+    }
+    return matched === ids.length
+  }
+
   getAllMessages(toHandle: string, limit = 20): MessageRow[] {
     return exposeMessageListTimestamps(
       this.db
@@ -3781,6 +3873,19 @@ export class OrchestrationDb {
     }
   }
 
+  markAsUndelivered(ids: string[]): void {
+    for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) {
+      const batch = ids.slice(offset, offset + MESSAGE_ID_UPDATE_BATCH_SIZE)
+      const placeholders = batch.map(() => '?').join(',')
+      this.db
+        .prepare(
+          `UPDATE messages SET delivered_at = NULL
+           WHERE read = 0 AND id IN (${placeholders})`
+        )
+        .run(...batch)
+    }
+  }
+
   markAsReadAndDelivered(ids: string[]): void {
     if (ids.length === 0) {
       return
@@ -6896,21 +7001,28 @@ export class OrchestrationDb {
       return undefined
     }
 
-    const actives = this.db
+    const exactPane = this.db
+      .prepare(
+        `SELECT * FROM dispatch_contexts
+         WHERE assignee_pane_key = ? AND status IN ('pending', 'dispatched')
+         ORDER BY rowid DESC LIMIT 1`
+      )
+      .get(assigneePaneKey) as DispatchContextRow | undefined
+    if (exactPane) {
+      return exactPane
+    }
+    if (!parsePaneKey(assigneePaneKey)) {
+      return undefined
+    }
+    return this.db
       .prepare(
         `SELECT * FROM dispatch_contexts
          WHERE assignee_pane_key IS NOT NULL
-           AND status IN ('pending', 'dispatched')
-           AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ?`
+           AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1
+           AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ?
+         ORDER BY rowid DESC LIMIT 1`
       )
-      .all(paneKeyMatchSuffix(assigneePaneKey)) as DispatchContextRow[]
-
-    for (const row of actives) {
-      if (row.assignee_pane_key && isEquivalentPaneKey(row.assignee_pane_key, assigneePaneKey)) {
-        return row
-      }
-    }
-    return undefined
+      .get(paneKeyMatchSuffix(assigneePaneKey)) as DispatchContextRow | undefined
   }
 
   getLatestDispatchForTerminal(handle: string): DispatchContextRow | undefined {
diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts
index f0bf56d1f7d..ca5120bd09f 100644
--- a/src/main/runtime/rpc/methods/orchestration.ts
+++ b/src/main/runtime/rpc/methods/orchestration.ts
@@ -1,5 +1,6 @@
 /* eslint-disable max-lines -- Why: RPC method definitions co-locate param schemas with handlers; splitting by method would scatter the shared enums and Zod transforms without reducing complexity. */
 import { z } from 'zod'
+import { setImmediate as yieldToEventLoop } from 'node:timers/promises'
 import { defineMethod, type RpcMethod } from '../core'
 import { OptionalFiniteNumber, OptionalString, OptionalBoolean, requiredString } from '../schemas'
 import {
@@ -43,6 +44,22 @@ const TASK_STATUSES: TaskStatus[] = [
   'blocked'
 ]
 
+async function routeAllDirectMessagePages(
+  routePage: () => { routedCount: number; hasMore: boolean },
+  signal?: AbortSignal
+): Promise<void> {
+  while (true) {
+    if (signal?.aborted) {
+      throw new OrchestrationError('request_aborted', 'Mailbox routing was cancelled.')
+    }
+    const page = routePage()
+    if (!page.hasMore) {
+      return
+    }
+    await yieldToEventLoop()
+  }
+}
+
 function getLifecycleGroupRecipientError(type: 'worker_done' | 'heartbeat'): string {
   return `${type} messages belong to one exact Dispatch and cannot target a group address.`
 }
@@ -725,6 +742,16 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
       const db = runtime.getOrchestrationDb()
       const handle = params.terminal ?? 'unknown'
       const typeFilter = parseMessageTypes(params.types)
+      const routeDirectSnapshot = async (
+        runId: string,
+        directHandle: string,
+        routePage: (throughSequence: number) => { routedCount: number; hasMore: boolean }
+      ): Promise<void> => {
+        const throughSequence = db.getLatestUnreadDirectMessageSequenceForRun(runId, directHandle)
+        if (throughSequence !== undefined) {
+          await routeAllDirectMessagePages(() => routePage(throughSequence), signal)
+        }
+      }
 
       // Why: a live runtime handle is authoritative; pane metadata is only the restart fallback.
       const paneKey = runtime.getTerminalPaneKey(handle) ?? params.terminalPaneKey
@@ -741,13 +768,29 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
         const generation = run.consumer_generation
         const address = `run:${run.id}`
         runtime.ensureOrchestrationFederationRelay(run.id)
-        const routedDirect = db.routeUnreadDirectMessagesToRunMailbox(run.id, handle)
-        if (
-          routedDirect.routedCount === 0 &&
-          run.coordinator_handle &&
-          run.coordinator_handle !== handle
-        ) {
-          db.routeUnreadDirectMessagesToRunMailbox(run.id, run.coordinator_handle)
+        await routeDirectSnapshot(run.id, handle, (throughSequence) =>
+          db.routeUnreadDirectMessagesToRunMailbox(run.id, handle, throughSequence)
+        )
+        const coordinatorHandle = run.coordinator_handle
+        if (coordinatorHandle && coordinatorHandle !== handle) {
+          await routeDirectSnapshot(run.id, coordinatorHandle, (throughSequence) =>
+            db.routeUnreadDirectMessagesToRunMailbox(run.id, coordinatorHandle, throughSequence)
+          )
+        }
+        revalidateLegacyCoordinator?.()
+        const currentRun = resolveRunScope(runtime, {
+          runId: run.id,
+          callerTerminalHandle: handle,
+          callerPaneKey: paneKey ?? undefined,
+          requireCurrentConsumer: true,
+          legacyCoordinatorRunId,
+          callerEvidence: orchestrationCompatibilityEvidence
+        })
+        if (currentRun.consumer_generation !== generation) {
+          throw new OrchestrationError(
+            'consumer_fenced',
+            'This mailbox consumer was replaced while routing pending mail.'
+          )
         }
 
         const acknowledged = params.ack
@@ -917,24 +960,68 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
           : undefined
       if (workerMailbox) {
         const address = `dispatch:${workerMailbox.dispatchId}`
-        if (activeDispatch) {
-          const routedDirect = db.routeUnreadDirectMessagesToDispatchMailbox(
-            activeDispatch.id,
-            activeDispatch.run_id,
-            handle
-          )
+        const revalidateWorkerMailbox = (): void => {
+          if (activeDispatch) {
+            const current = db.getActiveDispatchForIdentity(handle, paneKey ?? undefined)
+            if (current?.id === activeDispatch.id) {
+              return
+            }
+          } else if (remoteAttachment && paneKey) {
+            const current = db.findActiveRemoteAttachmentForPane(paneKey)
+            if (
+              current?.dispatch_id === remoteAttachment.dispatch_id &&
+              db.isRemoteAttachmentProcessCurrent({
+                dispatchId: current.dispatch_id,
+                paneKey,
+                processIncarnation: runtime.getTerminalProcessIncarnation(handle)
+              })
+            ) {
+              return
+            }
+          }
+          const latestDispatch = db.getDispatchContextById(workerMailbox.dispatchId)
+          const owningRunId =
+            latestDispatch?.run_id ?? activeDispatch?.run_id ?? workerMailbox.runId
           if (
-            routedDirect.routedCount === 0 &&
-            activeDispatch.assignee_handle &&
-            activeDispatch.assignee_handle !== handle
+            owningRunId &&
+            (!latestDispatch ||
+              (latestDispatch.status !== 'pending' && latestDispatch.status !== 'dispatched'))
           ) {
+            const routed = db.routeUnreadDispatchMailboxToRunMailbox(
+              workerMailbox.dispatchId,
+              owningRunId
+            )
+            for (const routedType of routed.types) {
+              runtime.notifyMessageArrived(`run:${owningRunId}`, routedType)
+            }
+          }
+          throw new OrchestrationError(
+            'dispatch_inactive',
+            `Dispatch ${workerMailbox.dispatchId} is no longer assigned to this worker.`
+          )
+        }
+        if (activeDispatch) {
+          await routeDirectSnapshot(activeDispatch.run_id, handle, (throughSequence) =>
             db.routeUnreadDirectMessagesToDispatchMailbox(
               activeDispatch.id,
               activeDispatch.run_id,
-              activeDispatch.assignee_handle
+              handle,
+              throughSequence
+            )
+          )
+          const assigneeHandle = activeDispatch.assignee_handle
+          if (assigneeHandle && assigneeHandle !== handle) {
+            await routeDirectSnapshot(activeDispatch.run_id, assigneeHandle, (throughSequence) =>
+              db.routeUnreadDirectMessagesToDispatchMailbox(
+                activeDispatch.id,
+                activeDispatch.run_id,
+                assigneeHandle,
+                throughSequence
+              )
             )
           }
         }
+        revalidateWorkerMailbox()
         const showAll = params.all === true || (params.unread === false && params.peek !== true)
         const messages = showAll
           ? db.getAllMessagesForHandle(address, 100, typeFilter)
@@ -958,6 +1045,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [
           timeoutMs: params.timeoutMs ?? undefined,
           signal
         })
+        revalidateWorkerMailbox()
         if (waitResult === 'timed_out' || waitResult === 'cancelled') {
           return {
             ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}),
diff --git a/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts b/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts
index 85c541cea2b..f9dc83c0967 100644
--- a/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts
+++ b/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts
@@ -233,6 +233,7 @@ type StoredMessageRow = {
 
 function makeOrchestrationDbStub(toHandle: () => string) {
   const rows: StoredMessageRow[] = []
+  const runMailbox = 'run:run_test'
   const markAsDelivered = vi.fn((ids: string[]) => {
     for (const row of rows) {
       if (ids.includes(row.id)) {
@@ -240,9 +241,18 @@ function makeOrchestrationDbStub(toHandle: () => string) {
       }
     }
   })
+  const markAsUndelivered = vi.fn((ids: string[]) => {
+    for (const row of rows) {
+      if (ids.includes(row.id) && row.read === 0) {
+        row.delivered_at = null
+      }
+    }
+  })
   return {
     rows,
+    runMailbox,
     markAsDelivered,
+    markAsUndelivered,
     insert(subject: string, type: StoredMessageRow['type'] = 'status'): void {
       rows.push({
         id: `msg_${rows.length + 1}`,
@@ -268,10 +278,34 @@ function makeOrchestrationDbStub(toHandle: () => string) {
         rows.filter((row) => row.to_handle === handle && row.read === 0 && !row.delivered_at),
       getUndeliveredUnreadMailboxHandles: () => [toHandle()],
       getActiveCoordinatorRun: () => null,
-      getCurrentRunForPane: () => undefined,
+      getCurrentRunForPane: () => ({ id: 'run_test' }),
+      getRun: () => ({ id: 'run_test', coordinator_handle: toHandle() }),
+      hasUndeliveredDirectMessageForRun: (runId: string, handle: string) =>
+        rows.some(
+          (row) =>
+            row.run_id === runId && row.to_handle === handle && row.read === 0 && !row.delivered_at
+        ),
+      routeUnreadDirectMessagesToRunMailbox: (runId: string, handle: string) => {
+        const routed = rows.filter(
+          (row) => row.run_id === runId && row.to_handle === handle && row.read === 0
+        )
+        for (const row of routed) {
+          row.to_handle = runMailbox
+        }
+        return {
+          routedCount: routed.length,
+          hasMore: false,
+          types: [...new Set(routed.map((row) => row.type))]
+        }
+      },
+      areUnreadMessages: (handle: string, ids: string[]) =>
+        ids.every((id) =>
+          rows.some((row) => row.id === id && row.to_handle === handle && row.read === 0)
+        ),
       // Consulted by onPtyExit's dispatch-failure path.
       getActiveDispatchForTerminal: () => null,
       markAsDelivered,
+      markAsUndelivered,
       close: () => {}
     }
   }
@@ -345,7 +379,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
 
     const pulled: string[] = []
     const checkResumed = runtime
-      .waitForMessage(handle, { typeFilter: ['worker_done'], timeoutMs: 60_000 })
+      .waitForMessage(stub.runMailbox, { typeFilter: ['worker_done'], timeoutMs: 60_000 })
       .then(() => {
         for (const row of stub.rows) {
           if (row.type === 'worker_done' && row.read === 0) {
@@ -391,7 +425,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
         })
     })
 
-    const waitPromise = runtime.waitForMessage(handle, {
+    const waitPromise = runtime.waitForMessage(stub.runMailbox, {
       typeFilter: ['worker_done'],
       timeoutMs: 60_000
     })
@@ -402,7 +436,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
 
     // The reserving waiter goes away, then its type finally arrives — and the
     // probe dedup drops this notify, so only the continuation can deliver it.
-    runtime.cancelMessageWaiters(handle)
+    runtime.cancelMessageWaiters(stub.runMailbox)
     await expect(waitPromise).resolves.toBe('cancelled')
     stub.insert('late completion', 'worker_done')
     runtime.notifyMessageArrived(handle, 'worker_done')
@@ -570,7 +604,8 @@ describe('push-on-idle orchestration delivery absence gate', () => {
 
       await vi.advanceTimersByTimeAsync(500)
       expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)
-      expect(stub.markAsDelivered).not.toHaveBeenCalled()
+      expect(stub.markAsDelivered).toHaveBeenCalledOnce()
+      expect(stub.markAsUndelivered).toHaveBeenCalledOnce()
       expect(stub.rows[0].delivered_at).toBeNull()
 
       // The replacement's own delivery starts a fresh flight and completes —
@@ -590,7 +625,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
       expect(payloadWrites).toHaveLength(2)
       await vi.advanceTimersByTimeAsync(500)
       expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(1)
-      expect(stub.markAsDelivered).toHaveBeenCalledOnce()
+      expect(stub.markAsDelivered).toHaveBeenCalledTimes(2)
       expect(stub.rows[0].delivered_at).toEqual(expect.any(String))
     } finally {
       vi.useRealTimers()
@@ -625,7 +660,8 @@ describe('push-on-idle orchestration delivery absence gate', () => {
 
       await vi.advanceTimersByTimeAsync(500)
       expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)
-      expect(stub.markAsDelivered).not.toHaveBeenCalled()
+      expect(stub.markAsDelivered).toHaveBeenCalledOnce()
+      expect(stub.markAsUndelivered).toHaveBeenCalledOnce()
       // No stray settle flushed the parked trigger into the dead pty.
       expect(write).toHaveBeenCalledTimes(1)
       expect(stub.rows.every((row) => row.delivered_at === null)).toBe(true)
@@ -655,7 +691,8 @@ describe('push-on-idle orchestration delivery absence gate', () => {
 
       await vi.advanceTimersByTimeAsync(500)
       expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)
-      expect(stub.markAsDelivered).not.toHaveBeenCalled()
+      expect(stub.markAsDelivered).toHaveBeenCalledOnce()
+      expect(stub.markAsUndelivered).toHaveBeenCalledOnce()
       expect(stub.rows[0].delivered_at).toBeNull()
     } finally {
       vi.useRealTimers()