fix(orchestration): scope sleeping mail invalidation to affected owners

src/main/runtime/agent-session/records/agent-session-crew-invalidation.ts

@@ -0,0 +1,52 @@
+import { sessionCrewChanges } from '../../orchestration/crew/crew-changes'
+import type { AgentSessionStoreState } from './agent-session-store-state'
+import type { AgentSessionStoreRowWrites } from './agent-session-store-draft'
+
+export function publishAgentSessionCrewChanges(
+  published: AgentSessionStoreState,
+  draft: AgentSessionStoreState,
+  writes: AgentSessionStoreRowWrites
+): void {
+  const topologyChanged =
+    writes.sessionTabs !== null ||
+    writes.records.remove.length > 0 ||
+    writes.records.upsert.some(([id]) => {
+      const before = published.records.get(id)
+      const after = draft.records.get(id)
+      return (
+        !before ||
+        !after ||
+        before.lease.claimStatus !== after.lease.claimStatus ||
+        before.provider !== after.provider ||
+        before.location !== after.location ||
+        before.conversationCommand !== after.conversationCommand
+      )
+    })
+  const renamed = writes.records.upsert
+    .filter(
+      ([id]) =>
+        published.records.get(id)?.conversationName !== draft.records.get(id)?.conversationName
+    )
+    .map(([id]) => id)
+  if (topologyChanged) {
+    const affected = new Set([...writes.records.remove, ...writes.records.upsert.map(([id]) => id)])
+    if (writes.sessionTabs) {
+      for (const id of [
+        ...(published.sessionTabs?.sessionIds() ?? []),
+        ...(draft.sessionTabs?.sessionIds() ?? [])
+      ]) {
+        if (published.sessionTabs?.tabIdFor(id) !== draft.sessionTabs?.tabIdFor(id)) {
+          affected.add(id)
+        }
+      }
+    }
+    sessionCrewChanges.notify(
+      Object.assign(new Set(['records'] as const), {
+        recordIds: [...affected],
+        trafficAddresses: new Set([...affected].map((id) => `orca_session_id:${id}`))
+      })
+    )
+  } else if (renamed.length) {
+    sessionCrewChanges.notify(Object.assign(new Set(['names'] as const), { recordIds: renamed }))
+  }
+}

src/main/runtime/agent-session/records/agent-session-record-store-slot.ts

@@ -61,7 +61,9 @@ export function openAgentSessionRecordStoreOnce(
       (opened) => {
         slot.store = opened.store
         if (opening === slot) {
-          sessionCrewChanges.notify(new Set(['topology', 'records']))
+          sessionCrewChanges.notify(
+            Object.assign(new Set(['topology', 'records'] as const), { mailGlobal: true as const })
+          )
         }
         return opened
       },
@@ -107,7 +109,9 @@ export async function releaseAgentSessionRecordStore(): Promise<OpenedAgentSessi
   const released = opening
   opening = null
   if (released) {
-    sessionCrewChanges.notify(new Set(['topology', 'records']))
+    sessionCrewChanges.notify(
+      Object.assign(new Set(['topology', 'records'] as const), { mailGlobal: true as const })
+    )
   }
   return released ? released.opened.catch(() => null) : null
 }

src/main/runtime/agent-session/records/agent-session-store-transactions.ts

@@ -1,4 +1,4 @@
-import { sessionCrewChanges } from '../../orchestration/crew/crew-changes'
+import { publishAgentSessionCrewChanges } from './agent-session-crew-invalidation'
 import type { JournalHostDatabase } from '../../../native-chat/agent-session-journal/database/journal-host-database'
 import type { JournalOperationReceipt } from '../../../native-chat/agent-session-journal/writing/journal-row-writer'
 import { journalOpenRefusalError } from '../../../native-chat/agent-session-journal/replay/journal-open-failure'
@@ -263,35 +263,7 @@ export class AgentSessionStoreTransactions {
           if (lineageChanged) {
             this.lineage = {}
           }
-          const topologyChanged =
-            writes.sessionTabs !== null ||
-            writes.records.remove.length > 0 ||
-            writes.records.upsert.some(([id]) => {
-              const before = published.records.get(id)
-              const after = draft.records.get(id)
-              return (
-                !before ||
-                !after ||
-                before.lease.claimStatus !== after.lease.claimStatus ||
-                before.provider !== after.provider ||
-                before.location !== after.location ||
-                before.conversationCommand !== after.conversationCommand
-              )
-            })
-          const renamed = writes.records.upsert
-            .filter(
-              ([id]) =>
-                published.records.get(id)?.conversationName !==
-                draft.records.get(id)?.conversationName
-            )
-            .map(([id]) => id)
-          if (topologyChanged) {
-            sessionCrewChanges.notify(new Set(['records']))
-          } else if (renamed.length) {
-            sessionCrewChanges.notify(
-              Object.assign(new Set(['names'] as const), { recordIds: renamed })
-            )
-          }
+          publishAgentSessionCrewChanges(published, draft, writes)
         }
       }
     }

src/main/runtime/orchestration/crew/crew-changes.ts

@@ -1,6 +1,8 @@
 export type CrewFacet = 'topology' | 'records' | 'status' | 'names' | 'traffic'
 export type CrewInvalidation = ReadonlySet<CrewFacet> & {
   readonly recordIds?: readonly string[]
+  readonly mailboxOwners?: ReadonlySet<string>
+  readonly mailGlobal?: true
   readonly trafficAddresses?: ReadonlySet<string>
 }
 

src/main/runtime/orchestration/crew/crew-mail-mutation-scope.ts

