feat(orchestration): a consuming check marks the batch it returns as handled

2026-11-13 09:27:29 +00:00 · 2026-10-28 00:12:37 -04:00
parent 884867d5ea
commit 8fbb395a32
22 changed files with 221 additions and 191 deletions
@@ -89,8 +89,8 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
    summary: "Check this agent's messages",
    usage:
      'orca orchestration check [--terminal <handle>] [--run <run_id>] [--ack <delivery_id>] [--unread | --peek | --all] [--types <type,...>] [--format] [--wait] [--timeout-ms <n>] [--retry-request <id>] [--json]\n' +
-      "  default: return the bound Run's oldest unacknowledged FIFO batch.\n" +
-      '  --ack: acknowledge the prior whole batch before checking/waiting.\n' +
+      "  default: return the bound Run's oldest unread FIFO batch and mark it handled.\n" +
+      '  --ack: optional; acknowledges a batch from an older Orca that does not mark it.\n' +
      '  --peek: return only unread messages without marking them read.\n' +
      '  --all: return every message for the handle; does not mark read.\n' +
      '  --wait: block until a matching message arrives or --timeout-ms expires.\n' +
@@ -117,7 +117,7 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
      'On Windows PowerShell, quote comma-separated type filters, e.g. --types "worker_done,escalation".',
      '--types is the wake condition for --wait; a returned Delivery is always the whole FIFO batch, so it is never filtered by type. Without --wait it has no effect on consuming checks. Only --peek and --all filter their rows.',
      '--format renders the returned rows as local text only; it never writes to another terminal.',
-      'A bound Run replays the same Delivery until --ack or all its messages are marked read, even with --types; process every message before acknowledging. Several --wait calls may wait on one Run inbox; waiting reserves nothing, listeners woken together get the same Delivery, and a repeated --ack is a no-op.'
+      'A consuming check marks the batch it returns as handled, so the next check returns newer mail; no --ack is needed. If the response was lost, rerun with the --retry-request it printed to see the same batch, or use --all to see every message. Several --wait calls may wait on one inbox; waiting reserves nothing, listeners woken together get the same Delivery, and --ack of a handled Delivery is a no-op.'
    ]
  },
  {
@@ -172,14 +172,16 @@ it("retains CLI work and mail across restart, refuses a worker's own Run, and co
    expect(await cli(coord, ['task-list'])).toMatchObject({
      result: { tasks: [{ id: task, status: 'dispatched', dispatch_id: dispatch }] }
    })
-    expect(check.parse(await cli(coord, ['check'])).result).toEqual(retainedBatch)
+    expect(await cli(coord, ['check'])).toMatchObject({ result: { count: 0, deliveryId: null } })
+    expect(await cli(coord, ['check', '--all'])).toMatchObject({
+      result: { runId: root, messages: [{ id: retainedMessage }] }
+    })

    const bindingBefore = db.getRunRaw(root)
    expect(await cli(worker, ['run-use', '--id', root], 1)).toMatchObject({
      error: { code: 'worker_cannot_coordinate_own_run', data: { effectsApplied: false } }
    })
    expect(db.getRunRaw(root)).toEqual(bindingBefore)
-    expect(check.parse(await cli(coord, ['check'])).result).toEqual(retainedBatch)
    expect(await cli(coord, ['check', '--ack', retainedBatch.deliveryId])).toMatchObject({
      result: { acknowledged: retainedBatch.deliveryId, count: 0 }
    })
@@ -221,11 +223,10 @@ it("retains CLI work and mail across restart, refuses a worker's own Run, and co
        }
      ]
    })
-    expect(db.getMessageById(reply)?.read).toBe(0)
+    expect(db.getMessageById(reply)?.read).toBe(1)
    expect(await cli(worker, ['check', '--ack', replyBatch.deliveryId])).toMatchObject({
      result: { runId: separate, acknowledged: replyBatch.deliveryId, count: 0 }
    })
-    expect(db.getMessageById(reply)?.read).toBe(1)
    await cli(coord, ['check', '--ack', questionBatch.deliveryId])

    const continuedDispatch = idOf(
@@ -27,7 +27,7 @@ describe('Dispatch mailbox Delivery', () => {
    }
  })

