fix(orchestration): deliver dispatch mail and replies to the current lead (#23325)

Deliver Dispatch mail and ordinary replies to the current lead Run, preserving original delivery receipts and pre-bind mail draining.

Based on Seongho Bae proposals #22977 and #22979.

Co-authored-by: Seongho Bae <seonghobae@me.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

src/main/runtime/rpc/methods/orchestration/messaging/check-methods.ts

@@ -89,7 +89,8 @@ export const ORCHESTRATION_CHECK_METHODS = [
           typeFilter,
           signal,
           activeDispatch,
-          remoteAttachment
+          remoteAttachment,
+          recordMutationReceipt
         })
       }
       const consumingCheck = params.peek !== true && params.all !== true && params.unread !== false

src/main/runtime/rpc/methods/orchestration/messaging/check-run-pending-mail.ts

@@ -0,0 +1,140 @@
+import type { MessageType, OrchestrationDb, RunRow } from '../../../../orchestration/db'
+import type { OrcaRuntimeService } from '../../../../orca-runtime'
+import { interruptedAcknowledgedCheck } from '../routing'
+import { checkWorkerMailbox } from './check-worker'
+import { currentDispatchAssigneeRun } from './recipient-routing'
+import { callerHoldsDispatchPane, dispatchFenced } from './dispatch-mailbox-fence'
+import { orchestrationCallerIdentity } from '../runs/run-scope'
+import type { OrchestrationSessionCaller } from '../../../../orchestration/orchestration-caller-identity'
+import type { CheckParams } from '../schemas'
+import type { z } from 'zod'
+
+export async function checkRunPendingMail(args: {
+  params: z.infer<typeof CheckParams>
+  runtime: OrcaRuntimeService
+  db: OrchestrationDb
+  run: RunRow
+  handle: string
+  paneKey: string | undefined
+  callerSession: OrchestrationSessionCaller | undefined
+  typeFilter: MessageType[] | undefined
+  signal: AbortSignal | undefined
+  revalidateConsumer: () => void
+  recordMutationReceipt: ((receipt: unknown) => void) | undefined
+}): Promise<{ acknowledged?: string; result?: unknown }> {
+  const {
+    params,
+    runtime,
+    db,
+    run,
+    handle,
+    paneKey,
+    typeFilter,
+    signal,
+    revalidateConsumer,
+    recordMutationReceipt
+  } = args
+  const generation = run.consumer_generation
+  const address = `run:${run.id}`
+  // Drain pre-bind mail through its original Dispatch owner, never by changing its Run.
+  const residual = !params.run ? db.getActiveDispatchForIdentity(handle, paneKey) : undefined
+  const caller = orchestrationCallerIdentity(runtime, {
+    handle,
+    paneKey,
+    session: args.callerSession
+  })
+  const residualAck =
+    params.ack &&
+    residual &&
+    db.getDeliveryRaw(params.ack)?.mailbox_handle === `dispatch:${residual.id}`
+      ? params.ack
+      : undefined
+  const acknowledgeRun = () =>
+    params.ack
+      ? db.acknowledgeRunDelivery({
+          runId: run.id,
+          consumerGeneration: generation,
+          deliveryId: params.ack
+        })
+      : undefined
+  let acknowledged: { delivery: { id: string } } | undefined = residualAck
+    ? undefined
+    : acknowledgeRun()
+  const recordAcknowledged = () => {
+    if (acknowledged) {
+      recordMutationReceipt?.(
+        interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'outcome_unknown')
+      )
+    }
+  }
+  recordAcknowledged()
+  if (
+    residual &&
+    residual.run_id !== run.id &&
+    (residual.assignee_orca_session_id === null ||
+      residual.assignee_orca_session_id === caller.orcaSessionId) &&
+    callerHoldsDispatchPane(residual, paneKey) &&
+    currentDispatchAssigneeRun(runtime, db, residual)?.id === run.id &&
+    (residualAck || !db.hasOutstandingMailboxDelivery(address))
+  ) {
+    const result = await checkWorkerMailbox({
+      params: { ...params, ack: residualAck, wait: false },
+      runtime,
+      db,
+      handle,
+      paneKey,
+      typeFilter,
+      signal,
+      activeDispatch: residual,
+      remoteAttachment: undefined,
+      wakeTypes: params.wait ? typeFilter : undefined,
+      // Accept the original owner's ack without creating a batch ahead of Run replay.
+      deferDelivery: () => db.hasOutstandingMailboxDelivery(address),
+      revalidateConsumer: () => {
+        revalidateConsumer()
+        const current = db.getActiveDispatchForIdentity(handle, paneKey)
+        if (
+          !current ||
+          current.id !== residual.id ||
+          currentDispatchAssigneeRun(runtime, db, current)?.id !== run.id
+        ) {
+          throw dispatchFenced()
+        }
+      },
+      recordMutationReceipt
+    })
+    if (result?.acknowledged) {
+      acknowledged = { delivery: { id: result.acknowledged } }
+    }
+    recordAcknowledged()
+    try {
+      revalidateConsumer()
+    } catch (error) {
+      if (acknowledged) {
+        return {
+          acknowledged: acknowledged.delivery.id,
+          result: interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'consumer_fenced')
+        }
+      }
+      throw error
+    }
+    const inspectingHistory =
+      params.all === true || (params.unread === false && params.peek !== true)
+    if (
+      result &&
+      result.count > 0 &&
+      (!inspectingHistory || db.getUnreadMessages(`dispatch:${residual.id}`).length > 0)
+    ) {
+      return {
+        acknowledged: acknowledged?.delivery.id,
+        result: { ...result, acknowledged: acknowledged?.delivery.id ?? null }
+      }
+    }
+  }
+
+  // A supplied Delivery outside this caller's current Dispatch must still fail acknowledgment.
+  if (params.ack && !acknowledged) {
+    acknowledgeRun()
+  }
+  return { acknowledged: acknowledged?.delivery.id }
+}

src/main/runtime/rpc/methods/orchestration/messaging/check-run.ts

@@ -10,6 +10,7 @@ import { resolveRunScope } from '../runs/run-scope'
 import type { CheckParams } from '../schemas'
 import type { z } from 'zod'
 import type { OrchestrationSessionCaller } from '../../../../orchestration/orchestration-caller-identity'
+import { checkRunPendingMail } from './check-run-pending-mail'
 
 type CheckParamsInput = z.infer<typeof CheckParams>
 
