fix(orchestration): native-chat agents get their mail notice right away, the same way for every agent (#28540)

* fix(orchestration): steer mail notices into a busy native chat's running turn

A native-chat coordinator's mail notice always queued as a card until the whole turn ended, so a
coordinator busy in a long command never saw its workers' reports. Agents that take a message
mid-turn (Claude, Codex, Pi) now get the notice at once; agents that would cancel their turn for
it (ACP) still queue it.

* fix(orchestration): drain mail that arrives while a notice is pending

* Send orchestration mail the same way to every chat agent

A mail notice is a plain send for every agent, as a person's own mid-turn message is.
While a pending approval or question waits on the person it is held, and the turn's
working edge after the answer sends it.

* Hold orchestration mail at live prompt admission

src/main/acp/acp-structured-mail-admission.test.ts

@@ -0,0 +1,54 @@
+import { afterEach, expect, it } from 'vitest'
+import { openAttachedHostRig, send, hello, framesOf } from './acp-structured-host.test-support'
+import { PROVIDER_SESSION, waitFor } from './acp-structured-adapter.test-support'
+import { closeProviderTimelineRigs } from '../native-chat/agent-session-timeline/provider-timeline-assembler-test-support'
+import { HOST_TEST_SESSION as SESSION } from '../native-chat/agent-session-wire/structured-agent-session-host-test-data'
+import { setStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry'
+import { commandReceiptScope } from '../native-chat/agent-session-journal/command-receipt-schema'
+import { createStructuredMailboxPointerHost } from '../runtime/orchestration/structured-mailbox-pointer-host'
+import { mintAgentSessionOperationId } from '../runtime/orchestration/structured-pointer-operation-id'
+import type { AgentJournalMessageItem } from '../../shared/agent-session-journal-types'
+
+afterEach(async () => {
+  setStructuredAgentSessionHost(null)
+  await closeProviderTimelineRigs()
+})
+
+it('holds mail without interrupting an answerable ACP question', async () => {
+  const { host, rig } = await openAttachedHostRig()
+  setStructuredAgentSessionHost(host)
+  await send(host, 'work')
+  await rig.frame('session/prompt')
+  void rig.child().agent.request(601, 'x.ai/ask_user_question', {
+    sessionId: PROVIDER_SESSION,
+    toolCallId: 'question-1',
+    questions: [{ question: 'Which file?', options: [{ label: 'a.ts' }] }]
+  })
+  const pointer = createStructuredMailboxPointerHost()
+  await waitFor(() => expect(pointer.hasActionablePromptNow(SESSION)).toBe(true))
+  const body: AgentJournalMessageItem = {
+    ...hello,
+    from: {
+      kind: 'agent',
+      senders: [],
+      orchestration: { message: 'mail-notice', mailbox: 'run:r1', dispatchId: null, messages: [] }
+    }
+  }
+  const operationId = mintAgentSessionOperationId(Date.now())
+  const expectedRuntimeFence = host.deps.store.getRecord(SESSION)!.lease.runtimeFence
+  expect(
+    await pointer.send({
+      sessionId: SESSION,
+      dispatchId: null,
+      operationId,
+      expectedRuntimeFence,
+      body
+    })
+  ).toEqual({ kind: 'held' })
+  expect((await host.journalSnapshot(SESSION)).submissions).toHaveLength(1)
+  expect(
+    host.deps.store.readCommandReceipt(commandReceiptScope('any', 'global'), operationId)
+  ).toMatchObject({ verdict: 'absent' })
+  expect(framesOf(rig.child(), 'session/cancel')).toHaveLength(0)
+  await host.close(SESSION, 'user-close')
+})

src/main/runtime/orca-runtime-structured-status-sink-wiring.test.ts

@@ -109,7 +109,8 @@ describe('mail dispatch through the current runtime', () => {
     })
 
     target.onStructuredSessionStatusForMail({ sessionId: 'session-mailbox', status: 'working' })
-    expect(events).toEqual([])
+    expect(events).toEqual(['journal:session-mailbox'])
+    events.length = 0
     target.onStructuredSessionStatusForMail({ sessionId: 'session-mailbox', status: 'idle' })
     expect(events).toEqual(['journal:session-mailbox', 'delivery:term_mailbox'])
   })

src/main/runtime/orchestration/agent-facing-parity.test.ts