@@ -0,0 +1,235 @@
+import { isOrcaSessionId } from '../../../../shared/orca-session-address'
+import type { OrchestrationDb } from '../db/orchestration-db'
+
+const dispatchMutations = new Set([
+  'settleRemoteAttachmentInRelayTransaction',
+  'completeDispatch',
+  'failDispatch',
+  'closeQuestionsForDispatch',
+  'markWorkerDispatchReady',
+  'failWorkerStart',
+  'markWorkerStartUnknown',
+  'abandonWorkerDispatch',
+  'beginWorkerStop',
+  'settleWorkerStop',
+  'markWorkerStopUnknown',
+  'reconcileMissingWorkerTerminal',
+  'failRemoteAttachment',
+  'markRemoteAttachmentReady',
+  'beginRemoteAttachmentStop',
+  'settleRemoteAttachmentStop',
+  'markRemoteAttachmentStopUnknown',
+  'requestRemoteAttachmentTerminalRelease'
+])
+const taskMutations = new Set([
+  'completeActiveDispatchesForTask',
+  'failActiveDispatchForTask',
+  'promoteReadyTasks',
+  'updateTaskStatus'
+])
+
+function fields(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null && !Array.isArray(value)
+}
+
+/** Capture both sides of transfers; nested mutations contribute their own scope before publication. */
+export class CrewMailMutationScope {
+  readonly trafficAddresses = new Set<string>()
+  readonly mailboxOwners = new Set<string>()
+  private readonly visited = new Set<string>()
+
+  get notice() {
+    return { trafficAddresses: this.trafficAddresses, mailboxOwners: this.mailboxOwners }
+  }
+
+  constructor(private readonly db: OrchestrationDb) {}
+
+  run(id: string): void {
+    this.trafficAddresses.add(`run:${id}`)
+    if (this.seen(`run:${id}`)) {
+      return
+    }
+    this.row(this.db.getRun(id))
+  }
+
+  dispatch(id: string): void {
+    this.trafficAddresses.add(`dispatch:${id}`)
+    if (this.seen(`dispatch:${id}`)) {
+      return
+    }
+    this.row(this.db.getDispatchContextById(id) ?? this.db.getRemoteDispatchAttachment(id))
+  }
+
+  private seen(id: string): boolean {
+    if (this.visited.has(id)) {
+      return true
+    }
+    this.visited.add(id)
+    return false
+  }
+
+  private task(id: string): void {
+    if (this.seen(`task:${id}`)) {
+      return
+    }
+    this.row(this.db.getTask(id))
+    for (const row of this.db.db
+      .prepare('SELECT id FROM dispatch_contexts WHERE task_id = ?')
+      .all(id)) {
+      if (typeof row.id === 'string') {
+        this.dispatch(row.id)
+      }
+    }
+  }
+
+  row(value: unknown): void {
+    if (Array.isArray(value)) {
+      for (const row of value) {
+        this.row(row)
+      }
+    } else if (fields(value)) {
+      for (const [key, item] of Object.entries(value)) {
+        if (typeof item === 'string') {
+          if (['runId', 'run_id', 'local_run_id'].includes(key)) {
+            this.run(item)
+          }
+          if (
+            [
+              'dispatchId',
+              'dispatch_id',
+              'owner_dispatch_id',
+              'toDispatchId',
+              'creatorDispatchId'
+            ].includes(key)
+          ) {
+            this.dispatch(item)
+          }
+          if (['taskId', 'task_id'].includes(key)) {
+            this.task(item)
+          }
+          if (
+            [
+              'paneKey',
+              'pane_key',
+              'fromPaneKey',
+              'toPaneKey',
+              'assigneePaneKey',
+              'assignee_pane_key',
+              'coordinatorPaneKey',
+              'coordinator_pane_key',
+              'creator_pane_key'
+            ].includes(key)
+          ) {
+            this.mailboxOwners.add(item)
+          }
+          if (
+            [
+              'mailboxHandle',
+              'from',
+              'to',
+              'from_handle',
+              'to_handle',
+              'assigneeHandle',
+              'assignee_handle',
+              'coordinatorHandle',
+              'coordinator_handle',
+              'terminalHandle',
+              'terminal_handle'
+            ].includes(key)
+          ) {
+            this.trafficAddresses.add(item)
+          }
+          if (
+            [
+              'coordinator_orca_session_id',
+              'assignee_orca_session_id',
+              'creator_orca_session_id'
+            ].includes(key)
+          ) {
+            this.trafficAddresses.add(`orca_session_id:${item}`)
+          }
+          if (['messageId', 'message_id'].includes(key)) {
+            this.row(this.db.getMessageById(item))
+          }
+          if (key === 'resourceId') {
+            this.row(this.db.getWorkerTerminalResource(item))
+          }
+        } else if (item && typeof item === 'object') {
+          this.row(item)
+        }
+      }
+      if (typeof value.id === 'string') {
+        if ('coordinator_pane_key' in value) {
+          this.trafficAddresses.add(`run:${value.id}`)
+        }
+        if ('assignee_pane_key' in value) {
+          this.trafficAddresses.add(`dispatch:${value.id}`)
+        }
+      }
+    }
+  }
+
+  capture(name: string, args: unknown[]): void {
+    const first = args[0]
+    this.row(args)
+    if (typeof first === 'string') {
+      if (dispatchMutations.has(name)) {
+        this.dispatch(first)
+      }
+      if (taskMutations.has(name)) {
+        this.task(first)
+      }
+      if (name === 'resolveGate' || name === 'timeoutGate') {
+        this.row(
+          this.db.db.prepare('SELECT run_id, task_id FROM decision_gates WHERE id = ?').get(first)
+        )
+      }
+      if (name === 'convertLifecycleMessageToRejection') {
+        this.row(this.db.getMessageById(first))
+      }
+      if (name === 'settleWorkerTerminalRelease') {
+        this.row(this.db.getWorkerTerminalResource(first))
+      }
+      if (name === 'routeDirectMessagePage') {
+        this.trafficAddresses.add(first)
+        if (typeof args[1] === 'string') {
+          this.run(args[1])
+        }
+        if (typeof args[2] === 'string') {
+          this.trafficAddresses.add(args[2])
+        }
+      }
+      if (name === 'routeUnreadDispatchMailboxToRunMailbox') {
+        this.dispatch(first)
+        if (typeof args[1] === 'string') {
+          this.run(args[1])
+        }
+      }
+      if (name === 'routeForeignDirectMessagesToOwnedMailboxes') {
+        this.trafficAddresses.add(first)
+      }
+    }
+    if (name === 'unbindOtherRunsForCoordinator' && fields(first)) {
+      const { paneKey, orcaSessionId, terminalHandle } = first
+      if (
+        (paneKey === null || typeof paneKey === 'string') &&
+        (orcaSessionId === null ||
+          (typeof orcaSessionId === 'string' && isOrcaSessionId(orcaSessionId))) &&
+        (terminalHandle === null || typeof terminalHandle === 'string')
+      ) {
+        this.row(this.db.runsBoundToCoordinator({ paneKey, orcaSessionId, terminalHandle }))
+      }
+    }
+    if (name === 'transitionLifecycle' && fields(first) && typeof first.id === 'string') {
+      if (first.entity === 'run') {
+        this.run(first.id)
+      }
+      if (first.entity === 'dispatch' || first.entity === 'worker') {
+        this.dispatch(first.id)
+      }
+      if (first.entity === 'task') {
+        this.task(first.id)
+      }
+    }
+  }
+}

