Keep orchestration notices flowing after an unacknowledged mail batch

config/scripts/vitest-sqlite-database-runtime-files.mjs

@@ -37,8 +37,15 @@ export const SQLITE_DATABASE_RUNTIME_INCLUDE = [
   'src/main/runtime/orchestration-cli-subprocess.test.ts',
   'src/main/runtime/orchestration-dispatch-mailbox-delivery.test.ts',
   'src/main/runtime/orchestration-mailbox-crash-recovery.test.ts',
+  'src/main/runtime/orchestration-mailbox-cold-park-idle.test.ts',
   'src/main/runtime/orchestration-mailbox-notification-consistency.test.ts',
+  'src/main/runtime/orchestration-mailbox-transport-settlement.test.ts',
+  'src/main/runtime/orchestration-mailbox-unacked-notice.test.ts',
   'src/main/runtime/orchestration-message-delivery-identity.test.ts',
+  'src/main/runtime/structured-chat-worker-unacked-notice.test.ts',
+  'src/main/runtime/structured-chat-coordinator-mail-queue.test.ts',
+  'src/main/runtime/structured-chat-coordinator-mail.test.ts',
+  'src/main/runtime/structured-chat-mail-queue-recovery.test.ts',
   'src/main/runtime/orchestration-messages-fake-parity.test.ts',
   'src/main/runtime/orchestration/agent-facing-parity.test.ts',
   'src/main/runtime/orchestration/coordinator-decision-gates.test.ts',
@@ -68,6 +75,7 @@ export const SQLITE_DATABASE_RUNTIME_INCLUDE = [
   'src/main/runtime/orchestration/db/hot-path-statement-compilation.test.ts',
   'src/main/runtime/orchestration/db/lifecycle-transition.test.ts',
   'src/main/runtime/orchestration/db/messages/mailbox-consumer-lifecycle-fencing.test.ts',
+  'src/main/runtime/orchestration/db/messages/replay-waiting-count.test.ts',
   'src/main/runtime/orchestration/db/remaining-dispatch-statement-compilation.test.ts',
   'src/main/runtime/orchestration/db/row-column-lists.test.ts',
   'src/main/runtime/orchestration/db/runs/run-coordinator-orca-session-binding.test.ts',
@@ -90,6 +98,7 @@ export const SQLITE_DATABASE_RUNTIME_INCLUDE = [
   'src/main/runtime/orchestration/mailbox-pointer-eligibility.test.ts',
   'src/main/runtime/orchestration/mailbox-pointer-release-query.test.ts',
   'src/main/runtime/orchestration/mailbox-pointer-stage.test.ts',
+  'src/main/runtime/orchestration/mailbox-pointer-partial-fetch.test.ts',
   'src/main/runtime/orchestration/mailbox-pointer-submit.test.ts',
   'src/main/runtime/orchestration/message-batch-atomicity.test.ts',
   'src/main/runtime/orchestration/mutation-receipt-capacity.test.ts',
@@ -143,6 +152,10 @@ export const SQLITE_DATABASE_RUNTIME_INCLUDE = [
   'src/main/runtime/rpc/methods/orchestration/federation/federation-worker-brief-agent-in-front.test.ts',
   'src/main/runtime/rpc/methods/orchestration/federation/federation.test.ts',
   'src/main/runtime/rpc/methods/orchestration/messaging/check-worker-federated-attachment.test.ts',
+  'src/main/runtime/rpc/methods/orchestration/messaging/check-replay-recovery.test.ts',
+  'src/main/runtime/rpc/methods/orchestration/messaging/check-worker-consumer-fencing.test.ts',
+  'src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-interleavings.test.ts',
+  'src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-safety.test.ts',
   'src/main/runtime/rpc/methods/orchestration/runs/migration-behavior.test.ts',
   'src/main/runtime/rpc/methods/orchestration/worker/agent-status-producer-census.test.ts',
   'src/main/runtime/rpc/methods/orchestration/worker/manual-dispatch-observation.test.ts',

src/cli/handlers/orchestration-check-replay-recovery.test.ts

@@ -0,0 +1,131 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type { HandlerContext } from '../dispatch'
+import { RuntimeClient } from '../runtime-client'
+import type { OrchestrationCheckOutput } from '../../shared/orchestration-check-output'
+
+const callMock = vi.hoisted(() => vi.fn())
+const printResultMock = vi.hoisted(() => vi.fn())
+vi.mock('../format', () => ({ printResult: printResultMock }))
+
+import { ORCHESTRATION_HANDLERS } from './orchestration'
+
+describe('CLI replay recovery across host versions', () => {
+  beforeEach(() => {
+    callMock.mockReset()
+    printResultMock.mockReset()
+    vi.stubEnv('ORCA_CLI_COMMAND', 'orca-ide')
+  })
+  afterEach(() => {
+    vi.unstubAllEnvs()
+    vi.restoreAllMocks()
+  })
+
+  function invoke(flags: [string, string | boolean][], json: boolean) {
+    const client = new RuntimeClient('/folder-workspace/.test-orca', 60_000, null, null, 'orca-ide')
+    vi.spyOn(client, 'call').mockImplementation(callMock)
+    const context: HandlerContext = {
+      flags: new Map([['terminal', 'term_original'], ...flags]),
+      client,
+      cwd: '/folder-workspace',
+      json
+    }
+    return ORCHESTRATION_HANDLERS['orchestration check'](context)
+  }
+  function printed(): OrchestrationCheckOutput {
+    const response: unknown = printResultMock.mock.calls[0]?.[0]
+    if (
+      typeof response === 'object' &&
+      response &&
+      'result' in response &&
+      typeof response.result === 'object' &&
+      response.result &&
+      'messages' in response.result &&
+      'count' in response.result
+    ) {
+      // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The mocked response is an orchestration check result and was prepared by the real handler.
+      return response.result as OrchestrationCheckOutput
+    }
+    throw new Error('Expected printed check receipt')
+  }
+  function oldHostResult() {
+    return {
+      runId: 'run_receipt_owner',
+      messages: [{ id: 'msg_old', from_handle: 'worker' }],
+      count: 1,
+      deliveryId: 'delivery_old',
+      replayed: true
+    }
+  }
+
+  it.each([false, true])(
+    'projects old-host recovery into actual JSON and text: json=%s',
+    async (json) => {
+      callMock.mockResolvedValue({ result: oldHostResult() })
+      await invoke([], json)
+      const result = printed()
+      expect(result.replayRecovery).toMatchObject({
+        ackCommand: 'orca-ide orchestration check --terminal term_original --ack delivery_old',
+        guidance:
+          'Process every message in this batch before acknowledging, then process the next batch returned.'
+      })
+      expect(result.replayRecovery).not.toHaveProperty('waitingCount')
+      expect(JSON.parse(JSON.stringify(result))).toHaveProperty('replayRecovery.ackCommand')
+      const formatter = printResultMock.mock.calls[0]?.[2]
+      expect(formatter?.(result)).toContain('Delivery delivery_old (replayed)')
+      expect(formatter?.(result)).not.toContain('0 unread messages')
+      expect(formatter?.(result)).not.toContain('--run run_receipt_owner')
+      expect(callMock).toHaveBeenCalledTimes(1)
+    }
+  )
+
+  it('preserves explicit Run selection while omitting unavailable backlog counts', async () => {
+    callMock.mockResolvedValue({ result: oldHostResult() })
+    await invoke(
+      [
+        ['run', 'run_selected'],
+        ['format', true]
+      ],
+      true
+    )
+    expect(printed().replayRecovery?.ackCommand).toBe(
+      'orca-ide orchestration check --terminal term_original --run run_selected --ack delivery_old'
+    )
+  })
+
+  it('keeps new-host recovery count and guidance in JSON and formatted output', async () => {
+    callMock.mockResolvedValue({
+      result: {
+        ...oldHostResult(),
+        formatted: 'Message body',
+        replayRecovery: {
+          waitingCount: 83,
+          ackCommand: 'orca-dev orchestration check --terminal term_original --ack delivery_old',
+          guidance:
+            'Process every message in this batch before acknowledging, then process the next batch returned.'
+        }
+      }
+    })
+    await invoke([['format', true]], true)
+    expect(printed().replayRecovery?.waitingCount).toBe(83)
+    expect(printResultMock.mock.calls[0]?.[2]?.(printed())).toContain(
+      '83 unread messages waiting behind this Delivery.'
+    )
+  })
+
+  it.each([
+    ['peek', true],
+    ['all', true]
+  ] as const)('omits fallback for inspection %s', async (flag, value) => {
+    callMock.mockResolvedValue({ result: oldHostResult() })
+    await invoke([[flag, value]], true)
+    expect(printed()).not.toHaveProperty('replayRecovery')
+  })
+
+  it('omits fallback for legacy recovery reads', async () => {
+    callMock.mockResolvedValue({
+      result: { ...oldHostResult(), legacyCompatibility: { readOnly: true, recovery: true } }
+    })
+    await invoke([], true)
+    expect(printed()).not.toHaveProperty('replayRecovery')
+  })
+})

src/cli/handlers/orchestration/message-check-handler.ts

@@ -6,6 +6,7 @@ import type { RuntimeRpcSuccess } from '../../runtime-client'
 import {
   formatOrchestrationCheckText,
   prepareOrchestrationCheckOutput,
+  type OrchestrationReplayRecovery,
   type LegacyCompatibilityResult,
   type OrchestrationMessageSummary as MessageSummary
 } from '../../../shared/orchestration-check-output'
@@ -20,6 +21,8 @@ type CheckResult = {
   count: number
   formatted?: string
   deliveryId?: string | null
+  replayed?: boolean
+  replayRecovery?: OrchestrationReplayRecovery
   runId?: string
   timedOut?: boolean
   cancelled?: boolean
@@ -72,7 +75,12 @@ export const ORCHESTRATION_CHECK_HANDLER: Record<string, CommandHandler> = {
     }
     result = {
       ...result,
-      result: prepareOrchestrationCheckOutput(result.result, callerLabel, flags.has('format'))
+      result: prepareOrchestrationCheckOutput(result.result, callerLabel, flags.has('format'), {
+        cliCommand: resolveCompatibilityCliCommand(),
+        terminal,
+        run: getOptionalStringFlag(flags, 'run'),
+        consuming: !peek && !flags.has('all')
+      })
     }
     printResult(result, json, (value) => formatOrchestrationCheckText(value, callerLabel))
     const compatibilityAck = result.result.legacyCompatibility?.ackMessageIds

src/main/native-chat/agent-session-wire/structured-agent-session-host.ts

@@ -331,6 +331,12 @@ export class StructuredAgentSessionHost {
   journalSnapshot = async (sessionId: string): Promise<AgentJournalSnapshot> =>
     (await this.lifetime.conversation(sessionId)).journal.snapshot()
 
+  /** Includes settled queue receipts, which the published card list omits. */
+  mailboxPointerSnapshot = async (sessionId: string) => {
+    const { journal } = await this.lifetime.conversation(sessionId)
+    return { snapshot: journal.snapshot(), queuedMessages: journal.queuedMessages.list() }
+  }
+
   subscribe = (input: AgentSessionSubscribeInput) => this.backgroundTasks.subscribe(input)
 
   settleLateDispatch = (input: Parameters<typeof settleStructuredAgentSessionLateDispatch>[1]) =>

src/main/runtime/orchestration-dispatch-mailbox-delivery.test.ts

@@ -88,7 +88,7 @@ describe('Dispatch mailbox Delivery', () => {
     const restarted = createRuntime(restartedDb)
     await driveToLiveIdle(restarted.runtime)
     await vi.advanceTimersByTimeAsync(2_500)
-    expect(pointerCount(restarted.write)).toBe(0)
+    expect(pointerCount(restarted.write)).toBe(1)
 
     const replayed = await checkBoundMailbox(restarted.runtime)
     expect(replayed).toMatchObject({

src/main/runtime/orchestration-mailbox-notification-consistency.test.ts

@@ -724,7 +724,7 @@ describe('orchestration notification mailbox consistency', () => {
       })
     )
     const notificationQuery = vi.spyOn(db, 'getUndeliveredUnreadMessages')
-    const submitRevalidation = vi.spyOn(db, 'areUnreadMessages')
+    const submitRevalidation = vi.spyOn(db, 'getMailboxPointerAttentionIds')
     const prepare = vi.spyOn(sqliteFor(db), 'prepare')
 
     await driveToLiveIdle(harness.runtime)
@@ -743,7 +743,7 @@ describe('orchestration notification mailbox consistency', () => {
     )
     const revalidationSql = prepare.mock.calls
       .map(([sql]) => sql)
-      .find((sql) => sql.includes('SELECT COUNT(*)') && sql.includes('id IN'))
+      .find((sql) => sql.includes('SELECT id FROM messages') && sql.includes('id IN'))
     expect(revalidationSql).toContain('INDEXED BY idx_messages_id')
     const revalidationPlan = sqliteFor(db)
       .prepare(`EXPLAIN QUERY PLAN ${revalidationSql}`)
@@ -833,7 +833,7 @@ describe('orchestration notification mailbox consistency', () => {
     db.close()
   })
 
-  it('does not point newer Run mail behind an outstanding Delivery', async () => {
+  it('points newer Run mail while preserving an outstanding Delivery replay', async () => {
     vi.useFakeTimers()
     const db = createDatabase('orca-mailbox-outstanding-')
     const harness = createRuntime(db)
@@ -855,7 +855,7 @@ describe('orchestration notification mailbox consistency', () => {
     await driveToLiveIdle(harness.runtime)
     const replayed = await checkBoundMailbox(harness.runtime)
 
-    expect(pointerCount(harness.write)).toBe(0)
+    expect(pointerCount(harness.write)).toBe(1)
     expect(replayed.messages).toEqual([expect.objectContaining({ id: firstMessage.id })])
     expect(replayed.deliveryId).toBe(firstDelivery.deliveryId)