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