src/main/runtime/orchestration/crew/crew-mutation-notifications.ts

@@ -1,3 +1,4 @@
+import { CrewMailMutationScope } from './crew-mail-mutation-scope'
 import type { CrewInvalidation } from './crew-changes'
 import type { OrchestrationDb } from '../db/orchestration-db'
 
@@ -89,10 +90,13 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP
   if (hasDomainMethod(prototype, 'settleWorkerTerminalRelease')) {
     const release = prototype.settleWorkerTerminalRelease
     prototype.settleWorkerTerminalRelease = function (this: OrchestrationDb, resourceId) {
+      const scope = new CrewMailMutationScope(this)
+      scope.capture('settleWorkerTerminalRelease', [resourceId])
       const before = this.getWorkerTerminalResource(resourceId)?.release_state
       const result = release.call(this, resourceId)
       if (before !== result.release_state) {
-        this.crewChanges.notify(new Set(['topology']))
+        scope.row(result)
+        this.crewChanges.notify(Object.assign(new Set(['topology'] as const), scope.notice))
       }
       return result
     }
@@ -100,10 +104,13 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP
   if (hasDomainMethod(prototype, 'settleDeadWorkerTerminalRelease')) {
     const release = prototype.settleDeadWorkerTerminalRelease
     prototype.settleDeadWorkerTerminalRelease = function (this: OrchestrationDb, params) {
+      const scope = new CrewMailMutationScope(this)
+      scope.row(params)
       const before = this.getWorkerTerminalResource(params.resourceId)?.release_state
       const result = release.call(this, params)
       if (before !== result.resource.release_state) {
-        this.crewChanges.notify(new Set(['topology']))
+        scope.row(result)
+        this.crewChanges.notify(Object.assign(new Set(['topology'] as const), scope.notice))
       }
       return result
     }
@@ -111,9 +118,12 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP
   if (hasDomainMethod(prototype, 'commitLegacyCompatibilityPrincipal')) {
     const commit = prototype.commitLegacyCompatibilityPrincipal
     prototype.commitLegacyCompatibilityPrincipal = function (this: OrchestrationDb, params) {
+      const scope = new CrewMailMutationScope(this)
+      scope.row(params)
       const result = commit.call(this, params)
       if (!result.duplicate) {
-        this.crewChanges.notify(new Set(['topology']))
+        scope.row(result)
+        this.crewChanges.notify(Object.assign(new Set(['topology'] as const), scope.notice))
       }
       return result
     }
@@ -123,7 +133,9 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP
     const known = this.localMutationCallerFingerprint
     const result = sourceIdentity.call(this)
     if (!known) {
-      this.crewChanges.notify()
+      this.crewChanges.notify(
+        Object.assign(new Set(['topology'] as const), { trafficAddresses: new Set<string>() })
+      )
     }
     return result
   }
@@ -133,7 +145,9 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP
     if (result.type !== 'heartbeat') {
       const notice: CrewInvalidation =
         result.delivery_contract === 'legacy_direct'
-          ? new Set(['topology'])
+          ? Object.assign(new Set(['topology'] as const), {
+              trafficAddresses: new Set([result.from_handle, result.to_handle])
+            })
           : Object.assign(new Set(['traffic'] as const), {
               trafficAddresses: new Set([result.from_handle, result.to_handle])
             })
@@ -150,7 +164,18 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP
       configurable: true,
       writable: true,
       value(this: OrchestrationDb, ...args: unknown[]) {
+        const before = new CrewMailMutationScope(this)
+        before.capture(name, args)
         const result = original.apply(this, args)
+        const after = new CrewMailMutationScope(this)
+        after.capture(name, args)
+        after.row(result)
+        for (const address of before.trafficAddresses) {
+          after.trafficAddresses.add(address)
+        }
+        for (const owner of before.mailboxOwners) {
+          after.mailboxOwners.add(owner)
+        }
         if (
           typeof result === 'object' &&
           result !== null &&
@@ -160,7 +185,13 @@ export function attachCrewMutationNotifications(ctor: { prototype: CrewMutationP
         ) {
           return result
         }
-        this.crewChanges.notify(new Set([trafficMutations.has(name) ? 'traffic' : 'topology']))
+        this.crewChanges.notify(
+          Object.assign(
+            new Set([trafficMutations.has(name) ? 'traffic' : 'topology'] as const),
+            after.notice,
+            name.startsWith('reset') ? { mailGlobal: true as const } : {}
+          )
+        )
         return result
       }
     })

src/main/runtime/orchestration/db/legacy/legacy-worker-completion.ts

@@ -90,10 +90,15 @@ export function setLegacyCompatibilityPrincipalStatus(
        WHERE id = ? AND status = 'committed'`
     )
     .run(status, id)
-  if (changed.changes) {
-    this.crewChanges.notify(new Set(['topology']))
+  const principal = this.getLegacyCompatibilityPrincipal(id)
+  if (changed.changes && principal) {
+    this.crewChanges.notify(
+      Object.assign(new Set(['topology'] as const), {
+        trafficAddresses: new Set([`run:${principal.run_id}`])
+      })
+    )
   }
-  return this.getLegacyCompatibilityPrincipal(id)
+  return principal
 }
 
 export function getLegacyOperationReceipt(

src/main/runtime/orchestration/db/messages/message-inbox.ts

@@ -1,3 +1,4 @@
+import { captureMessageMutation } from './message-mutation-invalidation'
 import { undeliveredInboxQuery, type UndeliveredInboxOptions } from './undelivered-inbox-query'
 import type { MessageType, MessageRow } from '../orchestration-records'
 import { exposeMessageTimestamps, exposeMessageListTimestamps } from '../utc-timestamp'
@@ -18,6 +19,7 @@ function runBatchedMessageMutation(
   if (ids.length === 0) {
     return
   }
+  const changed = captureMessageMutation(db, ids)
   db.db.exec(`SAVEPOINT ${MESSAGE_MUTATION_SAVEPOINT}`)
   try {
     for (let offset = 0; offset < ids.length; offset += MESSAGE_ID_UPDATE_BATCH_SIZE) {
@@ -31,6 +33,7 @@ function runBatchedMessageMutation(
     db.db.exec(`RELEASE ${MESSAGE_MUTATION_SAVEPOINT}`)
     throw error
   }
+  changed()
 }
 
 export function getUnreadMessages(

src/main/runtime/orchestration/db/messages/message-mutation-invalidation.ts

@@ -0,0 +1,54 @@
+import type { OrchestrationDb } from '../orchestration-db'
+
+export function captureMessageMutation(db: OrchestrationDb, ids: string[]): () => void {
+  const read = () => {
+    const rows = new Map<string, { address: string; state: string }>()
+    for (let offset = 0; offset < ids.length; offset += 500) {
+      const batch = ids.slice(offset, offset + 500)
+      for (const row of db.db
+        .prepare(`
+        SELECT id, to_handle, read, delivered_at, pointer_enter_pending FROM messages
+        WHERE id IN (${batch.map(() => '?').join(',')})
+      `)
+        .all(...batch)) {
+        if (typeof row.id === 'string' && typeof row.to_handle === 'string') {
+          rows.set(row.id, { address: row.to_handle, state: JSON.stringify(row) })
+        }
+      }
+    }
+    return rows
+  }
+  const before = read()
+  return () => {
+    const publish = () => {
+      const after = read()
+      const addresses = new Set<string>()
+      for (const [id, row] of before) {
+        if (after.get(id)?.state !== row.state) {
+          addresses.add(row.address)
+          const next = after.get(id)
+          if (next) {
+            addresses.add(next.address)
+          }
+        }
+      }
+      if (addresses.size) {
+        db.crewChanges.notify(
+          Object.assign(new Set(['traffic'] as const), { trafficAddresses: addresses })
+        )
+      }
+    }
+    // An outer transaction can still roll back; compare committed rows on the next turn.
+    if (db.db.isTransaction) {
+      queueMicrotask(() => {
+        try {
+          publish()
+        } catch {
+          /* The owning database can close after the transaction commits. */
+        }
+      })
+    } else {
+      publish()
+    }
+  }
+}

src/main/runtime/orchestration/db/runs/run-coordinator-mail-routing.ts

@@ -67,7 +67,11 @@ export function rememberRunCoordinatorHandle(
     )
     .run(runId, terminalHandle)
   if (changed.changes) {
-    this.crewChanges.notify(new Set(['topology']))
+    this.crewChanges.notify(
+      Object.assign(new Set(['topology'] as const), {
+        trafficAddresses: new Set([`run:${runId}`, terminalHandle])
+      })
+    )
   }
 }
 

src/main/runtime/orchestration/mailbox/routing/runtime-mail-wake-admission.ts

@@ -66,7 +66,7 @@ export abstract class RuntimeMailWakeAdmission extends OrcaRuntimeWithPruneMobil
       shellInFront: async (ptyId) =>
         (await this.readLaunchedAgentForeground(ptyId, record.agent)) === 'shell'
     })
-    // Host-side input fencing is required before a proven failed invocation may be stopped.
+    // TEMPORARY: STA-15118 requires host input fencing before automatic surviving-shell replacement.
     return false
   }
 
@@ -77,25 +77,32 @@ export abstract class RuntimeMailWakeAdmission extends OrcaRuntimeWithPruneMobil
       return null
     }
     const attempt = record.mailWake?.attempt
-    const pty = this.getPtyRecordForPaneKey(record.paneKey)
-    const liveShell =
-      !!attempt &&
-      pty?.connected === true &&
-      pty.ptyId === attempt.ptyId &&
-      pty.incarnationId === attempt.incarnationId &&
-      pty.launchToken === attempt.launchToken &&
-      pty.launchIncarnationId === attempt.incarnationId &&
-      pty.worktreeId === record.worktreeId &&
-      pty.connectionId === (record.connectionId ?? null) &&
-      this.mailWakeInvocations.ended(record) &&
-      (await this.readLaunchedAgentForeground(attempt.ptyId, record.agent)) === 'shell' &&
-      this.getPtyRecordForPaneKey(record.paneKey) === pty &&
-      this.sleepingMailRecord(record.paneKey) === record
+    const matchingAttempt = () => {
+      const pty = this.getPtyRecordForPaneKey(record.paneKey)
+      return (
+        !!attempt &&
+        pty?.connected === true &&
+        pty.ptyId === attempt.ptyId &&
+        pty.incarnationId === attempt.incarnationId &&
+        pty.launchToken === attempt.launchToken &&
+        pty.launchIncarnationId === attempt.incarnationId &&
+        pty.worktreeId === record.worktreeId &&
+        pty.connectionId === (record.connectionId ?? null) &&
+        this.terminalRunFacts.readProcessStartedAt(attempt.ptyId, attempt.incarnationId) !== null
+      )
+    }
+    const foreground =
+      attempt && matchingAttempt()
+        ? await this.readLaunchedAgentForeground(attempt.ptyId, record.agent)
+        : 'unknown'
+    const current = matchingAttempt() && this.sleepingMailRecord(record.paneKey) === record
+    const ended = this.mailWakeInvocations.ended(record)
+    const liveShell = current && ended && foreground === 'shell'
     return {
       kind: 'asleep',
       waking:
         mailWakeObligation(record) === 'automatic' && !this.sleptMailboxWake.failedToWake(record),
-      resuming: mailWakeObligation(record) === 'retained' && !liveShell,
+      resuming: current && !ended && foreground === 'agent',
       agentName: TUI_AGENT_DISPLAY_NAMES[record.agent],
       openTab: record.restoreOnTabOpenOnly === true,
       failed: this.sleptMailboxWake.failedToWake(record),

src/main/runtime/orchestration/mailbox/routing/sleeping-mail-owner-mailboxes.ts

@@ -0,0 +1,74 @@
+import type { SleptMailboxWake } from './slept-mailbox-wake'
+import type { OrchestrationDb } from '../../db/orchestration-db'
+import {
+  RUN_PANE_KEY_MATCH_SUFFIX_SQL,
+  DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL,
+  REMOTE_ATTACHMENT_PANE_KEY_MATCH_SUFFIX_SQL,
+  paneKeyMatchSuffix,
+  isEquivalentPaneKey
+} from '../../db/pane-key-match'
+import { potentiallyLiveRemoteAttachmentSql } from '../../db/federation/remote-attachment-liveness'
+import { parseOrcaSessionAddress } from '../../../../../shared/orca-session-address'
+
+/** Owner edits discover mail without enumerating unrelated unread mailboxes. */
+export function sleepingMailOwnerMailboxes(
+  db: OrchestrationDb,
+  pane: string | null,
+  address?: string
+): string[] {
+  const suffix = pane === null ? null : paneKeyMatchSuffix(pane)
+  const sessionId = parseOrcaSessionAddress(address)
+  const rows = db.db
+    .prepare(`
+    SELECT 'run:' || id AS mailbox, coordinator_pane_key AS owner FROM runs
+      WHERE coordinator_pane_key IS NOT NULL AND ${RUN_PANE_KEY_MATCH_SUFFIX_SQL} = ?
+    UNION SELECT 'dispatch:' || id AS mailbox, assignee_pane_key AS owner FROM dispatch_contexts
+      WHERE assignee_pane_key IS NOT NULL AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ?
+        AND status IN ('pending', 'dispatched')
+    UNION SELECT 'dispatch:' || dispatch_id AS mailbox, pane_key AS owner FROM remote_dispatch_attachments
+      WHERE pane_key IS NOT NULL AND ${REMOTE_ATTACHMENT_PANE_KEY_MATCH_SUFFIX_SQL} = ?
+        AND ${potentiallyLiveRemoteAttachmentSql()}
+    UNION SELECT 'run:' || id AS mailbox, NULL AS owner FROM runs WHERE coordinator_orca_session_id = ?
+    UNION SELECT 'run:' || run_id AS mailbox, NULL AS owner FROM run_coordinator_handles WHERE terminal_handle = ?
+    UNION SELECT 'dispatch:' || id AS mailbox, NULL AS owner FROM dispatch_contexts
+      WHERE (assignee_handle = ? OR assignee_orca_session_id = ?)
+        AND status IN ('pending', 'dispatched')
+  `)
+    .all(suffix, suffix, suffix, sessionId, address ?? null, address ?? null, sessionId)
+  return rows.flatMap((row) =>
+    typeof row.mailbox === 'string' &&
+    (pane === null ||
+      row.owner === null ||
+      (typeof row.owner === 'string' && isEquivalentPaneKey(pane, row.owner)))
+      ? [row.mailbox]
+      : []
+  )
+}
+
+export function refreshSleepingMailOwners(
+  wake: SleptMailboxWake,
+  db: OrchestrationDb,
+  owners: Iterable<string>,
+  addresses: Iterable<string>
+): Set<string> {
+  const affected = new Set(owners)
+  const mailboxes = new Set(addresses)
+  for (const pane of affected) {
+    for (const mailbox of [...wake.mailboxesForPane(pane), ...wake.discover(pane)]) {
+      mailboxes.add(mailbox)
+    }
+  }
+  for (const address of addresses) {
+    if (!address.startsWith('run:') && !address.startsWith('dispatch:')) {
+      for (const mailbox of sleepingMailOwnerMailboxes(db, null, address)) {
+        mailboxes.add(mailbox)
+      }
+    }
+  }
+  for (const mailbox of mailboxes) {
+    for (const pane of wake.refresh(mailbox)) {
+      affected.add(pane)
+    }
+  }
+  return affected
+}

src/main/runtime/orchestration/mailbox/routing/slept-mailbox-runtime.ts

@@ -1,3 +1,5 @@
+import { refreshSleepingMailOwners } from './sleeping-mail-owner-mailboxes'
+import { sessionCrewChanges, type CrewInvalidation } from '../../crew/crew-changes'
 import { replaceMailWakeRecord } from '../../../session-lifecycle/projection/retained-mail-wake-attempt'
 import { SleepingMailRecordIndex } from './sleeping-mail-record-index'
 import {
@@ -118,13 +120,16 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission
       { consumesTerminalSideEffects: false }
     )
     this.subscribeAgentStatusRowChanges((paneKey) => this.watchMailWakeInvocations(paneKey))
-    this.crewChanges.subscribe((facets) => {
+    const changed = (facets: CrewInvalidation) => {
       if (facets.has('traffic') || facets.has('topology') || facets.has('records')) {
         this.publishSleepingMail(
-          facets.has('topology') || facets.has('records') ? undefined : facets.trafficAddresses
+          facets.mailGlobal ? undefined : (facets.trafficAddresses ?? []),
+          facets.mailboxOwners
         )
       }
-    })
+    }
+    this.crewChanges.subscribe(changed)
+    sessionCrewChanges.subscribe(changed)
     this.store?.onWorkspaceSessionWritten?.((hostId) => {
       const hosts =
         hostId === null ? (this.store?.getWorkspaceSessionHostIds?.() ?? ['local']) : [hostId]
@@ -132,7 +137,7 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission
         const changed = this.refreshSleepingRecords(host)
         this.watchMailWakeInvocations()
         if (changed) {
-          queueMicrotask(() => this.retrySleptMailboxWakes())
+          queueMicrotask(() => this.publishSleepingMail([]))
         } else {
           this.publishSleepingMail([])
         }
@@ -221,7 +226,10 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission
   private sleepingMailPublicationQueued = false
   private sleepingMailPublication = ''
 
-  protected publishSleepingMail(mailboxes?: Iterable<string>): void {
+  protected publishSleepingMail(mailboxes?: Iterable<string>, owners?: Iterable<string>): void {
+    for (const owner of owners ?? []) {
+      this.changedSleepingPanes.add(owner)
+    }
     if (mailboxes) {
       for (const mailbox of mailboxes) {
         this.sleepingMailAddresses.add(mailbox)
@@ -240,7 +248,8 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission
           return
         }
         const global = this.globalSleepingMailInvalidation
-        const affected = new Set(this.changedSleepingPanes)
+        let affected = new Set(this.changedSleepingPanes)
+        const addresses = new Set(this.sleepingMailAddresses)
         this.changedSleepingPanes.clear()
         if (global && (this.sleepingRecords.size || this.sleptMailboxWake.hasTrackedMail)) {
           this.sleptMailboxWake.reconcile()
@@ -248,11 +257,12 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission
             affected.add(pane)
           }
         } else {
-          for (const mailbox of this.sleepingMailAddresses) {
-            for (const pane of this.sleptMailboxWake.refresh(mailbox)) {
-              affected.add(pane)
-            }
-          }
+          affected = refreshSleepingMailOwners(
+            this.sleptMailboxWake,
+            this._orchestrationDb,
+            affected,
+            addresses
+          )
         }
         this.globalSleepingMailInvalidation = false
         this.sleepingMailAddresses.clear()
@@ -262,11 +272,9 @@ export abstract class RuntimeSleptMailboxWakes extends RuntimeMailWakeAdmission
             await this.editMailWakeRecord(record, withoutMailWake(record))
           }
         }
-        if (global) {
-          for (const pane of this.sleepingRecords.keys()) {
-            for (const mailbox of this.sleptMailboxWake.mailboxesForPane(pane)) {
-              this.sleptMailboxWake.request(mailbox)
-            }
+        for (const pane of global ? this.sleepingRecords.keys() : affected) {
+          for (const mailbox of this.sleptMailboxWake.mailboxesForPane(pane)) {
+            this.sleptMailboxWake.request(mailbox)
           }
         }
         const settledIds = [...this.settledMailWakeIds]

src/main/runtime/orchestration/mailbox/routing/slept-mailbox-wake.test.ts

@@ -1,6 +1,8 @@
 import type { ExecutionHostId } from '../../../../../shared/execution-host'
 import type { WorkspaceSessionState } from '../../../../../shared/workspace-session/workspace-session-state-types'
 import { OrchestrationMailboxPointerDelivery } from '../terminal-pointer/mailbox-pointer-delivery'
+import { createRootDispatch } from '../../db/root-dispatch-test-fixture'
+import { ORCHESTRATION_REPLY_METHOD } from '../../../rpc/methods/orchestration/messaging/reply-method'
 import { MailWakeInvocations } from './mail-wake-invocations'
 import { admitsMailWakeRetry } from './mail-wake-retry-admission'
 import {
@@ -37,6 +39,24 @@ const record: SleepingAgentSessionRecord = {
 }
 const settled = () => new Promise<void>((resolve) => setImmediate(resolve))
 
+const mailNotifier = (
+  resume: RuntimeNotifier['resumeSleepingAgents'],
+  bell?: RuntimeNotifier['sleepingMailChanged']
+): RuntimeNotifier => ({
+  worktreesChanged: vi.fn(),
+  reposChanged: vi.fn(),
+  activateWorktree: vi.fn(),
+  createTerminal: vi.fn(),
+  splitTerminal: vi.fn(),
+  renameTerminal: vi.fn(),
+  focusTerminal: vi.fn(),
+  closeTerminal: vi.fn(),
+  terminalFitOverrideChanged: vi.fn(),
+  terminalDriverChanged: vi.fn(),
+  resumeSleepingAgents: resume,
+  sleepingMailChanged: bell
+})
+
 it('wakes only the automatic sleeper owed deliverable mail, coalesces it, and retries a failed wake on availability', async () => {
   const db = new OrchestrationDb(':memory:')
   try {
@@ -156,20 +176,7 @@ it('holds manual Sleep through the sender receipt and keeps its bell until mail
   )
   const resume = vi.fn(async () => true)
   const bell = vi.fn()
-  const notifier: RuntimeNotifier = {
-    worktreesChanged: vi.fn(),
-    reposChanged: vi.fn(),
-    activateWorktree: vi.fn(),
-    createTerminal: vi.fn(),
-    splitTerminal: vi.fn(),
-    renameTerminal: vi.fn(),
-    focusTerminal: vi.fn(),
-    closeTerminal: vi.fn(),
-    terminalFitOverrideChanged: vi.fn(),
-    terminalDriverChanged: vi.fn(),
-    resumeSleepingAgents: resume,
-    sleepingMailChanged: bell
-  }
+  const notifier = mailNotifier(resume, bell)
   try {
     runtime.setNotifier(notifier)
     runtime.syncWindowGraph(1, { tabs: [], leaves: [] })
@@ -454,11 +461,16 @@ it('suppresses a definitively confirmed late wake only for its obligation and re
     delivery.request(mailbox)
     await settled()
     expect(wake).toHaveBeenCalledTimes(1)
+    sleeping = null
+    delivery.refresh(mailbox)
+    sleeping = record
+    delivery.request(mailbox)
+    await settled()
+    expect(wake).toHaveBeenCalledTimes(1)
     db.markAsRead([message.id])
     delivery.refresh(mailbox)
-    expect(Reflect.get(delivery, 'succeeded').size).toBe(0)
-    expect(Reflect.get(delivery, 'failed').size).toBe(0)
-    expect(Reflect.get(delivery, 'attempted').size).toBe(0)
+    expect(delivery.failedToWake(retained)).toBe(false)
+    expect(delivery.heldWorktrees()).toEqual([])
 
     for (let i = 0; i < 20; i++) {
       sleeping = record
@@ -480,9 +492,8 @@ it('suppresses a definitively confirmed late wake only for its obligation and re
       db.markAsRead([row.id])
       delivery.refresh(handle)
     }
-    expect(Reflect.get(delivery, 'succeeded').size).toBe(0)
-    expect(Reflect.get(delivery, 'failed').size).toBe(0)
-    expect(Reflect.get(delivery, 'attempted').size).toBe(0)
+    expect(wake).toHaveBeenCalledTimes(21)
+    expect(delivery.failedToWake(record)).toBe(false)
     expect(delivery.heldWorktrees()).toEqual([])
   } finally {
     db.close()
@@ -499,7 +510,7 @@ it.each([10, 100])(
     const records: Record<string, SleepingAgentSessionRecord> = {}
     const mailboxes: string[] = []
     for (let i = 0; i < count; i++) {
-      const key = `tab-${i}:11111111-1111-4111-8111-111111111111`
+      const key = `tab-${i}:11111111-1111-4111-8111-${String(i).padStart(12, '0')}`
       records[key] = new Proxy(
         {
           ...record,
@@ -537,6 +548,13 @@ it.each([10, 100])(
         db.insertMessage({ runId: run.id, from: 'worker', to: mailbox, subject: 'done' })
       }
     }
+    const task = db.createTask({ spec: 'local completion', runId: mailboxes[0]!.slice(4) })
+    const dispatch = createRootDispatch(
+      db,
+      task.id,
+      'term_local_worker',
+      'tab-0:11111111-1111-4111-8111-000000000000'
+    )
     let session = { ...getDefaultWorkspaceSession(), sleepingAgentSessionsByPaneKey: records }
     const readSession = vi.fn(() => session)
     const runtime = new OrcaRuntimeService(
@@ -564,24 +582,20 @@ it.each([10, 100])(
       readSession.mockClear()
       statusChanged('unrelated:11111111-1111-4111-8111-111111111111')
       expect(visits).toBe(0)
-      statusChanged('tab-0:11111111-1111-4111-8111-111111111111')
+      statusChanged('tab-0:11111111-1111-4111-8111-000000000000')
       // A status event needs only this pane's attempt and PTY identity; no session inventory.
       expect(visits).toBeLessThanOrEqual(4)
       expect(readSession).not.toHaveBeenCalled()
       const inventory = vi.spyOn(db, 'getUndeliveredUnreadMailboxHandles')
       const fullRows = vi.spyOn(db, 'getUndeliveredUnreadMessages')
       const newest = vi.spyOn(db, 'getUndeliveredUnreadSequence')
-      runtime.crewChanges.notify(
-        Object.assign(new Set(['traffic'] as const), { trafficAddresses: new Set([mailboxes[0]!]) })
-      )
+      db.completeDispatch(dispatch.id)
       await settled()
       expect(inventory).not.toHaveBeenCalled()
       expect(fullRows).not.toHaveBeenCalled()
-      // Refresh and the custody check each need one scalar newest-sequence row for the affected mailbox.
-      expect(newest.mock.calls.length).toBeLessThanOrEqual(3)
-      expect(
-        newest.mock.results.filter((result) => result.value !== undefined).length
-      ).toBeLessThanOrEqual(3)
+      expect(newest.mock.calls.filter(([mailbox]) => mailbox !== mailboxes[0])).toEqual([])
+      // One affected Run: refresh, custody validation and each start-stage revalidation; no N-dependent work.
+      expect(newest.mock.calls.length).toBeLessThanOrEqual(8)
       expect(newest).toHaveBeenCalled()
       session = {
         ...session,
@@ -604,9 +618,18 @@ it.each([10, 100])(
       await settled()
       visits = 0
       readSession.mockClear()
-      statusChanged('tab-0:11111111-1111-4111-8111-111111111111')
+      statusChanged('tab-0:11111111-1111-4111-8111-000000000000')
       expect(visits).toBe(0)
       expect(readSession).not.toHaveBeenCalled()
+      session = { ...session, sleepingAgentSessionsByPaneKey: {} }
+      sessionWritten('local')
+      await settled()
+      inventory.mockClear()
+      newest.mockClear()
+      db.completeDispatch(dispatch.id)
+      await settled()
+      expect(inventory).not.toHaveBeenCalled()
+      expect(newest).not.toHaveBeenCalled()
     } finally {
       db.close()
     }
@@ -640,19 +663,7 @@ it('holds unknown dispatch custody but retries a positively undispatched wake on
   const resume = vi.fn(
     async (_workspace: string, _pane?: string, prepare?: boolean) => prepare === true
   )
-  const notifier: RuntimeNotifier = {
-    worktreesChanged: vi.fn(),
-    reposChanged: vi.fn(),
-    activateWorktree: vi.fn(),
-    createTerminal: vi.fn(),
-    splitTerminal: vi.fn(),
-    renameTerminal: vi.fn(),
-    focusTerminal: vi.fn(),
-    closeTerminal: vi.fn(),
-    terminalFitOverrideChanged: vi.fn(),
-    terminalDriverChanged: vi.fn(),
-    resumeSleepingAgents: resume
-  }
+  const notifier = mailNotifier(resume)
   try {
     const run = db.createRun({
       objective: 'refused launch',
@@ -666,7 +677,8 @@ it('holds unknown dispatch custody but retries a positively undispatched wake on
     runtime.syncWindowGraph(1, { tabs: [], leaves: [] })
     await settled()
     expect(await runtime.readSleepingMailHold(mailbox)).toMatchObject({
-      resuming: true,
+      resuming: false,
+      openTab: true,
       liveShell: false
     })
     runtime.retrySleptMailboxWakes()
@@ -689,3 +701,101 @@ it('holds unknown dispatch custody but retries a positively undispatched wake on
     db.close()
   }
 })
+
+it('clears the original workspace hold and custody when another terminal replies to its last mail', async () => {
+  const db = new OrchestrationDb(':memory:')
+  let session: WorkspaceSessionState = {
+    ...getDefaultWorkspaceSession(),
+    sleepingAgentSessionsByPaneKey: {
+      [paneKey]: {
+        ...record,
+        origin: 'quit',
+        restoreOnTabOpenOnly: true,
+        mailWake: { id: 'reply-custody', capturedAt: 1, startedAt: 100 }
+      }
+    }
+  }
+  let written: (host: ExecutionHostId | null) => void = () => {}
+  const runtime = new OrcaRuntimeService(
+    partialRuntimeStore({
+      getWorkspaceSession: () => session,
+      getWorkspaceSessionHostIds: () => ['local'],
+      setWorkspaceSession: (next) => {
+        session = next
+        written('local')
+      },
+      runDurableMutation: async (mutation) => mutation().value,
+      onWorkspaceSessionWritten: (listener) => {
+        written = listener
+        return () => {}
+      }
+    })
+  )
+  const bell = vi.fn()
+  runtime.setNotifier(
+    mailNotifier(
+      vi.fn(async () => false),
+      bell
+    )
+  )
+  try {
+    const run = db.createRun({
+      objective: 'reply consumes mail',
+      coordinatorHandle: 'term_asleep',
+      coordinatorPaneKey: paneKey
+    })
+    const mailbox = `run:${run.id}`
+    const sender = db.createRun({
+      objective: 'sender',
+      coordinatorHandle: 'term_sender',
+      coordinatorPaneKey: null
+    })
+    const original = db.insertMessage({
+      runId: run.id,
+      from: `run:${sender.id}`,
+      to: mailbox,
+      subject: 'question'
+    })
+    runtime.setOrchestrationDb(db)
+    runtime.retrySleptMailboxWakes()
+    await settled()
+    expect(await runtime.readSleepingMailHold(mailbox)).toMatchObject({
+      kind: 'asleep',
+      resuming: false
+    })
+    expect(bell).toHaveBeenLastCalledWith(['folder:workspace'])
+    const inventory = vi.spyOn(db, 'getUndeliveredUnreadMailboxHandles')
+    const originalInvalidation = vi.fn()
+    const unwatch = db.crewChanges.subscribe((event) => {
+      if (event.trafficAddresses?.has(mailbox)) {
+        originalInvalidation(db.getMessageById(original.id)?.read)
+      }
+    })
+    db.db.exec('BEGIN IMMEDIATE')
+    db.markAsRead([original.id])
+    expect(originalInvalidation).not.toHaveBeenCalled()
+    db.db.exec('ROLLBACK')
+    await settled()
+    expect(originalInvalidation).not.toHaveBeenCalled()
+    expect(await runtime.readSleepingMailHold(mailbox)).not.toBeNull()
+    await ORCHESTRATION_REPLY_METHOD.handler(
+      ORCHESTRATION_REPLY_METHOD.params.parse({
+        id: original.id,
+        from: 'term_other',
+        body: 'answer'
+      }),
+      { runtime }
+    )
+    await settled()
+    unwatch()
+    expect(originalInvalidation).toHaveBeenCalledWith(1)
+    expect(db.getMessageById(original.id)?.read).toBe(1)
+    expect(await runtime.readSleepingMailHold(mailbox)).toBeNull()
+    expect(session.sleepingAgentSessionsByPaneKey?.[paneKey]?.mailWake).toBeUndefined()
+    expect(inventory).not.toHaveBeenCalled()
+    expect(bell).toHaveBeenLastCalledWith([], ['reply-custody'])
+  } finally {
+    runtime.setNotifier(null)
+    db.close()
+  }
+})

src/main/runtime/orchestration/mailbox/routing/slept-mailbox-wake.ts

@@ -1,3 +1,4 @@
+import { sleepingMailOwnerMailboxes } from './sleeping-mail-owner-mailboxes'
 import { POTENTIALLY_LIVE_REMOTE_ATTACHMENT_STATES } from '../../db/federation/remote-attachment-liveness'
 import type { SleepingAgentSessionRecord } from '../../../../../shared/agent-session/lifecycle/agent-session-resume'
 import { agentProviderSessionsEqual } from '../../../../../shared/agent-session/lifecycle/agent-session-resume'
@@ -76,7 +77,11 @@ export class SleptMailboxWake {
   private selection(mailbox: string, reservedTypes?: ReadonlySet<string>) {
     const db = this.deps.db()
     const paneKey = db && mailboxOwnerPane(db, mailbox)
-    if (!db || !paneKey) {
+    if (
+      !db ||
+      !paneKey ||
+      (!mailWakeObligation(this.deps.record(paneKey)) && !this.succeeded.has(mailbox))
+    ) {
       return null
     }
     const sequence = selectOrchestrationPointerSequence({
@@ -90,7 +95,9 @@ export class SleptMailboxWake {
 
   refresh(mailbox: string): string[] {
     const before = this.owed.get(mailbox)
-    const next = this.selection(mailbox)
+    const selection = this.selection(mailbox)
+    const next =
+      selection && mailWakeObligation(this.deps.record(selection.paneKey)) ? selection : null
     if (before) {
       const handles = this.byPane.get(before.paneKey)
       handles?.delete(mailbox)
@@ -106,7 +113,11 @@ export class SleptMailboxWake {
       this.byPane.set(next.paneKey, handles)
     }
     const success = this.succeeded.get(mailbox)
-    if (!next || success?.paneKey !== next.paneKey || success.sequence !== next.sequence) {
+    if (
+      !selection ||
+      success?.paneKey !== selection.paneKey ||
+      success.sequence !== selection.sequence
+    ) {
       this.succeeded.delete(mailbox)
     }
     const attempted = this.attempted.get(mailbox)
@@ -126,12 +137,12 @@ export class SleptMailboxWake {
   }
 
   get hasTrackedMail(): boolean {
-    return this.owed.size > 0 || this.attempted.size > 0
+    return this.owed.size > 0 || this.attempted.size > 0 || this.succeeded.size > 0
   }
 
   reconcile(): void {
     const handles = new Set(this.deps.db()?.getUndeliveredUnreadMailboxHandles() ?? [])
-    for (const handle of this.owed.keys()) {
+    for (const handle of [...this.owed.keys(), ...this.succeeded.keys()]) {
       handles.add(handle)
     }
     for (const handle of handles) {
@@ -139,6 +150,13 @@ export class SleptMailboxWake {
     }
   }
 
+  discover(paneKey: string): readonly string[] {
+    const db = this.deps.db()
+    return db && mailWakeObligation(this.deps.record(paneKey))
+      ? sleepingMailOwnerMailboxes(db, paneKey)
+      : []
+  }
+
   mailboxesForPane(paneKey: string): readonly string[] {
     return [...(this.byPane.get(paneKey) ?? [])]
   }

src/main/runtime/workspaces/persisted-surfaces/crew-operations.ts

@@ -32,7 +32,9 @@ export abstract class OrcaRuntimeWithCrewOperations extends OrcaRuntimeWithOrche
       this.crewChanges.notify(facets)
     })
     this.orchestrationDbChanges.notify()
-    this.crewChanges.notify()
+    this.crewChanges.notify(
+      Object.assign(new Set(['topology'] as const), { mailGlobal: true as const })
+    )
   }
 
   private readonly crewReadModel = new CrewReadModel(