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

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