refactor(orchestration): derive delivery eligibility from messages (#19837)

* fix(orchestration): retire read deliveries and clarify mailbox recovery * fix(orchestration): simplify delivery recovery and update nudge contracts * test: align orchestration check help expectation * refactor(orchestration): derive delivery eligibility from messages * fix(orchestration): validate live consumers and simplify batch revocation * refactor(orchestration): keep deliveries.status and derive eligibility without a column drop The outstanding_deliveries view now reads status = 'outstanding' plus unread membership, so v41 only drops uniqueness from idx_deliveries_one_outstanding and adds the view and trigger. Older binaries can still open the database. Removes the column-drop migration, the v40 test fixture and hasColumn guards, the fenced skew probe, and the unrelated nudge-text change. * docs(orchestration): drop delivery storage reference The compatibility caveat it existed to explain no longer applies; the view and index comments carry the remaining rationale. * docs: revert unrelated formatter churn * test(orchestration): verify historical database downgrade round trip
2026-09-22 00:02:31 +00:00 · 2026-09-13 23:24:04 -04:00
parent 31db2774f8
commit 3ab2a1b91c
31 changed files with 995 additions and 63 deletions
@@ -64,7 +64,7 @@ describe('orchestration check command spec', () => {

    expect(checkSpec?.notes).toEqual(
      expect.arrayContaining([
-        '--types is the wake condition for --wait; a returned Delivery is always the whole FIFO batch, so it is never filtered by type. Only --peek and --all filter their rows.'
+        '--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.'
      ])
    )
  })
@@ -111,9 +111,9 @@ export const ORCHESTRATION_COMMAND_SPECS: CommandSpec[] = [
    ],
    notes: [
      '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. Only --peek and --all filter their rows.',
+      '--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; process every message before acknowledging.'
+      'A bound Run replays the same Delivery until --ack or all its messages are marked read, even with --types; process every message before acknowledging.'
    ]
  },
  {
@@ -17,4 +17,5 @@ export const LEGACY_CONTRACT_VERSION = 0
 export const CURRENT_CONTRACT_VERSION = ORCHESTRATION_CONTRACT_VERSION

 // Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 durable local mutation caller identity, v31 dispatch/resource identity links, v32 bounded worker-terminal recovery metadata, v33 durable mailbox pointer Enter state, v34 role-addressed mailbox deliveries, v35 mailbox delivery default and index-predicate repair, v36 dispatch mailbox consumer generation, v37 recorded dispatch creator identity, v39 structured session journal archives.
-export const SCHEMA_VERSION = 40
+// v41: derive outstanding deliveries from unread messages.
+export const SCHEMA_VERSION = 41
@@ -38,7 +38,7 @@ export function mintDispatchCapability(
        params.processIncarnation,
        params.dispatchId
      )
-    this.fenceOutstandingMailboxDelivery(`dispatch:${params.dispatchId}`)
+    this.fenceUnacknowledgedMailboxDeliveries(`dispatch:${params.dispatchId}`)
    this.db.exec('COMMIT')
  } catch (error) {
    this.db.exec('ROLLBACK')
@@ -1,7 +1,6 @@
 import { afterEach, beforeEach, describe, expect, it } from 'vitest'
 import { OrchestrationDb } from '../db'
 import { ORCHESTRATION_CONTRACT_VERSION } from '../../../../shared/protocol-version'
-import { ORCHESTRATION_LEGACY_RUN_ID } from '../../../../shared/orchestration-rpc-contract'
 import { createRootDispatch } from './root-dispatch-test-fixture'
 import type { DeliveryRow } from '../types'

@@ -35,11 +34,17 @@ describe('dispatch mailbox consumer fencing', () => {
    return { id: dispatch.id, runId: dispatch.run_id }
  }

-  function openDelivery(dispatchId: string, runId: string, generation: number) {
+  function openDelivery(
+    dispatchId: string,
+    runId: string,
+    generation: number,
+    consumerSource: 'dispatch' | 'attachment' = 'dispatch'
+  ) {
    return db.getOrCreateMailboxDelivery({
      runId,
      mailboxHandle: `dispatch:${dispatchId}`,
-      consumerGeneration: generation
+      consumerGeneration: generation,
+      consumerSource
    })
  }

@@ -169,9 +174,9 @@ describe('dispatch mailbox consumer fencing', () => {
      from: 'home-peer',
      to: `dispatch:${dispatchId}`,
      subject: 'relayed before attach',
-      runId: ORCHESTRATION_LEGACY_RUN_ID
+      runId: 'run-home'
    })
-    const stale = openDelivery(dispatchId, ORCHESTRATION_LEGACY_RUN_ID, 0)
+    const stale = openDelivery(dispatchId, 'run-home', 0, 'attachment')

    // The worker host holds no dispatch_contexts row for a federated Dispatch.
    expect(db.getDispatchContextById(dispatchId)).toBeUndefined()
@@ -71,7 +71,7 @@ export function prepareRemoteAttachmentAuthority(
        `Remote Dispatch ${params.dispatchId} is not starting.`
      )
    }
-    this.fenceOutstandingMailboxDelivery(`dispatch:${params.dispatchId}`)
+    this.fenceUnacknowledgedMailboxDeliveries(`dispatch:${params.dispatchId}`)
    if (params.terminalOwnership && !this.getWorkerTerminalResourceByOwner(params.dispatchId)) {
      const resource =
        params.terminalOwnership === 'external'
@@ -0,0 +1,234 @@
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it } from 'vitest'
+import { ORCHESTRATION_CONTRACT_VERSION } from '../../../../../shared/protocol-version'
+import { OrchestrationDb } from '../orchestration-db'
+import { createRootDispatch } from '../root-dispatch-test-fixture'
+
+type Settlement = 'local completion' | 'local failure' | 'remote stop' | 'remote failure'
+type DeliveryOperation = 'create' | 'acknowledge'
+
+describe('mailbox consumer lifecycle fencing', () => {
+  const connections: OrchestrationDb[] = []
+  const directories: string[] = []
+
+  afterEach(() => {
+    for (const db of connections.splice(0)) {
+      db.close()
+    }
+    for (const directory of directories.splice(0)) {
+      rmSync(directory, { recursive: true, force: true })
+    }
+  })
+
+  function open(path: string): OrchestrationDb {
+    const db = new OrchestrationDb(path)
+    connections.push(db)
+    return db
+  }
+
+  function databasePath(): string {
+    const directory = mkdtempSync(join(tmpdir(), 'orca-mailbox-consumer-lifecycle-'))
+    directories.push(directory)
+    return join(directory, 'orchestration.db')
+  }
+
+  function setup(settlement: Settlement): {
+    db: OrchestrationDb
+    peer: OrchestrationDb
+    messageId: string
+    params: {
+      runId: string
+      mailboxHandle: string
+      consumerGeneration: number
+      consumerSource: 'dispatch' | 'attachment'
+    }
+    settle: () => void
+  } {
+    const path = databasePath()
+    const db = open(path)
+    const run = db.createRun({
+      objective: 'Fence settled mailbox consumers',
+      coordinatorHandle: 'coord',
+      coordinatorPaneKey: 'tab:11111111-1111-4111-8111-111111111111'
+    })
+    const remote = settlement.startsWith('remote')
+    const dispatchId = remote
+      ? `ctx_${settlement.replace(' ', '_')}`
+      : createRootDispatch(db, db.createTask({ runId: run.id, spec: settlement }).id, 'worker').id
+    let consumerGeneration = 0
+
+    if (remote) {
+      db.createRemoteDispatchAttachment({
+        runId: run.id,
+        dispatchId,
+        taskId: `task_${dispatchId}`,
+        homePeerFingerprint: 'home-peer',
+        protocolVersion: ORCHESTRATION_CONTRACT_VERSION,
+        runtimeEpoch: 'epoch-1',
+        mutationReceipt: {
+          callerFingerprint: 'home-peer',
+          requestId: `request_${dispatchId}`,
+          method: 'orchestration.federationAttachStart',
+          payloadHash: `hash_${dispatchId}`
+        }
+      })
+      if (settlement === 'remote stop') {
+        db.prepareRemoteAttachmentAuthority({
+          dispatchId,
+          paneKey: 'worker:22222222-2222-4222-9222-222222222222',
+          processIncarnation: 'runtime:worker:1',
+          worktreeId: 'folder',
+          terminalHandle: 'worker',
+          setupState: 'not_applicable',
+          effects: []
+        })
+        db.markRemoteAttachmentReady(dispatchId)
+        consumerGeneration = 1
+      }
+    }
+
+    const mailboxHandle = `dispatch:${dispatchId}`
+    const message = db.insertMessage({
+      runId: run.id,
+      from: 'coord',
+      to: mailboxHandle,
+      subject: 'must remain unread'
+    })
+    const peer = open(path)
+    const settle = (): void => {
+      if (settlement === 'local completion') {
+        peer.completeDispatch(dispatchId)
+      } else if (settlement === 'local failure') {
+        peer.failDispatch(dispatchId, 'settled by peer')
+      } else if (settlement === 'remote stop') {
+        peer.beginRemoteAttachmentStop(dispatchId)
+        peer.settleRemoteAttachmentStop(dispatchId)
+      } else {
+        peer.failRemoteAttachment(dispatchId, 'peer_failure', 'settled by peer', false)
+      }
+    }
+
+    return {
+      db,
+      peer,
+      messageId: message.id,
+      params: {
+        runId: run.id,
+        mailboxHandle,
+        consumerGeneration,
+        consumerSource: remote ? 'attachment' : 'dispatch'
+      },
+      settle
+    }
+  }
+
+  function currentGeneration(
+    db: OrchestrationDb,
+    params: {
+      mailboxHandle: string
+      consumerSource: 'dispatch' | 'attachment'
+    }
+  ): number | undefined {
+    const dispatchId = params.mailboxHandle.slice('dispatch:'.length)
+    return params.consumerSource === 'dispatch'
+      ? db.getDispatchContextById(dispatchId)?.consumer_generation
+      : db.getRemoteDispatchAttachment(dispatchId)?.consumer_generation
+  }
+
+  it.each<{
+    operation: DeliveryOperation
+    settlement: Settlement
+  }>([
+    { operation: 'create', settlement: 'local completion' },
+    { operation: 'create', settlement: 'local failure' },
+    { operation: 'create', settlement: 'remote stop' },
+    { operation: 'create', settlement: 'remote failure' },
+    { operation: 'acknowledge', settlement: 'local completion' },
+    { operation: 'acknowledge', settlement: 'local failure' },
+    { operation: 'acknowledge', settlement: 'remote stop' },
+    { operation: 'acknowledge', settlement: 'remote failure' }
+  ])('rejects $operation after $settlement on another connection', ({ operation, settlement }) => {
+    const { db, peer, messageId, params, settle } = setup(settlement)
+    const delivery =
+      operation === 'acknowledge' ? db.getOrCreateMailboxDelivery(params)?.delivery : undefined
+
+    settle()
+
+    const operationCall = (): unknown =>
+      operation === 'create'
+        ? db.getOrCreateMailboxDelivery(params)
+        : db.acknowledgeMailboxDelivery({ ...params, deliveryId: delivery!.id })
+    expect(operationCall).toThrow(expect.objectContaining({ code: 'consumer_fenced' }))
+    expect(db.getMessageById(messageId)?.read).toBe(0)
+    expect(currentGeneration(peer, params)).toBe(params.consumerGeneration)
+    if (delivery) {
+      expect(db.getDeliveryRaw(delivery.id)?.acknowledged_at).toBeNull()
+    }
+  })
+
+  it.each(['start_unknown', 'stop_unknown'] as const)(
+    'keeps a remote %s attachment eligible to consume mail',
+    (state) => {
+      const path = databasePath()
+      const db = open(path)
+      const run = db.createRun({
+        objective: 'Preserve unverifiable remote consumers',
+        coordinatorHandle: 'coord',
+        coordinatorPaneKey: 'tab:11111111-1111-4111-8111-111111111111'
+      })
+      const dispatchId = `ctx_${state}`
+      db.createRemoteDispatchAttachment({
+        runId: run.id,
+        dispatchId,
+        taskId: `task_${state}`,
+        homePeerFingerprint: 'home-peer',
+        protocolVersion: ORCHESTRATION_CONTRACT_VERSION,
+        runtimeEpoch: 'epoch-1',
+        mutationReceipt: {
+          callerFingerprint: 'home-peer',
+          requestId: `request_${state}`,
+          method: 'orchestration.federationAttachStart',
+          payloadHash: `hash_${state}`
+        }
+      })
+      let consumerGeneration = 0
+      if (state === 'start_unknown') {
+        db.failRemoteAttachment(dispatchId, 'start_unknown', 'contact lost', true)
+      } else {
+        db.prepareRemoteAttachmentAuthority({
+          dispatchId,
+          paneKey: 'worker:22222222-2222-4222-9222-222222222222',
+          processIncarnation: 'runtime:worker:1',
+          worktreeId: 'folder',
+          terminalHandle: 'worker',
+          setupState: 'not_applicable',
+          effects: []
+        })
+        db.markRemoteAttachmentReady(dispatchId)
+        db.beginRemoteAttachmentStop(dispatchId)
+        db.markRemoteAttachmentStopUnknown(dispatchId, 'contact lost')
+        consumerGeneration = 1
+      }
+      const mailboxHandle = `dispatch:${dispatchId}`
+      const message = db.insertMessage({
+        runId: run.id,
+        from: 'coord',
+        to: mailboxHandle,
+        subject: 'still deliverable'
+      })
+
+      expect(
+        db
+          .getOrCreateMailboxDelivery({
+            runId: run.id,
+            mailboxHandle,
+            consumerGeneration,
+            consumerSource: 'attachment'
+          })
+          ?.messages.map((row) => row.id)
+      ).toEqual([message.id])
+    }
+  )
+})
@@ -0,0 +1,45 @@
+import type { OrchestrationDb } from '../orchestration-db'
+import { OrchestrationError } from '../../orchestration-error'
+import { potentiallyLiveRemoteAttachmentSql } from '../federation/remote-attachment-liveness'
+
+const ACTIVE_DISPATCH_CONSUMER_SQL = `
+  SELECT run_id, consumer_generation FROM dispatch_contexts
+  WHERE id = ? AND status IN ('pending', 'dispatched')
+`
+const ACTIVE_ATTACHMENT_CONSUMER_SQL = `
+  SELECT home_run_id AS run_id, consumer_generation FROM remote_dispatch_attachments
+  WHERE dispatch_id = ? AND ${potentiallyLiveRemoteAttachmentSql()}
+`
+
+// Validate inside the delivery transaction, so another connection cannot replace the consumer mid-check.
+export function requireMailboxConsumer(
+  db: OrchestrationDb,
+  params: {
+    runId: string
+    mailboxHandle: string
+    consumerGeneration: number
+    consumerSource?: 'dispatch' | 'attachment'
+  }
+): void {
+  if (params.mailboxHandle === `run:${params.runId}`) {
+    db.requireCurrentConsumer(params.runId, params.consumerGeneration)
+    return
+  }
+  const dispatchId = params.mailboxHandle.startsWith('dispatch:')
+    ? params.mailboxHandle.slice('dispatch:'.length)
+    : ''
+  // A loopback runtime has both records; use the counter belonging to the caller's attachment.
+  const sql =
+    params.consumerSource === 'attachment'
+      ? ACTIVE_ATTACHMENT_CONSUMER_SQL
+      : ACTIVE_DISPATCH_CONSUMER_SQL
+  const consumer = db.db.prepare(sql).get(dispatchId) as
+    | { run_id: string; consumer_generation: number }
+    | undefined
+  if (
+    consumer?.run_id !== params.runId ||
+    consumer.consumer_generation !== params.consumerGeneration
+  ) {
+    throw new OrchestrationError('consumer_fenced', 'This mailbox consumer has been replaced.')
+  }
+}
@@ -3,6 +3,7 @@ import { OrchestrationError } from '../../orchestration-error'
 import { generateId } from '../generated-id'
 import type { OrchestrationDb } from '../orchestration-db'
 import { exposeDeliveryTimestamps, exposeMessageListTimestamps } from '../utc-timestamp'
+import { requireMailboxConsumer } from './mailbox-consumer'
 import { ORCHESTRATION_DELIVERY_BATCH_LIMIT } from './mailbox-routing-page'

 export function getDeliveryRaw(this: OrchestrationDb, id: string): DeliveryRow | undefined {
@@ -29,9 +30,9 @@ export function getOrCreateMailboxDelivery(
    runId: string
    mailboxHandle: string
    consumerGeneration: number
+    consumerSource?: 'dispatch' | 'attachment'
    limit?: number
    wakeTypes?: MessageType[]
-    requireCurrentRunConsumer?: boolean
  }
 ): { delivery: DeliveryRow; messages: MessageRow[]; replayed: boolean } | undefined {
  const limit = Math.min(
@@ -40,11 +41,9 @@ export function getOrCreateMailboxDelivery(
  )
  this.db.exec('BEGIN IMMEDIATE')
  try {
-    if (params.requireCurrentRunConsumer) {
-      this.requireCurrentConsumer(params.runId, params.consumerGeneration)
-    }
+    requireMailboxConsumer(this, params)
    const existing = this.db
-      .prepare("SELECT * FROM deliveries WHERE mailbox_handle = ? AND status = 'outstanding'")
+      .prepare('SELECT * FROM outstanding_deliveries WHERE mailbox_handle = ?')
      .get(params.mailboxHandle) as DeliveryRow | undefined
    if (existing) {
      if (existing.consumer_generation !== params.consumerGeneration) {
@@ -115,15 +114,13 @@ export function acknowledgeMailboxDelivery(
    runId: string
    mailboxHandle: string
    consumerGeneration: number
+    consumerSource?: 'dispatch' | 'attachment'
    deliveryId: string
-    requireCurrentRunConsumer?: boolean
  }
 ): { delivery: DeliveryRow; duplicate: boolean } {
  this.db.exec('BEGIN IMMEDIATE')
  try {
-    if (params.requireCurrentRunConsumer) {
-      this.requireCurrentConsumer(params.runId, params.consumerGeneration)
-    }
+    requireMailboxConsumer(this, params)
    const delivery = this.getDeliveryRaw(params.deliveryId)
    if (
      !delivery ||
@@ -132,7 +129,7 @@ export function acknowledgeMailboxDelivery(
    ) {
      throw new OrchestrationError(
        'stale_delivery',
-        `Delivery ${params.deliveryId} does not belong to this mailbox.`
+        `Delivery ${params.deliveryId} does not belong to this mailbox. --ack requires a delivery_* ID returned by orchestration check; process the entire batch before acknowledging.`
      )
    }
    if (
@@ -180,14 +177,12 @@ export function hasOutstandingMailboxDelivery(
 ): boolean {
  return Boolean(
    this.db
-      .prepare(
-        "SELECT 1 FROM deliveries WHERE mailbox_handle = ? AND status = 'outstanding' LIMIT 1"
-      )
+      .prepare('SELECT 1 FROM outstanding_deliveries WHERE mailbox_handle = ? LIMIT 1')
      .get(mailboxHandle)
  )
 }

-export function fenceOutstandingMailboxDelivery(
+export function fenceUnacknowledgedMailboxDeliveries(
  this: OrchestrationDb,
  mailboxHandle: string
 ): void {
@@ -204,7 +199,7 @@ export type RoleMailboxDeliveryMethods = {
  getOrCreateMailboxDelivery: typeof getOrCreateMailboxDelivery
  acknowledgeMailboxDelivery: typeof acknowledgeMailboxDelivery
  hasOutstandingMailboxDelivery: typeof hasOutstandingMailboxDelivery
-  fenceOutstandingMailboxDelivery: typeof fenceOutstandingMailboxDelivery
+  fenceUnacknowledgedMailboxDeliveries: typeof fenceUnacknowledgedMailboxDeliveries
 }

 export function attachRoleMailboxDelivery(ctor: { prototype: object }): void {
@@ -214,6 +209,6 @@ export function attachRoleMailboxDelivery(ctor: { prototype: object }): void {
    getOrCreateMailboxDelivery,
    acknowledgeMailboxDelivery,
    hasOutstandingMailboxDelivery,
-    fenceOutstandingMailboxDelivery
+    fenceUnacknowledgedMailboxDeliveries
  })
 }
@@ -142,7 +142,7 @@ export function bindRun(
           WHERE id = ?`
        )
        .run(params.coordinatorHandle, params.coordinatorPaneKey, params.runId)
-      this.fenceOutstandingDelivery(params.runId)
+      this.fenceUnacknowledgedMailboxDeliveries(`run:${params.runId}`)
      if (params.takeoverLegacy || replacesLegacyCoordinator) {
        this.promoteLegacyCoordinatorMailForTakeover(params.runId, retainedCoordinatorHandle)
      }
@@ -32,8 +32,7 @@ export function getOrCreateRunDelivery(
    mailboxHandle: `run:${params.runId}`,
    consumerGeneration: params.consumerGeneration,
    limit: params.limit,
-    wakeTypes: params.wakeTypes,
-    requireCurrentRunConsumer: true
+    wakeTypes: params.wakeTypes
  })
 }

@@ -49,8 +48,7 @@ export function acknowledgeRunDelivery(
    runId: params.runId,
    mailboxHandle: `run:${params.runId}`,
    consumerGeneration: params.consumerGeneration,
-    deliveryId: params.deliveryId,
-    requireCurrentRunConsumer: true
+    deliveryId: params.deliveryId
  })
 }

@@ -141,7 +141,7 @@ export function unbindOtherRunsForPane(
           WHERE id = ?`
        )
        .run(run.id)
-      this.fenceOutstandingDelivery(run.id)
+      this.fenceUnacknowledgedMailboxDeliveries(`run:${run.id}`)
    }
  }
 }
@@ -152,10 +152,6 @@ export function requireRun(this: OrchestrationDb, runId: string): void {
  }
 }

-export function fenceOutstandingDelivery(this: OrchestrationDb, runId: string): void {
-  this.fenceOutstandingMailboxDelivery(`run:${runId}`)
-}
-
 export type RunLookupMethods = {
  getRun: typeof getRun
  getLegacyAdoptedRunMailboxOwner: typeof getLegacyAdoptedRunMailboxOwner
@@ -166,7 +162,6 @@ export type RunLookupMethods = {
  getRunRaw: typeof getRunRaw
  unbindOtherRunsForPane: typeof unbindOtherRunsForPane
  requireRun: typeof requireRun
-  fenceOutstandingDelivery: typeof fenceOutstandingDelivery
 }

 export function attachRunLookup(ctor: { prototype: object }): void {
@@ -179,7 +174,6 @@ export function attachRunLookup(ctor: { prototype: object }): void {
    runsBoundToPane,
    getRunRaw,
    unbindOtherRunsForPane,
-    requireRun,
-    fenceOutstandingDelivery
+    requireRun
  })
 }
@@ -1,10 +1,12 @@
 import type { OrchestrationDb } from '../orchestration-db'
 import { createCoreTablesSql } from './create-core-tables-sql'
 import { createGraphTablesSql } from './create-graph-tables-sql'
+import { DERIVED_DELIVERY_SCHEMA_SQL } from './migrate-v41'

 export function createTables(this: OrchestrationDb): void {
  this.db.exec(`${createCoreTablesSql()}\n${createGraphTablesSql()}`)
  this.createMailboxDeliveryIndexesIfPossible()
+  this.db.exec(DERIVED_DELIVERY_SCHEMA_SQL)
 }

 export type CreateTablesMethods = {
@@ -0,0 +1,269 @@
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it } from 'vitest'
+import Database from '../../../../sqlite/sync-database'
+import { OrchestrationDb } from '../orchestration-db'
+import { dropDerivedDeliverySchema } from './derived-delivery-test-fixture'
+import { resolveOrchestrationMigrationStartVersion } from '../../orchestration-schema-version-skew'
+import { createRootDispatch } from '../root-dispatch-test-fixture'
+import { SCHEMA_VERSION } from '../contract-constants'
+
+describe('derived delivery migration', () => {
+  const connections: OrchestrationDb[] = []
+  const directories: string[] = []
+  afterEach(() => {
+    for (const db of connections.splice(0)) {
+      db.close()
+    }
+    for (const directory of directories.splice(0)) {
+      rmSync(directory, { recursive: true, force: true })
+    }
+  })
+  function open(path: string) {
+    const db = new OrchestrationDb(path)
+    connections.push(db)
+    return db
+  }
+  function databasePath() {
+    const directory = mkdtempSync(join(tmpdir(), 'orca-derived-delivery-'))
+    directories.push(directory)
+    return join(directory, 'orchestration.db')
+  }
+
+  it('preserves batch identity and terminal facts while removing a persisted wedge', () => {
+    const path = databasePath()
+    const original = open(path)
+    const run = original.createRun({
+      objective: 'upgrade',
+      coordinatorHandle: 'coord',
+      coordinatorPaneKey: 'tab:11111111-1111-4111-8111-111111111111'
+    })
+    const params = { runId: run.id, consumerGeneration: run.consumer_generation }
+    const old = original.insertMessage({
+      runId: run.id,
+      from: 'worker',
+      to: `run:${run.id}`,
+      subject: 'old'
+    })
+    const batch = original.getDeliveryRaw(original.getOrCreateRunDelivery(params)!.delivery.id)!
+    const next = original.insertMessage({
+      runId: run.id,
+      from: 'worker',
+      to: `run:${run.id}`,
+      subject: 'next'
+    })
+    connections.pop()!.close()
+    const raw = new Database(path)
+    dropDerivedDeliverySchema(raw)
+    raw.prepare('UPDATE messages SET read = 1 WHERE id = ?').run(old.id)
+    raw.exec(`
+      INSERT INTO deliveries (id, run_id, mailbox_handle, consumer_generation, message_ids, status, created_at, acknowledged_at)
+      VALUES ('history_ack', '${run.id}', 'run:${run.id}', 1, '[]', 'acknowledged', '2026-01-01 00:00:00', '2026-01-02 00:00:00'),
+             ('history_fence', '${run.id}', 'run:${run.id}', 1, '[]', 'fenced', '2026-01-03 00:00:00', NULL);
+    `)
+    raw.pragma('user_version = 40')
+    raw.close()
+    const db = open(path)
+    expect(db.getDeliveryRaw(batch.id)).toEqual(batch)
+    expect(db.hasOutstandingRunDelivery(run.id)).toBe(false)
+    expect(db.getDeliveryRaw('history_ack')).toMatchObject({
+      acknowledged_at: '2026-01-02 00:00:00',
+      status: 'acknowledged'
+    })
+    expect(db.getDeliveryRaw('history_fence')).toMatchObject({
+      acknowledged_at: null,
+      status: 'fenced'
+    })
+    expect(() => db.acknowledgeRunDelivery({ ...params, deliveryId: 'history_fence' })).toThrow(
+      expect.objectContaining({ code: 'consumer_fenced' })
+    )
+    expect(db.getOrCreateRunDelivery(params)?.messages.map((message) => message.id)).toEqual([
+      next.id
+    ])
+    expect(db.getDeliveryRaw(batch.id)?.acknowledged_at).toBeNull()
+    expect(resolveOrchestrationMigrationStartVersion(db.db, SCHEMA_VERSION, SCHEMA_VERSION)).toBe(
+      SCHEMA_VERSION
+    )
+    const reopened = open(path)
+    expect(reopened.getOrCreateRunDelivery(params)?.messages.map((message) => message.id)).toEqual([
+      next.id
+    ])
+  })
+
+  it('enforces one active batch using the same derived view and permits history', () => {
+    const db = open(':memory:')
+    const run = db.createRun({
+      objective: 'constraint',
+      coordinatorHandle: 'coord',
+      coordinatorPaneKey: 'tab:11111111-1111-4111-8111-111111111111'
+    })
+    const message = db.insertMessage({
+      runId: run.id,
+      from: 'worker',
+      to: `run:${run.id}`,
+      subject: 'one'
+    })
+    const params = { runId: run.id, consumerGeneration: run.consumer_generation }
+    const first = db.getDeliveryRaw(db.getOrCreateRunDelivery(params)!.delivery.id)!
+    const insert = db.db.prepare(`INSERT INTO deliveries
+      (id, run_id, mailbox_handle, consumer_generation, message_ids, acknowledged_at, status)
+      VALUES (?, ?, ?, ?, ?, ?, ?)`)
+    const values = [run.id, `run:${run.id}`, run.consumer_generation, JSON.stringify([message.id])]
+    expect(() => insert.run('duplicate', ...values, null, 'outstanding')).toThrow(
+      'Mailbox already has an outstanding delivery'
+    )
+    expect(() =>
+      insert.run('ack_history', ...values, '2026-01-01 00:00:00', 'acknowledged')
+    ).not.toThrow()
+    expect(() => insert.run('fenced_history', ...values, null, 'fenced')).not.toThrow()
+    db.markAsRead([message.id])
+    expect(() => insert.run('consumed_history', ...values, null, 'outstanding')).not.toThrow()
+    expect(db.getDeliveryRaw(first.id)).toEqual(first)
+    expect(db.hasOutstandingRunDelivery(run.id)).toBe(false)
+  })
+
+  it('shares a single batch across connections and rejects replaced consumers after it is consumed', () => {
+    const path = databasePath()
+    const first = open(path)
+    const run = first.createRun({
+      objective: 'connections',
+      coordinatorHandle: 'coord',
+      coordinatorPaneKey: 'tab:11111111-1111-4111-8111-111111111111'
+    })
+    const params = { runId: run.id, consumerGeneration: run.consumer_generation }
+    const message = first.insertMessage({
+      runId: run.id,
+      from: 'worker',
+      to: `run:${run.id}`,
+      subject: 'one'
+    })
+    const second = open(path)
+    const batch = first.getOrCreateRunDelivery(params)!
+    expect(second.getOrCreateRunDelivery(params)?.delivery.id).toBe(batch.delivery.id)
+    second.markAsRead([message.id])
+    const replacement = second.bindRun({
+      runId: run.id,
+      coordinatorHandle: 'replacement',
+      coordinatorPaneKey: 'other:22222222-2222-4222-9222-222222222222'
+    })!
+    expect(first.getDeliveryRaw(batch.delivery.id)).toMatchObject({
+      status: 'fenced',
+      acknowledged_at: null
+    })
+    first.insertMessage({ runId: run.id, from: 'worker', to: `run:${run.id}`, subject: 'next' })
+    expect(() => first.getOrCreateRunDelivery(params)).toThrow(
+      expect.objectContaining({ code: 'consumer_fenced' })
+    )
+    expect(() =>
+      first.acknowledgeRunDelivery({ ...params, deliveryId: batch.delivery.id })
+    ).toThrow(expect.objectContaining({ code: 'consumer_fenced' }))
+    expect(
+      second.getOrCreateRunDelivery({
+        ...params,
+        consumerGeneration: replacement.consumer_generation
+      })?.messages[0].subject
+    ).toBe('next')
+  })
+
+  it.each(['dispatch', 'attachment'] as const)(
+    'fences a stale %s consumer across connections even after its batch is read',
+    (consumerSource) => {
+      const path = databasePath()
+      const db = open(path)
+      const run = db.createRun({
+        objective: 'worker connections',
+        coordinatorHandle: 'coord',
+        coordinatorPaneKey: 'tab:11111111-1111-4111-8111-111111111111'
+      })
+      const dispatchId =
+        consumerSource === 'dispatch'
+          ? createRootDispatch(db, db.createTask({ spec: 'work', runId: run.id }).id, 'worker').id
+          : 'ctx_remote'
+      if (consumerSource === 'attachment') {
+        db.createRemoteDispatchAttachment({
+          runId: run.id,
+          dispatchId,
+          taskId: 'task_remote',
+          homePeerFingerprint: 'peer',
+          runtimeEpoch: 'epoch',
+          protocolVersion: 1,
+          mutationReceipt: {
+            callerFingerprint: 'peer',
+            requestId: 'attach',
+            method: 'orchestration.federationAttachStart',
+            payloadHash: 'hash'
+          }
+        })
+      }
+      const mailboxHandle = `dispatch:${dispatchId}`
+      const message = db.insertMessage({
+        runId: run.id,
+        from: 'coord',
+        to: mailboxHandle,
+        subject: 'old'
+      })
+      const params = { runId: run.id, mailboxHandle, consumerGeneration: 0, consumerSource }
+      const batch = db.getOrCreateMailboxDelivery(params)!
+      const peer = open(path)
+      peer.markAsRead([message.id])
+      const authority = {
+        dispatchId,
+        paneKey: 'other:22222222-2222-4222-9222-222222222222',
+        processIncarnation: 'worker:2'
+      }
+      if (consumerSource === 'dispatch') {
+        peer.mintDispatchCapability(authority)
+      } else {
+        peer.prepareRemoteAttachmentAuthority({
+          ...authority,
+          worktreeId: 'folder',
+          terminalHandle: 'replacement',
+          setupState: 'not_applicable',
+          effects: []
+        })
+      }
+      expect(db.getDeliveryRaw(batch.delivery.id)).toMatchObject({
+        status: 'fenced',
+        acknowledged_at: null
+      })
+      peer.insertMessage({ runId: run.id, from: 'coord', to: mailboxHandle, subject: 'next' })
+      expect(() => db.getOrCreateMailboxDelivery(params)).toThrow(
+        expect.objectContaining({ code: 'consumer_fenced' })
+      )
+      expect(() =>
+        db.acknowledgeMailboxDelivery({ ...params, deliveryId: batch.delivery.id })
+      ).toThrow(expect.objectContaining({ code: 'consumer_fenced' }))
+      expect(
+        peer.getOrCreateMailboxDelivery({ ...params, consumerGeneration: 1 })?.messages[0].subject
+      ).toBe('next')
+    }
+  )
+
+  it('keeps the pre-v41 column and index shape a downgraded binary reads', () => {
+    const db = open(':memory:')
+    expect(
+      (db.db.pragma('table_info(deliveries)') as { name: string }[]).map((c) => c.name)
+    ).toContain('status')
+    const index = db.db
+      .prepare("SELECT sql FROM sqlite_master WHERE name = 'idx_deliveries_one_outstanding'")
+      .get() as { sql: string }
+    expect(index.sql).not.toContain('UNIQUE')
+    expect(index.sql).toContain("status = 'outstanding' AND mailbox_handle != ''")
+    // Why: a v40 binary probes exactly these objects before trusting the stamp; nothing it needs is gone.
+    expect(resolveOrchestrationMigrationStartVersion(db.db, SCHEMA_VERSION, 40)).toBe(
+      SCHEMA_VERSION
+    )
+  })
+
+  it('recreates a missing derived view on reopen without changing batch records', () => {
+    const path = databasePath()
+    const db = open(path)
+    db.db.exec('DROP VIEW outstanding_deliveries')
+    const reopened = open(path)
+    expect(reopened.hasOutstandingMailboxDelivery('run:missing')).toBe(false)
+    expect(
+      resolveOrchestrationMigrationStartVersion(reopened.db, SCHEMA_VERSION, SCHEMA_VERSION)
+    ).toBe(SCHEMA_VERSION)
+  })
+})
@@ -0,0 +1,9 @@
+import type { OrchestrationDb } from '../orchestration-db'
+
+// Tests that hand-edit the deliveries table must drop the derived objects first: SQLite refuses
+// DROP COLUMN / RENAME while a trigger or view still references the table. Reopening recreates them.
+export function dropDerivedDeliverySchema(db: OrchestrationDb['db']): void {
+  db.exec(
+    'DROP TRIGGER IF EXISTS trg_deliveries_one_outstanding; DROP VIEW IF EXISTS outstanding_deliveries;'
+  )
+}
@@ -0,0 +1,36 @@
+import type { OrchestrationDb } from '../orchestration-db'
+
+// Why: eligibility comes from unread membership, so a fully read but unacknowledged batch must not
+// block the next insert. The index keeps its pre-v41 name and predicate so downgraded binaries'
+// IF NOT EXISTS create and schema probes still pass; only uniqueness is dropped.
+export const OUTSTANDING_MAILBOX_INDEX_SQL = `
+  CREATE INDEX IF NOT EXISTS idx_deliveries_one_outstanding
+    ON deliveries(mailbox_handle) WHERE status = 'outstanding' AND mailbox_handle != '';
+`
+
+export const DERIVED_DELIVERY_SCHEMA_SQL = `
+  CREATE VIEW IF NOT EXISTS outstanding_deliveries AS
+    SELECT * FROM deliveries
+    WHERE status = 'outstanding'
+      AND EXISTS (
+        SELECT 1 FROM json_each(deliveries.message_ids) AS member
+        JOIN messages ON messages.id = member.value WHERE messages.read = 0
+      );
+  CREATE TRIGGER IF NOT EXISTS trg_deliveries_one_outstanding
+    AFTER INSERT ON deliveries
+    WHEN NEW.mailbox_handle != '' AND EXISTS (
+      SELECT 1 FROM outstanding_deliveries WHERE mailbox_handle = NEW.mailbox_handle LIMIT 1 OFFSET 1
+    )
+    BEGIN
+      SELECT RAISE(ABORT, 'Mailbox already has an outstanding delivery');
+    END;
+`
+
+export function migrateV41(this: OrchestrationDb, current: number): void {
+  if (current >= 41) {
+    return
+  }
+  this.db.exec(
+    `DROP INDEX IF EXISTS idx_deliveries_one_outstanding;\n${OUTSTANDING_MAILBOX_INDEX_SQL}`
+  )
+}
@@ -11,6 +11,7 @@ import { migrateV37 } from './migrate-v37'
 import { migrateV38 } from './migrate-v38'
 import { migrateV39 } from './migrate-v39'
 import { migrateV40 } from './migrate-v40'
+import { DERIVED_DELIVERY_SCHEMA_SQL, migrateV41 } from './migrate-v41'

 // Why: CREATE TABLE IF NOT EXISTS won't alter existing DBs; migrate in a txn that bumps user_version only on success (atomic all-or-nothing).
 export function migrate(this: OrchestrationDb): void {
@@ -22,6 +23,9 @@ export function migrate(this: OrchestrationDb): void {

  this.db.exec('BEGIN IMMEDIATE')
  try {
+    this.db.exec(
+      'DROP TRIGGER IF EXISTS trg_deliveries_one_outstanding; DROP VIEW IF EXISTS outstanding_deliveries;'
+    )
    applySchemaMigrationsV2ToV12.call(this, current)
    applySchemaMigrationsV13ToV30.call(this, current)
    migrateMailboxPointerEnterV33.call(this, current)
@@ -32,7 +36,11 @@ export function migrate(this: OrchestrationDb): void {
    migrateV38.call(this, current)
    migrateV39.call(this, current)
    migrateV40.call(this, current)
+    // Why: older steps recreate the unique index; v41 must run after them.
+    migrateV41.call(this, current)
    this.createMailboxDeliveryIndexesIfPossible()
+    // Why: rebuild steps above RENAME the table, which SQLite refuses while a view names it.
+    this.db.exec(DERIVED_DELIVERY_SCHEMA_SQL)
    this.db.pragma(`user_version = ${SCHEMA_VERSION}`)
    this.db.exec('COMMIT')
  } catch (err) {
@@ -1,4 +1,5 @@
 import type { OrchestrationDb } from '../orchestration-db'
+import { OUTSTANDING_MAILBOX_INDEX_SQL } from './migrate-v41'

 export function hasColumn(this: OrchestrationDb, table: string, column: string): boolean {
  const rows = this.db.pragma(`table_info(${table})`) as { name: string }[]
@@ -7,12 +8,7 @@ export function hasColumn(this: OrchestrationDb, table: string, column: string):

 export function createMailboxDeliveryIndexesIfPossible(this: OrchestrationDb): void {
  if (this.hasColumn('deliveries', 'mailbox_handle')) {
-    // Excluding '' trades the pre-v34 per-run one-outstanding backstop for downgraded binaries; the
-    // app-level BEGIN IMMEDIATE still serializes one process.
-    this.db.exec(`
-      CREATE UNIQUE INDEX IF NOT EXISTS idx_deliveries_one_outstanding
-        ON deliveries(mailbox_handle) WHERE status = 'outstanding' AND mailbox_handle != '';
-    `)
+    this.db.exec(OUTSTANDING_MAILBOX_INDEX_SQL)
  }
  const hasDeliveredAt = this.hasColumn('messages', 'delivered_at')
  if (hasDeliveredAt) {
@@ -75,7 +75,7 @@ export function prepareStartingWorkerAuthority(
        `Dispatch ${params.dispatchId} is not starting.`
      )
    }
-    this.fenceOutstandingMailboxDelivery(`dispatch:${params.dispatchId}`)
+    this.fenceUnacknowledgedMailboxDeliveries(`dispatch:${params.dispatchId}`)
    const workerUpdate = this.db
      .prepare(
        `UPDATE worker_dispatches
@@ -3,6 +3,7 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { afterEach, describe, expect, it } from 'vitest'
 import Database from '../../sqlite/sync-database'
+import { dropDerivedDeliverySchema } from './db/schema/derived-delivery-test-fixture'
 import { OrchestrationDb } from './db'
 import { SCHEMA_VERSION } from './db/contract-constants'
 import { createRootDispatch } from './db/root-dispatch-test-fixture'
@@ -48,6 +49,7 @@ describe('OrchestrationDb v35 to v36 migration', () => {
    seed.close()

    const raw = new Database(dbPath)
+    dropDerivedDeliverySchema(raw)
    raw.exec(`
      ALTER TABLE dispatch_contexts DROP COLUMN consumer_generation;
      ALTER TABLE remote_dispatch_attachments DROP COLUMN consumer_generation;
@@ -78,6 +80,7 @@ describe('OrchestrationDb v35 to v36 migration', () => {
  it('does not send a v35 stamp back to the pre-Run repair floor', () => {
    const v35 = createV35Database()
    const raw = new Database(v35.path)
+    dropDerivedDeliverySchema(raw)
    try {
      expect(resolveOrchestrationMigrationStartVersion(raw, 35, SCHEMA_VERSION)).toBe(35)
    } finally {
@@ -88,6 +91,7 @@ describe('OrchestrationDb v35 to v36 migration', () => {
  it('repairs a database stamped v36 that never got the columns', () => {
    const v35 = createV35Database()
    const raw = new Database(v35.path)
+    dropDerivedDeliverySchema(raw)
    raw.pragma('user_version = 36')
    try {
      // Why: the skew repair is the only thing that catches a partially-written v36.
@@ -3,6 +3,7 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { afterEach, describe, expect, it } from 'vitest'
 import Database from '../../sqlite/sync-database'
+import { dropDerivedDeliverySchema } from './db/schema/derived-delivery-test-fixture'
 import { OrchestrationDb } from './db'
 import { resolveOrchestrationMigrationStartVersion } from './orchestration-schema-version-skew'
 import { SCHEMA_VERSION } from './db/contract-constants'
@@ -26,6 +27,7 @@ describe('federation acknowledgment migration', () => {
    db = undefined

    const oldDb = new Database(dbPath)
+    dropDerivedDeliverySchema(oldDb)
    oldDb.exec('ALTER TABLE federated_dispatches DROP COLUMN to_home_acknowledged_sequence')
    oldDb.pragma('user_version = 26')
    expect(resolveOrchestrationMigrationStartVersion(oldDb, 26, 28)).toBe(26)
@@ -3,6 +3,7 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { afterEach, describe, expect, it } from 'vitest'
 import Database from '../../sqlite/sync-database'
+import { dropDerivedDeliverySchema } from './db/schema/derived-delivery-test-fixture'
 import { OrchestrationDb } from './db'
 import { resolveOrchestrationMigrationStartVersion } from './orchestration-schema-version-skew'
 import { SCHEMA_VERSION } from './db/contract-constants'
@@ -32,6 +33,7 @@ describe('nested worker depth migration (v30)', () => {
    fresh.close()

    const oldDb = new Database(dbPath)
+    dropDerivedDeliverySchema(oldDb)
    oldDb.exec('ALTER TABLE dispatch_contexts DROP COLUMN depth')
    oldDb.exec('ALTER TABLE remote_dispatch_attachments DROP COLUMN depth')
    oldDb.exec('ALTER TABLE remote_dispatch_attachments DROP COLUMN home_run_id')
@@ -95,6 +97,7 @@ describe('nested worker depth migration (v30)', () => {
    // replay migrations from v6 instead of starting at 29.
    const dbPath = createV29Database()
    const oldDb = new Database(dbPath)
+    dropDerivedDeliverySchema(oldDb)
    expect(resolveOrchestrationMigrationStartVersion(oldDb, 29, SCHEMA_VERSION)).toBe(29)
    oldDb.close()
  })
@@ -0,0 +1,146 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import { OrchestrationDb } from './db'
+import { createRootDispatch } from './db/root-dispatch-test-fixture'
+import { reconcileLifecycleMessage } from './lifecycle-reconciliation'
+
+describe('mailbox delivery consumption', () => {
+  let db: OrchestrationDb
+  afterEach(() => db?.close())
+
+  function setup() {
+    db = new OrchestrationDb(':memory:')
+    const run = db.createRun({
+      objective: 'Retired delivery',
+      coordinatorHandle: 'term_coord',
+      coordinatorPaneKey: 'tab:11111111-1111-4111-8111-111111111111'
+    })
+    const params = { runId: run.id, consumerGeneration: run.consumer_generation }
+    const insert = (subject: string) =>
+      db.insertMessage({ runId: run.id, from: 'worker', to: `run:${run.id}`, subject })
+    return { run, params, insert }
+  }
+
+  it('advances past a heartbeat batch when completion suppresses its contents', () => {
+    const { run, params } = setup()
+    const task = db.createTask({ runId: run.id, spec: 'work' })
+    const dispatch = createRootDispatch(db, task.id, 'worker')
+    const insert = (type: 'heartbeat' | 'worker_done') =>
+      db.insertMessage({
+        runId: run.id,
+        from: 'worker',
+        to: `run:${run.id}`,
+        subject: type,
+        type,
+        payload: JSON.stringify({ taskId: task.id, dispatchId: dispatch.id, outcome: 'succeeded' })
+      })
+    insert('heartbeat')
+    const first = db.getOrCreateRunDelivery(params)!
+    const done = insert('worker_done')
+    expect(reconcileLifecycleMessage(db, done).action).toBe('completed')
+    expect(db.getDeliveryRaw(first.delivery.id)?.acknowledged_at).toBeNull()
+    expect(db.hasOutstandingRunDelivery(run.id)).toBe(false)
+    expect(
+      db
+        .getOrCreateRunDelivery({ ...params, wakeTypes: ['worker_done'] })
+        ?.messages.map((m) => m.id)
+    ).toEqual([done.id])
+    expect(db.acknowledgeRunDelivery({ ...params, deliveryId: first.delivery.id }).duplicate).toBe(
+      false
+    )
+  })
+
+  it('ignores a fully read batch without rewriting it', () => {
+    const { params, insert } = setup()
+    const old = insert('old')
+    const first = db.getOrCreateRunDelivery(params)!
+    db.db.prepare('UPDATE messages SET read = 1 WHERE id = ?').run(old.id)
+    const next = insert('next')
+    const current = db.getOrCreateRunDelivery(params)!
+    expect(current.messages.map((m) => m.id)).toEqual([next.id])
+    expect(current.replayed).toBe(false)
+    expect(db.getDeliveryRaw(first.delivery.id)?.acknowledged_at).toBeNull()
+  })
+
+  it('preserves the entire replay batch while any member is unread', () => {
+    const { params, insert } = setup()
+    const a = insert('a')
+    const b = insert('b')
+    const first = db.getOrCreateRunDelivery(params)!
+    db.markAsReadAndDelivered([a.id])
+    insert('later')
+    const replay = db.getOrCreateRunDelivery(params)!
+    expect(replay.delivery.id).toBe(first.delivery.id)
+    expect(replay.messages.map((m) => m.id)).toEqual([a.id, b.id])
+    expect(replay.replayed).toBe(true)
+  })
+
+  it('derives eligibility again when a read transaction rolls back', () => {
+    const { params, insert } = setup()
+    const message = insert('old')
+    const first = db.getOrCreateRunDelivery(params)!
+    const before = db.getDeliveryRaw(first.delivery.id)
+    db.db.exec('BEGIN')
+    db.markAsReadAndDelivered([message.id])
+    expect(db.getDeliveryRaw(first.delivery.id)?.acknowledged_at).toBeNull()
+    expect(db.hasOutstandingRunDelivery(params.runId)).toBe(false)
+    db.db.exec('ROLLBACK')
+    expect(db.hasOutstandingRunDelivery(params.runId)).toBe(true)
+    expect(db.getDeliveryRaw(first.delivery.id)).toEqual(before)
+    expect(db.getMessageById(message.id)?.read).toBe(0)
+    expect(db.getDeliveryRaw(first.delivery.id)?.acknowledged_at).toBeNull()
+  })
+
+  it('explains the delivery ID contract for invalid acknowledgements without consuming mail', () => {
+    const { params, insert } = setup()
+    const message = insert('pending')
+    const first = db.getOrCreateRunDelivery(params)!
+    expect(() => db.acknowledgeRunDelivery({ ...params, deliveryId: message.id })).toThrow(
+      '--ack requires a delivery_* ID returned by orchestration check; process the entire batch before acknowledging.'
+    )
+    expect(db.getMessageById(message.id)?.read).toBe(0)
+    expect(db.getDeliveryRaw(first.delivery.id)?.acknowledged_at).toBeNull()
+  })
+
+  it('checks the consumer generation even when the prior batch is already read', () => {
+    const { run, params, insert } = setup()
+    const message = insert('old')
+    const first = db.getOrCreateRunDelivery(params)!
+    db.db.prepare('UPDATE messages SET read = 1 WHERE id = ?').run(message.id)
+    expect(() =>
+      db.getOrCreateMailboxDelivery({
+        ...params,
+        mailboxHandle: `run:${run.id}`,
+        consumerGeneration: params.consumerGeneration + 1
+      })
+    ).toThrow(expect.objectContaining({ code: 'consumer_fenced' }))
+    expect(db.getDeliveryRaw(first.delivery.id)?.acknowledged_at).toBeNull()
+  })
+
+  it.each(['markAsRead', 'markAsReadAndDelivered'] as const)(
+    '%s releases dispatch mail without changing a different mailbox',
+    (method) => {
+      const { run, params, insert } = setup()
+      insert('coordinator mail')
+      db.getOrCreateRunDelivery(params)!
+      const task = db.createTask({ runId: run.id, spec: 'worker mail' })
+      const dispatch = createRootDispatch(db, task.id, 'worker')
+      const mailboxHandle = `dispatch:${dispatch.id}`
+      const message = db.insertMessage({
+        runId: run.id,
+        from: 'term_coord',
+        to: mailboxHandle,
+        subject: 'worker mail'
+      })
+      const workerParams = {
+        ...params,
+        mailboxHandle,
+        consumerGeneration: dispatch.consumer_generation
+      }
+      db.getOrCreateMailboxDelivery(workerParams)!
+      db[method]([message.id])
+      expect(db.hasOutstandingMailboxDelivery(mailboxHandle)).toBe(false)
+      expect(db.getOrCreateMailboxDelivery(workerParams)).toBeUndefined()
+      expect(db.hasOutstandingRunDelivery(run.id)).toBe(true)
+    }
+  )
+})
@@ -0,0 +1,56 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import { OrchestrationDb } from './db'
+
+describe('delivery eligibility derived from messages', () => {
+  let db: OrchestrationDb
+  afterEach(() => db?.close())
+
+  function setup() {
+    db = new OrchestrationDb(':memory:')
+    const run = db.createRun({
+      objective: 'Derived delivery',
+      coordinatorHandle: 'coord',
+      coordinatorPaneKey: 'tab:11111111-1111-4111-8111-111111111111'
+    })
+    const params = { runId: run.id, consumerGeneration: run.consumer_generation }
+    const message = db.insertMessage({
+      runId: run.id,
+      from: 'worker',
+      to: `run:${run.id}`,
+      subject: 'old'
+    })
+    const first = db.getOrCreateRunDelivery(params)!
+    return { run, params, message, first }
+  }
+
+  it.each(['read mutation', 'lifecycle suppression', 'direct SQL'])(
+    '%s changes eligibility without updating the batch',
+    (path) => {
+      const { run, params, message, first } = setup()
+      const before = db.getDeliveryRaw(first.delivery.id)
+      if (path === 'read mutation') {
+        db.markAsRead([message.id])
+      } else if (path === 'lifecycle suppression') {
+        db.markAsReadAndDelivered([message.id])
+      } else {
+        db.db.prepare('UPDATE messages SET read = 1 WHERE id = ?').run(message.id)
+      }
+      expect(db.getDeliveryRaw(first.delivery.id)).toEqual(before)
+      expect(db.hasOutstandingRunDelivery(run.id)).toBe(false)
+      const changes = db.db.prepare('SELECT total_changes() AS n').get()
+      expect(db.getOrCreateRunDelivery(params)).toBeUndefined()
+      expect(db.db.prepare('SELECT total_changes() AS n').get()).toEqual(changes)
+      expect(db.getDeliveryRaw(first.delivery.id)?.acknowledged_at).toBeNull()
+    }
+  )
+
+  it('records only an actual acknowledgement, including after all messages were suppressed', () => {
+    const { params, message, first } = setup()
+    db.markAsReadAndDelivered([message.id])
+    expect(db.getDeliveryRaw(first.delivery.id)?.acknowledged_at).toBeNull()
+    const ack = { ...params, deliveryId: first.delivery.id }
+    expect(db.acknowledgeRunDelivery(ack).duplicate).toBe(false)
+    expect(db.getDeliveryRaw(first.delivery.id)?.acknowledged_at).not.toBeNull()
+    expect(db.acknowledgeRunDelivery(ack).duplicate).toBe(true)
+  })
+})
@@ -3,6 +3,7 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import Database from '../../sqlite/sync-database'
 import { LEGACY_RUN_ID, OrchestrationDb } from './db'
+import { dropDerivedDeliverySchema } from './db/schema/derived-delivery-test-fixture'
 import { createRootDispatch } from './db/root-dispatch-test-fixture'

 export type LegacyStorageCutoverFixture = {
@@ -175,6 +176,7 @@ export function createLegacyStorageCutoverFixture(): {
  first.close()

  const raw = new Database(dbPath)
+  dropDerivedDeliverySchema(raw)
  const legacyDeliveryId = 'delivery_legacy_outstanding'
  raw
    .prepare(
@@ -6,6 +6,7 @@ import Database from '../../sqlite/sync-database'
 import { LEGACY_CONTRACT_VERSION, LEGACY_RUN_ID, OrchestrationDb } from './db'
 import { resolveOrchestrationMigrationStartVersion } from './orchestration-schema-version-skew'
 import { createRootDispatch } from './db/root-dispatch-test-fixture'
+import { dropDerivedDeliverySchema } from './db/schema/derived-delivery-test-fixture'
 import { SCHEMA_VERSION } from './db/contract-constants'

 describe('OrchestrationDb version-skew migration', () => {
@@ -267,6 +268,7 @@ describe('OrchestrationDb version-skew migration', () => {
    db = undefined

    const raw = new Database(dbPath)
+    dropDerivedDeliverySchema(raw)
    raw.exec(`
      DROP INDEX idx_deliveries_one_outstanding;
      ALTER TABLE deliveries DROP COLUMN mailbox_handle;
@@ -303,6 +305,7 @@ describe('OrchestrationDb version-skew migration', () => {
    db = undefined

    const raw = new Database(dbPath)
+    dropDerivedDeliverySchema(raw)
    raw.exec(`
      DROP INDEX idx_deliveries_one_outstanding;
      ALTER TABLE deliveries DROP COLUMN mailbox_handle;
@@ -374,15 +377,7 @@ describe('OrchestrationDb version-skew migration', () => {
    expect(deliveryIndexes.map(({ name }) => name)).toEqual(
      expect.arrayContaining(['idx_deliveries_one_outstanding', 'idx_deliveries_run_created'])
    )
-    expect(() =>
-      db!.db
-        .prepare(
-          `INSERT INTO deliveries (
-             id, run_id, mailbox_handle, consumer_generation, message_ids
-           ) VALUES (?, ?, ?, ?, '[]')`
-        )
-        .run('delivery_v34_duplicate', run.id, `run:${run.id}`, run.consumer_generation)
-    ).toThrow(/UNIQUE constraint failed/)
+    expect(db.hasOutstandingRunDelivery(run.id)).toBe(false)
  })

  it('cleans additive lifecycle rows when a v30 writer resets tasks before re-upgrade', () => {
@@ -489,6 +484,7 @@ describe('OrchestrationDb version-skew migration', () => {
    db = undefined

    const raw = new Database(dbPath)
+    dropDerivedDeliverySchema(raw)
    raw.exec(`
      DROP TABLE deliveries;
      CREATE TABLE deliveries (
@@ -567,6 +563,7 @@ describe('OrchestrationDb version-skew migration', () => {
    db = undefined

    const raw = new Database(dbPath)
+    dropDerivedDeliverySchema(raw)
    raw.exec(`
      DROP INDEX IF EXISTS idx_deliveries_one_outstanding;
      CREATE UNIQUE INDEX idx_deliveries_one_outstanding
@@ -0,0 +1,46 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import { createOrchestrationRpcHarness } from '../rpc-test-harness'
+
+describe('Run delivery history', () => {
+  const h = createOrchestrationRpcHarness()
+  afterEach(() => h.cleanup())
+
+  it('does not label filtered history as an acknowledgeable delivery', async () => {
+    const { db, ctx, activeRunId } = h.setup()
+    const params = { terminal: 'term_coord', run: activeRunId, all: true }
+    db.insertMessage({
+      from: 'worker',
+      to: `run:${activeRunId}`,
+      runId: activeRunId,
+      subject: 'waiting'
+    })
+    expect(await h.call('orchestration.check', params, ctx)).toMatchObject({
+      count: 1
+    })
+    expect(db.hasOutstandingRunDelivery(activeRunId!)).toBe(false)
+    const delivery = db.getOrCreateRunDelivery({
+      runId: activeRunId!,
+      consumerGeneration: db.getRun(activeRunId!)!.consumer_generation
+    })!
+    db.insertMessage({
+      from: 'worker',
+      to: `run:${activeRunId}`,
+      runId: activeRunId,
+      subject: 'later completion',
+      type: 'worker_done'
+    })
+    const history = await h.call(
+      'orchestration.check',
+      {
+        ...params,
+        format: true,
+        types: 'worker_done'
+      },
+      ctx
+    )
+    expect(history).toMatchObject({ count: 1, messages: [{ subject: 'later completion' }] })
+    expect(history).not.toHaveProperty('deliveryId')
+    expect(db.hasOutstandingRunDelivery(activeRunId!)).toBe(true)
+    expect(db.getMessageById(delivery.messages[0].id)?.read).toBe(0)
+  })
+})
@@ -172,6 +172,7 @@ export async function checkWorkerMailbox(args: {
          runId: deliveryRunId,
          mailboxHandle: address,
          consumerGeneration: workerMailbox.generation,
+          consumerSource: activeDispatch ? 'dispatch' : 'attachment',
          deliveryId: params.ack
        })
      : undefined
@@ -181,16 +182,12 @@ export async function checkWorkerMailbox(args: {
  const showAll = params.all === true || (params.unread === false && params.peek !== true)
  const readPeek = () => db.getUnreadMessages(address, typeFilter)
  const readDelivery = (wakeTypes?: MessageType[]) => {
-    // Why: re-read live, or a re-attach landing on an await above mints a Delivery at a generation
-    // the row has already left, which then fences the legitimate worker on every later check.
-    if (readCurrentGeneration() !== workerMailbox.generation) {
-      throw dispatchFenced()
-    }
    try {
      return db.getOrCreateMailboxDelivery({
        runId: deliveryRunId,
        mailboxHandle: address,
        consumerGeneration: workerMailbox.generation,
+        consumerSource: activeDispatch ? 'dispatch' : 'attachment',
        wakeTypes
      })
    } catch (error) {
@@ -1,5 +1,8 @@
 import { describe, expect, it } from 'vitest'
-import { prepareOrchestrationCheckOutput } from './orchestration-check-output'
+import {
+  formatOrchestrationCheckText,
+  prepareOrchestrationCheckOutput
+} from './orchestration-check-output'

 describe('prepareOrchestrationCheckOutput', () => {
  it('keeps mixed read-only mail safe and current Run replies executable', () => {
@@ -40,3 +43,19 @@ describe('prepareOrchestrationCheckOutput', () => {
    expect(prepared.formatted).not.toContain('unsafe stale formatter output')
  })
 })
+
+describe('formatted delivery acknowledgment', () => {
+  it('retains the delivery ID above formatted message bodies', () => {
+    expect(
+      formatOrchestrationCheckText(
+        {
+          messages: [{ id: 'msg_one', from_handle: 'worker' }],
+          count: 1,
+          deliveryId: 'delivery_one',
+          formatted: 'Message body'
+        },
+        'term_coord'
+      )
+    ).toBe('Delivery delivery_one\nMessage body')
+  })
+})
@@ -81,7 +81,7 @@ export function formatOrchestrationCheckText(
    : ''
  const deliveryNotice = formatCurrentDeliveryNotice(prepared.legacyCompatibility?.currentDelivery)
  if (prepared.formatted) {
-    return `${legacyHeader}${prepared.formatted}${deliveryNotice}`
+    return `${legacyHeader}${prepared.deliveryId ? `Delivery ${prepared.deliveryId}\n` : ''}${prepared.formatted}${deliveryNotice}`
  }
  if (prepared.count === 0) {
    if (prepared.timedOut) {
@@ -0,0 +1,68 @@
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { expect, test } from 'vitest'
+import { OrchestrationDb } from '../../../src/main/runtime/orchestration/db'
+import { importReleaseCheckoutModule, materializeReleaseCheckout } from './release-checkout'
+
+// Pin the last pre-v41 implementation: this contract specifically exercises status-only readers.
+const PRE_V41 = 'aac38d698ff75ac4c8658addab48ef5a83617619'
+
+test('pre-v41 code opens, acknowledges and writes a v41 database, then current code reopens it', async () => {
+  const checkout = await materializeReleaseCheckout(PRE_V41)
+  const baseline = await importReleaseCheckoutModule(
+    checkout,
+    'src/main/runtime/orchestration/db.ts'
+  )
+  const OldDb = baseline.OrchestrationDb as typeof OrchestrationDb
+  const directory = mkdtempSync(join(tmpdir(), 'orca-delivery-downgrade-'))
+  const path = join(directory, 'orchestration.db')
+  let db: OrchestrationDb | undefined
+  try {
+    db = new OldDb(path)
+    expect(db.db.pragma('user_version', { simple: true })).toBe(40)
+    const run = db.createRun({
+      objective: 'downgrade round trip',
+      coordinatorHandle: 'coord',
+      coordinatorPaneKey: 'tab:11111111-1111-4111-8111-111111111111'
+    })
+    const params = { runId: run.id, consumerGeneration: run.consumer_generation }
+    const insert = (subject: string) =>
+      db!.insertMessage({
+        runId: run.id,
+        from: 'worker',
+        to: `run:${run.id}`,
+        subject
+      })
+    const oldMessage = insert('obsolete heartbeat')
+    const oldBatch = db.getOrCreateRunDelivery(params)!
+    db.close()
+
+    db = new OrchestrationDb(path)
+    db.markAsRead([oldMessage.id])
+    const completion = insert('completion')
+    const currentBatch = db.getOrCreateRunDelivery(params)!
+    expect(currentBatch.messages.map((message) => message.id)).toEqual([completion.id])
+    expect(currentBatch.delivery.id).not.toBe(oldBatch.delivery.id)
+    db.close()
+
+    db = new OldDb(path)
+    expect(db.db.pragma('user_version', { simple: true })).toBe(41)
+    // Old readers retain their original replay semantics, but can acknowledge either stored batch.
+    db.acknowledgeRunDelivery({ ...params, deliveryId: oldBatch.delivery.id })
+    expect(db.getOrCreateRunDelivery(params)?.delivery.id).toBe(currentBatch.delivery.id)
+    db.acknowledgeRunDelivery({ ...params, deliveryId: currentBatch.delivery.id })
+    const next = insert('written after downgrade')
+    const nextBatch = db.getOrCreateRunDelivery(params)!
+    expect(nextBatch.messages.map((message) => message.id)).toEqual([next.id])
+    db.close()
+
+    db = new OrchestrationDb(path)
+    expect(db.getOrCreateRunDelivery(params)?.delivery.id).toBe(nextBatch.delivery.id)
+    db.acknowledgeRunDelivery({ ...params, deliveryId: nextBatch.delivery.id })
+    expect(db.getOrCreateRunDelivery(params)).toBeUndefined()
+  } finally {
+    db?.close()
+    rmSync(directory, { recursive: true, force: true })
+  }
+})