refactor(orchestration): derive delivery eligibility from messages

src/cli/specs/orchestration.test.ts

@@ -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.'
       ])
     )
   })

src/cli/specs/orchestration.ts

@@ -109,9 +109,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.'
     ]
   },
   {

src/main/runtime/orchestration/db/contract-constants.ts

@@ -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

src/main/runtime/orchestration/db/dispatch-context/dispatch-capability.ts

@@ -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')

src/main/runtime/orchestration/db/dispatch-mailbox-consumer-fencing.test.ts

@@ -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()

src/main/runtime/orchestration/db/federation/remote-dispatch-attachment-authority.ts

@@ -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'

src/main/runtime/orchestration/db/messages/mailbox-consumer-lifecycle-fencing.test.ts

@@ -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])
+    }
+  )
+})

src/main/runtime/orchestration/db/messages/mailbox-consumer.ts

@@ -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.')
+  }
+}