From 5e53ea6e7176b48dc1fd63d9ec27c62c7c381f4c Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Tue, 18 Aug 2026 02:12:58 -0400 Subject: [PATCH] fix(orchestration): push Dispatch mailbox mail to idle workers (STA-3105) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Push-on-idle pointed only `run:` mailboxes, so mail addressed to `dispatch:` — the address coordinator→worker mail lands on — was reachable by an explicit `orca orchestration check` but never announced. A worker sat idle next to unread instructions until something else made it check, which in practice meant it never did. Three boundaries carried the same `run:`-only assumption: * pointer delivery refused any mailbox that was not `run:`; * the restore scan skipped `dispatch:` handles outright, so mail that landed while the app was down was never repointed; * rerouting a Dispatch mailbox to its Run mailbox carried the push stamp along, which would hide inherited mail from the coordinator's push now that Dispatch mail can be stamped at all. Pointability now covers both Run and Dispatch addresses, and a mailbox ownership change resets `delivered_at` because the new owner was never pointed at that mail. Ack semantics are unchanged: Dispatch mailboxes have no `deliveries` ledger, so `delivered_at` alone dedupes the push and `check` still consumes by marking rows read. Busy workers are deliberately still not interrupted — the mail is held and pointed at the next idle edge, matching Run mailbox behaviour. --- src/main/runtime/orca-runtime.ts | 6 +- ...rchestration-dispatch-mailbox-push.test.ts | 152 ++++++++++++++++++ ...estration-mailbox-detached-routing.test.ts | 7 +- .../db/messages/direct-mailbox-routing.ts | 5 +- .../runtime/orchestration/mailbox-address.ts | 7 + .../orchestration/mailbox-pointer-delivery.ts | 6 +- .../mailbox-pointer-eligibility.ts | 12 ++ 7 files changed, 182 insertions(+), 13 deletions(-) create mode 100644 src/main/runtime/orchestration-dispatch-mailbox-push.test.ts create mode 100644 src/main/runtime/orchestration/mailbox-address.ts diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 3872f25d03e..7b2e08d4a31 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -246,6 +246,7 @@ import { } from './orchestration/federation-ack-checkpoints' import { syncFederatedDispatch } from './orchestration/federation-sync' import { MailPointerRepointScheduler } from './orchestration/mail-pointer-repoint-scheduler' +import { isOrchestrationMailboxAddress } from './orchestration/mailbox-address' import { OrchestrationMailboxOwner } from './orchestration/mailbox-owner' import { OrchestrationMailboxNotificationCoordinator } from './orchestration/mailbox-notification-coordinator' import { OrchestrationMailboxDeliveryTarget } from './orchestration/mailbox-delivery-target' @@ -34345,10 +34346,7 @@ export class OrcaRuntimeService { } for (const handle of handles) { try { - if (handle.startsWith('dispatch:')) { - continue - } - if (handle.startsWith('run:')) { + if (isOrchestrationMailboxAddress(handle)) { this.mailPointerRepointScheduler.schedule(handle) continue } diff --git a/src/main/runtime/orchestration-dispatch-mailbox-push.test.ts b/src/main/runtime/orchestration-dispatch-mailbox-push.test.ts new file mode 100644 index 00000000000..72c44170e68 --- /dev/null +++ b/src/main/runtime/orchestration-dispatch-mailbox-push.test.ts @@ -0,0 +1,152 @@ +import { rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + checkBoundMailbox, + createDatabase, + createRuntime, + driveToLiveIdle, + LAUNCH_TOKEN, + PANE_KEY, + pointerCount, + PTY_ID, + TERMINAL_HANDLE, + temporaryDirectories +} from './orchestration-mailbox-notification-test-harness' + +vi.mock('electron', () => ({ + app: { getPath: vi.fn(() => tmpdir()), isPackaged: false }, + BrowserWindow: { fromId: vi.fn(() => null) }, + ipcMain: { on: vi.fn(), removeListener: vi.fn() }, + webContents: { fromId: vi.fn(() => null) } +})) + +const COORDINATOR_PANE_KEY = + '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' + +function dispatchToWorker(db: OrchestrationTestDb) { + const run = db.createRun({ + objective: 'Coordinator run', + coordinatorHandle: 'term_coordinator', + coordinatorPaneKey: COORDINATOR_PANE_KEY + }) + const task = db.createTask({ spec: 'Worker task', runId: run.id }) + return { run, dispatch: db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY) } +} + +type OrchestrationTestDb = ReturnType + +function sendToDispatch( + db: OrchestrationTestDb, + runId: string, + dispatchId: string, + subject: string +) { + return db.insertMessage({ + from: 'term_coordinator', + to: `dispatch:${dispatchId}`, + subject, + type: 'status', + runId, + deliveryContract: 'current_delivery' + }) +} + +describe('dispatch mailbox push-on-idle', () => { + afterEach(() => { + vi.useRealTimers() + for (const directory of temporaryDirectories.splice(0)) { + rmSync(directory, { recursive: true, force: true }) + } + }) + + it('points an idle worker at mail already waiting in its Dispatch mailbox', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-dispatch-push-idle-') + const harness = createRuntime(db) + const { run, dispatch } = dispatchToWorker(db) + sendToDispatch(db, run.id, dispatch.id, 'push authorization') + + await driveToLiveIdle(harness.runtime) + await vi.advanceTimersByTimeAsync(600) + + expect(pointerCount(harness.write)).toBe(1) + // The pointer must be backed by mail the worker's own check then returns. + await expect( + checkBoundMailbox(harness.runtime, { + terminal: TERMINAL_HANDLE, + paneKey: PANE_KEY, + launchToken: LAUNCH_TOKEN + }) + ).resolves.toMatchObject({ dispatchId: dispatch.id, count: 1 }) + db.close() + }) + + it('points a worker that is already idle when Dispatch mail arrives', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-dispatch-push-arrival-') + const harness = createRuntime(db) + const { run, dispatch } = dispatchToWorker(db) + + await driveToLiveIdle(harness.runtime) + sendToDispatch(db, run.id, dispatch.id, 'rule update') + harness.runtime.notifyMessageArrived(`dispatch:${dispatch.id}`, 'status') + await vi.advanceTimersByTimeAsync(600) + + expect(pointerCount(harness.write)).toBe(1) + db.close() + }) + + it('holds Dispatch mail for a busy worker until it returns to idle', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-dispatch-push-busy-') + const harness = createRuntime(db) + const { run, dispatch } = dispatchToWorker(db) + + await harness.runtime.listTerminals() + harness.runtime.onPtyData(PTY_ID, '\x1b]0;Codex working\x07', 1) + sendToDispatch(db, run.id, dispatch.id, 'urgent rule update') + harness.runtime.notifyMessageArrived(`dispatch:${dispatch.id}`, 'status') + await vi.advanceTimersByTimeAsync(600) + expect(pointerCount(harness.write)).toBe(0) + + harness.runtime.onPtyData(PTY_ID, '\x1b]0;Codex done\x07', 2) + await vi.advanceTimersByTimeAsync(600) + + expect(pointerCount(harness.write)).toBe(1) + db.close() + }) + + it('repoints Dispatch mail left unread when the database attaches', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-dispatch-push-restore-') + const harness = createRuntime(db) + const { run, dispatch } = dispatchToWorker(db) + + await driveToLiveIdle(harness.runtime) + // Mail that landed while the app was down raises no arrival notification. + sendToDispatch(db, run.id, dispatch.id, 'offline rule update') + harness.runtime.setOrchestrationDb(db) + await vi.advanceTimersByTimeAsync(3_000) + + expect(pointerCount(harness.write)).toBe(1) + db.close() + }) + + it('clears the push stamp when unread mail changes mailbox owner', () => { + const db = createDatabase('orca-dispatch-push-reroute-') + const { run, dispatch } = dispatchToWorker(db) + const message = sendToDispatch(db, run.id, dispatch.id, 'push authorization') + db.markAsDelivered([message.id]) + + db.routeUnreadDispatchMailboxToRunMailbox(dispatch.id, run.id) + + // The coordinator inherits the mail unpointed, so it must be pushable again. + expect(db.getMessageById(message.id)).toMatchObject({ + to_handle: `run:${run.id}`, + read: 0, + delivered_at: null + }) + db.close() + }) +}) diff --git a/src/main/runtime/orchestration-mailbox-detached-routing.test.ts b/src/main/runtime/orchestration-mailbox-detached-routing.test.ts index 654f01acef1..9d7ba352fef 100644 --- a/src/main/runtime/orchestration-mailbox-detached-routing.test.ts +++ b/src/main/runtime/orchestration-mailbox-detached-routing.test.ts @@ -32,7 +32,7 @@ describe('orchestration detached mailbox routing', () => { } }) - it('routes active worker direct mail without injecting an unpinned Dispatch pointer', async () => { + it('routes active worker direct mail into the Dispatch mailbox and points the worker at it', async () => { vi.useFakeTimers() const db = createDatabase('orca-mailbox-dispatch-') const harness = createRuntime(db) @@ -57,13 +57,14 @@ describe('orchestration detached mailbox routing', () => { await vi.advanceTimersByTimeAsync(500) const checked = await checkBoundMailbox(harness.runtime) - expect(pointerCount(harness.write)).toBe(0) + // The pointer must be backed by mail the worker's own check then returns. + expect(pointerCount(harness.write)).toBe(1) expect(checked).toMatchObject({ runId: run.id, dispatchId: dispatch.id, count: 1 }) expect(checked.messages).toEqual([expect.objectContaining({ id: message.id })]) expect(db.getMessageById(message.id)).toMatchObject({ to_handle: `dispatch:${dispatch.id}`, read: 1, - delivered_at: null + delivered_at: expect.any(String) }) db.close() }) diff --git a/src/main/runtime/orchestration/db/messages/direct-mailbox-routing.ts b/src/main/runtime/orchestration/db/messages/direct-mailbox-routing.ts index aa7d93fbea2..2e89b665a14 100644 --- a/src/main/runtime/orchestration/db/messages/direct-mailbox-routing.ts +++ b/src/main/runtime/orchestration/db/messages/direct-mailbox-routing.ts @@ -36,6 +36,7 @@ export function getLatestUnreadDirectMessageSequenceForRun( } // Why: change mailbox ownership without changing unread or acknowledgment state. +// The push stamp is owner-scoped, so it resets: the new owner was never pointed at this mail. export function routeDirectMessagePage( this: OrchestrationDb, mailboxHandle: string, @@ -76,7 +77,7 @@ export function routeDirectMessagePage( const placeholders = page.map(() => '?').join(',') const result = this.db .prepare( - `UPDATE messages INDEXED BY idx_messages_id SET to_handle = ? + `UPDATE messages INDEXED BY idx_messages_id SET to_handle = ?, delivered_at = NULL WHERE run_id = ? AND to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery' AND id IN (${placeholders})` ) @@ -143,7 +144,7 @@ export function routeUnreadDispatchMailboxToRunMailbox( const placeholders = page.map(() => '?').join(',') const result = this.db .prepare( - `UPDATE messages INDEXED BY idx_messages_id SET to_handle = ? + `UPDATE messages INDEXED BY idx_messages_id SET to_handle = ?, delivered_at = NULL WHERE run_id = ? AND to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery' AND id IN (${placeholders})` diff --git a/src/main/runtime/orchestration/mailbox-address.ts b/src/main/runtime/orchestration/mailbox-address.ts new file mode 100644 index 00000000000..48a665092e9 --- /dev/null +++ b/src/main/runtime/orchestration/mailbox-address.ts @@ -0,0 +1,7 @@ +export const RUN_MAILBOX_PREFIX = 'run:' +export const DISPATCH_MAILBOX_PREFIX = 'dispatch:' + +/** Shared Run/Dispatch mailboxes, as opposed to a direct terminal handle. */ +export function isOrchestrationMailboxAddress(handle: string): boolean { + return handle.startsWith(RUN_MAILBOX_PREFIX) || handle.startsWith(DISPATCH_MAILBOX_PREFIX) +} diff --git a/src/main/runtime/orchestration/mailbox-pointer-delivery.ts b/src/main/runtime/orchestration/mailbox-pointer-delivery.ts index 3efb121cc78..e9f2d7e92f8 100644 --- a/src/main/runtime/orchestration/mailbox-pointer-delivery.ts +++ b/src/main/runtime/orchestration/mailbox-pointer-delivery.ts @@ -3,6 +3,7 @@ import { ORCHESTRATION_DELIVERY_BATCH_LIMIT, type OrchestrationDb } from './db' import { formatMessagePointer } from './formatter' import type { OrchestrationMailboxDeliveryTarget } from './mailbox-delivery-target' import { + canPointAtMailbox, hasUnfilteredOrchestrationWaiter, messageTypeHasOrchestrationWaiter, shouldReleaseOrchestrationPointer, @@ -65,15 +66,12 @@ export class OrchestrationMailboxPointerDelivery