@@ -129,6 +129,7 @@ async function renderChatPointer(mailbox: string): Promise<string> {
     senderName: () => null,
     host: {
       readSessionFacts: async () => ({ submissions: [] }),
+      hasActionablePromptNow: () => false,
       currentFence: () => 1,
       currentContextClearOperationId: () => undefined,
       send: async (input) => {

src/main/runtime/orchestration/send-agent-turn.ts

@@ -41,6 +41,8 @@ export type StructuredSessionTurn = {
   operationId: string
   expectedRuntimeFence: number
   delivery: AgentTurnDelivery
+  /** Host-local admission for an automated send. */
+  beforeRun?: () => void
 }
 
 export type StructuredSessionTurnSend = {
@@ -134,7 +136,10 @@ async function sendStructuredSessionTurn(
     body: turn.body,
     delivery: turn.delivery === 'queue' ? 'queue-if-active' : undefined
   })
-  const result = await send.host.send({ callerKey: send.callerKey }, message)
+  const result = await send.host.send(
+    { callerKey: send.callerKey },
+    turn.beforeRun ? { ...message, beforeRun: turn.beforeRun } : message
+  )
   if (!result.ok) {
     return { kind: 'refused', refusal: result.refusal }
   }

src/main/runtime/orchestration/structured-mailbox-pointer-delivery.test.ts

@@ -9,6 +9,7 @@ import {
   type StructuredPointerSubmission
 } from './structured-pointer-operation-id'
 import type { StructuredWorkerIdentity } from '../structured-worker-identity'
+import { RuntimeMessageWaiters } from '../runtime-message-waiters'
 
 const IDENTITY: StructuredWorkerIdentity = {
   handle: 'structworker_1',
@@ -32,10 +33,12 @@ function harness(options: {
   /** The mailbox this worker owns; its own handle for direct peer mail outside a dispatch. */
   mailbox?: string
   dispatchId?: string | null
+  waiters?: RuntimeMessageWaiters
 }) {
   const mailbox = options.mailbox ?? 'dispatch:d1'
   const dispatchId = options.dispatchId === undefined ? 'd1' : options.dispatchId
   let attached = options.attached ?? true
+  let awaitingHuman = false
   // The session's recorded sends, as its journal reports them.
   let submissions: StructuredPointerSubmission[] = []
   // The mailbox's unread mail; a pointed message is no longer selected for a pointer.
@@ -49,9 +52,11 @@ function harness(options: {
     }
   })
   const send: StructuredMailboxPointerHost['send'] = vi.fn(async () =>
-    options.queued
-      ? { kind: 'queued' as const }
-      : { kind: 'sent' as const, state: options.dispatchState ?? ('accepted' as const) }
+    awaitingHuman
+      ? { kind: 'held' as const }
+      : options.queued
+        ? { kind: 'queued' as const }
+        : { kind: 'sent' as const, state: options.dispatchState ?? ('accepted' as const) }
   )
   const sendMock = vi.mocked(send)
   const readContext = vi.fn<StructuredMailboxPointerHost['currentContextClearOperationId']>(
@@ -63,7 +68,14 @@ function harness(options: {
     hasOutstandingMailboxDelivery: (handle: string) =>
       ((options.outstandingRunDelivery ?? false) && handle.startsWith('run:')) ||
       ((options.outstandingOwnDelivery ?? false) && !handle.startsWith('run:')),
-    getUndeliveredUnreadMessages: () => mail.filter((message) => !pointed.has(message.id)),
+    getUndeliveredUnreadMessages: (
+      _handle: string,
+      _filter: unknown,
+      input: { excludeTypes: readonly string[] }
+    ) =>
+      mail.filter(
+        (message) => !pointed.has(message.id) && !input.excludeTypes.includes(message.type)
+      ),
     markAsDelivered,
     getStructuredPointerOperation: (key: string) => stored.get(key),
     putStructuredPointerOperation: (row: StructuredPointerOperationRow) =>
@@ -72,13 +84,14 @@ function harness(options: {
   }
   const delivery = new OrchestrationStructuredMailboxPointerDelivery({
     getDb: () => db as never,
-    getMessageWaiters: () => undefined,
+    getMessageWaiters: (handle) => options.waiters?.get(handle),
     resolveStructuredTarget: (mailboxHandle) =>
       mailboxHandle === mailbox ? { sessionId: IDENTITY.sessionId, dispatchId } : null,
     getCliCommand: () => 'orca-dev',
     senderName: (party) => (party.address === 'term_coord' ? 'Coordinator' : null),
     host: {
       readSessionFacts: async () => (attached ? { submissions } : null),
+      hasActionablePromptNow: () => awaitingHuman,
       currentFence: () => 4,
       currentContextClearOperationId: readContext,
       send
@@ -93,11 +106,14 @@ function harness(options: {
     setAttached: (next: boolean) => {
       attached = next
     },
-    receive: (id: string, sequence: number) => {
-      mail.push({ id, type: 'status', sequence, from_handle: 'term_coord', run_id: 'run_1' })
+    receive: (id: string, sequence: number, type = 'status') => {
+      mail.push({ id, type, sequence, from_handle: 'term_coord', run_id: 'run_1' })
     },
     setSubmissions: (next: StructuredPointerSubmission[]) => {
       submissions = next
+    },
+    setAwaitingHuman: (next: boolean) => {
+      awaitingHuman = next
     }
   }
 }
@@ -211,6 +227,20 @@ describe('structured mailbox pointer delivery', () => {
     expect(markAsDelivered).toHaveBeenCalledWith(['m1'])
   })
 
+  it('rechecks a held notice when the answer edge lands during its host call', async () => {
+    const { delivery, send, markAsDelivered, setAwaitingHuman } = harness({})
+    setAwaitingHuman(true)
+    send.mockImplementationOnce(async () => {
+      setAwaitingHuman(false)
+      delivery.onJournalActivity('session-1')
+      return { kind: 'held' }
+    })
+    delivery.deliverForHandle('dispatch:d1')
+    await flush()
+    expect(send).toHaveBeenCalledTimes(2)
+    expect(markAsDelivered).toHaveBeenCalledWith(['m1'])
+  })
+
   it('sends the pointer through the chat, with who it is from, and counts it pointed once queued', async () => {
     const { delivery, send, markAsDelivered, stored } = harness({ queued: true })
     delivery.deliverForHandle('dispatch:d1')
@@ -250,6 +280,38 @@ describe('structured mailbox pointer delivery', () => {
     })
   })
 
+  it('points a second report after its filtered waiter expires during first-notice acceptance', async () => {
+    const waiters = new RuntimeMessageWaiters()
+    const { delivery, send, receive, markAsDelivered } = harness({
+      mailbox: 'run:run_1',
+      waiters
+    })
+    const waiting = waiters.wait('run:run_1', { typeFilter: ['worker_done'], timeoutMs: 40 })
+    const accepted = Promise.withResolvers<void>()
+    send.mockImplementationOnce(async () => {
+      await accepted.promise
+      return { kind: 'sent', state: 'accepted' }
+    })
+    delivery.deliverForHandle('run:run_1', new Set(['worker_done']))
+    await flush()
+    expect(send).toHaveBeenCalledTimes(1)
+    expect(await waiting).toBe('timed_out')
+    receive('m2', 4, 'worker_done')
+    delivery.deliverForHandle('run:run_1')
+    await flush()
+    expect(send).toHaveBeenCalledTimes(1)
+    accepted.resolve()
+    await vi.waitFor(() => expect(send).toHaveBeenCalledTimes(2))
+    expect(send.mock.calls[1]![0].body.from).toMatchObject({
+      orchestration: { messages: [{ messageId: 'm2' }] }
+    })
+    expect(markAsDelivered.mock.calls).toEqual([[['m1']], [['m2']]])
+    delivery.onJournalActivity('session-1')
+    delivery.deliverForHandle('run:run_1')
+    await flush()
+    expect(send).toHaveBeenCalledTimes(2)
+  })
+
   it('nudges the worker while its coordinator holds an unacked Run delivery', async () => {
     // The exact window in which a coordinator replies to its workers: it checked, is acting on the
     // batch, and has not acked yet. The gate is keyed on the handle being nudged, so the
@@ -442,6 +504,7 @@ describe('forgetting one settled worker', () => {
   function twoWorkerHarness() {
     let resolves = true
     let attached = false
+    const pointed = new Set<string>()
     const sessionByMailbox: Record<string, string> = {
       'dispatch:d1': 'session-1',
       'dispatch:d2': 'session-2'
@@ -453,10 +516,11 @@ describe('forgetting one settled worker', () => {
     const db = {
       getDispatchContextById: () => ({ run_id: 'run_1' }),
       hasOutstandingMailboxDelivery: () => false,
-      getUndeliveredUnreadMessages: () => [
-        { id: 'm1', type: 'status', sequence: 3, from_handle: 'term_coord', run_id: 'run_1' }
-      ],
-      markAsDelivered: vi.fn(),
+      getUndeliveredUnreadMessages: () =>
+        pointed.has('m1')
+          ? []
+          : [{ id: 'm1', type: 'status', sequence: 3, from_handle: 'term_coord', run_id: 'run_1' }],
+      markAsDelivered: vi.fn((ids: string[]) => ids.forEach((id) => pointed.add(id))),
       getStructuredPointerOperation: () => undefined,
       putStructuredPointerOperation: () => {},
       deleteStructuredPointerOperation: () => {}
@@ -474,6 +538,7 @@ describe('forgetting one settled worker', () => {
       senderName: () => null,
       host: {
         readSessionFacts: async () => (attached ? { submissions: [] } : null),
+        hasActionablePromptNow: () => false,
         currentFence: () => 4,
         currentContextClearOperationId: () => undefined,
         send
@@ -575,6 +640,7 @@ describe('a mailbox a /clear moves while its nudge is in flight', () => {
       senderName: () => null,
       host: {
         readSessionFacts: async () => ({ submissions: [] }),
+        hasActionablePromptNow: () => false,
         currentFence: () => 4,
         currentContextClearOperationId: () => contextClearOperationId,
         send

src/main/runtime/orchestration/structured-mailbox-pointer-delivery.ts

@@ -51,6 +51,7 @@ type ParkedPointerDelivery = {
 
 export type StructuredPointerSendOutcome =
   | { kind: 'sent'; state: StructuredDispatchState }
+  | { kind: 'held' }
   /** The chat's queue took it, as it takes a person's message. */
   | { kind: 'queued' }
   | { kind: 'unattached' }
@@ -63,6 +64,7 @@ export type StructuredPointerSessionFacts = {
 export type StructuredMailboxPointerHost = {
   /** `null` when the session cannot be read. */
   readSessionFacts: (sessionId: string) => Promise<StructuredPointerSessionFacts | null>
+  hasActionablePromptNow: (sessionId: string) => boolean | null
   send: (input: {
     sessionId: string
     dispatchId: string | null
@@ -168,56 +170,92 @@ export class OrchestrationStructuredMailboxPointerDelivery<
     reservedTypes?: ReadonlySet<string>,
     attemptedContexts = new Set<string>()
   ): Promise<void> {
-    const db = this.deps.getDb()
-    if (!db || this.inFlight.has(mailboxHandle)) {
-      return
-    }
-    // Don't re-nudge a mailbox whose consumer still holds an unacknowledged batch. The lookup is
-    // keyed on the exact handle being nudged, so a coordinator's own `run:` delivery is invisible
-    // to a worker's `dispatch:` gate and cannot suppress the nudges a coordinator sends its
-    // workers. Worth more here than in the PTY lane: a structured nudge costs a whole provider
-    // turn, not a line of text into a composer.
-    if (db.hasOutstandingMailboxDelivery?.(mailboxHandle)) {
-      return
-    }
-    const unread = selectOrchestrationPointerBatch({
-      db,
-      mailboxHandle,
-      waiters: this.deps.getMessageWaiters(mailboxHandle),
-      reservedTypes
-    })
-    if (unread.length === 0) {
+    if (this.inFlight.has(mailboxHandle)) {
       return
     }
     this.inFlight.add(mailboxHandle)
+    let current = target
+    let currentReservedTypes = reservedTypes
     let contextKey: string | undefined
+    let progressed = false
     try {
-      const session = await this.deps.host.readSessionFacts(target.sessionId)
-      const contextClearOperationId = this.deps.host.currentContextClearOperationId(
-        target.sessionId
-      )
-      contextKey = JSON.stringify([target.sessionId, contextClearOperationId])
-      if (attemptedContexts.has(contextKey)) {
-        return
+      for (;;) {
+        const db = this.deps.getDb()
+        // The consumer's own unacknowledged batch suppresses nudges only for this exact mailbox.
+        if (!db || db.hasOutstandingMailboxDelivery?.(mailboxHandle)) {
+          return
+        }
+        const unread = selectOrchestrationPointerBatch({
+          db,
+          mailboxHandle,
+          waiters: this.deps.getMessageWaiters(mailboxHandle),
+          reservedTypes: currentReservedTypes
+        })
+        if (unread.length === 0) {
+          return
+        }
+        this.parkedUntilJournalEdge.set(mailboxHandle, {
+          sessionId: current.sessionId,
+          reservedTypes: currentReservedTypes
+        })
+        const session = await this.deps.host.readSessionFacts(current.sessionId)
+        const contextClearOperationId = this.deps.host.currentContextClearOperationId(
+          current.sessionId
+        )
+        contextKey = JSON.stringify([current.sessionId, contextClearOperationId])
+        if (!progressed && attemptedContexts.has(contextKey)) {
+          return
+        }
+        attemptedContexts.add(contextKey)
+        let attempt = await this.attempt(
+          db,
+          mailboxHandle,
+          current,
+          unread,
+          currentReservedTypes,
+          session,
+          contextClearOperationId
+        )
+        while (
+          attempt === 'held' &&
+          this.deps.host.hasActionablePromptNow(current.sessionId) === false
+        ) {
+          attempt = await this.attempt(
+            db,
+            mailboxHandle,
+            current,
+            unread,
+            currentReservedTypes,
+            session,
+            contextClearOperationId
+          )
+        }
+        if (attempt === 'held') {
+          this.retain(mailboxHandle, current.sessionId, 'turn-unsettled', currentReservedTypes)
+          return
+        }
+        progressed = attempt
+        if (!progressed) {
+          return
+        }
+        // The arrival snapshot protects its first handoff; successors derive live waiter ownership.
+        currentReservedTypes = undefined
+        // Acceptance releases the snapshotted batch; rederive any mail that arrived meanwhile.
+        this.parkedUntilJournalEdge.delete(mailboxHandle)
+        const next = this.deps.resolveStructuredTarget(mailboxHandle)
+        if (!next) {
+          return
+        }
+        current = next
       }
-      attemptedContexts.add(contextKey)
-      await this.attempt(
-        db,
-        mailboxHandle,
-        target,
-        unread,
-        reservedTypes,
-        session,
-        contextClearOperationId
-      )
     } finally {
       this.inFlight.delete(mailboxHandle)
       // A thrown attempt follows too; its own failure is what still propagates.
       if (contextKey !== undefined) {
         await this.followChangedContext(
           mailboxHandle,
-          target,
-          reservedTypes,
+          current,
+          currentReservedTypes,
           attemptedContexts
         ).catch(() => undefined)
       }
@@ -258,16 +296,12 @@ export class OrchestrationStructuredMailboxPointerDelivery<
     reservedTypes: ReadonlySet<string> | undefined,
     session: StructuredPointerSessionFacts | null,
     contextClearOperationId: string | undefined
-  ): Promise<void> {
+  ): Promise<boolean | 'held'> {
     const sessionId = target.sessionId
-    if (!session) {
-      this.retain(mailboxHandle, sessionId, 'session-not-attached', reservedTypes)
-      return
-    }
     const fence = this.deps.host.currentFence(sessionId)
-    if (fence === null) {
+    if (!session || fence === null) {
       this.retain(mailboxHandle, sessionId, 'session-not-attached', reservedTypes)
-      return
+      return false
     }
     const body: AgentJournalMessageItem = {
       kind: 'message',
@@ -301,11 +335,11 @@ export class OrchestrationStructuredMailboxPointerDelivery<
       db.markAsDelivered(staged)
       db.deleteStructuredPointerOperation(mailboxHandle)
       this.sentOperationIds.delete(mailboxHandle)
-      return
+      return true
     }
     if (operation.kind === 'park') {
       this.retain(mailboxHandle, sessionId, 'turn-unsettled', reservedTypes)
-      return
+      return false
     }
     this.sentOperationIds.set(mailboxHandle, operation.operationId)
     const outcome = await this.deps.host.send({
@@ -315,21 +349,23 @@ export class OrchestrationStructuredMailboxPointerDelivery<
       expectedRuntimeFence: fence,
       body
     })
+    if (outcome.kind === 'held') {
+      return 'held'
+    }
     if (outcome.kind === 'unattached') {
       this.retain(mailboxHandle, sessionId, 'session-not-attached', reservedTypes)
-      return
+      return false
     }
     // A queued pointer is the chat's queue's to send, as a person's queued message is.
     if (outcome.kind === 'sent' && !structuredDispatchDelivered(outcome.state)) {
       // The row stays: resending under its id replays this verdict and starts nothing.
       this.retain(mailboxHandle, sessionId, retainReasonForDispatch(outcome.state), reservedTypes)
-      return
+      return false
     }
     db.markAsDelivered(staged)
-    // The nudge landed as its own turn, so the next settle edge is the natural retry point for
-    // anything that arrives while it runs.
     db.deleteStructuredPointerOperation(mailboxHandle)
     this.sentOperationIds.delete(mailboxHandle)
+    return true
   }
 
   /**

src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts

@@ -87,7 +87,8 @@ describe('structured mailbox pointer host', () => {
     const submissions = [{ clientMessageId: 'op1', dispatchState: 'unknown' }]
     hostRef.current = {
       deps: { store: { getRecord: () => null } },
-      journalSnapshot: () => ({ items: [], submissions })
+      journalSnapshot: () => ({ items: [], submissions }),
+      currentWork: () => null
     }
     expect(await createStructuredMailboxPointerHost().readSessionFacts('s1')).toEqual({
       submissions
@@ -115,10 +116,16 @@ describe('structured mailbox pointer host', () => {
           getRecord: () => ({ providerContextBoundary: { operationId: 'clear-1', afterFence: 2 } })
         }
       },
-      journalSnapshot: () => ({ items, submissions: [pointerSubmission('earlier', 2, 6), current] })
+      journalSnapshot: () => ({
+        items,
+        submissions: [pointerSubmission('earlier', 2, 6), current]
+      }),
+      currentWork: () => null
     }
     const host = createStructuredMailboxPointerHost()
-    expect(await host.readSessionFacts('s1')).toEqual({ submissions: [current] })
+    expect(await host.readSessionFacts('s1')).toEqual({
+      submissions: [current]
+    })
     expect(host.currentContextClearOperationId('s1')).toBe('clear-1')
   })
 
@@ -133,10 +140,13 @@ describe('structured mailbox pointer host', () => {
       journalSnapshot: () => ({
         items: [],
         submissions: [pointerSubmission('earlier', 2, 2), current]
-      })
+      }),
+      currentWork: () => null
     }
     const host = createStructuredMailboxPointerHost()
-    expect(await host.readSessionFacts('s1')).toEqual({ submissions: [current] })
+    expect(await host.readSessionFacts('s1')).toEqual({
+      submissions: [current]
+    })
     expect(host.currentContextClearOperationId('s1')).toBe('clear-1')
   })
 
@@ -192,14 +202,11 @@ describe('structured mailbox pointer host', () => {
     expect(send.mock.calls[0]![1]!.retryUnknown).toBeUndefined()
   })
 
-  it('asks a busy chat to queue the pointer as a card, with who it is from', async () => {
+  it('sends the pointer as a plain send for every agent, with who it is from', async () => {
     const send = vi.fn(
       async (_caller: unknown, _payload: { delivery?: string; body?: unknown }) => ({
         ok: true,
-        value: {
-          clientMessageId: 'op1',
-          queued: { messageId: 'op1', position: 0, state: 'waiting' }
-        }
+        value: { clientMessageId: 'op1', submission: { dispatchState: 'accepted' } }
       })
     )
     hostRef.current = { send }
@@ -211,11 +218,9 @@ describe('structured mailbox pointer host', () => {
         expectedRuntimeFence: 1,
         body: { kind: 'message', role: 'user', blocks: [], from: NOTICE_SOURCE }
       })
-    ).resolves.toEqual({ kind: 'queued' })
-    expect(send.mock.calls[0]![1]).toMatchObject({
-      delivery: 'queue-if-active',
-      body: { from: NOTICE_SOURCE }
-    })
+    ).resolves.toEqual({ kind: 'sent', state: 'accepted' })
+    expect(send.mock.calls[0]![1]).toMatchObject({ body: { from: NOTICE_SOURCE } })
+    expect(send.mock.calls[0]![1]!.delivery).toBeUndefined()
   })
 
   it('consumes mail once an accepted nudge is delivered while the worker starts (W10)', async () => {

src/main/runtime/orchestration/structured-mailbox-pointer-host.ts

@@ -17,7 +17,7 @@ import {
   structuredSessionGateFacts,
   type StructuredSessionGateFacts
 } from './structured-session-pointer-delivery'
-import { sendAgentTurn } from './send-agent-turn'
+import { sendAgentTurn, type StructuredSessionTurnOutcome } from './send-agent-turn'
 
 /** Names the dispatch a nudge belongs to on the operation row it writes. */
 export function structuredPointerCallerKey(dispatchId: string): string {
@@ -63,8 +63,8 @@ async function readPointerSessionFacts(
   if (!snapshot) {
     return null
   }
-  const boundary =
-    getStructuredAgentSessionHost()?.deps.store.getRecord(sessionId)?.providerContextBoundary
+  const host = getStructuredAgentSessionHost()
+  const boundary = host?.deps.store.getRecord(sessionId)?.providerContextBoundary
   return {
     submissions: boundary
       ? snapshot.submissions.filter((submission) => submission.fence > boundary.afterFence)
@@ -95,6 +95,10 @@ export function createStructuredMailboxPointerHost(): StructuredMailboxPointerHo
       return readPointerSessionFacts(sessionId)
     },
 
+    hasActionablePromptNow(sessionId) {
+      return getStructuredAgentSessionHost()?.currentWork(sessionId)?.hasActionablePrompt() ?? null
+    },
+
     currentFence(sessionId) {
       return (
         getStructuredAgentSessionHost()?.deps.store.getRecord(sessionId)?.lease.runtimeFence ?? null
@@ -111,21 +115,37 @@ export function createStructuredMailboxPointerHost(): StructuredMailboxPointerHo
       if (!host) {
         return { kind: 'unattached' }
       }
-      const outcome = await sendAgentTurn({
-        kind: 'structured-session',
-        host,
-        sessionId: input.sessionId,
-        callerKey: input.dispatchId
-          ? structuredPointerCallerKey(input.dispatchId)
-          : structuredSessionPointerCallerKey(input.sessionId),
-        turn: {
-          body: input.body,
-          // As a person's message is: a busy chat queues it as a card, sent when the turn ends.
-          delivery: 'queue',
-          operationId: input.operationId,
-          expectedRuntimeFence: input.expectedRuntimeFence
+      let outcome: StructuredSessionTurnOutcome
+      try {
+        outcome = await sendAgentTurn({
+          kind: 'structured-session',
+          host,
+          sessionId: input.sessionId,
+          callerKey: input.dispatchId
+            ? structuredPointerCallerKey(input.dispatchId)
+            : structuredSessionPointerCallerKey(input.sessionId),
+          turn: {
+            body: input.body,
+            // A person's plain send: every agent takes it the way it takes their own mid-turn message.
+            delivery: 'now',
+            operationId: input.operationId,
+            expectedRuntimeFence: input.expectedRuntimeFence,
+            beforeRun: () => {
+              if (
+                input.body.from?.orchestration?.message === 'mail-notice' &&
+                host.currentWork(input.sessionId)?.hasActionablePrompt()
+              ) {
+                throw new StructuredMailHeldError()
+              }
+            }
+          }
+        })
+      } catch (error) {
+        if (error instanceof StructuredMailHeldError) {
+          return { kind: 'held' }
         }
-      })
+        throw error
+      }
       switch (outcome.kind) {
         case 'refused':
           return outcome.refusal.code === AGENT_SESSION_NOT_ATTACHED.code
@@ -146,3 +166,5 @@ export function createStructuredMailboxPointerHost(): StructuredMailboxPointerHo
     }
   }
 }
+
+class StructuredMailHeldError extends Error {}

src/main/runtime/orchestration/structured-session-pointer-delivery.ts

@@ -2,7 +2,8 @@
  * What orchestration mail delivery reads of a host-owned structured ("native") agent session.
  *
  * A structured session has no PTY the pointer can be typed into, so the nudge travels as a session
- * turn instead of as bytes, and a busy session's own queue holds it until the turn ends.
+ * turn instead of as bytes: steered into a busy session's running turn, or held by its queue until
+ * the turn ends when its agent cannot take a message mid-turn.
  * Everything here is pure. Orchestration's database stays the source of truth: nothing here
  * consumes mail.
  */

src/main/runtime/orchestration/terminal-pane-mail-routing.ts

@@ -195,20 +195,24 @@ export abstract class OrcaRuntimeWithGetPtyRecordForPaneKey extends OrcaRuntimeW
   }
 
   /**
-   * Every structured session's status change reaches here. At its idle edge, retry what is parked
-   * on it and re-derive the mailboxes it owns, so mail it could not take earlier (mid-turn, closed)
-   * is pointed again. Workers and chats alike: this is not per-dispatch.
+   * Every structured session's status change reaches here. At its working and idle edges, retry
+   * what is parked on it; at idle also re-derive the mailboxes it owns, so mail it could not take
+   * earlier (closed) is pointed again. Workers and chats alike: this is not per-dispatch.
    */
   onStructuredSessionStatusForMail(summary: {
     sessionId: string
     status: 'working' | 'attention' | 'idle' | null
   }): void {
-    if (summary.status === 'working' || summary.status === 'attention') {
+    if (summary.status === 'attention') {
       return
     }
     // Logged, never thrown: the same status callback goes on to the first-turn workspace rename.
     try {
+      // Working also follows an answered prompt, so mail held behind the prompt is retried too.
       this.notifyStructuredSessionJournalActivity(summary.sessionId)
+      if (summary.status === 'working') {
+        return
+      }
       const openDb = () => this.getExistingOrchestrationDb()
       const deliver = (mailbox: string) => this.deliverPendingMessagesForHandle(mailbox)
       structuredSessionIdleEdgeMailboxes(summary.sessionId, openDb).forEach(deliver)

src/main/runtime/structured-chat-coordinator-fake-codex-fixture.ts

@@ -6,7 +6,6 @@ import type {
   CodexAppServerConnectionHandlers,
   openCodexAppServerConnection
 } from '../codex/codex-app-server-connection'
-import { refuseUnroutedSteer } from '../codex/codex-structured-dispatch-test-support'
 import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope'
 import { attachFingerprintFields } from '../native-chat/agent-session-wire/structured-agent-session-attach'
 import { isRecord } from './rpc/orchestration-session-caller-test-fixture'
@@ -20,6 +19,8 @@ export type FakeConnection = Omit<CodexAppServerConnection, 'closed'> & {
   /** Every JSON-RPC method the adapter called, in order. */
   methods: string[]
   turns: { clientUserMessageId: string; text: string }[]
+  /** Messages steered into a running turn, as Codex takes them. */
+  steers: { expectedTurnId: string; clientUserMessageId: string; text: string }[]
 }
 
 /** How the fake provider misbehaves; reset before each test. */
@@ -61,6 +62,7 @@ export function fakeCodex() {
       threadId: null,
       methods: [],
       turns: [],
+      steers: [],
       pid: 4321,
       closed: false,
       request: async (method, params) => {
@@ -114,7 +116,12 @@ export function fakeCodex() {
           }
         }
         if (method === 'turn/steer') {
-          return refuseUnroutedSteer(undefined)
+          connection.steers.push({
+            expectedTurnId: String(input.expectedTurnId),
+            clientUserMessageId: String(input.clientUserMessageId),
+            text: JSON.stringify(input.input)
+          })
+          return { turnId: input.expectedTurnId }
         }
         return {}
       },

src/main/runtime/structured-chat-coordinator-mail-queue.test.ts

@@ -1,193 +0,0 @@
-import './rpc/unused-default-rpc-methods.test-fixture'
-// A busy structured chat holds the orchestration pointer as a card in its own queue, sent when the
-// turn ends, as it holds a message the person sends then; the queue does nothing else with it. End
-// to end on the coordinator-mail rig.
-
-import { describe, expect, it, vi } from 'vitest'
-import { structuredAgentSessionMessageSendMutation } from '../../shared/structured-agent-session-send-mutation'
-import { operationId, type FakeConnection } from './structured-chat-coordinator-fake-codex-fixture'
-import { idOf } from './rpc/orchestration-session-caller-test-fixture'
-import {
-  COORDINATOR,
-  WORKER_2_PANE,
-  WAIT,
-  call,
-  coordinatorRunAndTask,
-  db,
-  finishWorker,
-  host,
-  observationClock,
-  openChat,
-  ptyPointer,
-  queuedCardTexts,
-  runtime,
-  sendUserMessage,
-  settleTurn,
-  turnText
-} from './structured-chat-coordinator-mail-rig.test-fixture'
-
-/** The person's turn, started and still running; resolves to its end. */
-async function runningUserTurn(chat: FakeConnection): Promise<() => Promise<void>> {
-  expect(await sendUserMessage(COORDINATOR, 'go')).toMatchObject({ ok: true })
-  await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT)
-  const notify = (method: string, params: unknown) => chat.handlers.onNotification?.(method, params)
-  notify('turn/started', { turn: { id: 'turn-1' } })
-  notify('item/completed', {
-    item: {
-      type: 'userMessage',
-      id: 'echo-go',
-      clientId: chat.turns[0]!.clientUserMessageId,
-      content: [{ type: 'text', text: 'go' }]
-    }
-  })
-  await host.flushStreamedEvents(COORDINATOR)
-  return async () => {
-    notify('turn/completed', { turn: { id: 'turn-1' } })
-    await host.flushStreamedEvents(COORDINATOR)
-  }
-}
-
-/** Idle edges with nothing owed: whatever they would send gets the time to show. */
-async function idleEdgesSettled(): Promise<void> {
-  for (let edge = 0; edge < 3; edge += 1) {
-    runtime.onStructuredSessionStatusForMail({ sessionId: COORDINATOR, status: 'idle' })
-    await observationClock.observe(100)
-  }
-}
-
-/** The chat's queue as its journal stores it. */
-function queuedRows() {
-  return host.collaboratorsForTests().sessions.get(COORDINATOR)?.journal.queuedMessages.list() ?? []
-}
-
-/** A second task, for a second worker result. */
-async function secondTask(): Promise<string> {
-  return idOf(
-    (await call('orchestration.taskCreate', { spec: 'more' }, { sessionId: COORDINATOR })).task
-  )
-}
-
-describe("a busy chat's orchestration pointer waits in its queue", () => {
-  it('queues the pointer as a card, with who it is from, and sends it once when the turn ends', async () => {
-    observationClock.start()
-    const chat = await openChat(COORDINATOR)
-    const { runId, taskId } = await coordinatorRunAndTask()
-    const endTurn = await runningUserTurn(chat)
-    const dispatchId = await finishWorker(taskId)
-    // The report was accepted, so its dispatch settled before its mail was named.
-    expect(db.getDispatchContextById(dispatchId)?.status).toBe('completed')
-    await vi.waitFor(
-      async () => expect(await queuedCardTexts()).toEqual([ptyPointer(`run:${runId}`)]),
-      WAIT
-    )
-    expect(chat.turns).toHaveLength(1)
-    const [card] = queuedRows()
-    const [mail] = db.getAllMessages(`run:${runId}`)
-    const from = {
-      kind: 'agent',
-      senders: [
-        {
-          party: { address: 'term_worker', terminalHandle: 'term_worker', orcaSessionId: null },
-          // Named by the task of the dispatch it just finished, through the real runtime's naming.
-          name: 'build it'
-        }
-      ],
-      orchestration: {
-        message: 'mail-notice',
-        mailbox: `run:${runId}`,
-        dispatchId: null,
-        messages: [{ messageId: mail!.id, runId, from: 'term_worker' }]
-      }
-    }
-    // On the card's body: the turn the queue sends carries it, and the provider never sees it.
-    expect(card?.body.from).toEqual(from)
-
-    await endTurn()
-    await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT)
-    expect(turnText(chat.turns[1]!)).toBe(ptyPointer(`run:${runId}`))
-    expect(JSON.stringify(chat.turns[1])).not.toContain('term_worker')
-    const sent = (await host.journalSnapshot(COORDINATOR)).items.filter(
-      (item) => item.body.kind === 'message' && item.body.from
-    )
-    expect(sent.map((item) => item.body.kind === 'message' && item.body.from)).toEqual([from])
-    expect(await queuedCardTexts()).toEqual([])
-    await settleTurn(COORDINATOR, 1)
-    await idleEdgesSettled()
-    expect(chat.turns).toHaveLength(2)
-  })
-
-  it('queues a second card for mail that arrives while the first waits, each counting its own mail', async () => {
-    observationClock.start()
-    const chat = await openChat(COORDINATOR)
-    const { runId, taskId } = await coordinatorRunAndTask()
-    const second = await secondTask()
-    const endTurn = await runningUserTurn(chat)
-    await finishWorker(taskId)
-    await vi.waitFor(async () => expect(await queuedCardTexts()).toHaveLength(1), WAIT)
-    await finishWorker(second, { handle: 'term_worker_2', paneKey: WORKER_2_PANE })
-    const pointer = ptyPointer(`run:${runId}`)
-    await vi.waitFor(async () => expect(await queuedCardTexts()).toEqual([pointer, pointer]), WAIT)
-
-    await endTurn()
-    await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT)
-    await settleTurn(COORDINATOR, 1)
-    await vi.waitFor(() => expect(chat.turns).toHaveLength(3), WAIT)
-    expect(turnText(chat.turns[2]!)).toBe(pointer)
-    await settleTurn(COORDINATOR, 2)
-    await idleEdgesSettled()
-    expect(chat.turns).toHaveLength(3)
-    expect(await queuedCardTexts()).toEqual([])
-  })
-
-  it("leaves the chat's own `check` as it is: the mail stays readable, and the card stays", async () => {
-    const chat = await openChat(COORDINATOR)
-    const { runId, taskId } = await coordinatorRunAndTask()
-    const endTurn = await runningUserTurn(chat)
-    await finishWorker(taskId)
-    await vi.waitFor(async () => expect(await queuedCardTexts()).toHaveLength(1), WAIT)
-    const [mail] = db.getAllMessages(`run:${runId}`)
-    expect(await call('orchestration.check', {}, { sessionId: COORDINATOR })).toMatchObject({
-      count: 1,
-      messages: [{ id: mail!.id }]
-    })
-    expect(await queuedCardTexts()).toEqual([ptyPointer(`run:${runId}`)])
-    await endTurn()
-  })
-
-  // The queue once refused every send past twenty cards, so a busy coordinator's mail locked the
-  // person out of their own chat.
-  it("still queues the person's message behind more than twenty mail cards", async () => {
-    const chat = await openChat(COORDINATOR)
-    const { runId, taskId } = await coordinatorRunAndTask()
-    await runningUserTurn(chat)
-    await finishWorker(taskId)
-    await vi.waitFor(async () => expect(await queuedCardTexts()).toHaveLength(1), WAIT)
-    for (let index = 1; index <= 24; index += 1) {
-      await call('orchestration.send', {
-        from: 'term_worker',
-        to: `run:${runId}`,
-        subject: `mail ${index}`
-      })
-      await vi.waitFor(async () => expect(await queuedCardTexts()).toHaveLength(index + 1), WAIT)
-    }
-    const sent = await host.send(
-      { callerKey: 'test-surface' },
-      {
-        ...structuredAgentSessionMessageSendMutation({
-          sessionId: COORDINATOR,
-          clientOperationId: operationId(),
-          expectedRuntimeFence: host.deps.store.getRecord(COORDINATOR)!.lease.runtimeFence,
-          body: {
-            kind: 'message',
-            role: 'user',
-            blocks: [{ type: 'text', text: 'human message' }]
-          },
-          delivery: 'queue-if-active'
-        }),
-        userSend: true
-      }
-    )
-    expect(sent).toMatchObject({ ok: true, value: { queued: { position: 26, state: 'waiting' } } })
-    expect((await queuedCardTexts()).at(-1)).toBe('human message')
-  })
-})

src/main/runtime/structured-chat-coordinator-mail.test.ts

@@ -410,10 +410,10 @@ describe('a worker result reaches the structured chat that coordinates it', () =
     }
   })
 
-  it("holds mail a refused turn left in doubt, then queues the next pointer behind it as the person's message would wait", async () => {
+  it('points mail a refused turn left in doubt again with the next result, counting both', async () => {
     // A failed turn/start cannot prove the turn never started, so the host records it `unknown`
-    // and a resend under its id replays that. A live doubt counts as work still owed, so the next
-    // result's pointer waits in the chat's queue, as a message the person sent then would.
+    // and a resend under its id replays that. The next result's pointer does not wait behind
+    // that doubt: it names both messages, and the chat's own `check` reads them.
     const chat = await openChat(COORDINATOR)
     const { runId, taskId } = await coordinatorRunAndTask()
     const second = await call(
@@ -429,14 +429,11 @@ describe('a worker result reaches the structured chat that coordinates it', () =
     expect(chat.turns).toHaveLength(0)
 
     await finishWorker(idOf(second.task), { handle: 'term_worker_2', paneKey: WORKER_2_PANE })
-    await vi.waitFor(
-      async () =>
-        expect(await queuedCardTexts()).toEqual([
-          formatMessagePointer(2, `run:${runId}`, localOrchestrationCliCommand()).trim()
-        ]),
-      WAIT
+    await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT)
+    expect(turnText(chat.turns[0]!)).toBe(
+      formatMessagePointer(2, `run:${runId}`, localOrchestrationCliCommand()).trim()
     )
-    expect(chat.turns).toHaveLength(0)
+    expect(await queuedCardTexts()).toEqual([])
     expect(codex.connections.length).toBe(before)
   })
 
@@ -459,53 +456,45 @@ describe('a worker result reaches the structured chat that coordinates it', () =
     expect(await userTexts(COORDINATOR)).toEqual([expect.stringMatching(POINTER)])
   })
 
-  it('points mail at the idle edge when it arrived mid-turn', async () => {
+  it('steers mail that arrives mid-turn into the running turn, with who it is from', async () => {
     const chat = await openChat(COORDINATOR)
     const { runId, taskId } = await coordinatorRunAndTask()
-    const first = await host.send(
-      { callerKey: 'test-surface' },
-      {
-        envelope: {
-          sessionId: COORDINATOR,
-          clientOperationId: operationId(),
-          expectedRuntimeFence: host.deps.store.getRecord(COORDINATOR)!.lease.runtimeFence,
-          payloadFingerprint: computeAgentSessionPayloadFingerprint({
-            method: 'agentSession.send',
-            sessionId: COORDINATOR,
-            fields: {
-              body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'go' }] }
-            }
-          })
-        },
-        body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'go' }] }
-      }
-    )
-    expect(first).toMatchObject({ ok: true })
-    // Accepted, then delivered: the provider sees the turn once the host hands it over.
+    expect(await sendUserMessage(COORDINATOR, 'go')).toMatchObject({ ok: true })
     await vi.waitFor(() => expect(chat.turns).toHaveLength(1), WAIT)
     const notify = (method: string, params: unknown) =>
       chat.handlers.onNotification?.(method, params)
+    const echo = (id: string, clientId: string, text: string) =>
+      notify('item/completed', {
+        item: { type: 'userMessage', id, clientId, content: [{ type: 'text', text }] }
+      })
     notify('turn/started', { turn: { id: 'turn-1' } })
-    notify('item/completed', {
-      item: {
-        type: 'userMessage',
-        id: 'echo-go',
-        clientId: chat.turns[0]!.clientUserMessageId,
-        content: [{ type: 'text', text: 'go' }]
-      }
-    })
+    echo('echo-go', chat.turns[0]!.clientUserMessageId, 'go')
     await host.flushStreamedEvents(COORDINATOR)
 
     await finishWorker(taskId)
-    await new Promise((resolve) => setTimeout(resolve, 20))
-    // The coordinator is mid-turn, so nothing is folded into that turn.
-    expect(chat.turns).toHaveLength(1)
 
+    // Into the turn that is running, as a terminal agent takes its typed nudge: no card, no
+    // second turn.
+    await vi.waitFor(() => expect(chat.steers).toHaveLength(1), WAIT)
+    const [steer] = chat.steers
+    expect(steer!.expectedTurnId).toBe('turn-1')
+    expect(steer!.text).toContain(ptyPointer(`run:${runId}`))
+    expect(await queuedCardTexts()).toEqual([])
+    const sent = (await host.journalSnapshot(COORDINATOR)).items.find(
+      (item) => item.body.kind === 'message' && item.body.from
+    )
+    expect(sent?.body).toMatchObject({
+      from: { kind: 'agent', orchestration: { message: 'mail-notice', mailbox: `run:${runId}` } }
+    })
+
+    echo('echo-pointer', steer!.clientUserMessageId, ptyPointer(`run:${runId}`))
     notify('turn/completed', { turn: { id: 'turn-1' } })
     await host.flushStreamedEvents(COORDINATOR)
-    await vi.waitFor(() => expect(chat.turns).toHaveLength(2), WAIT)
-    expect(chat.turns[1]!.text).toMatch(POINTER)
-    expect(chat.turns[1]!.text).toContain(runId)
+    // The idle edge finds the mail delivered: nothing is resent.
+    runtime.onStructuredSessionStatusForMail({ sessionId: COORDINATOR, status: 'idle' })
+    await new Promise((resolve) => setTimeout(resolve, 200))
+    expect(chat.turns).toHaveLength(1)
+    expect(chat.steers).toHaveLength(1)
   })
 })
 

src/renderer/src/components/native-chat/structured-agent-session-send-runtime.ts

@@ -104,7 +104,9 @@ export function settleRecorded(
   finish(
     entry,
     'recorded',
-    keep ? { phase: 'recorded', issued: true, ...(cursor ? { recordedCursor: cursor } : {}) } : undefined
+    keep
+      ? { phase: 'recorded', issued: true, ...(cursor ? { recordedCursor: cursor } : {}) }
+      : undefined
   )
 }
 

src/renderer/src/components/right-sidebar/AiVaultScanProgressStrip.test.tsx

@@ -35,9 +35,7 @@ const FINISHED: AiVaultScanProgress = {
 
 it('says the finished total in the strip and opens to one row per agent', async () => {
   render(<AiVaultScanProgressStrip progress={FINISHED} />)
-  expect(screen.getByText('102,452 files').parentElement?.textContent).toBe(
-    'Scanned 102,452 files'
-  )
+  expect(screen.getByText('102,452 files').parentElement?.textContent).toBe('Scanned 102,452 files')
   expect(screen.queryByText('Codex')).toBeNull()
 
   await userEvent.setup().click(screen.getByRole('button'))
@@ -120,7 +118,12 @@ it('shows a running scan at once but a finished one only with the list it produc
 
   act(() => push({ scanKey: 'scope-a', progress: FINISHED }))
   expect(aView.getByText('Scanned', { exact: false })).toBeTruthy()
-  act(() => push({ scanKey: 'scope-b', progress: { ...FINISHED, scanId: 'scan-b', stage: 'finding', settledAt: null } }))
+  act(() =>
+    push({
+      scanKey: 'scope-b',
+      progress: { ...FINISHED, scanId: 'scan-b', stage: 'finding', settledAt: null }
+    })
+  )
   expect(bView.getByText('Scanning', { exact: false })).toBeTruthy()
   expect(aView.getByText('Scanned', { exact: false })).toBeTruthy()
 

src/shared/structured-agent-session-send-settlement.ts

@@ -8,9 +8,9 @@ export function structuredAgentSessionReadCoversSend(
 ): boolean {
   return Boolean(
     cursor &&
-      acknowledged &&
-      cursor.epoch === acknowledged.epoch &&
-      cursor.sequence >= acknowledged.sequence
+    acknowledged &&
+    cursor.epoch === acknowledged.epoch &&
+    cursor.sequence >= acknowledged.sequence
   )
 }