From c2852cdb2d4a63fea9284a0d310cdbfe63e24f48 Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Thu, 13 Aug 2026 22:58:56 -0400 Subject: [PATCH] fix(orchestration): fence inactive mailbox snapshots --- config/reliability-gates.jsonc | 11 +-- ...rchestration-mailbox-routing-races.test.ts | 72 ++++++++++++++++++- src/main/runtime/orchestration/db.ts | 36 +++++++--- src/main/runtime/rpc/methods/orchestration.ts | 48 +++++++++---- 4 files changed, 140 insertions(+), 27 deletions(-) diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 32c1cf2b1a0..1df65b955b5 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -8805,10 +8805,10 @@ "providers": ["local", "daemon", "ssh", "wsl", "remote-runtime"], "coveredPlatforms": ["macos"], "coveredProviders": ["local", "daemon", "ssh"], - "coverageNotes": "A deterministic in-process runtime/SQLite/RPC harness covers two remint-equivalent panes across Run and Dispatch ownership changes, captures mocked PTY-controller writes, executes the production check, reopens durable production-shape state under a fresh runtime, and verifies Run Delivery acknowledgment plus worker-check consumption. Controlled daemon-socket and SSH-multiplexer settlements prove a rejected pointer write stays undelivered and restart-retryable before durable staging; a fake clock proves a wedged daemon write disconnects at its bounded deadline. A 151-row backlog proves live-idle reconciliation and inactive-Dispatch transfer stay within 50-row pages and cooperative event-loop turns; explicit checks cooperatively drain their start-of-check sequence snapshot before filtering or waiting. Controlled setImmediate races prove Run/Dispatch consumers are fenced after ownership changes, split current/stored handles both drain, cancellation stops direct-mail paging, and a proven-inactive Dispatch preserves every unread row in its reachable Run after page routing and filtered wait cancellation. The authority decision adds no wire field, renderer state, polling, or provider call. Paired remote-runtime, live SSH, WSL, Linux, Windows, and headed restart evidence remain gaps.", + "coverageNotes": "A deterministic in-process runtime/SQLite/RPC harness covers two remint-equivalent panes across Run and Dispatch ownership changes, captures mocked PTY-controller writes, executes the production check, reopens durable production-shape state under a fresh runtime, and verifies Run Delivery acknowledgment plus worker-check consumption. Controlled daemon-socket and SSH-multiplexer settlements prove a rejected pointer write stays undelivered and restart-retryable before durable staging; a fake clock proves a wedged daemon write disconnects at its bounded deadline. A 151-row backlog proves live-idle reconciliation and inactive-Dispatch transfer stay within 50-row pages and cooperative event-loop turns; explicit checks cooperatively drain their start-of-check sequence snapshot before filtering or waiting. Controlled setImmediate races prove Run/Dispatch consumers are fenced after ownership changes, split current/stored handles both drain, cancellation stops direct-mail paging, and an inactive-Dispatch cancellation notifies partial progress before a background continuation finishes only the captured snapshot. The authority decision adds no wire field, renderer state, polling, or provider call. Paired remote-runtime, live SSH, WSL, Linux, Windows, and headed restart evidence remain gaps.", "motivatingLinks": ["internal incident evidence: improve-vps-setup, 2026-08-10"], "invariant": "Every injected orchestration pointer must select the same mailbox as the exact check from that pane and expose at least the notified unread messages through that mailbox's acknowledgment semantics. Rebinding cannot let foreign direct mail wake a new owner, newer Run mail cannot be announced behind an outstanding fixed Delivery, and restart cannot replay pointer text already accepted by the PTY even if delayed Enter has not fired. Notification records only a durable pointer cursor; it never marks mail read or globally clears it. Direct and Dispatch mail without a mixed-version-compatible pinned consuming action must not synthesize a terminal turn.", - "oracle": "Create Run A with unread current-delivery mail addressed directly to its coordinator handle, bind the same pane to Run B, drive a controlled working-to-idle event, and require zero PTY pointers while Run B check returns zero and Run A mail stays unread. Reopen a database already containing that mismatch and require live-idle reconciliation to make the row actionable to Run A without pointing Run B. For same-Run mail, require one Run-pinned pointer, a bound Delivery containing that row, explicit acknowledgment, and no replay after a fresh runtime, including restart between accepted pointer payload and delayed Enter. Require direct and Dispatch mail to remain explicitly checkable without an unpinned pointer, reminted active-worker mail to retain Dispatch ownership, newer Run mail to stay suppressed behind an outstanding fixed Delivery, waiter reservations to remain single-consumer across routing, absent local PTYs to retain mail, and a 151-row mismatch to reconcile in 50-row pages with one continuation per turn. During explicit paging, deterministically rebind the Run or complete the Dispatch and require a fenced response; require the inactive Dispatch's coordinator to consume and acknowledge the preserved rows through four fixed Run Deliveries while migration remains 50-row paged; split current and stored handles and require one check to route both; add mail beyond the captured sequence and require finite snapshot completion; abort between direct-mail pages and require paging to stop. Hold a daemon socket settlement past its fake-clock deadline and require disconnect plus a retryable undelivered pointer.", + "oracle": "Create Run A with unread current-delivery mail addressed directly to its coordinator handle, bind the same pane to Run B, drive a controlled working-to-idle event, and require zero PTY pointers while Run B check returns zero and Run A mail stays unread. Reopen a database already containing that mismatch and require live-idle reconciliation to make the row actionable to Run A without pointing Run B. For same-Run mail, require one Run-pinned pointer, a bound Delivery containing that row, explicit acknowledgment, and no replay after a fresh runtime, including restart between accepted pointer payload and delayed Enter. Require direct and Dispatch mail to remain explicitly checkable without an unpinned pointer, reminted active-worker mail to retain Dispatch ownership, newer Run mail to stay suppressed behind an outstanding fixed Delivery, waiter reservations to remain single-consumer across routing, absent local PTYs to retain mail, and a 151-row mismatch to reconcile in 50-row pages with one continuation per turn. During explicit paging, deterministically rebind the Run or complete the Dispatch and require a fenced response; require the inactive Dispatch's coordinator to consume and acknowledge the preserved rows through four fixed Run Deliveries while migration remains 50-row paged; split current and stored handles and require one check to route both; add mail beyond the captured sequence and require finite snapshot completion; abort between direct-mail pages and require paging to stop. Abort an inactive-Dispatch migration after its first page, require one partial wake and bounded background completion of the captured rows, and require a later row to remain outside that snapshot. Hold a daemon socket settlement past its fake-clock deadline and require disconnect plus a retryable undelivered pointer.", "commands": [ "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-mailbox-routing-races.test.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration-mailbox-transport-settlement.test.ts src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts src/main/runtime/orchestration/formatter.test.ts src/main/providers/ssh-pty-provider.test.ts src/main/providers/ssh-pty-write.test.ts src/main/daemon/client.test.ts src/main/daemon/daemon-pty-router.test.ts src/main/daemon/degraded-daemon-pty-provider.test.ts", "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orca-runtime.test.ts src/main/runtime/terminal-send-stale-leaf-liveness.test.ts src/main/runtime/rpc/methods/orchestration.test.ts" @@ -8882,6 +8882,7 @@ "current and stored reminted handles both drain in one explicit check", "explicit paging terminates at its indexed start-of-check sequence snapshot and stops on cancellation", "a fenced inactive Dispatch transfers unread rows to its Run in 50-row pages and coalesces arrival types for supported consumption and acknowledgment", + "a cancelled inactive-Dispatch migration wakes partial progress and finishes only its captured snapshot in the background", "filtered Dispatch wait cancellation reconciles unrelated unread types before fencing", "unfiltered and filtered parked reservations union in either arrival order", "reminted Dispatch pane lookup uses a bounded partial expression index while exact legacy keys remain supported" @@ -8895,8 +8896,8 @@ "platform": "macos", "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-mailbox-routing-races.test.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration-mailbox-transport-settlement.test.ts src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts src/main/runtime/orchestration/formatter.test.ts src/main/providers/ssh-pty-provider.test.ts src/main/providers/ssh-pty-write.test.ts src/main/daemon/client.test.ts src/main/daemon/daemon-pty-router.test.ts src/main/daemon/degraded-daemon-pty-provider.test.ts", "result": "passed", - "durationSeconds": 4.85, - "summary": "227 tests passed across mailbox identity, direct and Dispatch pointer suppression, persisted reconciliation, 50-row paging and filtered waits, cross-PTY serialization, live ownership transfer, restart before delayed Enter, submit races, bounded daemon and SSH transport settlement, outstanding Deliveries, reminted Dispatch ownership, acknowledgment, remote pane remints, split-handle drains, ownership-change fencing, paged inactive-mailbox reachability after filtered waits, snapshot termination, cancellation, reservation ordering, and bounded pane lookup. The original 10-scenario mailbox oracle failed 10/10 on origin/main plus the test commit and again with the implementation commit explicitly reverted, including pointerCount > 0 with bound check count 0." + "durationSeconds": 6.86, + "summary": "228 tests passed across mailbox identity, direct and Dispatch pointer suppression, persisted reconciliation, 50-row paging and filtered waits, cross-PTY serialization, live ownership transfer, restart before delayed Enter, submit races, bounded daemon and SSH transport settlement, outstanding Deliveries, reminted Dispatch ownership, acknowledgment, remote pane remints, split-handle drains, ownership-change fencing, paged inactive-mailbox reachability, cancellation continuation, snapshot termination, reservation ordering, and bounded pane lookup. The original 10-scenario mailbox oracle failed 10/10 on origin/main plus the test commit and again with the implementation commit explicitly reverted, including pointerCount > 0 with bound check count 0." }, { "date": "2026-08-13", @@ -8922,7 +8923,7 @@ }, "performanceBudget": { "required": true, - "evidence": "Arrival ownership uses indexed identity and SELECT 1 LIMIT 1 probes, including exact and remint-stable Dispatch pane lookups. Live-idle direct-mail reconciliation is capped to 50 rows per event-loop turn with one coalesced setImmediate continuation; explicit checks capture an indexed maximum sequence per distinct owner handle, route only that finite snapshot with one cooperative setImmediate yield per page, and stop on request cancellation. Submit-time row revalidation uses the unique message-ID index, durable cursor writes update only the staged batch in 500-ID chunks, and outstanding-Delivery suppression is a bounded existence query. A proven-inactive Dispatch transfers 50 indexed rows per statement and yields between pages without a long write transaction. Daemon pointer writes have one five-second settlement timer per active flight and destroy the wedged socket before releasing the flight. No subprocesses, provider fanout, polling, persistent listeners, or renderer work were added." + "evidence": "Arrival ownership uses indexed identity and SELECT 1 LIMIT 1 probes, including exact and remint-stable Dispatch pane lookups. Live-idle direct-mail reconciliation is capped to 50 rows per event-loop turn with one coalesced setImmediate continuation; explicit checks capture an indexed maximum sequence per distinct owner handle, route only that finite snapshot with one cooperative setImmediate yield per page, and stop on request cancellation. Submit-time row revalidation uses the unique message-ID index, durable cursor writes update only the staged batch in 500-ID chunks, and outstanding-Delivery suppression is a bounded existence query. A proven-inactive Dispatch captures an indexed maximum sequence, transfers 50 indexed rows per statement, yields between pages, and on cancellation continues only that finite snapshot outside the RPC after notifying partial progress. Daemon pointer writes have one five-second settlement timer per active flight and destroy the wedged socket before releasing the flight. No subprocesses, provider fanout, polling, persistent listeners, or renderer work were added." }, "promotionCriteria": [ "Add isolated live restart and SSH or paired-runtime evidence.", diff --git a/src/main/runtime/orchestration-mailbox-routing-races.test.ts b/src/main/runtime/orchestration-mailbox-routing-races.test.ts index e40c031ad15..33116f69d0b 100644 --- a/src/main/runtime/orchestration-mailbox-routing-races.test.ts +++ b/src/main/runtime/orchestration-mailbox-routing-races.test.ts @@ -171,7 +171,10 @@ describe('orchestration mailbox routing races', () => { controller.abort() const response = await waiting - expect(response).toMatchObject({ ok: false, error: { code: 'dispatch_inactive' } }) + expect(response).toMatchObject({ ok: false, error: { code: 'runtime_error' } }) + await vi.waitFor(() => { + expect(db.getUnreadMessages(`dispatch:${dispatch.id}`)).toHaveLength(0) + }) const checked = await checkBoundMailbox(harness.runtime, { terminal: SECOND_TERMINAL_HANDLE, paneKey: SECOND_PANE_KEY, @@ -188,6 +191,73 @@ describe('orchestration mailbox routing races', () => { db.close() }) + it('finishes only the captured inactive-Dispatch snapshot after cancellation', async () => { + const db = createDatabase('orca-mailbox-dispatch-migration-cancel-') + const harness = createRuntime(db) + registerSecondPane(harness.runtime) + const run = db.createRun({ + objective: 'Dispatch migration cancellation', + coordinatorHandle: SECOND_TERMINAL_HANDLE, + coordinatorPaneKey: SECOND_PANE_KEY + }) + const task = db.createTask({ spec: 'Cancelled worker check', runId: run.id }) + const dispatch = db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY) + for (let index = 0; index < 151; index += 1) { + insertDirectRunMessage(db, run.id, `Before cancelled migration ${index}`) + } + const route = db.routeUnreadDirectMessagesToDispatchMailbox.bind(db) + let completed = false + vi.spyOn(db, 'routeUnreadDirectMessagesToDispatchMailbox').mockImplementation((...args) => { + const page = route(...args) + if (page.hasMore && !completed) { + completed = true + setImmediate(() => db.completeDispatch(dispatch.id)) + } + return page + }) + const controller = new AbortController() + const migrate = db.routeUnreadDispatchMailboxToRunMailbox.bind(db) + let cancelled = false + let postSnapshot: ReturnType | undefined + const migrationSpy = vi + .spyOn(db, 'routeUnreadDispatchMailboxToRunMailbox') + .mockImplementation((...args) => { + const page = migrate(...args) + if (page.hasMore && !cancelled) { + cancelled = true + postSnapshot = db.insertMessage({ + from: 'term_remote_worker', + to: `dispatch:${dispatch.id}`, + subject: 'After cancelled migration snapshot', + type: 'status', + runId: run.id + }) + setImmediate(() => controller.abort()) + } + return page + }) + const arrivalSpy = vi.spyOn(harness.runtime, 'notifyMessageArrived') + + const response = await dispatchMailboxCheck(harness.runtime, { signal: controller.signal }) + + expect(response).toMatchObject({ ok: false, error: { code: 'runtime_error' } }) + await vi.waitFor(() => { + expect(migrationSpy).toHaveBeenCalledTimes(4) + }) + expect(db.getUnreadMessages(`dispatch:${dispatch.id}`)).toEqual([ + expect.objectContaining({ id: postSnapshot?.id }) + ]) + expect(arrivalSpy).toHaveBeenCalledTimes(2) + expect(arrivalSpy).toHaveBeenNthCalledWith(1, `run:${run.id}`, 'status') + expect(arrivalSpy).toHaveBeenNthCalledWith(2, `run:${run.id}`, 'status') + expect( + sqliteFor(db) + .prepare('SELECT COUNT(*) AS count FROM messages WHERE to_handle = ?') + .get(`run:${run.id}`) + ).toEqual({ count: 151 }) + db.close() + }) + it('drains current and stored Run handles in one check', async () => { const db = createDatabase('orca-mailbox-split-run-handles-') const harness = createRuntime(db) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 8061b0f3ad3..aab8fc6614d 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -90,13 +90,13 @@ function parseWorkerTerminalPriorOwnerIds(value: string): string[] | null { const MESSAGE_ID_UPDATE_BATCH_SIZE = 500 export const ORCHESTRATION_DELIVERY_BATCH_LIMIT = 50 -export type DirectMailboxRoutingPage = { +export type MailboxRoutingPage = { routedCount: number hasMore: boolean types: MessageType[] } -export type ForeignDirectMailboxRoutingPage = DirectMailboxRoutingPage & { +export type ForeignDirectMailboxRoutingPage = MailboxRoutingPage & { mailboxes: { mailboxHandle: string; types: MessageType[] }[] } // Why: indexable pre-filter for isEquivalentPaneKey — equal strings and equal leaves both share the @@ -3644,7 +3644,7 @@ export class OrchestrationDb { runId: string, directHandle: string, throughSequence?: number - ): DirectMailboxRoutingPage { + ): MailboxRoutingPage { const throughClause = throughSequence === undefined ? '' : ' AND sequence <= ?' const params: (string | number)[] = [runId, directHandle] if (throughSequence !== undefined) { @@ -3681,7 +3681,7 @@ export class OrchestrationDb { runId: string, directHandle: string, throughSequence?: number - ): DirectMailboxRoutingPage { + ): MailboxRoutingPage { return this.routeDirectMessagePage(`run:${runId}`, runId, directHandle, throughSequence) } @@ -3690,7 +3690,7 @@ export class OrchestrationDb { runId: string, directHandle: string, throughSequence?: number - ): DirectMailboxRoutingPage { + ): MailboxRoutingPage { return this.routeDirectMessagePage( `dispatch:${dispatchId}`, runId, @@ -3701,16 +3701,23 @@ export class OrchestrationDb { routeUnreadDispatchMailboxToRunMailbox( dispatchId: string, - runId: string - ): DirectMailboxRoutingPage { + runId: string, + throughSequence?: number + ): MailboxRoutingPage { const dispatchMailbox = `dispatch:${dispatchId}` + const throughClause = throughSequence === undefined ? '' : ' AND sequence <= ?' + const params: (string | number)[] = [dispatchMailbox] + if (throughSequence !== undefined) { + params.push(throughSequence) + } + params.push(ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1) const rows = this.db .prepare( `SELECT id, type FROM messages INDEXED BY idx_messages_unread_current_inbox - WHERE to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery' + WHERE to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery'${throughClause} ORDER BY sequence LIMIT ?` ) - .all(dispatchMailbox, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1) as { + .all(...params) as { id: string type: MessageType }[] @@ -3729,6 +3736,17 @@ export class OrchestrationDb { } } + getLatestUnreadMessageSequence(mailboxHandle: string): number | undefined { + const row = this.db + .prepare( + `SELECT sequence FROM messages INDEXED BY idx_messages_unread_current_inbox + WHERE to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery' + ORDER BY sequence DESC LIMIT 1` + ) + .get(mailboxHandle) as { sequence: number } | undefined + return row?.sequence + } + private findActiveDispatchForDirectMessageOwner( runId: string, directHandle: string, diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index 7d93a8be742..831aa61e6c5 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -57,6 +57,9 @@ async function routeAllMailboxPages( return } await yieldToEventLoop() + if (signal?.aborted) { + throw new OrchestrationError('request_aborted', 'Mailbox routing was cancelled.') + } } } @@ -987,19 +990,40 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ (!latestDispatch || (latestDispatch.status !== 'pending' && latestDispatch.status !== 'dispatched')) ) { - const routedTypes = new Set() - await routeAllMailboxPages(() => { - const routed = db.routeUnreadDispatchMailboxToRunMailbox( - workerMailbox.dispatchId, - owningRunId - ) - for (const routedType of routed.types) { - routedTypes.add(routedType) + const throughSequence = db.getLatestUnreadMessageSequence(address) + if (throughSequence !== undefined) { + const routedTypes = new Set() + const routePage = (): { routedCount: number; hasMore: boolean } => { + const routed = db.routeUnreadDispatchMailboxToRunMailbox( + workerMailbox.dispatchId, + owningRunId, + throughSequence + ) + for (const routedType of routed.types) { + routedTypes.add(routedType) + } + return routed } - return routed - }) - for (const routedType of routedTypes) { - runtime.notifyMessageArrived(`run:${owningRunId}`, routedType) + const notifyRoutedTypes = (): void => { + for (const routedType of routedTypes) { + runtime.notifyMessageArrived(`run:${owningRunId}`, routedType) + } + routedTypes.clear() + } + try { + await routeAllMailboxPages(routePage, signal) + } catch (error) { + notifyRoutedTypes() + if (error instanceof OrchestrationError && error.code === 'request_aborted') { + setImmediate(() => { + void routeAllMailboxPages(routePage) + .catch(() => undefined) + .finally(notifyRoutedTypes) + }) + } + throw error + } + notifyRoutedTypes() } } throw new OrchestrationError(