diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc
index 6d7292f3bf3..fd8ee458c0b 100644
--- a/config/reliability-gates.jsonc
+++ b/config/reliability-gates.jsonc
@@ -8785,6 +8785,95 @@
       ],
       "demotionRule": "Demote or quarantine if the live harness flakes without a product bug or harness bug filed to the terminal-input owner."
     },
+    {
+      "id": "orchestration.notification-mailbox-consistency",
+      "title": "Orchestration pointers expose the mailbox their terminal check consumes",
+      "maturity": "experimental",
+      "protection": "partial",
+      "owner": "orchestration",
+      "layer": "runtime-db-rpc-contract",
+      "surfaces": [
+        "live-idle terminal notification",
+        "Run binding and rebinding",
+        "active Dispatch ownership",
+        "orchestration check and acknowledgment",
+        "outstanding Run Deliveries",
+        "runtime restart and durable pointer cursors"
+      ],
+      "platforms": ["macos", "linux", "windows"],
+      "providers": ["local", "daemon", "ssh", "wsl", "remote-runtime"],
+      "coveredPlatforms": ["macos"],
+      "coveredProviders": ["local"],
+      "coverageNotes": "A deterministic in-process runtime/SQLite/RPC harness fixes one terminal handle and pane across Run and Dispatch ownership changes, captures mocked PTY-controller writes, executes the production check, reopens the durable database under a fresh runtime, and verifies Run Delivery acknowledgment plus worker-check consumption. The host-side authority decision is provider-independent and adds no wire field, renderer state, polling, or PTY-provider call. Daemon, paired remote-runtime, 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 the database and repeat. For same-Run mail, require one submitted pointer, a bound Delivery containing that row, explicit acknowledgment, and no replay after a fresh runtime. Require an active worker's direct mail to route through its Dispatch check, suppress newer Run pointers behind an outstanding Delivery, normalize mail that arrived while its Run was displaced, avoid full mailbox materialization while working, and let bound waiters consume without a duplicate PTY pointer.",
+      "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",
+        "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/orchestration-run-delivery-db.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"
+      ],
+      "assertionRefs": [
+        {
+          "file": "src/main/runtime/orchestration-mailbox-notification-consistency.test.ts",
+          "assertions": [
+            "Run A direct mail cannot point after the pane binds Run B",
+            "a fresh runtime cannot replay the unactionable pointer",
+            "a delayed pointer submit cannot press Enter after the agent becomes working",
+            "same-Run direct mail is delivered and acknowledged through the bound Run",
+            "a submitted actionable pointer is durably deduplicated across restart without marking mail read",
+            "newer Run mail is not pointed behind an outstanding fixed Delivery",
+            "active worker direct mail is exposed by its Dispatch check",
+            "late direct mail is normalized when its displaced Run rebinds",
+            "a bound waiter wakes without a duplicate PTY pointer",
+            "working-agent arrival uses bounded ownership lookup without a full mailbox scan"
+          ]
+        }
+      ],
+      "evidenceRuns": [
+        {
+          "date": "2026-08-10",
+          "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",
+          "result": "passed",
+          "durationSeconds": 3.1,
+          "summary": "Candidate passed all 10 mailbox-identity, durable restart, submit-race, bounded-batch, outstanding-Delivery, Dispatch, acknowledgment, waiter, and bounded-arrival scenarios. The identical 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."
+        }
+      ],
+      "runtimeBudget": {
+        "p95Seconds": 15,
+        "scope": "deterministic runtime, SQLite, and RPC contract tests"
+      },
+      "flakeHistory": {
+        "status": "unknown",
+        "evidence": "First deterministic local macOS run only; no sleeps or wall-clock delivery assertions are used."
+      },
+      "redGreenEvidence": {
+        "status": "complete",
+        "evidence": "The identical 10-scenario oracle failed 10/10 on origin/main plus the oracle commit, passed 10/10 on the candidate, failed 10/10 after explicitly reverting the implementation commit, and passed again after restoration. Live topology coverage remains a separate promotion gap."
+      },
+      "performanceBudget": {
+        "required": true,
+        "evidence": "Arrival ownership uses indexed identity and SELECT 1 LIMIT 1 probes. Pointer selection is capped to 50 rows inside SQLite, submit-time row revalidation is forced through 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, new timers, subprocesses, provider calls, listeners, or renderer work were added."
+      },
+      "promotionCriteria": [
+        "Add isolated live restart and SSH or paired-runtime evidence.",
+        "Accumulate stable soak runs before considering blocking promotion."
+      ],
+      "knownGaps": [
+        "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.",
+        "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."
+    },
     {
       "id": "orchestration.worker-terminal-delivery",
       "title": "Workers remain visible and observable across workspace entry and app restart",
diff --git a/src/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts
index 3ee35d809df..4ef974f12f8 100644
--- a/src/main/runtime/orca-runtime.test.ts
+++ b/src/main/runtime/orca-runtime.test.ts
@@ -19589,7 +19589,7 @@ describe('OrcaRuntimeService', () => {
       const unread = db.getUnreadMessages(terminal.handle)
       expect(unread).toHaveLength(1)
       expect(unread[0].read).toBe(0)
-      expect(unread[0].delivered_at).toBeNull()
+      expect(unread[0].delivered_at).not.toBeNull()
       db.close()
     } finally {
       vi.useRealTimers()
@@ -19635,7 +19635,7 @@ describe('OrcaRuntimeService', () => {
       const unread = db.getUnreadMessages(terminal.handle)
       expect(unread).toHaveLength(1)
       expect(unread[0].read).toBe(0)
-      expect(unread[0].delivered_at).toBeNull()
+      expect(unread[0].delivered_at).not.toBeNull()
       db.close()
     } finally {
       vi.useRealTimers()
@@ -19676,7 +19676,7 @@ describe('OrcaRuntimeService', () => {
       const unread = db.getUnreadMessages(terminal.handle)
       expect(unread).toHaveLength(1)
       expect(unread[0].read).toBe(0)
-      expect(unread[0].delivered_at).toBeNull()
+      expect(unread[0].delivered_at).not.toBeNull()
       db.close()
     } finally {
       vi.useRealTimers()
@@ -34599,7 +34599,7 @@ describe('OrcaRuntimeService', () => {
       expect(write).not.toHaveBeenCalledWith('pty-1', expect.stringContaining('after wait'))
       await vi.advanceTimersByTimeAsync(500)
       expect(write).toHaveBeenCalledWith('pty-1', '\r')
-      expect(message.delivered_at).toBeNull()
+      expect(message.delivered_at).not.toBeNull()
       db.close()
     } finally {
       vi.useRealTimers()
@@ -34650,7 +34650,7 @@ describe('OrcaRuntimeService', () => {
       )
       await vi.advanceTimersByTimeAsync(500)
       expect(write).toHaveBeenCalledWith('pty-1', '\r')
-      expect(message.delivered_at).toBeNull()
+      expect(message.delivered_at).not.toBeNull()
 
       runtime.notifyMessageArrived('run:run_mailbox', 'worker_done')
       await Promise.resolve()
@@ -35062,7 +35062,7 @@ describe('OrcaRuntimeService', () => {
         expect.stringContaining('You have 1 orchestration message')
       )
       await vi.advanceTimersByTimeAsync(600)
-      expect(message.delivered_at).toBeNull()
+      expect(message.delivered_at).not.toBeNull()
       db.close()
     } finally {
       vi.useRealTimers()
@@ -35109,7 +35109,7 @@ describe('OrcaRuntimeService', () => {
         expect.stringContaining('You have 1 orchestration message')
       )
       await vi.advanceTimersByTimeAsync(600)
-      expect(message.delivered_at).toBeNull()
+      expect(message.delivered_at).not.toBeNull()
       db.close()
     } finally {
       vi.useRealTimers()
@@ -35215,7 +35215,7 @@ describe('OrcaRuntimeService', () => {
         '\nYou have 1 orchestration message. Run `orca orchestration check`.\n'
       )
       expect(payloads.some((data) => data.includes('reserved completion'))).toBe(false)
-      expect(status.delivered_at).toBeNull()
+      expect(status.delivered_at).not.toBeNull()
       expect(done.delivered_at).toBeNull()
       db.close()
     } finally {
@@ -35397,7 +35397,7 @@ describe('OrcaRuntimeService', () => {
         expect.stringContaining('You have 1 orchestration message')
       )
       await vi.advanceTimersByTimeAsync(600)
-      expect(message.delivered_at).toBeNull()
+      expect(message.delivered_at).not.toBeNull()
       db.close()
     } finally {
       vi.useRealTimers()
@@ -35451,7 +35451,7 @@ describe('OrcaRuntimeService', () => {
         expect.stringContaining('You have 1 orchestration message')
       )
       await vi.advanceTimersByTimeAsync(600)
-      expect(message.delivered_at).toBeNull()
+      expect(message.delivered_at).not.toBeNull()
       db.close()
     } finally {
       vi.useRealTimers()
@@ -35500,7 +35500,7 @@ describe('OrcaRuntimeService', () => {
         expect.stringContaining('You have 1 orchestration message')
       )
       await vi.advanceTimersByTimeAsync(600)
-      expect(message.delivered_at).toBeNull()
+      expect(message.delivered_at).not.toBeNull()
 
       // The filtered waiter stays blocked; the push did not consume its wake.
       await vi.advanceTimersByTimeAsync(5_000)
@@ -35639,7 +35639,7 @@ describe('OrcaRuntimeService', () => {
       // timer-only settle (CodeRabbit settling-timeout gap, #12584).
       await vi.advanceTimersByTimeAsync(3_000)
       expect(write).toHaveBeenCalledWith('pty-1', '\r')
-      expect(first.delivered_at).toBeNull()
+      expect(first.delivered_at).not.toBeNull()
       expect(
         write.mock.calls.filter(
           ([, payload]) =>
@@ -35648,9 +35648,9 @@ describe('OrcaRuntimeService', () => {
       ).toHaveLength(2)
       expect(write).toHaveBeenCalledWith(
         'pty-1',
-        expect.stringContaining('You have 2 orchestration messages')
+        expect.stringContaining('You have 1 orchestration message')
       )
-      expect(second.delivered_at).toBeNull()
+      expect(second.delivered_at).not.toBeNull()
       db.close()
     } finally {
       vi.useRealTimers()
diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts
index 7f3697454be..93057f172bd 100644
--- a/src/main/runtime/orca-runtime.ts
+++ b/src/main/runtime/orca-runtime.ts
@@ -118,7 +118,7 @@ import { isAbsolute, join, resolve } from 'node:path'
 import { mkdir, readFile, readdir, rm, stat } from 'node:fs/promises'
 import { resolveWorktreeCreateBase } from '../worktree-create-base'
 import { resolveWorktreeAddBaseRef } from '../../shared/worktree/base-ref'
-import { OrchestrationDb } from './orchestration/db'
+import { ORCHESTRATION_DELIVERY_BATCH_LIMIT, OrchestrationDb } from './orchestration/db'
 import { reconcileRequestedWorkerTerminalReleases } from './orchestration/worker-terminal-release-reconciliation'
 import {
   classifyWorkerTerminalProcessIncarnation,
@@ -32356,13 +32356,24 @@ export class OrcaRuntimeService {
     let terminalHandle = handle
     if (!this.handles.has(terminalHandle)) {
       const runId = handle.startsWith('run:') ? handle.slice('run:'.length) : ''
-      const coordinatorHandle = runId
+      const dispatchId = handle.startsWith('dispatch:') ? handle.slice('dispatch:'.length) : ''
+      const dispatch = dispatchId
+        ? this._orchestrationDb?.getDispatchContextById?.(dispatchId)
+        : undefined
+      const remoteAttachment =
+        dispatchId && !dispatch
+          ? this._orchestrationDb?.getRemoteDispatchAttachment?.(dispatchId)
+          : undefined
+      const ownerPaneKey = dispatch?.assignee_pane_key ?? remoteAttachment?.pane_key
+      const ownerHandle = runId
         ? this._orchestrationDb?.getRun(runId)?.coordinator_handle
-        : null
-      if (!coordinatorHandle || !this.handles.has(coordinatorHandle)) {
+        : ((ownerPaneKey ? this.getTerminalHandleForPaneKey(ownerPaneKey) : null) ??
+          dispatch?.assignee_handle ??
+          remoteAttachment?.terminal_handle)
+      if (!ownerHandle || !this.handles.has(ownerHandle)) {
         return
       }
-      terminalHandle = coordinatorHandle
+      terminalHandle = ownerHandle
     }
     try {
       const { leaf } = this.getLiveLeafForHandle(terminalHandle)
@@ -32372,7 +32383,10 @@ export class OrcaRuntimeService {
       // would type a message plus Enter into a working agent. Seeded state waits
       // for a live observation to authorize it.
       if (leaf.lastAgentStatus === 'idle' && leaf.lastAgentStatusObservedLive) {
-        this.deliverPendingMessages(leaf, { mailboxHandle: handle, reservedTypes })
+        const mailboxHandle = this.resolveActionableMailboxForLeaf(leaf, handle)
+        if (mailboxHandle) {
+          this.deliverPendingMessages(leaf, { mailboxHandle, reservedTypes })
+        }
       }
     } catch {
       // Unknown/stale handles can't be pointed now; the persisted message stays available via explicit check or future idle delivery.
@@ -32397,13 +32411,129 @@ export class OrcaRuntimeService {
   }
 
   private deliverPendingMessagesForLeaf(leaf: RuntimeLeafRecord): void {
-    this.deliverPendingMessages(leaf)
-    if (!this._orchestrationDb) {
-      return
+    const mailboxHandle = this.resolveActionableMailboxForLeaf(leaf)
+    if (mailboxHandle) {
+      this.deliverPendingMessages(leaf, { mailboxHandle })
     }
-    const run = this._orchestrationDb.getCurrentRunForPane?.(`${leaf.tabId}:${leaf.leafId}`)
+  }
+
+  private resolveActionableMailboxForLeaf(
+    leaf: RuntimeLeafRecord,
+    requestedMailbox?: string,
+    options: { requireRequestedMail?: boolean; routeDirectMail?: boolean } = {}
+  ): string | null {
+    if (!this._orchestrationDb) {
+      return null
+    }
+    const terminalHandle = this.handleByLeafKey.get(this.getLeafKey(leaf.tabId, leaf.leafId))
+    if (!terminalHandle) {
+      return null
+    }
+    const paneKey = `${leaf.tabId}:${leaf.leafId}`
+    const run = this._orchestrationDb.getCurrentRunForPane?.(paneKey)
     if (run) {
-      this.deliverPendingMessages(leaf, { mailboxHandle: `run:${run.id}` })
+      const runMailbox = `run:${run.id}`
+      if (
+        requestedMailbox &&
+        requestedMailbox !== terminalHandle &&
+        requestedMailbox !== runMailbox
+      ) {
+        return null
+      }
+      const hasDirectRunMail =
+        this._orchestrationDb.hasUndeliveredDirectMessageForRun?.(run.id, terminalHandle) ?? false
+      if (
+        options.requireRequestedMail &&
+        requestedMailbox === terminalHandle &&
+        !hasDirectRunMail
+      ) {
+        return null
+      }
+      if (hasDirectRunMail && options.routeDirectMail !== false) {
+        this._orchestrationDb.routeUnreadDirectMessagesToRunMailbox?.(run.id, terminalHandle)
+      }
+      return runMailbox
+    }
+
+    const dispatch = this._orchestrationDb.getActiveDispatchForIdentity?.(terminalHandle, paneKey)
+    if (dispatch) {
+      const dispatchMailbox = `dispatch:${dispatch.id}`
+      if (
+        requestedMailbox &&
+        requestedMailbox !== terminalHandle &&
+        requestedMailbox !== dispatchMailbox
+      ) {
+        return null
+      }
+      const hasDirectDispatchMail =
+        this._orchestrationDb.hasUndeliveredDirectMessageForRun?.(
+          dispatch.run_id,
+          terminalHandle
+        ) ?? false
+      if (
+        options.requireRequestedMail &&
+        requestedMailbox === terminalHandle &&
+        !hasDirectDispatchMail
+      ) {
+        return null
+      }
+      if (hasDirectDispatchMail && options.routeDirectMail !== false) {
+        this._orchestrationDb.routeUnreadDirectMessagesToDispatchMailbox?.(
+          dispatch.id,
+          dispatch.run_id,
+          terminalHandle
+        )
+      }
+      return dispatchMailbox
+    }
+
+    const remoteAttachment = this._orchestrationDb.findActiveRemoteAttachmentForPane?.(paneKey)
+    if (remoteAttachment) {
+      const isCurrent = this._orchestrationDb.isRemoteAttachmentProcessCurrent?.({
+        dispatchId: remoteAttachment.dispatch_id,
+        paneKey,
+        processIncarnation: this.getTerminalProcessIncarnation(terminalHandle)
+      })
+      if (!isCurrent || (options.requireRequestedMail && requestedMailbox === terminalHandle)) {
+        return null
+      }
+      const dispatchMailbox = `dispatch:${remoteAttachment.dispatch_id}`
+      return !requestedMailbox || requestedMailbox === dispatchMailbox ? dispatchMailbox : null
+    }
+
+    return !requestedMailbox || requestedMailbox === terminalHandle ? terminalHandle : null
+  }
+
+  private resolveArrivedMessageMailboxes(handle: string): {
+    mailboxHandle: string | null
+    forwarded: { runId: string; types: string[] }[]
+  } {
+    if (!this.handles.has(handle) || !this._orchestrationDb) {
+      return { mailboxHandle: handle, forwarded: [] }
+    }
+    try {
+      const { leaf } = this.getLiveLeafForHandle(handle)
+      const ownerMailbox = this.resolveActionableMailboxForLeaf(leaf, undefined, {
+        routeDirectMail: false
+      })
+      const ownerRunId = ownerMailbox?.startsWith('run:')
+        ? ownerMailbox.slice('run:'.length)
+        : ownerMailbox?.startsWith('dispatch:')
+          ? this._orchestrationDb.getDispatchContextById?.(ownerMailbox.slice('dispatch:'.length))
+              ?.run_id
+          : undefined
+      const forwarded = ownerRunId
+        ? (this._orchestrationDb.routeForeignDirectMessagesToRunMailboxes?.(handle, ownerRunId) ??
+          [])
+        : []
+      return {
+        mailboxHandle: this.resolveActionableMailboxForLeaf(leaf, handle, {
+          requireRequestedMail: true
+        }),
+        forwarded
+      }
+    } catch {
+      return { mailboxHandle: null, forwarded: [] }
     }
   }
 
@@ -32412,6 +32542,15 @@ export class OrcaRuntimeService {
     if (!handle.startsWith('dispatch:')) {
       this.mailPointerRepointScheduler.schedule(handle)
     }
+    const { mailboxHandle, forwarded } = this.resolveArrivedMessageMailboxes(handle)
+    for (const routed of forwarded) {
+      for (const routedType of routed.types) {
+        this.notifyMessageArrived(`run:${routed.runId}`, routedType)
+      }
+    }
+    if (!mailboxHandle) {
+      return
+    }
     // Why: push-on-idle is driven by status transitions; a message that
     // arrives while the recipient is already idle never sees a transition, so
     // deliver now (#12536). deliverPendingMessagesForHandle no-ops when the
@@ -32423,7 +32562,7 @@ export class OrcaRuntimeService {
     // never returns this row — check re-reads under the same filter on timeout
     // — so treating it as the consumer strands the message in exactly the
     // already-idle state #12536 is about.
-    const waiters = this.messageWaitersByHandle.get(handle)
+    const waiters = this.messageWaitersByHandle.get(mailboxHandle)
     // Why: don't wake a coordinator waiting for worker_done/escalation on heartbeat noise it would misread as idleness.
     const consumers = waiters
       ? [...waiters].filter(
@@ -32444,7 +32583,7 @@ export class OrcaRuntimeService {
       // synchronous push would inject rows the resolved check is about to return.
       // Deferring one hop puts the push behind any already-queued check, and it
       // re-reads undelivered rows when it runs so nothing strands.
-      queueMicrotask(() => this.deliverPendingMessagesForHandle(handle, reservedTypes))
+      queueMicrotask(() => this.deliverPendingMessagesForHandle(mailboxHandle, reservedTypes))
       return
     }
     for (const waiter of consumers) {
@@ -33094,7 +33233,33 @@ export class OrcaRuntimeService {
     }
   }
 
-  // Why: normal delivery stays event-driven; the bounded mailbox retry only repairs missed liveness edges.
+  private canSubmitMessagePointer(
+    leaf: RuntimeLeafRecord,
+    mailboxHandle: string,
+    messages: readonly { id: string; type: string }[]
+  ): boolean {
+    if (this.resolveActionableMailboxForLeaf(leaf) !== mailboxHandle) {
+      return false
+    }
+    if (
+      mailboxHandle.startsWith('run:') &&
+      this._orchestrationDb?.hasOutstandingRunDelivery?.(mailboxHandle.slice('run:'.length))
+    ) {
+      return false
+    }
+    const waiters = this.messageWaitersByHandle.get(mailboxHandle)
+    if (messages.some((message) => messageTypeHasLiveWaiter(waiters, message.type))) {
+      return false
+    }
+    return (
+      this._orchestrationDb?.areUndeliveredUnreadMessages?.(
+        mailboxHandle,
+        messages.map((message) => message.id)
+      ) ?? true
+    )
+  }
+
+  // Why: push-on-idle delivery is event-driven (no polling) because the runtime owns both the message store and terminal status detection.
   private deliverPendingMessages(
     leaf: RuntimeLeafRecord,
     options: {
@@ -33113,6 +33278,13 @@ export class OrcaRuntimeService {
     }
     const mailboxHandle = options.mailboxHandle ?? handle
 
+    if (
+      mailboxHandle.startsWith('run:') &&
+      this._orchestrationDb.hasOutstandingRunDelivery?.(mailboxHandle.slice('run:'.length))
+    ) {
+      return
+    }
+
     if (leaf.ptyId && this.messageDeliveryFlightsByPtyId.has(leaf.ptyId)) {
       let parked = this.parkedMessageRedeliveriesByPtyId.get(leaf.ptyId)
       if (!parked) {
@@ -33134,7 +33306,19 @@ export class OrcaRuntimeService {
     // still-blocked case; reservedTypes carries the notify-time snapshot for a
     // waiter resolved later in the same drain, which is already gone from the map.
     const waiters = this.messageWaitersByHandle.get(mailboxHandle)
-    const pending = this._orchestrationDb.getUndeliveredUnreadMessages(mailboxHandle)
+    if ([...(waiters ?? [])].some((waiter) => !waiter.typeFilter)) {
+      return
+    }
+    const excludedTypes = new Set(options.reservedTypes)
+    for (const waiter of waiters ?? []) {
+      for (const type of waiter.typeFilter ?? []) {
+        excludedTypes.add(type)
+      }
+    }
+    const pending = this._orchestrationDb.getUndeliveredUnreadMessages(mailboxHandle, undefined, {
+      excludeTypes: [...excludedTypes],
+      limit: ORCHESTRATION_DELIVERY_BATCH_LIMIT
+    })
     const pendingIds = new Set(pending.map((message) => message.id))
     const pointedIds = this.pointedMessageIdsByHandle.get(mailboxHandle)
     if (pointedIds) {
@@ -33147,11 +33331,13 @@ export class OrcaRuntimeService {
         this.pointedMessageIdsByHandle.delete(mailboxHandle)
       }
     }
-    const unread = pending.filter(
-      (message) =>
-        !options.reservedTypes?.has(message.type) &&
-        !messageTypeHasLiveWaiter(waiters, message.type)
-    )
+    const unread = pending
+      .filter(
+        (message) =>
+          !options.reservedTypes?.has(message.type) &&
+          !messageTypeHasLiveWaiter(waiters, message.type)
+      )
+      .slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT)
     if (unread.length === 0) {
       return
     }
@@ -33255,30 +33441,39 @@ export class OrcaRuntimeService {
       const tabTitle = this.tabs.get(leaf.tabId)?.title
       if (isCursorAgentOrchestrationTarget(leaf, tabTitle)) {
         // Why: Cursor Agent treats injected PTY text as editable prompt input, so submitting must stay under user control.
+        this._orchestrationDb.markAsDelivered(unread.map((message) => message.id))
         return
       }
 
       // Why: agent TUIs can swallow a \r in the same PTY write; submit separately after a delay.
       flight.enterTimer = setTimeout(() => {
-        try {
-          // Why current state, not the closure: graph resync replaces leaf
-          // objects, so the captured record can read writable=true after the
-          // pty died, and an exit retire may have superseded this flight.
-          if (this.messageDeliveryFlightsByPtyId.get(deliveryPtyId) !== flight) {
-            return
-          }
-          const currentLeaf = this.leaves.get(this.getLeafKey(leaf.tabId, leaf.leafId))
-          if (!currentLeaf || currentLeaf.ptyId !== deliveryPtyId || !currentLeaf.writable) {
-            return
-          }
-          this.ptyController?.write(deliveryPtyId, '\r')
-        } catch {
-          // Terminal may have closed during the delay; mail remains queued for check.
-        } finally {
-          // Why finally: every outcome — submit, refusal, throw — ends the flight,
-          // and settle re-runs any trigger parked during it so nothing strands.
-          this.settlePendingMessageDelivery(deliveryPtyId, flight)
-        }
+        void this.isLeafPtyProvenAbsent(deliveryPtyId)
+          .then((absent) => {
+            // Why current state, not the closure: ownership, graph records, or the
+            // process can change during the Enter delay and liveness probe.
+            if (absent || this.messageDeliveryFlightsByPtyId.get(deliveryPtyId) !== flight) {
+              return
+            }
+            const currentLeaf = this.leaves.get(this.getLeafKey(leaf.tabId, leaf.leafId))
+            if (
+              !currentLeaf ||
+              currentLeaf.ptyId !== deliveryPtyId ||
+              !currentLeaf.writable ||
+              currentLeaf.lastAgentStatus !== 'idle' ||
+              !currentLeaf.lastAgentStatusObservedLive ||
+              !this.canSubmitMessagePointer(currentLeaf, mailboxHandle, unread)
+            ) {
+              return
+            }
+            const submitted = this.ptyController?.write(deliveryPtyId, '\r') ?? false
+            if (submitted) {
+              this._orchestrationDb?.markAsDelivered(unread.map((message) => message.id))
+            }
+          })
+          .catch(() => {
+            // Terminal liveness is uncertain; mail remains queued for explicit check.
+          })
+          .finally(() => this.settlePendingMessageDelivery(deliveryPtyId, flight))
       }, 500)
       settlesInEnterCallback = true
     } finally {
diff --git a/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts b/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts
index 42cd972a1ee..fc3625ffd89 100644
--- a/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts
+++ b/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts
@@ -3,6 +3,7 @@ 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 { RpcDispatcher } from './rpc/dispatcher'
@@ -30,6 +31,10 @@ function createDatabase(prefix: string): OrchestrationDb {
   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,
@@ -243,6 +248,7 @@ describe('orchestration notification mailbox consistency', () => {
         runId: run.id
       })
     )
+    const notificationQuery = vi.spyOn(db, 'getUndeliveredUnreadMessages')
 
     await driveToLiveIdle(harness.runtime)
     await vi.advanceTimersByTimeAsync(500)
@@ -253,6 +259,15 @@ describe('orchestration notification mailbox consistency', () => {
       PTY_ID,
       expect.stringContaining('You have 50 orchestration messages')
     )
+    expect(notificationQuery.mock.results[0]?.value).toHaveLength(50)
+    const revalidationPlan = sqliteFor(db)
+      .prepare(
+        `EXPLAIN QUERY PLAN SELECT COUNT(*) FROM messages INDEXED BY idx_messages_id
+         WHERE to_handle = ? AND read = 0 AND delivered_at IS NULL
+           AND delivery_contract = 'current_delivery' AND id IN (?)`
+      )
+      .all(`run:${run.id}`, messages[0]!.id) as { detail: string }[]
+    expect(revalidationPlan.map((row) => row.detail).join(' ')).toContain('idx_messages_id')
     expect(checked).toMatchObject({ runId: run.id, count: 50 })
     expect(checked.deliveryId).toBeTruthy()
     expect(checked.messages).toEqual(
diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts
index 0613be520fa..b0d6f039b71 100644
--- a/src/main/runtime/orchestration/db.ts
+++ b/src/main/runtime/orchestration/db.ts
@@ -87,12 +87,16 @@ function parseWorkerTerminalPriorOwnerIds(value: string): string[] | null {
   }
 }
 
+const MESSAGE_ID_UPDATE_BATCH_SIZE = 500
+export const ORCHESTRATION_DELIVERY_BATCH_LIMIT = 50
+
 // Why: indexable pre-filter for isEquivalentPaneKey — equal strings and equal leaves both share the
 // text after the first ':', so this narrows candidates without deciding equivalence itself.
 const RUN_PANE_KEY_MATCH_SUFFIX_SQL =
   "substr(coordinator_pane_key, instr(coordinator_pane_key, ':') + 1)"
 const DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL =
   "substr(assignee_pane_key, instr(assignee_pane_key, ':') + 1)"
+const REMOTE_ATTACHMENT_PANE_KEY_MATCH_SUFFIX_SQL = "substr(pane_key, instr(pane_key, ':') + 1)"
 
 function paneKeyMatchSuffix(paneKey: string): string {
   const colon = paneKey.indexOf(':')
@@ -510,6 +514,12 @@ export class OrchestrationDb {
         created_at              TEXT NOT NULL DEFAULT (datetime('now')),
         updated_at              TEXT NOT NULL DEFAULT (datetime('now'))
       );
+      CREATE INDEX IF NOT EXISTS idx_remote_dispatch_attachments_active_pane
+        ON remote_dispatch_attachments(pane_key)
+        WHERE state IN ('starting', 'ready');
+      CREATE INDEX IF NOT EXISTS idx_remote_dispatch_attachments_active_pane_suffix
+        ON remote_dispatch_attachments(${REMOTE_ATTACHMENT_PANE_KEY_MATCH_SUFFIX_SQL})
+        WHERE state IN ('starting', 'ready') AND pane_key IS NOT NULL;
 
       CREATE TABLE IF NOT EXISTS federation_relay_items (
         dispatch_id   TEXT NOT NULL,
@@ -1967,7 +1977,10 @@ export class OrchestrationDb {
     recovery: boolean
   } {
     const principal = this.requireLegacyMailPrincipal(params.principalId)
-    const limit = Math.min(Math.max(params.limit ?? 50, 1), 50)
+    const limit = Math.min(
+      Math.max(params.limit ?? ORCHESTRATION_DELIVERY_BATCH_LIMIT, 1),
+      ORCHESTRATION_DELIVERY_BATCH_LIMIT
+    )
     const addressSql =
       principal.role === 'worker' ? '(m.to_handle = ? OR m.to_handle = ?)' : 'm.to_handle = ?'
     const addressParams =
@@ -2364,6 +2377,13 @@ export class OrchestrationDb {
         )
       }
       this.unbindOtherRunsForPane(params.coordinatorPaneKey, params.runId)
+      for (const handle of new Set(
+        [run.coordinator_handle, params.coordinatorHandle].filter((value): value is string =>
+          Boolean(value)
+        )
+      )) {
+        this.routeUnreadDirectMessagesToRunMailbox(params.runId, handle)
+      }
       if (
         (params.takeoverLegacy && !takeoverAlreadyApplied) ||
         !sameBinding ||
@@ -2470,6 +2490,9 @@ export class OrchestrationDb {
   private unbindOtherRunsForPane(paneKey: string, exceptRunId?: string): void {
     for (const run of this.runsBoundToPane(paneKey)) {
       if (run.id !== exceptRunId) {
+        if (run.coordinator_handle) {
+          this.routeUnreadDirectMessagesToRunMailbox(run.id, run.coordinator_handle)
+        }
         this.db
           .prepare(
             `UPDATE runs
@@ -3494,29 +3517,38 @@ export class OrchestrationDb {
   }
 
   // Why: delivered_at IS NULL filter — push-on-idle delivers each row at most once; read (set only by check) wouldn't prevent replay.
-  getUndeliveredUnreadMessages(toHandle: string, types?: MessageType[]): MessageRow[] {
-    if (types && types.length > 0) {
-      const placeholders = types.map(() => '?').join(',')
-      return exposeMessageListTimestamps(
-        this.db
-          .prepare(
-            `SELECT * FROM messages
-             WHERE to_handle = ? AND read = 0 AND delivered_at IS NULL
-               AND delivery_contract = 'current_delivery'
-               AND type IN (${placeholders}) ORDER BY sequence`
-          )
-          .all(toHandle, ...types) as MessageRow[]
-      )
+  getUndeliveredUnreadMessages(
+    toHandle: string,
+    types?: MessageType[],
+    options?: { excludeTypes?: readonly string[]; limit?: number }
+  ): MessageRow[] {
+    const conditions = [
+      'to_handle = ?',
+      'read = 0',
+      'delivered_at IS NULL',
+      "delivery_contract = 'current_delivery'"
+    ]
+    const params: (string | number)[] = [toHandle]
+    if (types?.length) {
+      conditions.push(`type IN (${types.map(() => '?').join(',')})`)
+      params.push(...types)
+    }
+    if (options?.excludeTypes?.length) {
+      conditions.push(`type NOT IN (${options.excludeTypes.map(() => '?').join(',')})`)
+      params.push(...options.excludeTypes)
+    }
+    const limitSql = options?.limit === undefined ? '' : ' LIMIT ?'
+    if (options?.limit !== undefined) {
+      params.push(Math.max(1, Math.floor(options.limit)))
     }
     return exposeMessageListTimestamps(
       this.db
         .prepare(
           `SELECT * FROM messages
-           WHERE to_handle = ? AND read = 0 AND delivered_at IS NULL
-             AND delivery_contract = 'current_delivery'
-           ORDER BY sequence`
+           WHERE ${conditions.join(' AND ')}
+           ORDER BY sequence${limitSql}`
         )
-        .all(toHandle) as MessageRow[]
+        .all(...params) as MessageRow[]
     )
   }
 
@@ -3532,6 +3564,106 @@ export class OrchestrationDb {
     ).map((row) => row.to_handle)
   }
 
+  hasUndeliveredDirectMessageForRun(runId: string, directHandle: string): boolean {
+    return Boolean(
+      this.db
+        .prepare(
+          `SELECT 1 FROM messages
+           WHERE run_id = ? AND to_handle = ? AND read = 0 AND delivered_at IS NULL
+             AND delivery_contract = 'current_delivery'
+           LIMIT 1`
+        )
+        .get(runId, directHandle)
+    )
+  }
+
+  hasOutstandingRunDelivery(runId: string): boolean {
+    return Boolean(
+      this.db
+        .prepare("SELECT 1 FROM deliveries WHERE run_id = ? AND status = 'outstanding' LIMIT 1")
+        .get(runId)
+    )
+  }
+
+  // Why: change mailbox ownership without changing unread or acknowledgment state.
+  routeUnreadDirectMessagesToRunMailbox(runId: string, directHandle: string): number {
+    const result = this.db
+      .prepare(
+        `UPDATE messages
+         SET to_handle = ?
+         WHERE run_id = ? AND to_handle = ? AND read = 0
+           AND delivery_contract = 'current_delivery'`
+      )
+      .run(`run:${runId}`, runId, directHandle)
+    return Number(result.changes)
+  }
+
+  routeUnreadDirectMessagesToDispatchMailbox(
+    dispatchId: string,
+    runId: string,
+    directHandle: string
+  ): number {
+    const result = this.db
+      .prepare(
+        `UPDATE messages
+         SET to_handle = ?
+         WHERE run_id = ? AND to_handle = ? AND read = 0
+           AND delivery_contract = 'current_delivery'`
+      )
+      .run(`dispatch:${dispatchId}`, runId, directHandle)
+    return Number(result.changes)
+  }
+
+  routeForeignDirectMessagesToRunMailboxes(
+    directHandle: string,
+    currentRunId: string
+  ): { runId: string; types: MessageType[] }[] {
+    const routedRows = this.db
+      .prepare(
+        `SELECT DISTINCT messages.run_id, messages.type FROM messages
+         WHERE to_handle = ? AND run_id <> ? AND read = 0 AND delivered_at IS NULL
+           AND delivery_contract = 'current_delivery'
+           AND EXISTS (SELECT 1 FROM runs WHERE runs.id = messages.run_id AND runs.legacy = 0)`
+      )
+      .all(directHandle, currentRunId) as { run_id: string; type: MessageType }[]
+    if (routedRows.length === 0) {
+      return []
+    }
+    this.db
+      .prepare(
+        `UPDATE messages
+         SET to_handle = 'run:' || run_id
+         WHERE to_handle = ? AND run_id <> ? AND read = 0 AND delivered_at IS NULL
+           AND delivery_contract = 'current_delivery'
+           AND EXISTS (SELECT 1 FROM runs WHERE runs.id = messages.run_id AND runs.legacy = 0)`
+      )
+      .run(directHandle, currentRunId)
+    const byRun = new Map<string, Set<MessageType>>()
+    for (const row of routedRows) {
+      const types = byRun.get(row.run_id) ?? new Set<MessageType>()
+      types.add(row.type)
+      byRun.set(row.run_id, types)
+    }
+    return [...byRun].map(([runId, types]) => ({ runId, types: [...types] }))
+  }
+
+  areUndeliveredUnreadMessages(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 delivered_at IS NULL
+             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
@@ -3557,13 +3689,13 @@ export class OrchestrationDb {
 
   // Why: use datetime('now') so delivered_at matches the space-format UTC shape of the table's other timestamps for correct ordering (§3.2).
   markAsDelivered(ids: string[]): void {
-    if (ids.length === 0) {
-      return
+    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 = datetime('now') WHERE id IN (${placeholders})`)
+        .run(...batch)
     }
-    const placeholders = ids.map(() => '?').join(',')
-    this.db
-      .prepare(`UPDATE messages SET delivered_at = datetime('now') WHERE id IN (${placeholders})`)
-      .run(...ids)
   }
 
   markAsReadAndDelivered(ids: string[]): void {
@@ -4982,13 +5114,24 @@ export class OrchestrationDb {
   }
 
   findActiveRemoteAttachmentForPane(paneKey: string): RemoteDispatchAttachmentRow | undefined {
+    const exact = this.db
+      .prepare(
+        `SELECT * FROM remote_dispatch_attachments
+         WHERE state IN ('starting', 'ready') AND pane_key = ?
+         ORDER BY rowid DESC LIMIT 1`
+      )
+      .get(paneKey) as RemoteDispatchAttachmentRow | undefined
+    if (exact) {
+      return exact
+    }
     const rows = this.db
       .prepare(
         `SELECT * FROM remote_dispatch_attachments
          WHERE state IN ('starting', 'ready') AND pane_key IS NOT NULL
+           AND ${REMOTE_ATTACHMENT_PANE_KEY_MATCH_SUFFIX_SQL} = ?
          ORDER BY rowid DESC`
       )
-      .all() as RemoteDispatchAttachmentRow[]
+      .all(paneKeyMatchSuffix(paneKey)) as RemoteDispatchAttachmentRow[]
     return rows.find((row) => row.pane_key && isEquivalentPaneKey(row.pane_key, paneKey))
   }
 
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 f64e960615b..ceb00f869f9 100644
--- a/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts
+++ b/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts
@@ -489,18 +489,22 @@ describe('push-on-idle orchestration delivery absence gate', () => {
       await vi.advanceTimersByTimeAsync(0)
       expect(pointerWrites()).toHaveLength(1)
 
-      // Enter settles the flight and re-runs the parked trigger, arming a fresh
-      // probe for the newer sequence.
+      // The submit-time liveness probe settles the flight, then the parked
+      // trigger arms its own delivery probe for the newer sequence.
       await vi.advanceTimersByTimeAsync(500)
       resolveProbe(null)
       await vi.advanceTimersByTimeAsync(0)
+      resolveProbe(null)
+      await vi.advanceTimersByTimeAsync(0)
 
       expect(pointerWrites()).toHaveLength(2)
-      expect(pointerWrites()[1]?.[1]).toContain('You have 2 orchestration messages')
+      expect(pointerWrites()[1]?.[1]).toContain('You have 1 orchestration message')
 
       await vi.advanceTimersByTimeAsync(500)
-      expect(stub.markAsDelivered).not.toHaveBeenCalled()
-      expect(stub.rows.every((row) => row.delivered_at === null)).toBe(true)
+      resolveProbe(null)
+      await vi.advanceTimersByTimeAsync(0)
+      expect(stub.markAsDelivered).toHaveBeenCalledTimes(2)
+      expect(stub.rows.every((row) => row.delivered_at !== null)).toBe(true)
     } finally {
       vi.useRealTimers()
     }
@@ -533,11 +537,11 @@ describe('push-on-idle orchestration delivery absence gate', () => {
       // while the newer sequence authorizes exactly one fresh pointer.
       await vi.advanceTimersByTimeAsync(500)
       expect(pointerWrites()).toHaveLength(2)
-      expect(pointerWrites()[1]?.[1]).toContain('You have 2 orchestration messages')
+      expect(pointerWrites()[1]?.[1]).toContain('You have 1 orchestration message')
 
       await vi.advanceTimersByTimeAsync(500)
-      expect(stub.markAsDelivered).not.toHaveBeenCalled()
-      expect(stub.rows.every((row) => row.delivered_at === null)).toBe(true)
+      expect(stub.markAsDelivered).toHaveBeenCalledTimes(2)
+      expect(stub.rows.every((row) => row.delivered_at !== null)).toBe(true)
     } finally {
       vi.useRealTimers()
     }
@@ -582,8 +586,8 @@ 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).not.toHaveBeenCalled()
-      expect(stub.rows[0].delivered_at).toBeNull()
+      expect(stub.markAsDelivered).toHaveBeenCalledOnce()
+      expect(stub.rows[0].delivered_at).not.toBeNull()
     } finally {
       vi.useRealTimers()
     }