@@ -38,8 +39,7 @@ export async function checkRunMailbox(args: {
     signal,
     legacyCoordinatorRunId,
     revalidateLegacyCoordinator,
-    orchestrationCompatibilityEvidence,
-    recordMutationReceipt
+    orchestrationCompatibilityEvidence
   } = args
   const routeDirectSnapshot = async (
     runId: string,
@@ -72,41 +72,45 @@ export async function checkRunMailbox(args: {
       db.routeUnreadDirectMessagesToRunMailbox(run.id, coordinatorHandle, throughSequence)
     )
   }
-  revalidateLegacyCoordinator?.()
-  const currentRun = resolveRunScope(runtime, {
-    runId: run.id,
-    callerTerminalHandle: handle,
-    callerPaneKey: paneKey,
-    callerSession,
-    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 revalidateConsumer = (): void => {
+    revalidateLegacyCoordinator?.()
+    const currentRun = resolveRunScope(runtime, {
+      runId: run.id,
+      callerTerminalHandle: handle,
+      callerPaneKey: paneKey,
+      callerSession,
+      requireCurrentConsumer: true,
+      legacyCoordinatorRunId,
+      callerEvidence: orchestrationCompatibilityEvidence
+    })
+    if (currentRun.consumer_generation !== generation) {
+      throw new OrchestrationError(
+        'consumer_fenced',
+        'This mailbox consumer was replaced while routing pending mail.'
+      )
+    }
   }
+  revalidateConsumer()
 
-  const acknowledged = params.ack
-    ? db.acknowledgeRunDelivery({
-        runId: run.id,
-        consumerGeneration: generation,
-        deliveryId: params.ack
-      })
-    : undefined
-  if (acknowledged) {
-    recordMutationReceipt?.(
-      interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'outcome_unknown')
-    )
+  const pending = await checkRunPendingMail({ ...args, run, revalidateConsumer })
+  try {
+    revalidateConsumer()
+  } catch (error) {
+    if (pending.acknowledged) {
+      return interruptedAcknowledgedCheck(run.id, pending.acknowledged, 'consumer_fenced')
+    }
+    throw error
   }
+  if (pending.result) {
+    return pending.result
+  }
+  const acknowledged = pending.acknowledged
   if (params.all || (params.unread === false && !params.peek)) {
     const messages = db.getRunMailboxHistory(run.id, 100, typeFilter)
     const result = {
       messages: exposeMessages(messages),
       count: messages.length,
-      acknowledged: acknowledged?.delivery.id ?? null
+      acknowledged: acknowledged ?? null
     }
     if (params.format || params.inject) {
       return {
@@ -122,7 +126,7 @@ export async function checkRunMailbox(args: {
     runId: run.id,
     messages: exposeMessages(messages),
     count: messages.length,
-    acknowledged: acknowledged?.delivery.id ?? null,
+    acknowledged: acknowledged ?? null,
     ...(params.format || params.inject
       ? { formatted: messages.map(formatMessageBanner).join('\n\n') }
       : {})
@@ -142,7 +146,7 @@ export async function checkRunMailbox(args: {
       messages: exposeMessages(current.messages),
       count: current.messages.length,
       replayed: current.replayed,
-      acknowledged: acknowledged?.delivery.id ?? null,
+      acknowledged: acknowledged ?? null,
       timedOut: false,
       cancelled: false,
       connectionLost: false,
@@ -160,7 +164,7 @@ export async function checkRunMailbox(args: {
       deliveryId: null,
       messages: [],
       count: 0,
-      acknowledged: acknowledged?.delivery.id ?? null,
+      acknowledged: acknowledged ?? null,
       timedOut: false,
       cancelled: false,
       connectionLost: false
@@ -174,17 +178,17 @@ export async function checkRunMailbox(args: {
     exclusive: true
   })
   try {
-    revalidateLegacyCoordinator?.()
+    revalidateConsumer()
   } catch (error) {
     if (!acknowledged) {
       throw error
     }
-    return interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'consumer_fenced')
+    return interruptedAcknowledgedCheck(run.id, acknowledged, 'consumer_fenced')
   }
   const latestRun = db.getRun(run.id)
   if (!latestRun || latestRun.consumer_generation !== generation) {
     if (acknowledged) {
-      return interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'consumer_fenced')
+      return interruptedAcknowledgedCheck(run.id, acknowledged, 'consumer_fenced')
     }
     throw new OrchestrationError(
       'consumer_fenced',
@@ -193,7 +197,7 @@ export async function checkRunMailbox(args: {
   }
   if (waitResult === 'waiter_exists') {
     if (acknowledged) {
-      return interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'waiter_exists')
+      return interruptedAcknowledgedCheck(run.id, acknowledged, 'waiter_exists')
     }
     throw new OrchestrationError(
       'waiter_exists',
@@ -209,7 +213,7 @@ export async function checkRunMailbox(args: {
       deliveryId: null,
       messages: [],
       count: 0,
-      acknowledged: acknowledged?.delivery.id ?? null,
+      acknowledged: acknowledged ?? null,
       timedOut: true,
       cancelled: false,
       connectionLost: false
@@ -229,7 +233,7 @@ export async function checkRunMailbox(args: {
       deliveryId: null,
       messages: [],
       count: 0,
-      acknowledged: acknowledged?.delivery.id ?? null,
+      acknowledged: acknowledged ?? null,
       timedOut: false,
       cancelled: true,
       connectionLost: signal?.aborted === true
@@ -246,7 +250,7 @@ export async function checkRunMailbox(args: {
     messages: exposeMessages(current?.messages ?? []),
     count: current?.messages.length ?? 0,
     replayed: current?.replayed ?? false,
-    acknowledged: acknowledged?.delivery.id ?? null,
+    acknowledged: acknowledged ?? null,
     timedOut: false,
     cancelled: false,
     connectionLost: false,

src/main/runtime/rpc/methods/orchestration/messaging/check-worker.ts

@@ -5,6 +5,8 @@ import { formatMessageBanner } from '../../../../orchestration/formatter'
 import { exposeMessages } from './mailbox-message-receipt'
 import { routeAllMailboxPages } from '../schemas'
 import { asDispatchFence, callerHoldsDispatchPane, dispatchFenced } from './dispatch-mailbox-fence'
+import { interruptedAcknowledgedCheck } from '../routing'
+import { currentDispatchAssigneeRun } from './recipient-routing'
 import type { CheckParams } from '../schemas'
 import type { z } from 'zod'
 
@@ -24,7 +26,11 @@ export async function checkWorkerMailbox(args: {
   signal: AbortSignal | undefined
   activeDispatch: ActiveDispatch | undefined
   remoteAttachment: RemoteAttachment | undefined
-}): Promise<unknown> {
+  wakeTypes?: MessageType[]
+  revalidateConsumer?: () => void
+  deferDelivery?: () => boolean
+  recordMutationReceipt?: (receipt: unknown) => void
+}) {
   const {
     params,
     runtime,
@@ -54,6 +60,7 @@ export async function checkWorkerMailbox(args: {
   }
   const deliveryRunId = workerMailbox.runId
   db.requireRun(deliveryRunId)
+  const mailboxIdentity = { runId: deliveryRunId, dispatchId: workerMailbox.dispatchId }
   const address = `dispatch:${workerMailbox.dispatchId}`
   // Why: a federated worker host has no dispatch_contexts row, so its generation lives on the
   // remote_dispatch_attachments row instead.
@@ -165,6 +172,7 @@ export async function checkWorkerMailbox(args: {
     }
   }
   await revalidateWorkerMailbox()
+  args.revalidateConsumer?.()
   let acknowledged
   try {
     acknowledged = params.ack
@@ -179,9 +187,17 @@ export async function checkWorkerMailbox(args: {
   } catch (error) {
     throw asDispatchFence(error)
   }
+  if (acknowledged) {
+    args.recordMutationReceipt?.(
+      interruptedAcknowledgedCheck(deliveryRunId, acknowledged.delivery.id, 'outcome_unknown')
+    )
+  }
   const showAll = params.all === true || (params.unread === false && params.peek !== true)
   const readPeek = () => db.getUnreadMessages(address, typeFilter)
   const readDelivery = (wakeTypes?: MessageType[]) => {
+    if (args.deferDelivery?.()) {
+      return undefined
+    }
     try {
       return db.getOrCreateMailboxDelivery({
         runId: deliveryRunId,
@@ -197,8 +213,7 @@ export async function checkWorkerMailbox(args: {
   if (showAll) {
     const messages = db.getAllMessagesForHandle(address, 100, typeFilter)
     return {
-      ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}),
-      dispatchId: workerMailbox.dispatchId,
+      ...mailboxIdentity,
       messages: exposeMessages(messages),
       count: messages.length,
       acknowledged: acknowledged?.delivery.id ?? null,
@@ -211,8 +226,7 @@ export async function checkWorkerMailbox(args: {
     const messages = readPeek()
     if (messages.length > 0 || !params.wait) {
       return {
-        ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}),
-        dispatchId: workerMailbox.dispatchId,
+        ...mailboxIdentity,
         messages: exposeMessages(messages),
         count: messages.length,
         acknowledged: acknowledged?.delivery.id ?? null,
@@ -222,11 +236,10 @@ export async function checkWorkerMailbox(args: {
       }
     }
   } else {
-    const current = readDelivery(params.wait ? typeFilter : undefined)
+    const current = readDelivery(params.wait ? typeFilter : args.wakeTypes)
     if (current || !params.wait) {
       return {
-        ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}),
-        dispatchId: workerMailbox.dispatchId,
+        ...mailboxIdentity,
         deliveryId: current?.delivery.id ?? null,
         messages: exposeMessages(current?.messages ?? []),
         count: current?.messages.length ?? 0,
@@ -241,19 +254,22 @@ export async function checkWorkerMailbox(args: {
       }
     }
   }
-  const waitResult = await runtime.waitForMessage(address, {
-    typeFilter: typeFilter as string[] | undefined,
-    timeoutMs: params.timeoutMs ?? undefined,
-    signal
-  })
+  // Binding can happen during recovery, before run-create/run-use can cancel this wait.
+  const waitResult =
+    activeDispatch && currentDispatchAssigneeRun(runtime, db, activeDispatch)
+      ? 'cancelled'
+      : await runtime.waitForMessage(address, {
+          typeFilter: typeFilter as string[] | undefined,
+          timeoutMs: params.timeoutMs ?? undefined,
+          signal
+        })
   await revalidateWorkerMailbox()
   if (readCurrentGeneration() !== workerMailbox.generation) {
     throw dispatchFenced()
   }
   if (waitResult === 'timed_out' || waitResult === 'cancelled') {
     return {
-      ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}),
-      dispatchId: workerMailbox.dispatchId,
+      ...mailboxIdentity,
       messages: [],
       count: 0,
       acknowledged: acknowledged?.delivery.id ?? null,
@@ -265,8 +281,7 @@ export async function checkWorkerMailbox(args: {
   if (params.peek) {
     const arrived = readPeek()
     return {
-      ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}),
-      dispatchId: workerMailbox.dispatchId,
+      ...mailboxIdentity,
       messages: exposeMessages(arrived),
       count: arrived.length,
       acknowledged: acknowledged?.delivery.id ?? null,
@@ -277,8 +292,7 @@ export async function checkWorkerMailbox(args: {
   }
   const arrived = readDelivery(typeFilter)
   return {
-    ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}),
-    dispatchId: workerMailbox.dispatchId,
+    ...mailboxIdentity,
     deliveryId: arrived?.delivery.id ?? null,
     messages: exposeMessages(arrived?.messages ?? []),
     count: arrived?.messages.length ?? 0,

src/main/runtime/rpc/methods/orchestration/messaging/dispatch-recipient-identity.test.ts

@@ -0,0 +1,137 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { testOrcaSessionId } from '../../../../../../shared/orca-session-address-test-fixture'
+import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture'
+import { createOrchestrationRpcHarness } from '../rpc-test-harness'
+
+const PANE = 'tab_lead:22222222-2222-4222-9222-222222222222'
+const OTHER = 'tab_other:33333333-3333-4333-8333-333333333333'
+const SESSION = testOrcaSessionId('4bd46b4a-035b-41dd-a122-a9c29122ff11')
+
+describe.each([false, true])('Dispatch recipient identity (settled=%s)', (settled) => {
+  const h = createOrchestrationRpcHarness()
+  let state: ReturnType<typeof h.setup>
+  let dispatch: ReturnType<typeof createRootDispatch>
+
+  beforeEach(() => {
+    state = h.setup()
+    const task = state.db.createTask({ spec: 'lead' })
+    dispatch = createRootDispatch(state.db, task.id, 'term_lead', PANE)
+    if (settled) {
+      state.db.completeDispatch(dispatch.id)
+    }
+    vi.spyOn(state.runtime, 'getLiveTerminalPaneKey').mockImplementation((handle) =>
+      handle === 'term_lead' ? PANE : h.coordinatorPaneKey
+    )
+  })
+  afterEach(() => h.cleanup())
+
+  function send() {
+    return h.call(
+      'orchestration.send',
+      { from: 'term_coord', to: `dispatch:${dispatch.id}`, subject: 'follow up' },
+      state.ctx
+    )
+  }
+  function sessionRun() {
+    state.db.db
+      .prepare('UPDATE dispatch_contexts SET assignee_orca_session_id = ? WHERE id = ?')
+      .run(SESSION, dispatch.id)
+    return state.db.createRun({
+      objective: 'session lead',
+      coordinatorHandle: 'term_lead',
+      coordinatorPaneKey: null,
+      coordinatorOrcaSessionId: SESSION
+    })
+  }
+  async function expectRun(runId: string) {
+    if (settled) {
+      await expect(send()).rejects.toMatchObject({
+        code: 'dispatch_inactive',
+        message: expect.stringContaining(`Send to run:${runId} instead`)
+      })
+      expect(state.db.getInbox()).toEqual([])
+    } else {
+      expect(await send()).toMatchObject({ message: { to_handle: `run:${runId}`, run_id: runId } })
+    }
+  }
+  async function expectNoRedirect(unrelatedRun: string) {
+    if (settled) {
+      await expect(send()).rejects.toMatchObject({
+        code: 'dispatch_inactive',
+        message: expect.not.stringContaining(unrelatedRun)
+      })
+      expect(state.db.getInbox()).toEqual([])
+    } else {
+      expect(await send()).toMatchObject({
+        message: { to_handle: `dispatch:${dispatch.id}`, run_id: dispatch.run_id }
+      })
+    }
+  }
+
+  it('uses a durable session binding without a live pane', async () => {
+    const run = sessionRun()
+    vi.mocked(state.runtime.getLiveTerminalPaneKey).mockReturnValue(null)
+    await expectRun(run.id)
+  })
+
+  it('uses the recorded session instead of an unrelated Run now occupying the saved pane', async () => {
+    const run = sessionRun()
+    state.db.createRun({
+      objective: 'new occupant',
+      coordinatorHandle: 'term_other',
+      coordinatorPaneKey: PANE
+    })
+    await expectRun(run.id)
+  })
+
+  it('ignores an old session column left behind by an older binary rebind', async () => {
+    const run = sessionRun()
+    state.db.db
+      .prepare(
+        'UPDATE runs SET coordinator_handle = ?, coordinator_pane_key = ?, consumer_generation = consumer_generation + 1 WHERE id = ?'
+      )
+      .run('term_other', OTHER, run.id)
+    await expectNoRedirect(run.id)
+  })
+
+  it('does not follow a closed handle to another occupant of its old pane', async () => {
+    const run = state.db.createRun({
+      objective: 'new occupant',
+      coordinatorHandle: 'term_other',
+      coordinatorPaneKey: PANE
+    })
+    vi.mocked(state.runtime.getLiveTerminalPaneKey).mockReturnValue(null)
+    await expectNoRedirect(run.id)
+  })
+
+  it.each(['replacement:pty:2', null])(
+    'does not redirect with a replaced or unverifiable process: %s',
+    async (process) => {
+      const run = state.db.createRun({
+        objective: 'pane run',
+        coordinatorHandle: 'term_lead',
+        coordinatorPaneKey: PANE
+      })
+      state.db.db
+        .prepare('UPDATE dispatch_contexts SET process_incarnation = ? WHERE id = ?')
+        .run('original:pty:1', dispatch.id)
+      vi.mocked(state.runtime.getTerminalProcessIncarnation).mockReturnValue(process)
+      await expectNoRedirect(run.id)
+      expect(state.db.getDispatchContextById(dispatch.id)?.status).toBe(
+        settled ? 'completed' : dispatch.status
+      )
+    }
+  )
+
+  it('accepts a reminted tab half with the same pane leaf and process', async () => {
+    const run = state.db.createRun({
+      objective: 'same pane',
+      coordinatorHandle: 'term_lead',
+      coordinatorPaneKey: PANE
+    })
+    vi.mocked(state.runtime.getLiveTerminalPaneKey).mockReturnValue(
+      PANE.replace('tab_lead', 'tab_restored')
+    )
+    await expectRun(run.id)
+  })
+})

src/main/runtime/rpc/methods/orchestration/messaging/message-methods.ts

@@ -12,6 +12,7 @@ import {
 import { exposeMessage } from './mailbox-message-receipt'
 import { resolveOrchestrationParty } from '../../../../orchestration/orchestration-party'
 import { recordReceiptBeforeNudge, replayMutationNudge } from './mutation-replay-nudge'
+import { resolveReplyRecipient } from './recipient-routing'
 import {
   ReplyParams,
   InboxParams,
@@ -113,15 +114,20 @@ export const ORCHESTRATION_MESSAGE_METHODS = [
         )
       }
 
+      const recipient = resolveReplyRecipient({
+        runtime,
+        db,
+        originalFrom: original.from_handle,
+        originalRunId: original.run_id
+      })
       db.markAsRead([original.id])
-
       const reply = db.insertMessage({
         from: params.from ?? original.to_handle,
-        to: original.from_handle,
+        to: recipient.to,
         subject: `Re: ${original.subject}`,
         body: params.body,
         threadId: original.thread_id ?? original.id,
-        runId: original.run_id
+        runId: recipient.runId
       })
 
       const receipt = { message: exposeMessage(reply) }

src/main/runtime/rpc/methods/orchestration/messaging/recipient-routing.ts

@@ -5,6 +5,8 @@ import type { OrcaRuntimeService } from '../../../../orca-runtime'
 import { readStructuredAgentSessionRecord } from '../../../../structured-worker-authority'
 import { structuredWorkerHostScope } from '../../../../structured-worker-identity'
 import { resolveOrchestrationParty } from '../../../../orchestration/orchestration-party'
+import { isEquivalentPaneKey } from '../../../../orchestration/db/pane-key-match'
+import { CURRENT_CONTRACT_VERSION } from '../../../../orchestration/db/contract-constants'
 
 const ACTIVE_DISPATCH_STATUSES: readonly DispatchStatus[] = ['pending', 'dispatched']
 
@@ -15,20 +17,123 @@ const ACTIVE_DISPATCH_STATUSES: readonly DispatchStatus[] = ['pending', 'dispatc
  * mailbox, so accepting the message reports success for a delivery that cannot
  * happen. Federated targets keep their own liveness check.
  */
-export function assertDispatchMailboxDeliverable(db: OrchestrationDb, dispatchId: string): void {
+export function assertDispatchMailboxDeliverable(
+  runtime: OrcaRuntimeService,
+  db: OrchestrationDb,
+  dispatchId: string
+): void {
   const dispatch = db.getDispatchContextById(dispatchId)
   if (!dispatch || ACTIVE_DISPATCH_STATUSES.includes(dispatch.status)) {
     return
   }
+  const recipientRun = currentDispatchAssigneeRun(runtime, db, dispatch)?.id ?? dispatch.run_id
   throw new OrchestrationError(
     'dispatch_inactive',
-    `Dispatch ${dispatchId} is ${dispatch.status}; its worker will never read that mailbox. Send to run:${dispatch.run_id} instead, or start a new Dispatch for follow-up work.`
+    `Dispatch ${dispatchId} is ${dispatch.status}; its worker will never read that mailbox. Send to run:${recipientRun} instead, or start a new Dispatch for follow-up work.`
   )
 }
 
+// A saved pane alone cannot identify its occupant after reuse.
+export function currentDispatchAssigneeRun(
+  runtime: OrcaRuntimeService,
+  db: OrchestrationDb,
+  dispatch: DispatchContextRow
+) {
+  if (
+    dispatch.contract_version !== CURRENT_CONTRACT_VERSION ||
+    db.getFederatedDispatch(dispatch.id)
+  ) {
+    return undefined
+  }
+  if (dispatch.assignee_orca_session_id !== null) {
+    return db.getCurrentRunForCoordinator({
+      terminalHandle: dispatch.assignee_handle,
+      paneKey: null,
+      orcaSessionId: dispatch.assignee_orca_session_id
+    })
+  }
+  if (dispatch.assignee_handle === null) {
+    return undefined
+  }
+  const paneKey = runtime.getLiveTerminalPaneKey(dispatch.assignee_handle)
+  if (
+    !paneKey ||
+    (dispatch.assignee_pane_key && !isEquivalentPaneKey(dispatch.assignee_pane_key, paneKey)) ||
+    (dispatch.process_incarnation !== null &&
+      runtime.getTerminalProcessIncarnation(dispatch.assignee_handle) !==
+        dispatch.process_incarnation)
+  ) {
+    return undefined
+  }
+  return db.getCurrentRunForPane(paneKey)
+}
+
+// Nested coordinators receive new mail where their current Run check waits.
+export function resolveRunBoundDispatchRecipient(
+  runtime: OrcaRuntimeService,
+  db: OrchestrationDb,
+  dispatchId: string,
+  explicitRunId?: string
+): { to: string; runId: string; warning: SendRecipientWarning } | undefined {
+  const dispatch = db.getDispatchContextById(dispatchId)
+  if (!dispatch || !ACTIVE_DISPATCH_STATUSES.includes(dispatch.status)) {
+    return undefined
+  }
+  const boundRun = currentDispatchAssigneeRun(runtime, db, dispatch)
+  if (!boundRun || boundRun.id === dispatch.run_id) {
+    return undefined
+  }
+  const recipient = `dispatch:${dispatchId}`
+  const mismatch = runMismatch(recipient, boundRun.id, explicitRunId)
+  if (mismatch && !mismatch.ok) {
+    throw new OrchestrationError(mismatch.code, mismatch.message)
+  }
+  return {
+    to: `run:${boundRun.id}`,
+    runId: boundRun.id,
+    warning: {
+      code: 'recipient_run_bound_redirect',
+      recipient,
+      message: `${recipient} is assigned to a terminal that now coordinates Run ${boundRun.id}; queued for run:${boundRun.id}, the mailbox that terminal reads.`
+    }
+  }
+}
+
+// Replies share send routing; unresolved historical senders keep their original address.
+export function resolveReplyRecipient(params: {
+  runtime: OrcaRuntimeService
+  db: OrchestrationDb
+  originalFrom: string
+  originalRunId: string | undefined
+}): { to: string; runId: string | undefined } {
+  const { runtime, db, originalFrom, originalRunId } = params
+  const unchanged = { to: originalFrom, runId: originalRunId }
+  if (originalFrom.startsWith('run:')) {
+    return { to: originalFrom, runId: originalFrom.slice('run:'.length) }
+  }
+  if (originalFrom.startsWith('dispatch:')) {
+    const dispatchId = originalFrom.slice('dispatch:'.length)
+    // Federation owns its own recipient and liveness checks.
+    if (db.getFederatedDispatch(dispatchId)) {
+      return unchanged
+    }
+    assertDispatchMailboxDeliverable(runtime, db, dispatchId)
+    const runBound = resolveRunBoundDispatchRecipient(runtime, db, dispatchId)
+    return runBound ?? unchanged
+  }
+  const recipient = resolveBareOrchestrationRecipient({
+    runtime,
+    db,
+    handle: originalFrom,
+    senderRunId: originalRunId
+  })
+  return recipient.ok ? { to: recipient.to, runId: recipient.runId ?? originalRunId } : unchanged
+}
+
 export type SendRecipientWarning = {
   code:
     | 'legacy_terminal_recipient'
+    | 'recipient_run_bound_redirect'
     | 'recipient_unreachable'
     | 'recipient_ambiguous'
     | 'recipient_run_mismatch'

src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-history.test.ts

@@ -0,0 +1,69 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture'
+import { createOrchestrationRpcHarness } from '../rpc-test-harness'
+
+const LEAD = 'tab_lead:22222222-2222-4222-9222-222222222222'
+function deliveryId(result: unknown): string {
+  if (
+    typeof result === 'object' &&
+    result &&
+    'deliveryId' in result &&
+    typeof result.deliveryId === 'string'
+  ) {
+    return result.deliveryId
+  }
+  throw new Error('Expected a Delivery')
+}
+
+describe('Run history after residual Dispatch acknowledgment', () => {
+  const h = createOrchestrationRpcHarness()
+  let state: ReturnType<typeof h.setup>
+  beforeEach(() => {
+    state = h.setup()
+    vi.mocked(state.runtime.getTerminalPaneKey).mockImplementation((handle) =>
+      handle === 'term_lead' ? LEAD : h.coordinatorPaneKey
+    )
+  })
+  afterEach(() => h.cleanup())
+  it.each([
+    { mode: { all: true }, acknowledgeTogether: false },
+    { mode: { all: true }, acknowledgeTogether: true },
+    { mode: { unread: false }, acknowledgeTogether: false },
+    { mode: { unread: false }, acknowledgeTogether: true }
+  ])('shows Run history after old mail is read: %j', async ({ mode, acknowledgeTogether }) => {
+    const task = state.db.createTask({ spec: 'nested lead' })
+    const dispatch = createRootDispatch(state.db, task.id, 'term_lead', LEAD)
+    const run = state.db.createRun({
+      objective: 'child Run',
+      coordinatorHandle: 'term_lead',
+      coordinatorPaneKey: LEAD
+    })
+    const old = state.db.insertMessage({
+      from: 'term_coord',
+      to: `dispatch:${dispatch.id}`,
+      runId: dispatch.run_id,
+      subject: 'old mail'
+    })
+    const check = (params = {}) =>
+      h.call('orchestration.check', { terminal: 'term_lead', ...params }, state.ctx)
+    expect(await check(mode)).toMatchObject({ runId: dispatch.run_id, messages: [{ id: old.id }] })
+    expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false)
+    const delivery = deliveryId(await check())
+    if (!acknowledgeTogether) {
+      await check({ ack: delivery })
+    }
+    const fresh = state.db.insertMessage({
+      from: 'term_child',
+      to: `run:${run.id}`,
+      runId: run.id,
+      subject: 'new mail'
+    })
+    expect(
+      await check({ ...mode, ...(acknowledgeTogether ? { ack: delivery } : {}) })
+    ).toMatchObject({ runId: run.id, messages: [{ id: fresh.id }] })
+    expect(state.db.getMessageById(old.id)?.read).toBe(1)
+    expect(state.db.getMessageById(fresh.id)?.read).toBe(0)
+    expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false)
+    expect(state.db.hasOutstandingMailboxDelivery(`run:${run.id}`)).toBe(false)
+  })
+})

src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-interleavings.test.ts

@@ -0,0 +1,175 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture'
+import { ORCHESTRATION_DELIVERY_BATCH_LIMIT } from '../../../../orchestration/db/messages/mailbox-routing-page'
+import { createOrchestrationRpcHarness } from '../rpc-test-harness'
+
+const LEAD = 'tab_lead:22222222-2222-4222-9222-222222222222'
+const OTHER = 'tab_other:33333333-3333-4333-8333-333333333333'
+
+function deliveryId(result: unknown): string {
+  if (
+    typeof result === 'object' &&
+    result !== null &&
+    'deliveryId' in result &&
+    typeof result.deliveryId === 'string'
+  ) {
+    return result.deliveryId
+  }
+  throw new Error('Expected a Delivery')
+}
+
+describe('Run binding during Dispatch checks', () => {
+  const h = createOrchestrationRpcHarness()
+  let state: ReturnType<typeof h.setup>
+  let dispatch: ReturnType<typeof createRootDispatch>
+
+  beforeEach(() => {
+    state = h.setup()
+    vi.mocked(state.runtime.getTerminalPaneKey).mockImplementation((handle) =>
+      handle === 'term_lead' ? LEAD : handle === 'term_coord' ? h.coordinatorPaneKey : OTHER
+    )
+    dispatch = createRootDispatch(
+      state.db,
+      state.db.createTask({ spec: 'lead' }).id,
+      'term_lead',
+      LEAD
+    )
+  })
+  afterEach(() => {
+    state.runtime.cancelMessageWaiters(`dispatch:${dispatch.id}`)
+    h.cleanup()
+    vi.restoreAllMocks()
+  })
+
+  function check(params: Record<string, unknown> = {}) {
+    return h.call('orchestration.check', { terminal: 'term_lead', ...params }, state.ctx)
+  }
+  function residual(subject: string) {
+    return state.db.insertMessage({
+      from: 'term_coord',
+      to: `dispatch:${dispatch.id}`,
+      runId: dispatch.run_id,
+      subject
+    })
+  }
+  async function bind(method: 'runCreate' | 'runUse') {
+    const params =
+      method === 'runCreate'
+        ? { objective: 'child' }
+        : {
+            id: state.db.createRun({
+              objective: 'adopt',
+              coordinatorHandle: 'term_other',
+              coordinatorPaneKey: OTHER
+            }).id
+          }
+    await h.call(`orchestration.${method}`, { from: 'term_lead', ...params }, state.ctx)
+    const run = state.db.getCurrentRunForPane(LEAD)
+    if (!run) {
+      throw new Error('Expected bound Run')
+    }
+    return run
+  }
+
+  describe.each(['runCreate', 'runUse'] as const)('%s', (method) => {
+    it.each(['parked', 'recovering'] as const)(
+      'cancels a %s Dispatch wait and preserves its ack',
+      async (phase) => {
+        const old = residual('before bind')
+        const ack = deliveryId(await check())
+        const record = vi.fn()
+        state.ctx.recordMutationReceipt = record
+        const wait = vi.spyOn(state.runtime, 'waitForMessage')
+        const waiting = check({ ack, wait: true, timeoutMs: 500 })
+        if (phase === 'parked') {
+          await vi.waitFor(() =>
+            expect(wait).toHaveBeenCalledWith(`dispatch:${dispatch.id}`, expect.anything())
+          )
+        }
+        const run = await bind(method)
+        await h.call(
+          'orchestration.send',
+          {
+            from: 'term_coord',
+            to: `dispatch:${dispatch.id}`,
+            subject: 'after bind'
+          },
+          state.ctx
+        )
+        expect(await waiting).toMatchObject({ acknowledged: ack, cancelled: true, timedOut: false })
+        expect(record).toHaveBeenCalledWith(expect.objectContaining({ acknowledged: ack }))
+        expect(state.db.getMessageById(old.id)).toMatchObject({ run_id: dispatch.run_id, read: 1 })
+        expect(await check()).toMatchObject({
+          runId: run.id,
+          messages: [{ subject: 'after bind' }]
+        })
+        if (phase === 'recovering') {
+          expect(wait).not.toHaveBeenCalled()
+        }
+      }
+    )
+  })
+
+  it.each(['before', 'during recovery'] as const)(
+    'replays Run delivery created %s residual acknowledgment',
+    async (phase) => {
+      const ids = Array.from(
+        { length: ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1 },
+        (_, i) => residual(`old ${i}`).id
+      )
+      const last = ids.at(-1)
+      if (!last) {
+        throw new Error('Expected residual tail')
+      }
+      const ack = deliveryId(await check())
+      const run = await bind('runCreate')
+      state.db.insertMessage({
+        from: 'term_child',
+        to: `run:${run.id}`,
+        runId: run.id,
+        subject: 'Run batch'
+      })
+      let runDelivery: string | undefined
+      const readRun = () => {
+        const result = state.db.getOrCreateRunDelivery({
+          runId: run.id,
+          consumerGeneration: run.consumer_generation
+        })
+        if (!result) {
+          throw new Error('Expected Run delivery')
+        }
+        runDelivery = result.delivery.id
+      }
+      if (phase === 'before') {
+        runDelivery = deliveryId(await check({ run: run.id }))
+      } else {
+        state.db.db.prepare('UPDATE messages SET to_handle = ? WHERE id = ?').run('term_lead', last)
+        const route = state.db.routeUnreadDirectMessagesToDispatchMailbox.bind(state.db)
+        vi.spyOn(state.db, 'routeUnreadDirectMessagesToDispatchMailbox').mockImplementationOnce(
+          (...args) => {
+            const result = route(...args)
+            readRun()
+            return result
+          }
+        )
+      }
+      const result = await check({ ack })
+      expect(result).toMatchObject({
+        runId: run.id,
+        deliveryId: runDelivery,
+        acknowledged: ack,
+        replayed: true
+      })
+      expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false)
+      for (const id of ids.slice(0, -1)) {
+        expect(state.db.getMessageById(id)).toMatchObject({ read: 1, run_id: dispatch.run_id })
+      }
+      expect(state.db.getMessageById(last)).toMatchObject({ read: 0, run_id: dispatch.run_id })
+      expect(await check({ ack })).toMatchObject({ deliveryId: runDelivery, replayed: true })
+      const tail = await check({ ack: runDelivery })
+      expect(tail).toMatchObject({ runId: dispatch.run_id, messages: [{ id: last }], count: 1 })
+      expect(await check({ ack: deliveryId(tail) })).toMatchObject({ count: 0 })
+      expect(state.db.getMessageById(last)?.read).toBe(1)
+    }
+  )
+})

src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-safety.test.ts

@@ -0,0 +1,290 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture'
+import { createOrchestrationRpcHarness } from '../rpc-test-harness'
+
+const LEAD = 'tab_lead:22222222-2222-4222-9222-222222222222'
+const OTHER = 'tab_other:33333333-3333-4333-8333-333333333333'
+
+function deliveryId(result: unknown): string {
+  if (
+    typeof result === 'object' &&
+    result &&
+    'deliveryId' in result &&
+    typeof result.deliveryId === 'string'
+  ) {
+    return result.deliveryId
+  }
+  throw new Error('Expected a Delivery')
+}
+
+describe('Run-bound lead mailbox boundaries', () => {
+  const h = createOrchestrationRpcHarness()
+  let state: ReturnType<typeof h.setup>
+  let dispatch: ReturnType<typeof createRootDispatch>
+  let leadRun: ReturnType<typeof state.db.createRun>
+
+  beforeEach(() => {
+    state = h.setup()
+    vi.mocked(state.runtime.getTerminalPaneKey).mockImplementation((handle) =>
+      handle === 'term_lead' ? LEAD : handle === 'term_coord' ? h.coordinatorPaneKey : OTHER
+    )
+    const task = state.db.createTask({ spec: 'nested lead' })
+    dispatch = createRootDispatch(state.db, task.id, 'term_lead', LEAD)
+    leadRun = state.db.createRun({
+      objective: 'child Run',
+      coordinatorHandle: 'term_lead',
+      coordinatorPaneKey: LEAD
+    })
+  })
+  afterEach(() => h.cleanup())
+
+  function check(params: Record<string, unknown> = {}) {
+    return h.call('orchestration.check', { terminal: 'term_lead', ...params }, state.ctx)
+  }
+  function residual(subject: string, type: 'status' | 'question' = 'status') {
+    return state.db.insertMessage({
+      from: 'term_coord',
+      to: `dispatch:${dispatch.id}`,
+      runId: dispatch.run_id,
+      subject,
+      type
+    })
+  }
+  function runMail(subject = 'Run report') {
+    return state.db.insertMessage({
+      from: 'term_child',
+      to: `run:${leadRun.id}`,
+      runId: leadRun.id,
+      subject,
+      type: 'question'
+    })
+  }
+  function handoff() {
+    state.db.bindRun({
+      runId: leadRun.id,
+      coordinatorHandle: 'term_other',
+      coordinatorPaneKey: OTHER
+    })
+  }
+
+  it('recovers old raw-handle replies only from the active Dispatch Run', async () => {
+    const old = residual('old reply')
+    state.db.db.prepare('UPDATE messages SET to_handle = ? WHERE id = ?').run('term_lead', old.id)
+    const foreign = state.db.createRun({
+      objective: 'unrelated',
+      coordinatorHandle: 'term_other',
+      coordinatorPaneKey: OTHER
+    })
+    const privateMail = state.db.insertMessage({
+      from: 'term_other',
+      to: 'term_lead',
+      runId: foreign.id,
+      subject: 'foreign'
+    })
+    runMail()
+    const first = await check()
+    expect(first).toMatchObject({ runId: dispatch.run_id, messages: [{ id: old.id }] })
+    expect(await check({ ack: deliveryId(first) })).toMatchObject({
+      runId: leadRun.id,
+      messages: [{ subject: 'Run report' }]
+    })
+    expect(state.db.getMessageById(privateMail.id)).toMatchObject({ read: 0, run_id: foreign.id })
+  })
+
+  it('treats types as a wake condition and replays the entire residual FIFO batch', async () => {
+    residual('older status')
+    residual('decision needed', 'question')
+    runMail()
+    const first = await check({ wait: true, types: 'question' })
+    expect(first).toMatchObject({
+      messages: [{ subject: 'older status' }, { subject: 'decision needed' }]
+    })
+    expect(await check({ wait: true, types: 'worker_done' })).toMatchObject({
+      deliveryId: deliveryId(first),
+      replayed: true
+    })
+    expect(await check({ ack: deliveryId(first) })).toMatchObject({
+      runId: leadRun.id,
+      acknowledged: deliveryId(first),
+      messages: [{ subject: 'Run report' }]
+    })
+  })
+
+  it('does not filter a non-waiting consuming residual check', async () => {
+    residual('status')
+    expect(await check({ types: 'question' })).toMatchObject({ messages: [{ subject: 'status' }] })
+  })
+
+  it.each([{ peek: true }, { all: true }])(
+    'filters residual inspection without consuming it: %j',
+    async (mode) => {
+      const status = residual('status')
+      residual('question', 'question')
+      expect(await check({ ...mode, types: 'question' })).toMatchObject({
+        messages: [{ subject: 'question' }]
+      })
+      expect(state.db.getMessageById(status.id)?.read).toBe(0)
+      expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false)
+    }
+  )
+
+  it('replays an outstanding Run batch before exposing older nonmatching residual mail', async () => {
+    residual('status')
+    runMail()
+    const first = await check({ wait: true, types: 'question' })
+    expect(first).toMatchObject({ runId: leadRun.id })
+    expect(await check()).toMatchObject({ deliveryId: deliveryId(first), replayed: true })
+    expect(await check({ ack: deliveryId(first) })).toMatchObject({
+      acknowledged: deliveryId(first),
+      messages: [{ subject: 'status' }]
+    })
+  })
+
+  it('keeps explicit Run checks scoped and refuses the parent Run', async () => {
+    const pending = residual('parent mail')
+    runMail()
+    expect(await check({ run: leadRun.id })).toMatchObject({
+      runId: leadRun.id,
+      messages: [{ subject: 'Run report' }]
+    })
+    await expect(check({ run: dispatch.run_id })).rejects.toMatchObject({ code: 'consumer_fenced' })
+    expect(state.db.getMessageById(pending.id)?.read).toBe(0)
+  })
+
+  it('fences a Run handoff during residual direct-mail recovery before delivery or ack', async () => {
+    const pending = residual('raw before bind')
+    state.db.db
+      .prepare('UPDATE messages SET to_handle = ? WHERE id = ?')
+      .run('term_lead', pending.id)
+    const route = state.db.routeUnreadDirectMessagesToDispatchMailbox.bind(state.db)
+    vi.spyOn(state.db, 'routeUnreadDirectMessagesToDispatchMailbox').mockImplementationOnce(
+      (...args) => {
+        const result = route(...args)
+        handoff()
+        return result
+      }
+    )
+    await expect(check()).rejects.toMatchObject({ code: 'consumer_fenced' })
+    expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false)
+    expect(state.db.getMessageById(pending.id)?.read).toBe(0)
+  })
+
+  it('records a residual acknowledgment before a Run handoff interrupts the following wait', async () => {
+    residual('ack me')
+    const first = await check()
+    const record = vi.fn()
+    state.ctx.recordMutationReceipt = record
+    vi.spyOn(state.runtime, 'waitForMessage').mockImplementationOnce(async () => {
+      expect(record).toHaveBeenCalledWith(
+        expect.objectContaining({ acknowledged: deliveryId(first) })
+      )
+      handoff()
+      return 'cancelled'
+    })
+    expect(await check({ ack: deliveryId(first), wait: true })).toMatchObject({
+      acknowledged: deliveryId(first),
+      waitInterrupted: 'consumer_fenced',
+      messages: []
+    })
+    expect(state.db.getUnreadMessages(`dispatch:${dispatch.id}`)).toEqual([])
+  })
+
+  it('retains the acknowledged receipt if the wait transport throws', async () => {
+    residual('ack me')
+    const first = await check()
+    const record = vi.fn()
+    state.ctx.recordMutationReceipt = record
+    vi.spyOn(state.runtime, 'waitForMessage').mockRejectedValueOnce(
+      new Error('connection interrupted')
+    )
+    await expect(check({ ack: deliveryId(first), wait: true })).rejects.toThrow(
+      'connection interrupted'
+    )
+    expect(record).toHaveBeenCalledWith(
+      expect.objectContaining({ acknowledged: deliveryId(first) })
+    )
+    expect(state.db.getUnreadMessages(`dispatch:${dispatch.id}`)).toEqual([])
+  })
+
+  it('does not give a reused process the previous assignee mail', async () => {
+    state.db.mintDispatchCapability({
+      dispatchId: dispatch.id,
+      paneKey: LEAD,
+      processIncarnation: 'old:pty:1'
+    })
+    const pending = residual('old process mail')
+    runMail()
+    expect(await check()).toMatchObject({
+      runId: leadRun.id,
+      messages: [{ subject: 'Run report' }]
+    })
+    expect(state.db.getMessageById(pending.id)?.read).toBe(0)
+    expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false)
+  })
+
+  it('refuses an explicit sender Run that would silently redirect into another Run', async () => {
+    await expect(
+      h.call(
+        'orchestration.send',
+        {
+          from: 'term_coord',
+          to: `dispatch:${dispatch.id}`,
+          run: dispatch.run_id,
+          subject: 'wrong scope'
+        },
+        state.ctx
+      )
+    ).rejects.toMatchObject({ code: 'recipient_run_mismatch' })
+    expect(state.db.getInbox()).toEqual([])
+  })
+
+  it('wakes a parked Run check for Dispatch mail sent after binding', async () => {
+    const waiter = vi.spyOn(state.runtime, 'waitForMessage')
+    const waiting = check({ wait: true, timeoutMs: 1_000 })
+    await vi.waitFor(() =>
+      expect(waiter).toHaveBeenCalledWith(`run:${leadRun.id}`, expect.anything())
+    )
+    await h.call(
+      'orchestration.send',
+      { from: 'term_coord', to: `dispatch:${dispatch.id}`, subject: 'wake up' },
+      state.ctx
+    )
+    expect(await waiting).toMatchObject({ timedOut: false, messages: [{ subject: 'wake up' }] })
+  })
+
+  it('keeps a canonical Run reply in the recipient Run even across an original thread Run', async () => {
+    const note = state.db.insertMessage({
+      from: `run:${leadRun.id}`,
+      to: `run:${dispatch.run_id}`,
+      runId: dispatch.run_id,
+      subject: 'report'
+    })
+    expect(
+      await h.call(
+        'orchestration.reply',
+        { from: 'term_coord', id: note.id, body: 'decision' },
+        state.ctx
+      )
+    ).toMatchObject({ message: { to_handle: `run:${leadRun.id}`, run_id: leadRun.id } })
+    expect(await check()).toMatchObject({ messages: [{ subject: 'Re: report' }] })
+  })
+
+  it('refuses replies to an inactive canonical Dispatch before reading or inserting mail', async () => {
+    const note = state.db.insertMessage({
+      from: `dispatch:${dispatch.id}`,
+      to: `run:${dispatch.run_id}`,
+      runId: dispatch.run_id,
+      subject: 'old report'
+    })
+    state.db.completeDispatch(dispatch.id)
+    await expect(
+      h.call(
+        'orchestration.reply',
+        { from: 'term_coord', id: note.id, body: 'too late' },
+        state.ctx
+      )
+    ).rejects.toMatchObject({ code: 'dispatch_inactive' })
+    expect(state.db.getMessageById(note.id)?.read).toBe(0)
+    expect(state.db.getInbox()).toHaveLength(1)
+  })
+})

src/main/runtime/rpc/methods/orchestration/messaging/run-bound-recipient-mail.test.ts

@@ -0,0 +1,189 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import type { RpcContext } from '../../../core'
+import type { OrchestrationDb, RunRow } from '../../../../orchestration/db'
+import type { OrcaRuntimeService } from '../../../../orca-runtime'
+import type { DispatchContextRow } from '../../../../orchestration/types'
+import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture'
+import { createOrchestrationRpcHarness } from '../rpc-test-harness'
+
+// A lead is dispatched by a root coordinator and then coordinates its own Run from the same pane.
+// That pane's `check` reads its own Run mailbox, so mail meant for it must land there.
+describe('mail for a lead whose pane coordinates its own Run', () => {
+  const h = createOrchestrationRpcHarness()
+  const coordPane = 'tab_coord:11111111-1111-4111-8111-111111111111'
+  const leadPane = 'tab_lead:22222222-2222-4222-9222-222222222222'
+  let db: OrchestrationDb
+  let runtime: OrcaRuntimeService
+  let ctx: RpcContext
+  let rootRun: RunRow
+  let dispatch: DispatchContextRow
+
+  afterEach(() => {
+    h.cleanup()
+  })
+
+  function setup(): void {
+    ;({ db, runtime, ctx } = h.setup(false))
+    vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) =>
+      handle === 'term_coord' ? coordPane : handle === 'term_lead' ? leadPane : null
+    )
+    rootRun = db.createRun({
+      objective: 'root',
+      coordinatorHandle: 'term_coord',
+      coordinatorPaneKey: coordPane
+    })
+    const task = db.createTask({ spec: 'lead the sub-project', runId: rootRun.id })
+    dispatch = createRootDispatch(db, task.id, 'term_lead', leadPane)
+  }
+
+  function bindLeadRun(): RunRow {
+    return db.createRun({
+      objective: 'lead',
+      coordinatorHandle: 'term_lead',
+      coordinatorPaneKey: leadPane
+    })
+  }
+
+  async function call(name: string, params: Record<string, unknown>) {
+    return h.call(name, params, ctx)
+  }
+
+  async function leadInbox(params: Record<string, unknown> = {}): Promise<unknown> {
+    return call('orchestration.check', { terminal: 'term_lead', ...params })
+  }
+
+  function deliveryIdOf(result: unknown): string {
+    if (
+      typeof result === 'object' &&
+      result !== null &&
+      'deliveryId' in result &&
+      typeof result.deliveryId === 'string'
+    ) {
+      return result.deliveryId
+    }
+    throw new Error('check returned no delivery')
+  }
+
+  it('routes dispatch:<id> mail to the Run the assignee pane now coordinates', async () => {
+    setup()
+    const leadRun = bindLeadRun()
+
+    const result = await call('orchestration.send', {
+      from: 'term_coord',
+      to: `dispatch:${dispatch.id}`,
+      subject: 'Follow-up for the lead'
+    })
+
+    expect(result).toMatchObject({
+      message: { to_handle: `run:${leadRun.id}`, run_id: leadRun.id },
+      warnings: [{ code: 'recipient_run_bound_redirect' }]
+    })
+    expect(await leadInbox()).toMatchObject({ messages: [{ subject: 'Follow-up for the lead' }] })
+  })
+
+  it('still reads dispatch mail that arrived before the pane bound its own Run', async () => {
+    setup()
+    db.insertMessage({
+      from: 'term_coord',
+      to: `dispatch:${dispatch.id}`,
+      subject: 'Sent before the lead bound a Run',
+      runId: rootRun.id
+    })
+    const leadRun = bindLeadRun()
+    db.insertMessage({
+      from: 'term_worker',
+      to: `run:${leadRun.id}`,
+      subject: 'Sub-worker report',
+      runId: leadRun.id
+    })
+
+    const first = await leadInbox()
+    expect(first).toMatchObject({ messages: [{ subject: 'Sent before the lead bound a Run' }] })
+
+    const second = await leadInbox({ ack: deliveryIdOf(first) })
+    expect(second).toMatchObject({ messages: [{ subject: 'Sub-worker report' }] })
+  })
+
+  it('keeps the --types wake condition when older Dispatch mail does not match it', async () => {
+    setup()
+    db.insertMessage({
+      from: 'term_coord',
+      to: `dispatch:${dispatch.id}`,
+      subject: 'Older status note',
+      type: 'status',
+      runId: rootRun.id
+    })
+    const leadRun = bindLeadRun()
+    db.insertMessage({
+      from: 'term_worker',
+      to: `run:${leadRun.id}`,
+      subject: 'Sub-worker finished',
+      type: 'worker_done',
+      runId: leadRun.id
+    })
+
+    const woke = await leadInbox({ wait: true, types: 'worker_done', timeoutMs: 500 })
+
+    expect(woke).toMatchObject({
+      runId: leadRun.id,
+      messages: [{ subject: 'Sub-worker finished' }]
+    })
+  })
+
+  it('delivers a reply to a Run-bound sender and wakes its waiting check', async () => {
+    setup()
+    const leadRun = bindLeadRun()
+    const report = db.insertMessage({
+      from: 'term_lead',
+      to: `run:${rootRun.id}`,
+      subject: 'Lead report',
+      runId: rootRun.id
+    })
+
+    const waiting = leadInbox({ wait: true, timeoutMs: 2_000 })
+    const reply = await call('orchestration.reply', {
+      id: report.id,
+      from: 'term_coord',
+      body: 'Decision'
+    })
+
+    expect(reply).toMatchObject({
+      message: { to_handle: `run:${leadRun.id}`, run_id: leadRun.id }
+    })
+    expect(await waiting).toMatchObject({
+      timedOut: false,
+      messages: [{ subject: 'Re: Lead report' }]
+    })
+  })
+
+  it('keeps dispatch:<id> mail on the Dispatch mailbox while the assignee has no Run', async () => {
+    setup()
+
+    const result = await call('orchestration.send', {
+      from: 'term_coord',
+      to: `dispatch:${dispatch.id}`,
+      subject: 'Plain worker follow-up'
+    })
+
+    expect(result).toMatchObject({ message: { to_handle: `dispatch:${dispatch.id}` } })
+    expect(result).not.toHaveProperty('warnings')
+  })
+
+  it('keeps a reply on the raw handle when the sender has no Run or live pane', async () => {
+    setup()
+    const note = db.insertMessage({
+      from: 'term_offline',
+      to: `run:${rootRun.id}`,
+      subject: 'Offline note',
+      runId: rootRun.id
+    })
+
+    const reply = await call('orchestration.reply', {
+      id: note.id,
+      from: 'term_coord',
+      body: 'Ack'
+    })
+
+    expect(reply).toMatchObject({ message: { to_handle: 'term_offline', run_id: rootRun.id } })
+  })
+})

src/main/runtime/rpc/methods/orchestration/messaging/send-methods.ts

@@ -7,6 +7,7 @@ import { resolveMessageRun } from '../routing'
 import {
   assertDispatchMailboxDeliverable,
   resolveBareOrchestrationRecipient,
+  resolveRunBoundDispatchRecipient,
   type SendRecipientWarning
 } from './recipient-routing'
 import {
@@ -162,7 +163,18 @@ export const ORCHESTRATION_SEND_METHODS = [
             : undefined
         // Federated targets perform their own liveness check before relaying.
         if (addressedDispatchId && !federatedTarget) {
-          assertDispatchMailboxDeliverable(db, addressedDispatchId)
+          assertDispatchMailboxDeliverable(runtime, db, addressedDispatchId)
+          const runBound = resolveRunBoundDispatchRecipient(
+            runtime,
+            db,
+            addressedDispatchId,
+            params.run
+          )
+          if (runBound) {
+            to = runBound.to
+            messageRunId = runBound.runId
+            sendWarnings.push(runBound.warning)
+          }
         }
         const federatedControl = sendFederatedControlMail({
           params,

src/main/runtime/rpc/methods/orchestration/messaging/settled-dispatch-mail.test.ts

@@ -1,4 +1,4 @@
-import { afterEach, describe, expect, it } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import type { RpcContext } from '../../../core'
 import type { OrchestrationDb } from '../../../../orchestration/db'
 import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture'
@@ -58,6 +58,38 @@ describe('orchestration.send to a settled Dispatch mailbox', () => {
     ).rejects.toThrow(new RegExp(`run:${dispatch.run_id}`))
   })
 
+  it('names the Run a settled assignee now coordinates, not the sender Run', async () => {
+    setup()
+    const leadPane = 'tab_lead:22222222-2222-4222-9222-222222222222'
+    vi.spyOn(ctx.runtime, 'getLiveTerminalPaneKey').mockReturnValue(leadPane)
+    const task = db.createTask({ spec: 'lead that settled and kept coordinating' })
+    const dispatch = createRootDispatch(db, task.id, 'term_lead', leadPane)
+    db.completeDispatch(dispatch.id)
+    const leadRun = db.createRun({
+      objective: 'lead',
+      coordinatorHandle: 'term_lead',
+      coordinatorPaneKey: leadPane
+    })
+
+    const rejection = call('orchestration.send', {
+      from: 'term_coord',
+      to: `dispatch:${dispatch.id}`,
+      subject: 'One more thing'
+    })
+
+    await expect(rejection).rejects.toMatchObject({ code: 'dispatch_inactive' })
+    await expect(rejection).rejects.toThrow(new RegExp(`run:${leadRun.id}`))
+    await expect(rejection).rejects.not.toThrow(new RegExp(`run:${dispatch.run_id}`))
+    expect(db.getInbox()).toEqual([])
+    await expect(
+      call('orchestration.check', { terminal: 'term_coord', run: leadRun.id })
+    ).rejects.toMatchObject({ code: 'consumer_fenced' })
+    // Run addresses are already visible to callers; a hint grants no consuming authority.
+    expect(await call('orchestration.runShow', { id: leadRun.id })).toMatchObject({
+      run: { id: leadRun.id }
+    })
+  })
+
   it('does not write an undeliverable message row', async () => {
     setup()
     const task = db.createTask({ spec: 'worker that already reported' })

src/main/runtime/rpc/methods/orchestration/runs/runs.ts

@@ -2,6 +2,9 @@ import { defineMethod } from '../../../core'
 import { OrchestrationError } from '../../../../orchestration/orchestration-error'
 import { assertCallerHandleMatchesEvidence, resolveOrchestrationCaller } from './run-scope'
 import { exposeRun } from './run-receipt'
+import type { OrcaRuntimeService } from '../../../../orca-runtime'
+import type { OrchestrationCallerIdentity } from '../../../../orchestration/orchestration-caller-identity'
+import { currentDispatchAssigneeRun } from '../messaging/recipient-routing'
 import {
   RunCreateParams,
   RunCurrentParams,
@@ -10,6 +13,18 @@ import {
   RunUseParams
 } from '../../../../../../shared/rpc-contract/orchestration-runs-params'
 
+function cancelBoundDispatchWaiters(
+  runtime: OrcaRuntimeService,
+  caller: OrchestrationCallerIdentity,
+  runId: string
+): void {
+  const db = runtime.getOrchestrationDb()
+  const dispatch = db.getActiveDispatchForIdentity(caller.address, caller.paneKey ?? undefined)
+  if (dispatch && currentDispatchAssigneeRun(runtime, db, dispatch)?.id === runId) {
+    runtime.cancelMessageWaiters(`dispatch:${dispatch.id}`)
+  }
+}
+
 export const ORCHESTRATION_RUN_METHODS = [
   defineMethod({
     name: 'orchestration.runCreate',
@@ -30,6 +45,7 @@ export const ORCHESTRATION_RUN_METHODS = [
         coordinatorOrcaSessionId: caller.orcaSessionId
       })
       runtime.cancelMessageWaiters(params.from)
+      cancelBoundDispatchWaiters(runtime, caller, run.id)
       if (priorRun) {
         runtime.cancelMessageWaiters(`run:${priorRun.id}`)
       }
@@ -86,6 +102,7 @@ export const ORCHESTRATION_RUN_METHODS = [
         )
       }
       runtime.cancelMessageWaiters(params.from)
+      cancelBoundDispatchWaiters(runtime, caller, run.id)
       runtime.cancelMessageWaiters(`run:${params.id}`)
       if (priorRun && priorRun.id !== params.id) {
         runtime.cancelMessageWaiters(`run:${priorRun.id}`)

src/main/runtime/rpc/orchestration-session-coordinator.test.ts

@@ -91,13 +91,30 @@ describe('a structured chat coordinates through the same verbs as a terminal', (
       to: WORKER_HANDLE,
       subject: 'more'
     })
-    expect(outbound).toMatchObject({ from_handle: ADDRESS_X, run_id: runId })
+    expect(outbound).toMatchObject({
+      from_handle: ADDRESS_X,
+      to_handle: `dispatch:${idOf(dispatch)}`,
+      run_id: runId
+    })
 
     const replied = await as(SESSION_X, 'orchestration.reply', {
       id: idOf(inbound),
       body: 'ack'
     })
-    expect(replied).toMatchObject({ message: { from_handle: ADDRESS_X, to_handle: WORKER_HANDLE } })
+    expect(replied).toMatchObject({
+      message: {
+        from_handle: ADDRESS_X,
+        to_handle: `dispatch:${idOf(dispatch)}`,
+        run_id: runId,
+        thread_id: idOf(inbound)
+      }
+    })
+    expect(await as(undefined, 'orchestration.check', { terminal: WORKER_HANDLE })).toMatchObject({
+      runId,
+      dispatchId: idOf(dispatch),
+      count: 2,
+      messages: [{ id: idOf(outbound) }, { id: idOf(replied.message) }]
+    })
 
     const { gate } = await as(SESSION_X, 'orchestration.gateCreate', {
       task: taskId,
@@ -255,16 +272,55 @@ describe('a structured chat coordinates through the same verbs as a terminal', (
       )
     )
     await vi.waitFor(() => expect(waiter).toHaveBeenCalledWith(`run:${runId}`, expect.anything()))
+    const deliver = vi.spyOn(h.db, 'getOrCreateRunDelivery')
+    const acknowledge = vi.spyOn(h.db, 'acknowledgeRunDelivery')
 
     await as(SESSION_Y, 'orchestration.runUse', { id: runId })
 
-    expect(await waiting).toMatchObject({
-      ok: false,
-      error: {
-        code: 'consumer_fenced',
-        message: 'This mailbox consumer was replaced while waiting.'
-      }
+    const fenced = await waiting
+    expect(fenced).toMatchObject({ ok: false, error: { code: 'consumer_fenced' } })
+    expect(fenced).not.toHaveProperty('result')
+
+    const { message } = await as(undefined, 'orchestration.send', {
+      from: WORKER_HANDLE,
+      to: ADDRESS_Y,
+      subject: 'after takeover',
+      run: runId
     })
+    expect(
+      await h.dispatch(
+        orchestrationRequest('orchestration.check', { run: runId }, { sessionId: SESSION_X })
+      )
+    ).toMatchObject({ ok: false, error: { code: 'consumer_fenced' } })
+    expect(deliver).not.toHaveBeenCalled()
+
+    const replacement = await as(SESSION_Y, 'orchestration.check', {})
+    expect(replacement).toMatchObject({
+      runId,
+      count: 1,
+      messages: [{ id: idOf(message), to_handle: `run:${runId}`, run_id: runId }]
+    })
+    const ack = replacement.deliveryId
+    if (typeof ack !== 'string') {
+      throw new Error('Expected a Run Delivery')
+    }
+    expect(
+      await h.dispatch(
+        orchestrationRequest('orchestration.check', { run: runId, ack }, { sessionId: SESSION_X })
+      )
+    ).toMatchObject({ ok: false, error: { code: 'consumer_fenced' } })
+    expect(acknowledge).not.toHaveBeenCalled()
+    expect(h.db.getDeliveryRaw(ack)?.acknowledged_at).toBeNull()
+    expect(await as(SESSION_Y, 'orchestration.check', {})).toMatchObject({
+      deliveryId: ack,
+      replayed: true,
+      messages: [{ id: idOf(message) }]
+    })
+    expect(await as(SESSION_Y, 'orchestration.check', { ack })).toMatchObject({
+      acknowledged: ack,
+      count: 0
+    })
+    expect(h.db.getDeliveryRaw(ack)?.status).toBe('acknowledged')
   })
 
   it('stops counting a coordinator Orca session id once an older binary rebinds the Run to a terminal', async () => {

src/main/runtime/rpc/orchestration-session-run-binding.test.ts

@@ -0,0 +1,138 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  mintStructuredWorkerHandle,
+  mintStructuredWorkerPaneKey,
+  structuredWorkerIdentities,
+  structuredWorkerProcessIncarnation
+} from '../structured-worker-identity'
+import {
+  createSessionCallerHarness,
+  orchestrationRequest,
+  resultOf,
+  idOf,
+  SESSION_X,
+  SESSION_Y,
+  type SessionCallerHarness
+} from './orchestration-session-caller-test-fixture'
+
+const hostRef = vi.hoisted((): { current: unknown } => ({ current: null }))
+vi.mock('../../native-chat/agent-session-wire/structured-agent-session-registry', () => ({
+  getStructuredAgentSessionHost: () => hostRef.current
+}))
+
+function deliveryId(result: Record<string, unknown>): string {
+  if (typeof result.deliveryId === 'string') {
+    return result.deliveryId
+  }
+  throw new Error('Expected a Delivery')
+}
+
+describe('structured lead Run binding through the RPC dispatcher', () => {
+  let h: SessionCallerHarness
+  let dispatchId: string
+
+  function call(sessionId: string, method: string, params: Record<string, unknown> = {}) {
+    return h
+      .dispatch(orchestrationRequest(`orchestration.${method}`, params, { sessionId }))
+      .then(resultOf)
+  }
+
+  beforeEach(async () => {
+    h = createSessionCallerHarness(hostRef)
+    const handle = mintStructuredWorkerHandle()
+    const paneKey = mintStructuredWorkerPaneKey(SESSION_Y)
+    structuredWorkerIdentities.register({
+      handle,
+      sessionId: SESSION_Y,
+      agent: 'claude',
+      paneKey,
+      processIncarnation: structuredWorkerProcessIncarnation(SESSION_Y),
+      worktreeId: 'folder-workspace',
+      hostScope: { kind: 'local', hostId: 'local' }
+    })
+    const runId = idOf((await call(SESSION_X, 'runCreate', { objective: 'parent' })).run)
+    dispatchId = h.db.createDispatchContext({
+      taskId: h.db.createTask({ runId, spec: 'structured lead' }).id,
+      assigneeHandle: handle,
+      assigneePaneKey: paneKey,
+      processIncarnation: structuredWorkerProcessIncarnation(SESSION_Y),
+      creator: { kind: 'session', orcaSessionId: SESSION_X },
+      maxDepth: Number.MAX_SAFE_INTEGER
+    }).id
+  })
+  afterEach(() => {
+    h.runtime.cancelMessageWaiters(`dispatch:${dispatchId}`)
+    h.close()
+    vi.restoreAllMocks()
+  })
+
+  it.each(['runCreate', 'runUse'] as const)(
+    'cancels a parked session Dispatch check on %s',
+    async (method) => {
+      const params =
+        method === 'runCreate'
+          ? { objective: 'child' }
+          : {
+              id: h.db.createRun({
+                objective: 'unbound',
+                coordinatorHandle: null,
+                coordinatorPaneKey: null
+              }).id
+            }
+      const wait = vi.spyOn(h.runtime, 'waitForMessage')
+      const waiting = call(SESSION_Y, 'check', { wait: true, timeoutMs: 500 })
+      await vi.waitFor(() =>
+        expect(wait).toHaveBeenCalledWith(`dispatch:${dispatchId}`, expect.anything())
+      )
+      const child = idOf((await call(SESSION_Y, method, params)).run)
+      expect(await waiting).toMatchObject({ cancelled: true, timedOut: false })
+      const sent = await call(SESSION_X, 'send', {
+        to: `dispatch:${dispatchId}`,
+        subject: 'after bind'
+      })
+      expect(sent.message).toMatchObject({ to_handle: `run:${child}`, run_id: child })
+      expect(await call(SESSION_Y, 'check')).toMatchObject({
+        messages: [{ subject: 'after bind' }]
+      })
+      expect(h.db.getInbox()).toHaveLength(1)
+    }
+  )
+
+  it.each(['cancel', 'throw'] as const)(
+    'replays a pre-bind ack receipt after the following wait ends by %s',
+    async (ending) => {
+      await call(SESSION_X, 'send', { to: `dispatch:${dispatchId}`, subject: 'ack me' })
+      const ack = deliveryId(await call(SESSION_Y, 'check'))
+      const acknowledge = vi.spyOn(h.db, 'acknowledgeMailboxDelivery')
+      const wait = vi.spyOn(h.runtime, 'waitForMessage').mockImplementationOnce(async () => {
+        await call(SESSION_Y, 'runCreate', { objective: 'child' })
+        if (ending === 'throw') {
+          throw new Error('wait transport interrupted')
+        }
+        return 'cancelled'
+      })
+      const request = orchestrationRequest(
+        'orchestration.check',
+        { ack, wait: true, timeoutMs: 500 },
+        {
+          sessionId: SESSION_Y,
+          requestId: `bind-during-ack-${ending}`
+        }
+      )
+      const first = await h.dispatch(request)
+      if (ending === 'throw') {
+        expect(first).toMatchObject({ ok: false, error: { code: 'runtime_error' } })
+      } else {
+        expect(resultOf(first)).toMatchObject({ acknowledged: ack, count: 0, cancelled: true })
+      }
+      const replay = resultOf(await h.dispatch(request))
+      expect(replay).toMatchObject({ acknowledged: ack, count: 0, mutation: { replayed: true } })
+      if (ending === 'throw') {
+        expect(replay.waitInterrupted).toBe('outcome_unknown')
+      }
+      expect(acknowledge).toHaveBeenCalledTimes(1)
+      expect(wait).toHaveBeenCalledTimes(1)
+      expect(h.db.getDeliveryRaw(ack)?.status).toBe('acknowledged')
+    }
+  )
+})