Merge remote-tracking branch 'origin/main' into brennanb2025/unify-agent-session-launch

cloud/apps/relay/src/assignment-store.ts

@@ -362,6 +362,9 @@ export const REGIONAL_REHOME_QUARANTINE_FAILURES = 3
 export const REGIONAL_REHOME_QUARANTINE_MS = 15 * 60_000
 const REGIONAL_REHOME_QUARANTINE_EXCLUSION_LIMIT = 50
 const REGIONAL_REHOME_QUARANTINE_MEMORY_LIMIT = 1_000
+// Consecutive drain-dispatch failures that latch the durable control off. Any
+// drain receipt and any enable reset it, so it reads "dispatch is broken now".
+const REGIONAL_REHOME_FAILURE_BUDGET = 3
 const REGIONAL_REHOME_OBSERVATION_MS = 24 * 60 * 60_000
 const ASSIGNMENT_LOCK_RETRY_MAX_DELAY_MS = 50
 type AssignmentInventoryScope = 'none' | 'general' | 'all'
@@ -4992,6 +4995,20 @@ export class RelayAssignmentStore {
           now
         ]
       )
+      if (input.enabled) {
+        // A budget spent under a previous enable is not evidence about this one.
+        // Without this an old counter latches the fresh enable straight back off
+        // on its first transient failure.
+        await transaction.query(
+          `INSERT INTO relay_region_rehome_worker_state
+           (worker_id, next_dispatch_at, paused_until, consecutive_failures, updated_at)
+           VALUES ('global', 0, 0, 0, ?)
+           ON CONFLICT (worker_id) DO UPDATE
+             SET paused_until = 0, consecutive_failures = 0,
+                 updated_at = excluded.updated_at`,
+          [now]
+        )
+      }
       const updated = (
         await transaction.query(
           `SELECT * FROM relay_region_rehome_control WHERE control_id = 'global'`
@@ -5940,7 +5957,11 @@ export class RelayAssignmentStore {
 
   async recordRegionalRehomeDispatchFailure(attemptId: string): Promise<void> {
     const now = this.now()
-    await this.database.transaction(async (transaction) => {
+    const disableLog = await this.database.transaction(async (transaction) => {
+      // Match claim and enable ordering before a spent budget updates the control.
+      await transaction.queryLocked(
+        `SELECT * FROM relay_region_rehome_control WHERE control_id = 'global'`
+      )
       const worker = (
         await transaction.queryLocked(
           `SELECT * FROM relay_region_rehome_worker_state WHERE worker_id = 'global'`
@@ -5952,49 +5973,44 @@ export class RelayAssignmentStore {
           [attemptId]
         )
       )[0]
-      if (!worker || !attempt) return
-      await this.incrementRegionalRehomeWorkerFailure(transaction, worker, now)
-    })
-  }
-
-  async recordRegionalRehomeWorkerFailure(): Promise<void> {
-    const now = this.now()
-    await this.database.transaction(async (transaction) => {
-      await transaction.query(
-        `INSERT INTO relay_region_rehome_worker_state
-         (worker_id, next_dispatch_at, paused_until, consecutive_failures, updated_at)
-         VALUES ('global', 0, 0, 0, ?)
-         ON CONFLICT (worker_id) DO NOTHING`,
-        [now]
-      )
-      const worker = (
-        await transaction.queryLocked(
-          `SELECT * FROM relay_region_rehome_worker_state WHERE worker_id = 'global'`
-        )
-      )[0]!
-      await this.incrementRegionalRehomeWorkerFailure(transaction, worker, now)
+      if (!worker || !attempt) return null
+      return await this.incrementRegionalRehomeWorkerFailure(transaction, worker, now)
     })
+    // Logged after the commit so a rollback cannot fabricate the record.
+    if (disableLog) console.warn(JSON.stringify(disableLog))
   }
 
+  // Returns the durable disable this failure caused, for the caller to log once
+  // its transaction commits; null when the budget survives or was already spent.
   private async incrementRegionalRehomeWorkerFailure(
     transaction: RelayDatabase,
     worker: SqlRow,
     now: number
-  ): Promise<void> {
+  ): Promise<Record<string, string | number> | null> {
     const failures = integer(worker, 'consecutive_failures') + 1
+    const spent = failures >= REGIONAL_REHOME_FAILURE_BUDGET
     await transaction.query(
       `UPDATE relay_region_rehome_worker_state
        SET consecutive_failures = ?, paused_until = ?, updated_at = ?
        WHERE worker_id = 'global'`,
-      [failures, failures >= 3 ? now + 5 * 60_000 : 0, now]
+      [failures, spent ? now + 5 * 60_000 : 0, now]
     )
-    if (failures >= 3) {
-      await transaction.query(
-        `UPDATE relay_region_rehome_control
-         SET generation = generation + 1, enabled = 0, updated_at = ?
-         WHERE control_id = 'global' AND enabled = 1`,
-        [now]
-      )
+    if (!spent) return null
+    const disabled = await transaction.query(
+      `UPDATE relay_region_rehome_control
+       SET generation = generation + 1, enabled = 0, updated_at = ?
+       WHERE control_id = 'global' AND enabled = 1
+       RETURNING generation`,
+      [now]
+    )
+    // The disable is otherwise invisible: inspection only shows enabled=false and
+    // nothing records that the failure budget, not an operator, turned it off.
+    if (disabled.length === 0) return null
+    return {
+      event: 'orca_relay_regional_rehome_failure_budget_disabled',
+      controlGeneration: integer(disabled[0]!, 'generation'),
+      consecutiveFailures: failures,
+      now
     }
   }
 

cloud/apps/relay/src/regional-rehome-postgres.test.ts

@@ -1,4 +1,4 @@
-import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 import { RelayAssignmentStore } from './assignment-store.js'
 import { openRelayDatabase, type RelayDatabase } from './database.js'
 import {
@@ -267,6 +267,91 @@ describePostgres('PostgreSQL regional rehoming', () => {
     )).toEqual([{ count: '1' }])
   })
 
+  it('serializes an enable with a budget-exhausting failure without retries', async () => {
+    const context = await fixture()
+    const attempt = await context.store.claimRegionalRehome()
+    await context.store.recordRegionalRehomeDispatchFailure(attempt!.attemptId)
+    await context.store.recordRegionalRehomeDispatchFailure(attempt!.attemptId)
+    const locked = Promise.withResolvers<void>()
+    const release = Promise.withResolvers<void>()
+    const primaryTransaction = primary.transaction.bind(primary)
+    const secondaryTransaction = secondary.transaction.bind(secondary)
+    let enableTransactions = 0
+    let failureTransactions = 0
+    let enablePid = 0
+    let failurePid = 0
+    const enableSpy = vi.spyOn(primary, 'transaction').mockImplementation((operation, options) =>
+      primaryTransaction(async (transaction) => {
+        enableTransactions++
+        enablePid = Number((await transaction.query('SELECT pg_backend_pid() AS pid'))[0]!.pid)
+        return await operation({
+          dialect: 'postgres',
+          query: transaction.query.bind(transaction),
+          queryLocked: async (sql, params, lockOptions) => {
+            const rows = await transaction.queryLocked(sql, params, lockOptions)
+            if (sql.includes('FROM relay_region_rehome_control')) {
+              locked.resolve()
+              await release.promise
+            }
+            return rows
+          },
+          transaction: transaction.transaction.bind(transaction),
+          close: transaction.close.bind(transaction)
+        })
+      }, options)
+    )
+    const failureSpy = vi.spyOn(secondary, 'transaction').mockImplementation((operation, options) =>
+      secondaryTransaction(async (transaction) => {
+        failureTransactions++
+        failurePid = Number((await transaction.query('SELECT pg_backend_pid() AS pid'))[0]!.pid)
+        return await operation(transaction)
+      }, options)
+    )
+    const enable = context.store.applyRegionalRehomeControl({
+      expectedGeneration: 1,
+      enabled: true,
+      notBefore: context.now(),
+      ratePerMinute: 10,
+      preferenceMaxAgeMs: 24 * 60 * 60_000,
+      hostCooldownMs: 7 * 24 * 60 * 60_000,
+      drainGraceMs: 60_000
+    })
+    let failure: Promise<void> | undefined
+    let outcomes: PromiseSettledResult<unknown>[] = []
+    try {
+      await Promise.race([
+        locked.promise,
+        enable.then(() => {
+          throw new Error('enable completed before the control lock')
+        })
+      ])
+      failure = context.competingStore.recordRegionalRehomeDispatchFailure(attempt!.attemptId)
+      // Observe the actual PostgreSQL wait before letting enable acquire the worker row.
+      await vi.waitFor(async () => {
+        expect(failurePid).not.toBe(0)
+        const rows = await primary.query('SELECT pg_blocking_pids(?) AS blockers', [failurePid])
+        expect(rows[0]!.blockers).toContain(enablePid)
+      }, { interval: 10, timeout: 800 })
+    } finally {
+      release.resolve()
+      outcomes = await Promise.allSettled([enable, ...(failure ? [failure] : [])])
+      enableSpy.mockRestore()
+      failureSpy.mockRestore()
+    }
+    expect(outcomes.map((outcome) => outcome.status)).toEqual(['fulfilled', 'fulfilled'])
+    expect({ enableTransactions, failureTransactions }).toEqual({
+      enableTransactions: 1,
+      failureTransactions: 1
+    })
+    expect(await context.store.inspectRegionalRehomeControl()).toMatchObject({
+      generation: 2,
+      enabled: true
+    })
+    expect(await primary.query(
+      `SELECT consecutive_failures, paused_until FROM relay_region_rehome_worker_state`
+    )).toEqual([{ consecutive_failures: '1', paused_until: '0' }])
+  })
+
   it('increments the disable generation once across competing directors', async () => {
     const context = await fixture()
     const disabled = await Promise.all([

cloud/apps/relay/src/regional-rehome-store.test.ts

@@ -1709,6 +1709,77 @@ describe('regional rehome assignment state', () => {
     expect(await context.store.claimRegionalRehome()).toBeNull()
     await context.database.close()
   })
+
+  it('clears a stale failure budget when the control is enabled again', async () => {
+    const context = await setup()
+    await activatePreferredSource(context, {
+      userId: 'user-1',
+      relayHostId: 'abcdefghijklmnop'
+    })
+    const attempt = await context.store.claimRegionalRehome()
+    for (let index = 0; index < 3; index++) {
+      await context.store.recordRegionalRehomeDispatchFailure(attempt!.attemptId)
+    }
+    expect(await workerState(context)).toMatchObject({ consecutiveFailures: 3 })
+    const latched = await context.store.inspectRegionalRehomeControl()
+    expect(latched).toMatchObject({ generation: 2, enabled: false })
+
+    await context.store.applyRegionalRehomeControl({
+      expectedGeneration: latched.generation,
+      enabled: true,
+      notBefore: context.now(),
+      ratePerMinute: 10,
+      preferenceMaxAgeMs: 24 * 60 * 60_000,
+      hostCooldownMs: 7 * 24 * 60 * 60_000,
+      drainGraceMs: 60 * 60_000
+    })
+
+    // A budget spent under the previous enable is not evidence about this one.
+    expect(await workerState(context)).toMatchObject({
+      consecutiveFailures: 0,
+      pausedUntil: 0
+    })
+    // One transient failure must not latch the fresh enable straight back off.
+    await context.store.recordRegionalRehomeDispatchFailure(attempt!.attemptId)
+    expect(await context.store.inspectRegionalRehomeControl()).toMatchObject({
+      generation: 3,
+      enabled: true
+    })
+    await context.database.close()
+  })
+
+  it('reports the durable disable when the failure budget latches the control off', async () => {
+    const context = await setup()
+    await activatePreferredSource(context, {
+      userId: 'user-1',
+      relayHostId: 'abcdefghijklmnop'
+    })
+    const attempt = await context.store.claimRegionalRehome()
+    const warnings = collectEventWarnings(
+      'orca_relay_regional_rehome_failure_budget_disabled'
+    )
+    try {
+      for (let index = 0; index < 5; index++) {
+        await context.store.recordRegionalRehomeDispatchFailure(attempt!.attemptId)
+      }
+    } finally {
+      warnings.restore()
+    }
+
+    // Only the transition is reported; later failures find the control already off.
+    expect(warnings.entries).toEqual([
+      expect.objectContaining({
+        event: 'orca_relay_regional_rehome_failure_budget_disabled',
+        controlGeneration: 2,
+        consecutiveFailures: 3
+      })
+    ])
+    expect(await context.store.inspectRegionalRehomeControl()).toMatchObject({
+      generation: 2,
+      enabled: false
+    })
+    await context.database.close()
+  })
 })
 
 class TransactionCountingDatabase implements RelayDatabase {
@@ -2066,3 +2137,18 @@ class CellInventoryLockProbe {
     return decorate(database)
   }
 }
+
+async function workerState(
+  context: Context
+): Promise<{ consecutiveFailures: number; pausedUntil: number }> {
+  const row = (
+    await context.database.query(
+      `SELECT consecutive_failures, paused_until
+       FROM relay_region_rehome_worker_state WHERE worker_id = 'global'`
+    )
+  )[0]!
+  return {
+    consecutiveFailures: Number(row.consecutive_failures),
+    pausedUntil: Number(row.paused_until)
+  }
+}

cloud/apps/relay/src/regional-rehome-worker.test.ts

@@ -30,11 +30,9 @@ describe('regional rehome worker', () => {
     }
     const claimRegionalRehome = vi.fn().mockResolvedValueOnce(null).mockResolvedValue(attempt)
     const recordRegionalRehomeDrainReceipt = vi.fn().mockResolvedValue(true)
-    const recordRegionalRehomeWorkerFailure = vi.fn().mockResolvedValue(undefined)
     const assignments = {
       claimRegionalRehome,
-      recordRegionalRehomeDrainReceipt,
-      recordRegionalRehomeWorkerFailure
+      recordRegionalRehomeDrainReceipt
     } as unknown as RelayAssignmentStore
     const requests: Array<{ url: string; init?: RequestInit }> = []
     const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
@@ -99,8 +97,7 @@ describe('regional rehome worker', () => {
     const claimRegionalRehome = vi.fn().mockResolvedValueOnce(null).mockResolvedValue(attempt)
     const assignments = {
       claimRegionalRehome,
-      recordRegionalRehomeDispatchFailure: vi.fn().mockResolvedValue(undefined),
-      recordRegionalRehomeWorkerFailure: vi.fn().mockResolvedValue(undefined)
+      recordRegionalRehomeDispatchFailure: vi.fn().mockResolvedValue(undefined)
     } as unknown as RelayAssignmentStore
     const worker = startRegionalRehomeWorker(config(), assignments, {
       now: () => now,
@@ -118,7 +115,36 @@ describe('regional rehome worker', () => {
     expect(assignments.recordRegionalRehomeDispatchFailure).toHaveBeenCalledWith(
       '11111111-1111-4111-8111-111111111111'
     )
-    expect(assignments.recordRegionalRehomeWorkerFailure).not.toHaveBeenCalled()
+  })
+
+  it('keeps a failed poll out of the durable dispatch-failure budget', async () => {
+    let now = 0
+    const claimRegionalRehome = vi
+      .fn()
+      .mockResolvedValueOnce(null)
+      .mockRejectedValue(new Error('Connection terminated due to connection timeout'))
+    const recordRegionalRehomeDispatchFailure = vi.fn().mockResolvedValue(undefined)
+    const assignments = {
+      claimRegionalRehome,
+      recordRegionalRehomeDispatchFailure
+    } as unknown as RelayAssignmentStore
+    const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
+    const worker = startRegionalRehomeWorker(config(), assignments, {
+      now: () => now,
+      safetySnapshot: () => safety(now),
+      intervalMs: 60_000
+    })!
+    await settleWorker()
+    now = 1_000
+    await expect(worker.run()).resolves.toBeUndefined()
+    worker.stop()
+
+    // The poll never claimed an attempt, so nothing was drained and nothing may
+    // be charged to the budget that latches the durable control off.
+    expect(recordRegionalRehomeDispatchFailure).not.toHaveBeenCalled()
+    expect(warn.mock.calls.map((call) => JSON.parse(String(call[0])).event)).toEqual([
+      'orca_relay_regional_rehome_poll_failed'
+    ])
   })
 
   it('passes unsafe process telemetry to the durable claim gate', async () => {

cloud/apps/relay/src/regional-rehome-worker.ts

@@ -97,13 +97,19 @@ export function startRegionalRehomeWorker(
         })
       )
     } catch (error) {
-      await (attemptId
-        ? assignments.recordRegionalRehomeDispatchFailure(attemptId)
-        : assignments.recordRegionalRehomeWorkerFailure()
-      ).catch(() => undefined)
+      // Only a claimed attempt was drained. A poll that failed before the claim
+      // - a pool timeout on the once-a-second control read - dispatched nothing,
+      // so it must not spend the budget that latches the durable control off.
+      if (attemptId) {
+        await assignments
+          .recordRegionalRehomeDispatchFailure(attemptId)
+          .catch(() => undefined)
+      }
       console.warn(
         JSON.stringify({
-          event: 'orca_relay_regional_rehome_dispatch_failed',
+          event: attemptId
+            ? 'orca_relay_regional_rehome_dispatch_failed'
+            : 'orca_relay_regional_rehome_poll_failed',
           reason: error instanceof Error ? error.message : 'unknown'
         })
       )

src/main/runtime/orchestration/mailbox-pointer-delivery.ts

@@ -9,6 +9,10 @@ import {
   OrchestrationMailboxPointerState,
   type OrchestrationMailboxDeliveryFlight
 } from './mailbox-pointer-state'
+import {
+  MAILBOX_POINTER_RESERVED,
+  MAILBOX_POINTER_WRITE_ATTEMPTED
+} from './db/messages/mailbox-pointer-enter-state'
 import { resumePendingOrchestrationMailboxPointer } from './mailbox-pointer-resume'
 import { stageOrchestrationMailboxPointer } from './mailbox-pointer-stage'
 
@@ -163,7 +167,19 @@ export class OrchestrationMailboxPointerDelivery<TWaiter extends OrchestrationMe
       clearTimeout(flight.enterTimer)
     }
     if (flight?.stagedMessageIds.length) {
-      this.deps.getDb()?.markAsUndelivered(flight.stagedMessageIds)
+      const db = this.deps.getDb()
+      if (db && flight.processIncarnation) {
+        // Why: the Enter timer was just cleared, so a reserved or merely-written pointer provably
+        // never submitted and is released. An attempted Enter may already have landed, so it stays
+        // at its phase for the resume path to revalidate rather than being sent a second time.
+        db.releaseMailboxPointerEnter(
+          flight.stagedMessageIds,
+          { ptyId, processIncarnation: flight.processIncarnation },
+          [MAILBOX_POINTER_RESERVED, MAILBOX_POINTER_WRITE_ATTEMPTED]
+        )
+      } else {
+        db?.markAsUndelivered(flight.stagedMessageIds)
+      }
     }
     for (const mailboxHandle of releasedMailboxes) {
       this.redrive(mailboxHandle, true)

src/main/runtime/orchestration/mailbox-pointer-stage.test.ts

@@ -200,3 +200,72 @@ describe('mailbox pointer staging watermark', () => {
     db.close()
   })
 })
+
+describe('retiring a pty mid-delivery', () => {
+  // Why: an Enter that was already written may have landed. Releasing it would send the same
+  // mail a second time, so only phases that provably never submitted become redeliverable.
+  it('leaves an attempted Enter at its phase instead of making it redeliverable', async () => {
+    vi.useFakeTimers()
+    const db = new OrchestrationDb(':memory:')
+    const settlements: ((settlement: WriteSettlement) => void)[] = []
+    const writePty = vi.fn(
+      () =>
+        new Promise<WriteSettlement>((resolve) => {
+          settlements.push(resolve)
+        }) as unknown as WriteSettlement
+    )
+    try {
+      const message = db.insertMessage({ from: 'a', to: 'run:run-1', subject: 'mail' })
+      const delivery = new OrchestrationMailboxPointerDelivery(pointerDeps(db, writePty) as never)
+      delivery.deliver(LEAF, { mailboxHandle: 'run:run-1' })
+
+      // Settle the pointer write, so the pane reaches WRITE_ATTEMPTED and arms the Enter.
+      settlements[0]?.(WRITE_ACCEPTED)
+      await vi.advanceTimersByTimeAsync(0)
+      expect(db.getMessageById(message.id)?.pointer_enter_pending).toBe(2)
+
+      // Fire the Enter but never settle it: this is the ambiguous state.
+      await vi.advanceTimersByTimeAsync(600)
+      expect(db.getMessageById(message.id)?.pointer_enter_pending).toBe(3)
+
+      delivery.retirePty('pty-1')
+      expect(db.getMessageById(message.id)).toMatchObject({
+        pointer_enter_pending: 3,
+        read: 0
+      })
+    } finally {
+      db.close()
+      vi.useRealTimers()
+    }
+  })
+
+  it('releases a pointer whose Enter never fired', async () => {
+    vi.useFakeTimers()
+    const db = new OrchestrationDb(':memory:')
+    const settlements: ((settlement: WriteSettlement) => void)[] = []
+    const writePty = vi.fn(
+      () =>
+        new Promise<WriteSettlement>((resolve) => {
+          settlements.push(resolve)
+        }) as unknown as WriteSettlement
+    )
+    try {
+      const message = db.insertMessage({ from: 'a', to: 'run:run-1', subject: 'mail' })
+      const delivery = new OrchestrationMailboxPointerDelivery(pointerDeps(db, writePty) as never)
+      delivery.deliver(LEAF, { mailboxHandle: 'run:run-1' })
+      settlements[0]?.(WRITE_ACCEPTED)
+      await vi.advanceTimersByTimeAsync(0)
+      expect(db.getMessageById(message.id)?.pointer_enter_pending).toBe(2)
+
+      delivery.retirePty('pty-1')
+      expect(db.getMessageById(message.id)).toMatchObject({
+        pointer_enter_pending: 0,
+        read: 0,
+        delivered_at: null
+      })
+    } finally {
+      db.close()
+      vi.useRealTimers()
+    }
+  })
+})

src/main/runtime/orchestration/mailbox-pointer-stage.ts

@@ -58,6 +58,7 @@ export function stageOrchestrationMailboxPointer<TWaiter extends OrchestrationMe
     return
   }
   const flight = args.state.beginFlight(ptyId)
+  flight.processIncarnation = expectedTarget.processIncarnation
   flight.stagedMessageIds = args.messages.map((message) => message.id)
   try {
     if (

src/main/runtime/orchestration/mailbox-pointer-state.ts

@@ -6,6 +6,8 @@ export type OrchestrationMailboxDeliveryFlight = {
   submitEnter: (() => void) | null
   deferredUntilIdle: boolean
   idleObservedWhileDeferred: boolean
+  /** The incarnation that staged this flight, so retirement can name the rows it owns. */
+  processIncarnation?: string
 }
 
 export type ParkedOrchestrationMailboxDelivery = {

src/main/runtime/terminal-send-stale-leaf-liveness.test.ts

@@ -386,6 +386,7 @@ function makeOrchestrationDbStub(toHandle: () => string) {
     runMailbox,
     markAsDelivered,
     markAsUndelivered,
+    releaseMailboxPointerEnter,
     stageMailboxPointerEnter,
     insert(subject: string, type: StoredMessageRow['type'] = 'status'): void {
       rows.push({
@@ -759,7 +760,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
       await vi.advanceTimersByTimeAsync(500)
       expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)
       expect(stub.stageMailboxPointerEnter).toHaveBeenCalledOnce()
-      expect(stub.markAsUndelivered).toHaveBeenCalledOnce()
+      expect(stub.releaseMailboxPointerEnter).toHaveBeenCalledOnce()
       expect(stub.rows[0].delivered_at).toBeNull()
 
       // The replacement's own delivery starts a fresh flight and completes —
@@ -804,7 +805,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
       await vi.advanceTimersByTimeAsync(500)
       expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)
       expect(stub.stageMailboxPointerEnter).toHaveBeenCalledOnce()
-      expect(stub.markAsUndelivered).toHaveBeenCalledOnce()
+      expect(stub.releaseMailboxPointerEnter).toHaveBeenCalledOnce()
       // No stray settle flushed the parked trigger into the dead pty.
       expect(write).toHaveBeenCalledTimes(1)
       expect(stub.rows.every((row) => row.delivered_at === null)).toBe(true)
@@ -835,7 +836,7 @@ describe('push-on-idle orchestration delivery absence gate', () => {
       await vi.advanceTimersByTimeAsync(500)
       expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)
       expect(stub.stageMailboxPointerEnter).toHaveBeenCalledOnce()
-      expect(stub.markAsUndelivered).toHaveBeenCalledOnce()
+      expect(stub.releaseMailboxPointerEnter).toHaveBeenCalledOnce()
       expect(stub.rows[0].delivered_at).toBeNull()
     } finally {
       vi.useRealTimers()