-  it('wakes once, replays after restart, and leaves concurrent guidance for the next ack', async () => {
+  it('wakes once and hands newer guidance after restart to the next check without an ack', async () => {
    vi.useFakeTimers()
    const directory = mkdtempSync(join(tmpdir(), 'orca-dispatch-delivery-'))
    temporaryDirectories.push(directory)
@@ -68,7 +68,8 @@ describe('Dispatch mailbox Delivery', () => {
    const issued = await checkBoundMailbox(first.runtime)
    expect(issued).toMatchObject({ dispatchId: dispatch.id, count: 1, replayed: false })
    expect(issued.messages).toEqual([expect.objectContaining({ id: firstMessage.id })])
-    expect(firstDb.getWorkerAttentionFacts(dispatch.id, Date.now()).pendingGuidance).toBe(true)
+    expect(firstDb.getMessageById(firstMessage.id)?.read).toBe(1)
+    expect(firstDb.getWorkerAttentionFacts(dispatch.id, Date.now()).pendingGuidance).toBe(false)
    firstDb.insertMessage({
      from: run.coordinator_handle!,
      to: address,
@@ -81,28 +82,20 @@ describe('Dispatch mailbox Delivery', () => {
    const restarted = createRuntime(restartedDb)
    await driveToLiveIdle(restarted.runtime)
    await vi.advanceTimersByTimeAsync(2_500)
-    // The held batch defers the concurrent mail's nudge, and that nudge names the batch to ack.
-    expect(
-      restarted.write.mock.calls.filter(([, payload]) =>
-        String(payload).includes(`--ack ${issued.deliveryId}`)
-      )
-    ).toHaveLength(1)
    expect(pointerCount(restarted.write)).toBe(1)
+    expect(
+      restarted.write.mock.calls.filter(([, payload]) => String(payload).includes('--ack'))
+    ).toHaveLength(0)

-    const replayed = await checkBoundMailbox(restarted.runtime)
-    expect(replayed).toMatchObject({
-      dispatchId: dispatch.id,
-      deliveryId: issued.deliveryId,
-      count: 1,
-      replayed: true
-    })
-    const next = await checkBoundMailbox(restarted.runtime, { ack: replayed.deliveryId! })
+    const next = await checkBoundMailbox(restarted.runtime)
+    expect(next).toMatchObject({ dispatchId: dispatch.id, count: 1, replayed: false })
    expect(next.messages).toEqual([expect.objectContaining({ subject: 'Concurrent follow-up' })])
-    expect(restartedDb.getMessageById(firstMessage.id)?.read).toBe(1)
-    expect(restartedDb.getWorkerAttentionFacts(dispatch.id, Date.now()).pendingGuidance).toBe(true)
-
-    await checkBoundMailbox(restarted.runtime, { ack: next.deliveryId! })
    expect(restartedDb.getWorkerAttentionFacts(dispatch.id, Date.now()).pendingGuidance).toBe(false)
+    // An older CLI's explicit ack of an already handled batch stays a harmless no-op.
+    expect(await checkBoundMailbox(restarted.runtime, { ack: issued.deliveryId! })).toMatchObject({
+      acknowledged: issued.deliveryId,
+      count: 0
+    })
    restartedDb.close()
  })

@@ -159,10 +152,6 @@ describe('Dispatch mailbox Delivery', () => {
      )
      const delivery = await checkBoundMailbox(restarted.runtime)
      expect(delivery.messages).toEqual([expect.objectContaining({ id: message.id })])
-      expect(restartedDb.getWorkerAttentionFacts(dispatch.id, Date.now()).pendingGuidance).toBe(
-        true
-      )
-      await checkBoundMailbox(restarted.runtime, { ack: delivery.deliveryId! })
      expect(restartedDb.getWorkerAttentionFacts(dispatch.id, Date.now()).pendingGuidance).toBe(
        false
      )
@@ -55,11 +55,9 @@ describe('orchestration detached mailbox routing', () => {
    expect(checked.messages).toEqual([expect.objectContaining({ id: message.id })])
    expect(db.getMessageById(message.id)).toMatchObject({
      to_handle: `dispatch:${dispatch.id}`,
-      read: 0,
+      read: 1,
      delivered_at: expect.any(String)
    })
-    await checkBoundMailbox(harness.runtime, { ack: checked.deliveryId! })
-    expect(db.getMessageById(message.id)?.read).toBe(1)
    db.close()
  })

@@ -32,7 +32,6 @@ import {
  WORKTREE_ID
 } from './orchestration-mailbox-notification-test-harness'
 import { RpcDispatcher } from './rpc/dispatcher'
-import { localOrchestrationCliCommand } from './orchestration/cli-command'
 import { ORCHESTRATION_METHODS } from './rpc/methods/orchestration'
 import { MAILBOX_POINTER_WRITE_ATTEMPTED } from './orchestration/db/messages/mailbox-pointer-enter-state'

@@ -762,7 +761,7 @@ describe('orchestration notification mailbox consistency', () => {
    )
    expect(db.getMessageById(messages[0]!.id)).toMatchObject({
      to_handle: `run:${run.id}`,
-      read: 0,
+      read: 1,
      delivered_at: expect.any(String)
    })
    expect(db.getMessageById(messages[50]!.id)?.delivered_at).toBeNull()
@@ -837,47 +836,25 @@ describe('orchestration notification mailbox consistency', () => {
    db.close()
  })

-  it('points newer Run mail behind an unacked Delivery once, naming the Delivery', async () => {
+  it('points newer Run mail with the normal text when the coordinator never acked', async () => {
    vi.useFakeTimers()
    const db = new OrchestrationDb(':memory:')
    const harness = createRuntime(db)
-    const run = createBoundRun(db, 'Outstanding Delivery Run')
+    const run = createBoundRun(db, 'Never-acked Run')
    const insert = (subject: string) =>
      db.insertMessage({ from: 'term_worker', to: `run:${run.id}`, subject, runId: run.id })
    const firstMessage = insert('First status')
    const firstDelivery = await checkBoundMailbox(harness.runtime)
+    expect(firstDelivery.messages).toEqual([expect.objectContaining({ id: firstMessage.id })])
    const newerMessage = insert('Newer status')

-    const enters = () => harness.write.mock.calls.filter(([, payload]) => payload === '\r').length
    await driveToLiveIdle(harness.runtime)
    const pointers = harness.write.mock.calls.filter(([, payload]) => isMailboxPointer(payload))
    expect(pointers).toHaveLength(1)
-    expect(String(pointers[0]![1])).toContain(
-      `after delivery ${firstDelivery.deliveryId}, which you have not acknowledged. If you no longer have it, run \`${localOrchestrationCliCommand()} orchestration check --run ${run.id}\` to see it again; once it is handled, run \`${localOrchestrationCliCommand()} orchestration check --run ${run.id} --ack ${firstDelivery.deliveryId}\``
-    )
+    expect(String(pointers[0]![1])).toContain('You have 1 orchestration message.')
+    expect(String(pointers[0]![1])).not.toContain('acknowledged')

-    // A turn that starts inside the Enter delay keeps the reminder until it ends.
-    harness.runtime.onPtyData(PTY_ID, '\x1b]0;Codex working\x07', 3)
-    await harness.runtime.serializeMainTerminalBuffer(PTY_ID)
-    await vi.advanceTimersByTimeAsync(500)
-    expect(enters()).toBe(0)
-    harness.runtime.onPtyData(PTY_ID, '\x1b]0;Codex done\x07', 4)
-    await harness.runtime.serializeMainTerminalBuffer(PTY_ID)
-    await vi.advanceTimersByTimeAsync(500)
-    expect(enters()).toBe(1)
-    expect(db.getMessageById(newerMessage.id)?.delivered_at).toEqual(expect.any(String))
-
-    harness.runtime.onPtyData(PTY_ID, '\x1b]0;Codex working\x07', 5)
-    harness.runtime.onPtyData(PTY_ID, '\x1b]0;Codex done\x07', 6)
-    await harness.runtime.serializeMainTerminalBuffer(PTY_ID)
-    await vi.advanceTimersByTimeAsync(500)
-    expect(pointerCount(harness.write)).toBe(1)
-    expect(enters()).toBe(1)
-
-    const replayed = await checkBoundMailbox(harness.runtime)
-    expect(replayed.messages).toEqual([expect.objectContaining({ id: firstMessage.id })])
-    expect(replayed.deliveryId).toBe(firstDelivery.deliveryId)
-    const next = await checkBoundMailbox(harness.runtime, { ack: firstDelivery.deliveryId! })
+    const next = await checkBoundMailbox(harness.runtime)
    expect(next.messages).toEqual([expect.objectContaining({ id: newerMessage.id })])
    expect(next.deliveryId).not.toBe(firstDelivery.deliveryId)
    db.close()
@@ -218,38 +218,33 @@ describe('STA-4325 message and delivery identity', () => {
    })

    await driveToLiveIdle(runtime)
-    const [first, concurrent] = await Promise.all([check(runtime), check(runtime)])
    const expectedIds = [status.id, dispatch.id, done.id]
-
-    expect(pointerPayloads(write)).toEqual([expect.stringContaining('3 orchestration messages')])
-    expect(first).toMatchObject({ runId: run.id, count: 3, replayed: false })
-    expect(concurrent).toMatchObject({ runId: run.id, count: 3, replayed: true })
-    expect(first.deliveryId).toBeTruthy()
-    expect(concurrent.deliveryId).toBe(first.deliveryId)
-    expect(first.messages.map((message) => message.id)).toEqual(expectedIds)
-    expect(concurrent.messages.map((message) => message.id)).toEqual(expectedIds)
-
    const peeked = await check(runtime, { peek: true, unread: false })
    const filtered = await check(runtime, { peek: true, unread: false, types: 'worker_done' })
-    const all = await check(runtime, { all: true })
    expect(peeked.messages.map((message) => message.id)).toEqual(
      expect.arrayContaining(expectedIds)
    )
    expect(peeked.count).toBe(3)
    expect(filtered.messages.map((message) => message.id)).toEqual([done.id])
-    expect(all.messages.map((message) => message.id)).toEqual(expect.arrayContaining(expectedIds))
-    expect(all.count).toBe(3)
-
-    const deliveryRowsBeforeAck = sqliteFor(db)
-      .prepare('SELECT id, status, message_ids FROM deliveries ORDER BY rowid')
-      .all() as { id: string; status: string; message_ids: string }[]
-    expect(deliveryRowsBeforeAck).toEqual([
-      { id: first.deliveryId, status: 'outstanding', message_ids: JSON.stringify(expectedIds) }
-    ])
    for (const id of expectedIds) {
      expect(db.getMessageById(id)).toMatchObject({ to_handle: `run:${run.id}`, read: 0 })
    }

+    const [first, concurrent] = await Promise.all([check(runtime), check(runtime)])
+    expect(pointerPayloads(write)).toEqual([expect.stringContaining('3 orchestration messages')])
+    expect(first).toMatchObject({ runId: run.id, count: 3, replayed: false })
+    expect(concurrent).toMatchObject({ runId: run.id, count: 0, deliveryId: null })
+    expect(first.deliveryId).toBeTruthy()
+    expect(first.messages.map((message) => message.id)).toEqual(expectedIds)
+    const all = await check(runtime, { all: true })
+    expect(all.messages.map((message) => message.id)).toEqual(expect.arrayContaining(expectedIds))
+    expect(all.count).toBe(3)
+    expect(
+      sqliteFor(db).prepare('SELECT id, status, message_ids FROM deliveries ORDER BY rowid').all()
+    ).toEqual([
+      { id: first.deliveryId, status: 'acknowledged', message_ids: JSON.stringify(expectedIds) }
+    ])
+
    const acknowledged = await check(runtime, { ack: first.deliveryId })
    expect(acknowledged).toMatchObject({
      count: 0,
@@ -271,7 +266,7 @@ describe('STA-4325 message and delivery identity', () => {
    db.close()
  })

-  it('replays one outstanding Delivery across restart and wakes a filtered waiter once', async () => {
+  it('keeps a handled Delivery handled across restart and wakes a filtered waiter once', async () => {
    vi.useFakeTimers()
    const fixture = createDatabase('orca-sta-4325-restart-')
    const firstRuntime = createRuntime(fixture.db)
@@ -297,16 +292,11 @@ describe('STA-4325 message and delivery identity', () => {
    const restarted = createRuntime(reopened, REMINTED_TERMINAL_HANDLE)
    await driveToLiveIdle(restarted.runtime)
    const afterRestart = await check(restarted.runtime, { terminal: REMINTED_TERMINAL_HANDLE })
-    expect(afterRestart).toMatchObject({ deliveryId: beforeRestart.deliveryId, replayed: true })
-    expect(afterRestart.messages.map((message) => message.id)).toEqual([status.id])
+    expect(afterRestart).toMatchObject({ deliveryId: null, count: 0 })
    expect(sqliteFor(reopened).prepare('SELECT id FROM deliveries').all()).toEqual([
      { id: beforeRestart.deliveryId }
    ])

-    await check(restarted.runtime, {
-      terminal: REMINTED_TERMINAL_HANDLE,
-      ack: afterRestart.deliveryId
-    })
    const waiting = check(restarted.runtime, {
      terminal: REMINTED_TERMINAL_HANDLE,
      wait: true,
@@ -64,9 +64,9 @@ Don't post to Slack, GitHub, or other channels during the run; report through th
  # you send worker_done, so a redirect lands before the task settles.
  orca orchestration check --terminal orca_session_id:WORKER --json

-  # After processing every message, ack the returned deliveryId; this also reads
-  # the next batch. Repeat until count is 0, then resume. Unacked batches replay
-  # and hide newer mail.
+  # Each check marks the batch it returns as handled; repeat until count is 0,
+  # then resume. Only if a batch lacks "deliveryAcknowledged": true (an older
+  # Orca), ack its deliveryId so it does not replay and hide newer mail:
  orca orchestration check --terminal orca_session_id:WORKER --ack <delivery_id> --json
 \`\`\`

@@ -154,9 +154,9 @@ Don't post to Slack, GitHub, or other channels during the run; report through th
  # you send worker_done, so a redirect lands before the task settles.
  orca orchestration check --terminal term_WORKER --json

-  # After processing every message, ack the returned deliveryId; this also reads
-  # the next batch. Repeat until count is 0, then resume. Unacked batches replay
-  # and hide newer mail.
+  # Each check marks the batch it returns as handled; repeat until count is 0,
+  # then resume. Only if a batch lacks "deliveryAcknowledged": true (an older
+  # Orca), ack its deliveryId so it does not replay and hide newer mail:
  orca orchestration check --terminal term_WORKER --ack <delivery_id> --json
 \`\`\`

@@ -173,6 +173,54 @@ export function acknowledgeMailboxDelivery(
  }
 }

+// A consuming read hands over the batch it returns; a lost response replays from its request receipt.
+export function takeMailboxDelivery(
+  this: OrchestrationDb,
+  params: Parameters<OrchestrationDb['getOrCreateMailboxDelivery']>[0]
+): { delivery: DeliveryRow; messages: MessageRow[]; replayed: boolean } | undefined {
+  const current = this.getOrCreateMailboxDelivery(params)
+  if (!current) {
+    return undefined
+  }
+  const { delivery } = this.acknowledgeMailboxDelivery({
+    ...params,
+    deliveryId: current.delivery.id
+  })
+  return { ...current, delivery }
+}
+
+// Global rather than per-mailbox so reading it is a rowid lookup, not a table scan.
+export function getDeliveryCursor(this: OrchestrationDb): number {
+  const row = this.db.prepare('SELECT MAX(rowid) AS cursor FROM deliveries').get() as {
+    cursor: number | null
+  }
+  return row.cursor ?? 0
+}
+
+// Waiters woken by the same arrival share its batch, so a forgotten waiter cannot swallow it.
+export function getMailboxDeliveryTakenAfter(
+  this: OrchestrationDb,
+  mailboxHandle: string,
+  consumerGeneration: number,
+  cursor: number
+): { delivery: DeliveryRow; messages: MessageRow[]; replayed: boolean } | undefined {
+  const delivery = this.db
+    .prepare(
+      `SELECT * FROM deliveries
+       WHERE rowid > ? AND mailbox_handle = ? AND consumer_generation = ?
+         AND status = 'acknowledged'
+       ORDER BY rowid ASC LIMIT 1`
+    )
+    .get(cursor, mailboxHandle, consumerGeneration) as DeliveryRow | undefined
+  return delivery
+    ? {
+        delivery: exposeDeliveryTimestamps(delivery),
+        messages: this.getDeliveryMessages(delivery),
+        replayed: true
+      }
+    : undefined
+}
+
 export function hasOutstandingMailboxDelivery(
  this: OrchestrationDb,
  mailboxHandle: string
@@ -216,6 +264,9 @@ export type RoleMailboxDeliveryMethods = {
  getDeliveryMessages: typeof getDeliveryMessages
  getOrCreateMailboxDelivery: typeof getOrCreateMailboxDelivery
  acknowledgeMailboxDelivery: typeof acknowledgeMailboxDelivery
+  takeMailboxDelivery: typeof takeMailboxDelivery
+  getDeliveryCursor: typeof getDeliveryCursor
+  getMailboxDeliveryTakenAfter: typeof getMailboxDeliveryTakenAfter
  hasOutstandingMailboxDelivery: typeof hasOutstandingMailboxDelivery
  getOutstandingMailboxDelivery: typeof getOutstandingMailboxDelivery
  fenceUnacknowledgedMailboxDeliveries: typeof fenceUnacknowledgedMailboxDeliveries
@@ -227,6 +278,9 @@ export function attachRoleMailboxDelivery(ctor: { prototype: object }): void {
    getDeliveryMessages,
    getOrCreateMailboxDelivery,
    acknowledgeMailboxDelivery,
+    takeMailboxDelivery,
+    getDeliveryCursor,
+    getMailboxDeliveryTakenAfter,
    hasOutstandingMailboxDelivery,
    getOutstandingMailboxDelivery,
    fenceUnacknowledgedMailboxDeliveries
@@ -36,6 +36,22 @@ export function getOrCreateRunDelivery(
  })
 }

+export function takeRunDelivery(
+  this: OrchestrationDb,
+  params: {
+    runId: string
+    consumerGeneration: number
+    wakeTypes?: MessageType[]
+  }
+): { delivery: DeliveryRow; messages: MessageRow[]; replayed: boolean } | undefined {
+  return this.takeMailboxDelivery({
+    runId: params.runId,
+    mailboxHandle: `run:${params.runId}`,
+    consumerGeneration: params.consumerGeneration,
+    wakeTypes: params.wakeTypes
+  })
+}
+
 export function acknowledgeRunDelivery(
  this: OrchestrationDb,
  params: {
@@ -119,6 +135,7 @@ export function hasOutstandingRunDelivery(this: OrchestrationDb, runId: string):
 export type RunDeliveryMethods = {
  requireCurrentConsumer: typeof requireCurrentConsumer
  getOrCreateRunDelivery: typeof getOrCreateRunDelivery
+  takeRunDelivery: typeof takeRunDelivery
  acknowledgeRunDelivery: typeof acknowledgeRunDelivery
  getRunMailboxHistory: typeof getRunMailboxHistory
  getUnreadRunMailbox: typeof getUnreadRunMailbox
@@ -129,6 +146,7 @@ export function attachRunDelivery(ctor: { prototype: object }): void {
  Object.assign(ctor.prototype, {
    requireCurrentConsumer,
    getOrCreateRunDelivery,
+    takeRunDelivery,
    acknowledgeRunDelivery,
    getRunMailboxHistory,
    getUnreadRunMailbox,
@@ -127,9 +127,9 @@ ${heartbeat}  # Ask the coordinator a question and block until it answers.
  # you send worker_done, so a redirect lands before the task settles.
  ${cli} orchestration check --terminal ${params.workerHandle} --json

-  # After processing every message, ack the returned deliveryId; this also reads
-  # the next batch. Repeat until count is 0, then resume. Unacked batches replay
-  # and hide newer mail.
+  # Each check marks the batch it returns as handled; repeat until count is 0,
+  # then resume. Only if a batch lacks "deliveryAcknowledged": true (an older
+  # Orca), ack its deliveryId so it does not replay and hide newer mail:
  ${cli} orchestration check --terminal ${params.workerHandle} --ack <delivery_id> --json
 \`\`\`

@@ -133,7 +133,7 @@ export async function checkRunMailbox(args: {
  })
  const readPeek = () => db.getUnreadRunMailbox(run.id, 100, typeFilter)
  const readDelivery = (wakeTypes?: MessageType[]) =>
-    db.getOrCreateRunDelivery({ runId: run.id, consumerGeneration: generation, wakeTypes })
+    db.takeRunDelivery({ runId: run.id, consumerGeneration: generation, wakeTypes })
  let peeked = params.peek ? readPeek() : []
  if (params.peek && peeked.length > 0) {
    return peekResult(peeked)
@@ -146,6 +146,7 @@ export async function checkRunMailbox(args: {
      messages: exposeMessages(current.messages),
      count: current.messages.length,
      replayed: current.replayed,
+      deliveryAcknowledged: true,
      acknowledged: acknowledged ?? null,
      timedOut: false,
      cancelled: false,
@@ -171,6 +172,7 @@ export async function checkRunMailbox(args: {
    }
  }

+  const deliveryCursor = db.getDeliveryCursor()
  const waitResult = await runtime.waitForMessage(address, {
    typeFilter: typeFilter as string[] | undefined,
    timeoutMs: params.timeoutMs ?? undefined,
@@ -233,13 +235,15 @@ export async function checkRunMailbox(args: {
    peeked = readPeek()
    return { ...peekResult(peeked), timedOut: false, cancelled: false, connectionLost: false }
  }
-  current = readDelivery(typeFilter)
+  current =
+    readDelivery(typeFilter) ?? db.getMailboxDeliveryTakenAfter(address, generation, deliveryCursor)
  return {
    runId: run.id,
    deliveryId: current?.delivery.id ?? null,
    messages: exposeMessages(current?.messages ?? []),
    count: current?.messages.length ?? 0,
    replayed: current?.replayed ?? false,
+    ...(current ? { deliveryAcknowledged: true } : {}),
    acknowledged: acknowledged ?? null,
    timedOut: false,
    cancelled: false,
@@ -84,30 +84,23 @@ describe('orchestration.check on a re-attached Dispatch', () => {
      code: 'consumer_fenced',
      message: expect.stringContaining('no longer owns its Dispatch')
    })
-    expect(db.getUnreadMessages(`dispatch:${dispatchId}`)).toHaveLength(1)
  })

-  it('hands the live worker a fresh Delivery with the same unread mail', async () => {
+  it('leaves mail the replaced process already took in the live worker history', async () => {
    const dispatchId = attachedDispatchWithMail()
-    const staleDelivery = (await check(PANE_A)).deliveryId
+    await check(PANE_A)
    reattach(dispatchId)

-    const live = await check(PANE_B)
-    expect(live.deliveryId).not.toBe(staleDelivery)
-    expect(live.replayed).toBe(false)
-    expect(live.messages.map((message) => message.subject)).toEqual(['do the work'])
-
-    await check(PANE_B, { ack: live.deliveryId })
-    expect(db.getUnreadMessages(`dispatch:${dispatchId}`)).toEqual([])
+    expect(await check(PANE_B)).toMatchObject({ deliveryId: null, count: 0 })
+    const history = await check(PANE_B, { all: true })
+    expect(history.messages.map((message) => message.subject)).toEqual(['do the work'])
  })

  it('keeps serving a worker whose process restarted without a re-attach', async () => {
    attachedDispatchWithMail()
    const first = await check(PANE_A)

-    const replay = await check(PANE_A)
-    expect(replay.deliveryId).toBe(first.deliveryId)
-    expect(replay.replayed).toBe(true)
+    expect(await check(PANE_A)).toMatchObject({ deliveryId: null, count: 0 })
    await expect(check(PANE_A, { ack: first.deliveryId })).resolves.toMatchObject({
      acknowledged: first.deliveryId
    })
@@ -117,6 +110,12 @@ describe('orchestration.check on a re-attached Dispatch', () => {
    const dispatchId = attachedDispatchWithMail()
    await check(PANE_A)
    reattach(dispatchId)
+    db.insertMessage({
+      from: 'term_coord',
+      to: `dispatch:${dispatchId}`,
+      subject: 'next step',
+      runId: db.getDispatchContextById(dispatchId)!.run_id
+    })

    await expect(check(PANE_A)).rejects.toMatchObject({
      code: 'consumer_fenced',
@@ -125,8 +124,7 @@ describe('orchestration.check on a re-attached Dispatch', () => {
    expect(db.getUnreadMessages(`dispatch:${dispatchId}`)).toHaveLength(1)

    const live = await check(PANE_B)
-    expect(live.messages.map((message) => message.subject)).toEqual(['do the work'])
-    await check(PANE_B, { ack: live.deliveryId })
+    expect(live.messages.map((message) => message.subject)).toEqual(['next step'])
    expect(db.getUnreadMessages(`dispatch:${dispatchId}`)).toEqual([])
  })

@@ -95,7 +95,7 @@ describe('orchestration.check on a federated attachment across a restart', () =>
    store.markRemoteAttachmentReady(dispatchId)
  }

-  it('replays the coordinator instruction and takes its ack after the app restarts', async () => {
+  it('keeps a handled coordinator instruction handled and acks idempotently after the app restarts', async () => {
    directory = mkdtempSync(join(tmpdir(), 'orca-federated-check-'))
    const path = join(directory, 'orchestration.db')

@@ -123,12 +123,11 @@ describe('orchestration.check on a federated attachment across a restart', () =>
    expect(deliveryId).not.toBeNull()
    ;(db as OrchestrationDb).close()

-    // The worker's process outlives the app; its instruction is still unacknowledged.
+    // The worker's process outlives the app; an older CLI may still ack the handled batch.
    const second = launch(path)
-    const replayed = await check(second)
-    expect(replayed.deliveryId).toBe(deliveryId)
-    expect(replayed.replayed).toBe(true)
-    expect(replayed.messages.map((message) => message.id)).toEqual(['msg_federated_1'])
+    expect(await check(second)).toMatchObject({ deliveryId: null, count: 0 })
+    const history = await check(second, { all: true })
+    expect(history.messages.map((message) => message.id)).toEqual(['msg_federated_1'])

    const acknowledged = await check(second, { ack: deliveryId })
    expect(acknowledged.acknowledged).toBe(deliveryId)
@@ -1,4 +1,4 @@
-import type { MessageType, OrchestrationDb } from '../../../../orchestration/db'
+import type { MessageRow, MessageType, OrchestrationDb } from '../../../../orchestration/db'
 import type { OrcaRuntimeService } from '../../../../orca-runtime'
 import { OrchestrationError } from '../../../../orchestration/orchestration-error'
 import { formatMessageBanner } from '../../../../orchestration/formatter'
@@ -191,6 +191,10 @@ export async function checkWorkerMailbox(args: {
      interruptedAcknowledgedCheck(deliveryRunId, acknowledged.delivery.id, 'outcome_unknown')
    )
  }
+  const formatted = (messages: MessageRow[]) =>
+    params.format || params.inject
+      ? { formatted: messages.map(formatMessageBanner).join('\n\n') }
+      : {}
  const showAll = params.all === true || (params.unread === false && params.peek !== true)
  const readPeek = () => db.getUnreadMessages(address, typeFilter)
  const readDelivery = (wakeTypes?: MessageType[]) => {
@@ -198,7 +202,7 @@ export async function checkWorkerMailbox(args: {
      return undefined
    }
    try {
-      return db.getOrCreateMailboxDelivery({
+      return db.takeMailboxDelivery({
        runId: deliveryRunId,
        mailboxHandle: address,
        consumerGeneration: workerMailbox.generation,
@@ -216,9 +220,7 @@ export async function checkWorkerMailbox(args: {
      messages: exposeMessages(messages),
      count: messages.length,
      acknowledged: acknowledged?.delivery.id ?? null,
-      ...(params.format || params.inject
-        ? { formatted: messages.map(formatMessageBanner).join('\n\n') }
-        : {})
+      ...formatted(messages)
    }
  }
  if (params.peek) {
@@ -229,9 +231,7 @@ export async function checkWorkerMailbox(args: {
        messages: exposeMessages(messages),
        count: messages.length,
        acknowledged: acknowledged?.delivery.id ?? null,
-        ...(params.format || params.inject
-          ? { formatted: messages.map(formatMessageBanner).join('\n\n') }
-          : {})
+        ...formatted(messages)
      }
    }
  } else {
@@ -243,16 +243,16 @@ export async function checkWorkerMailbox(args: {
        messages: exposeMessages(current?.messages ?? []),
        count: current?.messages.length ?? 0,
        replayed: current?.replayed ?? false,
+        ...(current ? { deliveryAcknowledged: true } : {}),
        acknowledged: acknowledged?.delivery.id ?? null,
        timedOut: false,
        cancelled: false,
        connectionLost: false,
-        ...(params.format || params.inject
-          ? { formatted: current?.messages.map(formatMessageBanner).join('\n\n') ?? '' }
-          : {})
+        ...formatted(current?.messages ?? [])
      }
    }
  }
+  const deliveryCursor = db.getDeliveryCursor()
  // Binding can happen during recovery, before run-create/run-use can cancel this wait.
  const waitResult =
    activeDispatch && currentDispatchAssigneeRun(runtime, db, activeDispatch)
@@ -284,21 +284,20 @@ export async function checkWorkerMailbox(args: {
      messages: exposeMessages(arrived),
      count: arrived.length,
      acknowledged: acknowledged?.delivery.id ?? null,
-      ...(params.format || params.inject
-        ? { formatted: arrived.map(formatMessageBanner).join('\n\n') }
-        : {})
+      ...formatted(arrived)
    }
  }
-  const arrived = readDelivery(typeFilter)
+  const arrived =
+    readDelivery(typeFilter) ??
+    db.getMailboxDeliveryTakenAfter(address, workerMailbox.generation, deliveryCursor)
  return {
    ...mailboxIdentity,
    deliveryId: arrived?.delivery.id ?? null,
    messages: exposeMessages(arrived?.messages ?? []),
    count: arrived?.messages.length ?? 0,
    replayed: arrived?.replayed ?? false,
+    ...(arrived ? { deliveryAcknowledged: true } : {}),
    acknowledged: acknowledged?.delivery.id ?? null,
-    ...(params.format || params.inject
-      ? { formatted: arrived?.messages.map(formatMessageBanner).join('\n\n') ?? '' }
-      : {})
+    ...formatted(arrived?.messages ?? [])
  }
 }
@@ -282,11 +282,6 @@ describe('orchestration RPC methods', () => {
        code: 'stale_delivery',
        message: `${queued.id} is a message id, not a delivery id. Acknowledge the batch with the deliveryId field from the check response; process the entire batch before acknowledging.`
      })
-      expect(db.getMessageById(queued.id)).toMatchObject({ id: queued.id, read: 0 })
-      expect(await call('orchestration.check', { terminal: 'term_coord' })).toMatchObject({
-        deliveryId,
-        messages: [{ id: queued.id }]
-      })
      expect(
        await call('orchestration.check', { terminal: 'term_coord', ack: deliveryId })
      ).toMatchObject({ acknowledged: deliveryId })
@@ -400,7 +395,7 @@ describe('orchestration RPC methods', () => {
      expect(result.messages[0].type).toBe('worker_done')
      expect(db.getTask(task.id)?.status).toBe('completed')
      expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed')
-      expect(db.getUnreadMessages(`run:${activeRunId}`)).toHaveLength(1)
+      expect(db.getUnreadMessages(`run:${activeRunId}`)).toHaveLength(0)
      const taskList = (await call('orchestration.taskList', {})) as {
        tasks: {
          id: string
@@ -423,12 +418,11 @@ describe('orchestration RPC methods', () => {
      const repeated = (await call('orchestration.check', {
        terminal: 'term_coord',
        types: 'worker_done'
-      })) as { count: number; deliveryId: string }
-      expect(repeated.count).toBe(1)
-      expect(repeated.deliveryId).toBe(result.deliveryId)
+      })) as { count: number }
+      expect(repeated.count).toBe(0)
      const acknowledged = (await call('orchestration.check', {
        terminal: 'term_coord',
-        ack: repeated.deliveryId,
+        ack: result.deliveryId,
        types: 'worker_done'
      })) as { count: number }
      expect(acknowledged.count).toBe(0)
@@ -116,7 +116,7 @@ describe('Run binding during Dispatch checks', () => {
  })

  it.each(['before', 'during recovery'] as const)(
-    'replays Run delivery created %s residual acknowledgment',
+    'serves residual mail around a Run delivery created %s residual acknowledgment',
    async (phase) => {
      const ids = Array.from(
        { length: ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1 },
@@ -158,23 +158,24 @@ describe('Run binding during Dispatch checks', () => {
          }
        )
      }
-      const result = await check({ ack })
-      expect(result).toMatchObject({
-        runId: run.id,
-        deliveryId: runDelivery,
-        acknowledged: ack,
-        replayed: true
-      })
+      const tail = { runId: dispatch.run_id, messages: [{ id: last }], count: 1 }
+      if (phase === 'before') {
+        expect(await check({ ack })).toMatchObject({ ...tail, acknowledged: ack })
+      } else {
+        // A batch left outstanding by another consumer is replayed once, then handled.
+        expect(await check({ ack })).toMatchObject({
+          runId: run.id,
+          deliveryId: runDelivery,
+          acknowledged: ack,
+          replayed: true
+        })
+        expect(await check()).toMatchObject(tail)
+      }
      expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false)
-      for (const id of ids.slice(0, -1)) {
+      for (const id of ids) {
        expect(state.db.getMessageById(id)).toMatchObject({ read: 1, run_id: dispatch.run_id })
      }
-      expect(state.db.getMessageById(last)).toMatchObject({ read: 0, run_id: dispatch.run_id })
-      expect(await check({ ack })).toMatchObject({ deliveryId: runDelivery, replayed: true })
-      const tail = await check({ ack: runDelivery })
-      expect(tail).toMatchObject({ runId: dispatch.run_id, messages: [{ id: last }], count: 1 })
-      expect(await check({ ack: deliveryId(tail) })).toMatchObject({ count: 0 })
-      expect(state.db.getMessageById(last)?.read).toBe(1)
+      expect(await check({ ack: runDelivery })).toMatchObject({ count: 0 })
    }
  )
 })
@@ -447,7 +447,7 @@ describe('Run-bound lead mailbox boundaries', () => {
    expect(state.db.getMessageById(privateMail.id)).toMatchObject({ read: 0, run_id: foreign.id })
  })

-  it('treats types as a wake condition and replays the entire residual FIFO batch', async () => {
+  it('treats types as a wake condition and returns the entire residual FIFO batch', async () => {
    residual('older status')
    residual('decision needed', 'question')
    runMail()
@@ -455,10 +455,6 @@ describe('Run-bound lead mailbox boundaries', () => {
    expect(first).toMatchObject({
      messages: [{ subject: 'older status' }, { subject: 'decision needed' }]
    })
-    expect(await check({ wait: true, types: 'worker_done' })).toMatchObject({
-      deliveryId: deliveryId(first),
-      replayed: true
-    })
    expect(await check({ ack: deliveryId(first) })).toMatchObject({
      runId: leadRun.id,
      acknowledged: deliveryId(first),
@@ -484,16 +480,12 @@ describe('Run-bound lead mailbox boundaries', () => {
    }
  )

-  it('replays an outstanding Run batch before exposing older nonmatching residual mail', async () => {
+  it('hands older nonmatching residual mail to the check after a filtered Run wait', async () => {
    residual('status')
    runMail()
    const first = await check({ wait: true, types: 'question' })
    expect(first).toMatchObject({ runId: leadRun.id })
-    expect(await check()).toMatchObject({ deliveryId: deliveryId(first), replayed: true })
-    expect(await check({ ack: deliveryId(first) })).toMatchObject({
-      acknowledged: deliveryId(first),
-      messages: [{ subject: 'status' }]
-    })
+    expect(await check()).toMatchObject({ messages: [{ subject: 'status' }] })
  })

  it('keeps explicit Run checks scoped and refuses the parent Run', async () => {
@@ -195,23 +195,22 @@ describe('current orchestration authority precedence', () => {
        count: 1
      }
    })
-    expect(harness.db.getMessageById(message.id)?.read).toBe(0)
+    expect(harness.db.getMessageById(message.id)?.read).toBe(1)

    const deliveryId = (response as { result: { deliveryId: string } }).result.deliveryId
+    // A response lost in transit replays from the request receipt when retried with the same id.
    const replayed = await harness.dispatcher.dispatch(
      request(
        'orchestration.check',
        { terminal: CURRENT_WORKER_HANDLE },
        currentEvidence('worker'),
-        'current-worker-check-replay'
+        'current-worker-check'
      )
    )
-
    expect(replayed).toMatchObject({
      ok: true,
-      result: { deliveryId, replayed: true, messages: [{ id: message.id }], count: 1 }
+      result: { deliveryId, messages: [{ id: message.id }], count: 1 }
    })
-    expect(harness.db.getMessageById(message.id)?.read).toBe(0)

    const acknowledged = await harness.dispatcher.dispatch(
      request(
@@ -311,11 +311,9 @@ describe('a structured chat coordinates through the same verbs as a terminal', (
      )
    ).toMatchObject({ ok: false, error: { code: 'consumer_fenced' } })
    expect(acknowledge).not.toHaveBeenCalled()
-    expect(h.db.getDeliveryRaw(ack)?.acknowledged_at).toBeNull()
    expect(await as(SESSION_Y, 'orchestration.check', {})).toMatchObject({
-      deliveryId: ack,
-      replayed: true,
-      messages: [{ id: idOf(message) }]
+      deliveryId: null,
+      count: 0
    })
    expect(await as(SESSION_Y, 'orchestration.check', { ack })).toMatchObject({
      acknowledged: ack,
@@ -551,11 +551,10 @@ describe('OrcaRuntimeRpcServer', () => {
        db.insertMessage({ from: 'term_worker', to: `run:${run.id}`, subject: 'go', runId: run.id })
        runtime.notifyMessageArrived(`run:${run.id}`)
        await Promise.all([b.done, c.done])
-        const consumerGeneration = run.consumer_generation
-        const deliveryId = db.getOrCreateRunDelivery({ runId: run.id, consumerGeneration })
-          ?.delivery.id
        const results = [b, c].map((s) => s.frames.find((f) => f.ok !== undefined)?.result)
-        expect(results).toMatchObject([{ deliveryId }, { deliveryId }])
+        const batch = { deliveryId: expect.stringMatching(/^delivery_/), count: 1 }
+        expect(results).toMatchObject([batch, batch])
+        expect(db.db.prepare('SELECT COUNT(*) AS n FROM deliveries').get()).toEqual({ n: 1 })
        await waitFor(() => server['activeLongPolls'] === 0)
      } finally {
        db.close()
@@ -52,10 +52,24 @@ describe('formatted delivery acknowledgment', () => {
          messages: [{ id: 'msg_one', from_handle: 'worker' }],
          count: 1,
          deliveryId: 'delivery_one',
+          deliveryAcknowledged: true,
          formatted: 'Message body'
        },
        'term_coord'
      )
    ).toBe('Delivery delivery_one\nMessage body')
  })
+
+  it('tells the caller to ack when an older runtime did not', () => {
+    expect(
+      formatOrchestrationCheckText(
+        {
+          messages: [{ id: 'msg_one', from_handle: 'worker', subject: 'done' }],
+          count: 1,
+          deliveryId: 'delivery_one'
+        },
+        'term_coord'
+      )
+    ).toContain('rerun this check with --ack delivery_one.')
+  })
 })
@@ -37,6 +37,8 @@ export type OrchestrationCheckOutput = {
  count: number
  formatted?: string
  deliveryId?: string | null
+  // Older runtimes omit it and replay the batch until an explicit --ack.
+  deliveryAcknowledged?: boolean
  timedOut?: boolean
  cancelled?: boolean
  connectionLost?: boolean
@@ -80,7 +82,11 @@ export function formatOrchestrationCheckText(
        ? '[LEGACY RECOVERY REPLAY — MAY HAVE BEEN SEEN]\n'
        : '[LEGACY COMPATIBILITY]\n'
    : ''
-  const deliveryNotice = formatCurrentDeliveryNotice(prepared.legacyCompatibility?.currentDelivery)
+  const deliveryNotice =
+    formatCurrentDeliveryNotice(prepared.legacyCompatibility?.currentDelivery) +
+    (prepared.deliveryId && prepared.deliveryAcknowledged !== true
+      ? `\nThis Orca version does not acknowledge batches automatically. After handling every message, rerun this check with --ack ${prepared.deliveryId}.`
+      : '')
  if (prepared.formatted) {
    return `${legacyHeader}${prepared.deliveryId ? `Delivery ${prepared.deliveryId}\n` : ''}${prepared.formatted}${deliveryNotice}`
  }