From ddac8cbe7ca656f30416f65878834a84d2327809 Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Mon, 10 Aug 2026 23:56:41 -0400 Subject: [PATCH] fix(orchestration): bound mailbox reconciliation --- config/reliability-gates.jsonc | 20 +- src/main/daemon/daemon-pty-adapter.ts | 3 +- src/main/daemon/daemon-pty-router.ts | 4 +- .../daemon/degraded-daemon-pty-provider.ts | 4 +- src/main/ipc/pty.ts | 3 +- src/main/providers/local-pty-provider.test.ts | 4 +- src/main/providers/local-pty-provider.ts | 11 +- src/main/providers/pty-provider-contract.ts | 2 +- src/main/runtime/orca-runtime.test.ts | 4 +- src/main/runtime/orca-runtime.ts | 143 +++++++---- ...n-mailbox-notification-consistency.test.ts | 232 +++++++++++++++++- src/main/runtime/orchestration/db.ts | 179 ++++++++++---- .../runtime/orchestration/formatter.test.ts | 4 +- src/main/runtime/orchestration/formatter.ts | 7 +- .../orchestration-worker-dispatch-db.test.ts | 3 + src/main/runtime/rpc/methods/orchestration.ts | 26 ++ .../terminal-send-stale-leaf-liveness.test.ts | 7 +- 17 files changed, 526 insertions(+), 130 deletions(-) diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 8d06f36bff7..e171124eed1 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -8804,10 +8804,10 @@ "providers": ["local", "daemon", "ssh", "wsl", "remote-runtime"], "coveredPlatforms": ["macos"], "coveredProviders": ["local"], - "coverageNotes": "A deterministic in-process runtime/SQLite/RPC harness fixes one terminal handle and pane across Run and Dispatch ownership changes, captures mocked PTY-controller writes, executes the production check, reopens the durable database under a fresh runtime, and verifies Run Delivery acknowledgment plus worker-check consumption. The host-side authority decision is provider-independent and adds no wire field, renderer state, polling, or PTY-provider call. Daemon, paired remote-runtime, 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 a durable production-shape mismatch under a fresh runtime, and verifies Run Delivery acknowledgment plus worker-check consumption. A 151-row backlog proves reconciliation stays within one 50-row page and one coalesced mailbox wake per event-loop turn. The authority decision adds no wire field, renderer state, polling, or provider call. Daemon, paired remote-runtime, 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 a successfully submitted pointer. Notification records only a durable pointer cursor; it never marks mail read or globally clears it.", - "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 the database and repeat. For same-Run mail, require one submitted pointer, a bound Delivery containing that row, explicit acknowledgment, and no replay after a fresh runtime. Require an active worker's direct mail to route through its Dispatch check, suppress newer Run pointers behind an outstanding Delivery, normalize mail that arrived while its Run was displaced, avoid full mailbox materialization while working, and let bound waiters consume without a duplicate PTY 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. Require reminted active-worker mail to retain Dispatch ownership, suppress newer Run pointers behind an outstanding Delivery, preserve waiter reservations, reject absent local PTYs, and reconcile a 151-row mismatch in 50-row pages with one mailbox wake per turn.", "commands": [ "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.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" @@ -8826,13 +8826,18 @@ "assertions": [ "Run A direct mail cannot point after the pane binds Run B", "a fresh runtime cannot replay the unactionable pointer", + "a fresh runtime reconciles a database that already contains the production mismatch", + "large persisted mismatches route in 50-row pages with coalesced per-mailbox wakes", + "Run pointers pin the existing mixed-version-compatible --run identity", "a delayed pointer submit cannot press Enter after the agent becomes working", + "an explicit check releases its staged pointer reservation without redrive", + "a provider-refused Enter remains undelivered and retryable", "a staged pointer follows its Run to a new live pane without cross-PTY duplication", "equivalent panes serialize one mailbox while newer mail arrives during Enter delay", "same-Run direct mail is delivered and acknowledged through the bound Run", "a submitted actionable pointer is durably deduplicated across restart without marking mail read", "newer Run mail is not pointed behind an outstanding fixed Delivery", - "active worker direct mail is exposed by its Dispatch check", + "active worker direct mail is exposed by its Dispatch check after handle remint", "late direct mail is normalized when its displaced Run rebinds", "a bound waiter wakes without a duplicate PTY pointer", "working-agent arrival uses bounded ownership lookup without a full mailbox scan" @@ -8852,8 +8857,8 @@ "platform": "macos", "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-mailbox-notification-consistency.test.ts src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts", "result": "passed", - "durationSeconds": 3.1, - "summary": "31 tests passed across mailbox identity, cross-PTY serialization, live ownership transfer, durable restart, submit races, bounded batches, outstanding Deliveries, Dispatch ownership, acknowledgment, waiters, Run Delivery, and remote pane remints. 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": 5.1, + "summary": "37 tests passed across mailbox identity, persisted-upgrade reconciliation, 50-row paging and coalesced wakes, cross-PTY serialization, live ownership transfer, durable restart, submit races, outstanding Deliveries, reminted Dispatch ownership, acknowledgment, waiters, Run Delivery, and remote pane remints. 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-10", @@ -8875,11 +8880,11 @@ }, "redGreenEvidence": { "status": "complete", - "evidence": "The identical 10-scenario oracle failed 10/10 on origin/main plus the oracle commit, passed 10/10 on the candidate, failed 10/10 after explicitly reverting the implementation commit, and passed again after restoration. Live topology coverage remains a separate promotion gap." + "evidence": "The identical original 10-scenario oracle failed 10/10 on origin/main plus the oracle commit, passed 10/10 on the candidate, failed 10/10 after explicitly reverting the implementation commit, and passed again after restoration. Added restart-upgrade, reminted-Dispatch, pointer-pinning, refusal, and bounded-backlog scenarios are candidate regressions; live topology coverage remains a separate promotion gap." }, "performanceBudget": { "required": true, - "evidence": "Arrival ownership uses indexed identity and SELECT 1 LIMIT 1 probes. Pointer selection is capped to 50 rows inside SQLite, submit-time row revalidation is forced through the unique message-ID index, durable cursor writes update only the submitted batch in 500-ID chunks, and outstanding-Delivery suppression is a bounded existence query. No polling, new timers, subprocesses, provider calls, listeners, or renderer work were added." + "evidence": "Arrival ownership uses indexed identity and SELECT 1 LIMIT 1 probes. Pointer selection and direct-mail reconciliation are capped to 50 rows inside SQLite; a single setImmediate continuation yields between pages and coalesces each mailbox/type wake. Submit-time row revalidation uses the unique message-ID index, durable cursor writes update only the submitted batch in 500-ID chunks, and outstanding-Delivery suppression is a bounded existence query. No polling, subprocesses, provider calls, persistent timers, listeners, or renderer work were added." }, "promotionCriteria": [ "Add isolated live restart and SSH or paired-runtime evidence.", @@ -8888,6 +8893,7 @@ "knownGaps": [ "No headed Electron restart journey was required to prove the runtime/PTY/check boundary.", "No daemon, paired remote-runtime, live SSH, WSL, Linux, or Windows journey has been collected.", + "Daemon and SSH PTY writes remain transport-accepted rather than owner-acknowledged; capability-negotiated owner acknowledgments are separate remote-protocol work.", "Passive renderer-native unread UI and opt-in coordinator wake policy remain separate product work." ], "demotionRule": "Demote or quarantine if the deterministic oracle flakes without a product or harness bug owned by orchestration." diff --git a/src/main/daemon/daemon-pty-adapter.ts b/src/main/daemon/daemon-pty-adapter.ts index 4d47720e159..bc59a5eff0b 100644 --- a/src/main/daemon/daemon-pty-adapter.ts +++ b/src/main/daemon/daemon-pty-adapter.ts @@ -1064,7 +1064,7 @@ export class DaemonPtyAdapter implements IPtyProvider { } } - write(id: string, data: string): void { + write(id: string, data: string): boolean { this.markSessionDirty(id) // Why recoverable and not just active: rejecting a write asks the pane to remount, // which only helps if this endpoint can come back. A legacy adapter has no respawn, @@ -1086,6 +1086,7 @@ export class DaemonPtyAdapter implements IPtyProvider { this.reconnectAfterWriteFailure() throw new PtyWriteUnavailableError(`Daemon PTY "${id}" is awaiting recovery`) } + return delivered } resize(id: string, cols: number, rows: number): void { diff --git a/src/main/daemon/daemon-pty-router.ts b/src/main/daemon/daemon-pty-router.ts index 5104268cbef..f42b8056a1f 100644 --- a/src/main/daemon/daemon-pty-router.ts +++ b/src/main/daemon/daemon-pty-router.ts @@ -89,8 +89,8 @@ export class DaemonPtyRouter implements IPtyProvider { return await this.ownerResolver.probe(id) } - write(id: string, data: string): void { - this.adapterFor(id).write(id, data) + write(id: string, data: string): boolean { + return this.adapterFor(id).write(id, data) } resize(id: string, cols: number, rows: number): void { diff --git a/src/main/daemon/degraded-daemon-pty-provider.ts b/src/main/daemon/degraded-daemon-pty-provider.ts index 545f66ca892..a113a82bd41 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.ts @@ -108,8 +108,8 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { this.sessionProviders.get(ptyId) ?? this.findProviderForExistingSession(ptyId) )?.providesAgentSessionOwnerListings?.(ptyId) === true - write(id: string, data: string): void { - this.providerFor(id).write(id, data) + write(id: string, data: string): boolean | void { + return this.providerFor(id).write(id, data) } resize(id: string, cols: number, rows: number): void { diff --git a/src/main/ipc/pty.ts b/src/main/ipc/pty.ts index 3356b6f7147..ce024131700 100644 --- a/src/main/ipc/pty.ts +++ b/src/main/ipc/pty.ts @@ -5430,8 +5430,7 @@ export function registerPtyHandlers( }, write: (ptyId, data) => { try { - getProviderForPty(ptyId).write(ptyId, data) - return true + return getProviderForPty(ptyId).write(ptyId, data) !== false } catch { return false } diff --git a/src/main/providers/local-pty-provider.test.ts b/src/main/providers/local-pty-provider.test.ts index a0cd84fc478..9ef2f47be1c 100644 --- a/src/main/providers/local-pty-provider.test.ts +++ b/src/main/providers/local-pty-provider.test.ts @@ -1454,12 +1454,12 @@ describe('LocalPtyProvider', () => { describe('write', () => { it('writes data to the PTY process', async () => { const { id } = await provider.spawn({ cols: 80, rows: 24 }) - provider.write(id, 'hello') + expect(provider.write(id, 'hello')).toBe(true) expect(mockProc.write).toHaveBeenCalledWith('hello') }) it('is a no-op for unknown PTY ids', () => { - provider.write('nonexistent', 'hello') + expect(provider.write('nonexistent', 'hello')).toBe(false) expect(mockProc.write).not.toHaveBeenCalled() }) }) diff --git a/src/main/providers/local-pty-provider.ts b/src/main/providers/local-pty-provider.ts index 55d8776d915..569583752ab 100644 --- a/src/main/providers/local-pty-provider.ts +++ b/src/main/providers/local-pty-provider.ts @@ -1150,15 +1150,20 @@ export class LocalPtyProvider implements IPtyProvider { hasPty(id: string): boolean { return ptyProcesses.has(id) } - write(id: string, data: string): void { + write(id: string, data: string): boolean { // Cooked PTYs echo private DSR/OSC replies; CPR/DA remain immediate (#13137, #7329). if (extractOnlyCookedEchoSafeQueryReplies(data)) { const ingress = startupIngressByPty.get(id) if (ingress?.answerLiveQueryReply(data)) { - return + return true } } - ptyProcesses.get(id)?.write(data) + const proc = ptyProcesses.get(id) + if (!proc) { + return false + } + proc.write(data) + return true } resize(id: string, cols: number, rows: number): void { ptyProcesses.get(id)?.resize(cols, rows) diff --git a/src/main/providers/pty-provider-contract.ts b/src/main/providers/pty-provider-contract.ts index cade47d84d2..49e110e5e10 100644 --- a/src/main/providers/pty-provider-contract.ts +++ b/src/main/providers/pty-provider-contract.ts @@ -126,7 +126,7 @@ export type IPtyProvider = { hasPty?: (id: string) => boolean /** Exact provider readback: false only when the provider answered that the PTY is absent. */ probePtyLiveness?: (id: string) => Promise - write(id: string, data: string): void + write(id: string, data: string): boolean | void resize(id: string, cols: number, rows: number): void /** * Producer-side flow control: stop/restart reading the underlying PTY so a diff --git a/src/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts index abf2f2f1e5c..893d9326ac2 100644 --- a/src/main/runtime/orca-runtime.test.ts +++ b/src/main/runtime/orca-runtime.test.ts @@ -34642,7 +34642,7 @@ describe('OrcaRuntimeService', () => { runtime.onPtyData('pty-1', '\x1b]0;Codex done\x07', 101) expect(write).toHaveBeenCalledWith( 'pty-1', - '\nYou have 1 orchestration message. Run `orca orchestration check`.\n' + '\nYou have 1 orchestration message. Run `orca orchestration check --run run_mailbox`.\n' ) expect(write).not.toHaveBeenCalledWith( 'pty-1', @@ -34990,7 +34990,7 @@ describe('OrcaRuntimeService', () => { await vi.waitFor(() => { expect(write).toHaveBeenCalledWith( 'pty-1', - '\nYou have 1 orchestration message. Run `orca orchestration check`.\n' + '\nYou have 1 orchestration message. Run `orca orchestration check --run run_codex_native_title`.\n' ) }) db.close() diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 03853430e89..c857a0d22c6 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -2915,6 +2915,7 @@ export class OrcaRuntimeService { string, ReadonlySet | null >() + private readonly pendingForeignDirectReconciliationsByLeafKey = new Set() private syntheticTerminalHandles = new Set() private detachedPreAllocatedLeaves = new Map() private graphSyncCallbacks: (() => void)[] = [] @@ -32420,12 +32421,76 @@ export class OrcaRuntimeService { } private deliverPendingMessagesForLeaf(leaf: RuntimeLeafRecord): void { + this.notifyForwardedOrchestrationMailboxes(this.routeForeignDirectMessagesForLeaf(leaf)) const mailboxHandle = this.resolveActionableMailboxForLeaf(leaf) if (mailboxHandle) { this.deliverPendingMessages(leaf, { mailboxHandle }) } } + private scheduleForeignDirectReconciliation(leaf: RuntimeLeafRecord): void { + const leafKey = this.getLeafKey(leaf.tabId, leaf.leafId) + if (this.pendingForeignDirectReconciliationsByLeafKey.has(leafKey)) { + return + } + this.pendingForeignDirectReconciliationsByLeafKey.add(leafKey) + setImmediate(() => { + this.pendingForeignDirectReconciliationsByLeafKey.delete(leafKey) + const currentLeaf = this.leaves.get(leafKey) + if (!currentLeaf || currentLeaf.ptyId !== leaf.ptyId) { + return + } + this.notifyForwardedOrchestrationMailboxes( + this.routeForeignDirectMessagesForLeaf(currentLeaf) + ) + }) + } + + private routeForeignDirectMessagesForLeaf( + leaf: RuntimeLeafRecord + ): { mailboxHandle: string; types: string[] }[] { + if (!this._orchestrationDb) { + return [] + } + const leafKey = this.getLeafKey(leaf.tabId, leaf.leafId) + const handle = this.handleByLeafKey.get(leafKey) + if (!handle) { + return [] + } + const paneKey = `${leaf.tabId}:${leaf.leafId}` + const ownerMailbox = this.resolveActionableMailboxForLeaf(leaf, undefined, { + routeDirectMail: false + }) + const ownerRunId = ownerMailbox?.startsWith('run:') + ? ownerMailbox.slice('run:'.length) + : ownerMailbox?.startsWith('dispatch:') + ? this._orchestrationDb.getDispatchContextById?.(ownerMailbox.slice('dispatch:'.length)) + ?.run_id + : undefined + if (!ownerRunId) { + return [] + } + const routed = this._orchestrationDb.routeForeignDirectMessagesToOwnedMailboxes?.( + handle, + ownerRunId, + paneKey + ) + if (routed?.hasMore) { + this.scheduleForeignDirectReconciliation(leaf) + } + return routed?.mailboxes ?? [] + } + + private notifyForwardedOrchestrationMailboxes( + forwarded: readonly { mailboxHandle: string; types: string[] }[] + ): void { + for (const routed of forwarded) { + for (const routedType of routed.types) { + this.notifyMessageArrived(routed.mailboxHandle, routedType) + } + } + } + private resolveActionableMailboxForLeaf( leaf: RuntimeLeafRecord, requestedMailbox?: string, @@ -32515,31 +32580,18 @@ export class OrcaRuntimeService { private resolveArrivedMessageMailboxes(handle: string): { mailboxHandle: string | null - forwarded: { runId: string; types: string[] }[] + forwarded: { mailboxHandle: string; types: string[] }[] } { if (!this.handles.has(handle) || !this._orchestrationDb) { return { mailboxHandle: handle, forwarded: [] } } try { const { leaf } = this.getLiveLeafForHandle(handle) - const ownerMailbox = this.resolveActionableMailboxForLeaf(leaf, undefined, { - routeDirectMail: false - }) - const ownerRunId = ownerMailbox?.startsWith('run:') - ? ownerMailbox.slice('run:'.length) - : ownerMailbox?.startsWith('dispatch:') - ? this._orchestrationDb.getDispatchContextById?.(ownerMailbox.slice('dispatch:'.length)) - ?.run_id - : undefined - const forwarded = ownerRunId - ? (this._orchestrationDb.routeForeignDirectMessagesToRunMailboxes?.(handle, ownerRunId) ?? - []) - : [] return { mailboxHandle: this.resolveActionableMailboxForLeaf(leaf, handle, { requireRequestedMail: true }), - forwarded + forwarded: this.routeForeignDirectMessagesForLeaf(leaf) } } catch { return { mailboxHandle: null, forwarded: [] } @@ -32552,11 +32604,7 @@ export class OrcaRuntimeService { this.mailPointerRepointScheduler.schedule(handle) } const { mailboxHandle, forwarded } = this.resolveArrivedMessageMailboxes(handle) - for (const routed of forwarded) { - for (const routedType of routed.types) { - this.notifyMessageArrived(`run:${routed.runId}`, routedType) - } - } + this.notifyForwardedOrchestrationMailboxes(forwarded) if (!mailboxHandle) { return } @@ -33243,7 +33291,6 @@ export class OrcaRuntimeService { this.pointedMessageMailboxHandlesByPtyId.delete(prior.ptyId) } } - this.lastPointedMessageSequenceByHandle.set(mailboxHandle, sequence) this.pointedMessageWatermarkOwnerByHandle.set(mailboxHandle, { ptyId, sequence, @@ -33265,9 +33312,6 @@ export class OrcaRuntimeService { return false } this.pointedMessageWatermarkOwnerByHandle.delete(mailboxHandle) - if (this.lastPointedMessageSequenceByHandle.get(mailboxHandle) === sequence) { - this.lastPointedMessageSequenceByHandle.delete(mailboxHandle) - } const mailboxes = this.pointedMessageMailboxHandlesByPtyId.get(ptyId) mailboxes?.delete(mailboxHandle) if (mailboxes?.size === 0) { @@ -33294,13 +33338,13 @@ export class OrcaRuntimeService { reservedTypes: ReadonlySet | undefined ): void { const prior = this.parkedMessageRedeliveryTypesByMailboxHandle.get(mailboxHandle) - if (prior === null || reservedTypes === undefined) { + if (prior === null || (prior === undefined && reservedTypes === undefined)) { this.parkedMessageRedeliveryTypesByMailboxHandle.set(mailboxHandle, null) return } this.parkedMessageRedeliveryTypesByMailboxHandle.set( mailboxHandle, - new Set([...(prior ?? []), ...reservedTypes]) + new Set([...(prior ?? []), ...(reservedTypes ?? [])]) ) } @@ -33354,30 +33398,30 @@ export class OrcaRuntimeService { this.pointedMessageMailboxHandlesByPtyId.delete(ptyId) } - private canSubmitMessagePointer( + private getMessagePointerSubmitDecision( leaf: RuntimeLeafRecord, mailboxHandle: string, messages: readonly { id: string; type: string }[] - ): boolean { + ): 'submit' | 'release' { if (this.resolveActionableMailboxForLeaf(leaf) !== mailboxHandle) { - return false + return 'release' } if ( mailboxHandle.startsWith('run:') && this._orchestrationDb?.hasOutstandingRunDelivery?.(mailboxHandle.slice('run:'.length)) ) { - return false + return 'release' } const waiters = this.messageWaitersByHandle.get(mailboxHandle) if (messages.some((message) => messageTypeHasLiveWaiter(waiters, message.type))) { - return false + return 'release' } - return ( - this._orchestrationDb?.areUndeliveredUnreadMessages?.( - mailboxHandle, - messages.map((message) => message.id) - ) ?? true - ) + return (this._orchestrationDb?.areUndeliveredUnreadMessages?.( + mailboxHandle, + messages.map((message) => message.id) + ) ?? true) + ? 'submit' + : 'release' } // Why: push-on-idle delivery is event-driven (no polling) because the runtime owns both the message store and terminal status detection. @@ -33485,7 +33529,8 @@ export class OrcaRuntimeService { if (newestSequence === undefined) { return } - const pointedSequence = this.lastPointedMessageSequenceByHandle.get(mailboxHandle) ?? -1 + const pointedSequence = + this.pointedMessageWatermarkOwnerByHandle.get(mailboxHandle)?.sequence ?? -1 if (newestSequence <= pointedSequence) { const owner = this.pointedMessageWatermarkOwnerByHandle.get(mailboxHandle) const leafKey = this.getLeafKey(leaf.tabId, leaf.leafId) @@ -33556,12 +33601,9 @@ export class OrcaRuntimeService { const deliveryPtyId = leaf.ptyId const flight: { enterTimer: ReturnType | null } = { enterTimer: null } this.messageDeliveryFlightsByPtyId.set(deliveryPtyId, flight) - // Why: every sync outcome — failed write, Cursor branch, or a throw — - // must end the flight here, or a leaked flag parks this pty's deliveries - // forever. Only an armed Enter hands settling to its own callback. let settlesInEnterCallback = false try { - const payload = formatMessagePointer(unread.length) + const payload = formatMessagePointer(unread.length, mailboxHandle) const wrote = this.ptyController?.write(deliveryPtyId, payload) ?? false if (!wrote) { return @@ -33585,7 +33627,6 @@ export class OrcaRuntimeService { const tabTitle = this.tabs.get(leaf.tabId)?.title if (isCursorAgentOrchestrationTarget(leaf, tabTitle)) { - // Why: Cursor Agent treats injected PTY text as editable prompt input, so submitting must stay under user control. let delivered = false try { this._orchestrationDb.markAsDelivered(unread.map((message) => message.id)) @@ -33605,11 +33646,10 @@ export class OrcaRuntimeService { flight.enterTimer = setTimeout(() => { let clearAndRedrive = false let delivered = false + let releaseWithoutRedrive = false let finalizeReservation = true void this.isLeafPtyProvenAbsent(deliveryPtyId) .then((absent) => { - // Why current state, not the closure: ownership, graph records, or the - // process can change during the Enter delay and liveness probe. if (absent) { clearAndRedrive = true return @@ -33629,11 +33669,16 @@ export class OrcaRuntimeService { } if ( currentLeaf.lastAgentStatus !== 'idle' || - !currentLeaf.lastAgentStatusObservedLive || - !this.canSubmitMessagePointer(currentLeaf, mailboxHandle, unread) + !currentLeaf.lastAgentStatusObservedLive ) { return } + if ( + this.getMessagePointerSubmitDecision(currentLeaf, mailboxHandle, unread) === 'release' + ) { + releaseWithoutRedrive = true + return + } const submitted = this.ptyController?.write(deliveryPtyId, '\r') ?? false if (submitted) { this._orchestrationDb?.markAsDelivered(unread.map((message) => message.id)) @@ -33647,7 +33692,7 @@ export class OrcaRuntimeService { let released = false if (finalizeReservation) { released = - delivered || clearAndRedrive + delivered || clearAndRedrive || releaseWithoutRedrive ? this.clearPointedMessageWatermark(mailboxHandle, newestSequence, deliveryPtyId) : this.deactivatePointedMessageWatermark( mailboxHandle, @@ -33656,7 +33701,7 @@ export class OrcaRuntimeService { ) } this.settlePendingMessageDelivery(deliveryPtyId, flight) - if (released) { + if (released && !releaseWithoutRedrive) { this.redrivePendingMessageMailbox(mailboxHandle, clearAndRedrive) } }) diff --git a/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts b/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts index 67deadece50..f4902ce06d0 100644 --- a/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts +++ b/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts @@ -256,6 +256,88 @@ describe('orchestration notification mailbox consistency', () => { db.close() }) + it('reconciles the persisted production mismatch after runtime restart', async () => { + vi.useFakeTimers() + const directory = mkdtempSync(join(tmpdir(), 'orca-mailbox-upgrade-restart-')) + temporaryDirectories.push(directory) + const dbPath = join(directory, 'orchestration.db') + const oldDb = new OrchestrationDb(dbPath) + const runA = createBoundRun(oldDb, 'Persisted Run A') + const message = insertDirectRunMessage(oldDb, runA.id, 'Persisted Run A completion') + createBoundRun(oldDb, 'Persisted Run B') + sqliteFor(oldDb) + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(TERMINAL_HANDLE, message.id) + oldDb.close() + + const restartedDb = new OrchestrationDb(dbPath) + const restarted = createRuntime(restartedDb) + await driveToLiveIdle(restarted.runtime) + + expect(pointerCount(restarted.write)).toBe(0) + expect(restartedDb.getMessageById(message.id)).toMatchObject({ + to_handle: `run:${runA.id}`, + read: 0, + delivered_at: null + }) + + registerSecondPane(restarted.runtime) + restartedDb.bindRun({ + runId: runA.id, + coordinatorHandle: SECOND_TERMINAL_HANDLE, + coordinatorPaneKey: SECOND_PANE_KEY + }) + restarted.runtime.onPtyData(SECOND_PTY_ID, '\x1b]0;Codex working\x07', 1) + restarted.runtime.onPtyData(SECOND_PTY_ID, '\x1b]0;Codex done\x07', 2) + const checked = await checkBoundMailbox(restarted.runtime, { + terminal: SECOND_TERMINAL_HANDLE, + paneKey: SECOND_PANE_KEY, + launchToken: SECOND_LAUNCH_TOKEN + }) + + expect(checked).toMatchObject({ runId: runA.id, count: 1 }) + expect(checked.messages).toEqual([expect.objectContaining({ id: message.id })]) + restartedDb.close() + }) + + it('pages a large persisted mismatch and coalesces each mailbox wake', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-mailbox-paged-reconciliation-') + const harness = createRuntime(db) + const runA = createBoundRun(db, 'Backlog Run A') + const messages = Array.from({ length: 151 }, (_, index) => + insertDirectRunMessage(db, runA.id, `Backlog message ${index}`) + ) + createBoundRun(db, 'Backlog Run B') + const placeholders = messages.map(() => '?').join(',') + sqliteFor(db) + .prepare(`UPDATE messages SET to_handle = ? WHERE id IN (${placeholders})`) + .run(TERMINAL_HANDLE, ...messages.map((message) => message.id)) + const notify = vi.spyOn(harness.runtime, 'notifyMessageArrived') + const directRemaining = (): number => + ( + sqliteFor(db) + .prepare('SELECT COUNT(*) AS count FROM messages WHERE to_handle = ?') + .get(TERMINAL_HANDLE) as { count: number } + ).count + + await driveToLiveIdle(harness.runtime) + expect(directRemaining()).toBe(101) + expect(notify).toHaveBeenCalledTimes(1) + + await vi.advanceTimersToNextTimerAsync() + expect(directRemaining()).toBe(51) + expect(notify).toHaveBeenCalledTimes(2) + await vi.advanceTimersToNextTimerAsync() + expect(directRemaining()).toBe(1) + expect(notify).toHaveBeenCalledTimes(3) + await vi.advanceTimersToNextTimerAsync() + expect(directRemaining()).toBe(0) + expect(notify).toHaveBeenCalledTimes(4) + expect(pointerCount(harness.write)).toBe(0) + db.close() + }) + it('does not submit or replay a pointer after ownership changes during Enter delay', async () => { vi.useFakeTimers() const directory = mkdtempSync(join(tmpdir(), 'orca-mailbox-restart-')) @@ -289,6 +371,52 @@ describe('orchestration notification mailbox consistency', () => { restartedDb.close() }) + it('fences the pointed mailbox instead of checking a rebound empty Run', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-mailbox-post-submit-rebind-') + const harness = createRuntime(db) + const runA = createBoundRun(db, 'Run A') + insertDirectRunMessage(db, runA.id, 'Run A completion') + + await driveToLiveIdle(harness.runtime) + await vi.advanceTimersByTimeAsync(500) + expect(harness.write).toHaveBeenCalledWith( + PTY_ID, + expect.stringContaining(`orchestration check --run ${runA.id}`) + ) + + db.bindRun({ + runId: runA.id, + coordinatorHandle: 'term_new_coordinator', + coordinatorPaneKey: + '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' + }) + const runB = createBoundRun(db, 'Run B') + const response = await new RpcDispatcher({ + runtime: harness.runtime, + methods: ORCHESTRATION_METHODS + }).dispatch({ + id: 'req-pointed-mailbox-fenced', + authToken: 'test-auth-token', + method: 'orchestration.check', + orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, + orchestrationCompatibilityEvidence: { + terminalHandle: TERMINAL_HANDLE, + paneKey: PANE_KEY, + launchToken: LAUNCH_TOKEN + }, + params: { terminal: TERMINAL_HANDLE, run: runA.id } + }) + + expect(response).toMatchObject({ + ok: false, + error: { code: 'consumer_fenced' } + }) + const generic = await checkBoundMailbox(harness.runtime) + expect(generic).toMatchObject({ runId: runB.id, count: 0 }) + db.close() + }) + it('releases a staged pointer when its Run moves to another live pane', async () => { vi.useFakeTimers() const db = createDatabase('orca-mailbox-live-rebind-') @@ -393,6 +521,35 @@ describe('orchestration notification mailbox consistency', () => { db.close() }) + it('releases staged pointer state when an explicit check owns the batch', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-mailbox-explicit-check-') + const harness = createRuntime(db) + const run = createBoundRun(db, 'Explicit-check Run') + insertDirectRunMessage(db, run.id, 'Checked before Enter') + + await driveToLiveIdle(harness.runtime) + expect(pointerCount(harness.write)).toBe(1) + const checked = await checkBoundMailbox(harness.runtime) + expect(checked).toMatchObject({ runId: run.id, count: 1 }) + + await vi.advanceTimersByTimeAsync(500) + const internals = harness.runtime as unknown as { + pointedMessageWatermarkOwnerByHandle: Map + pointedMessageMailboxHandlesByPtyId: Map> + parkedMessageRedeliveryTypesByMailboxHandle: Map + } + expect(internals.pointedMessageWatermarkOwnerByHandle.size).toBe(0) + expect(internals.pointedMessageMailboxHandlesByPtyId.size).toBe(0) + expect(internals.parkedMessageRedeliveryTypesByMailboxHandle.size).toBe(0) + + const redrive = vi.spyOn(harness.runtime, 'deliverPendingMessagesForHandle') + harness.runtime.onPtyExit(PTY_ID, 0) + await Promise.resolve() + expect(redrive).not.toHaveBeenCalled() + db.close() + }) + it('routes same-Run direct mail through the bound check and acknowledgment path', async () => { vi.useFakeTimers() const db = createDatabase('orca-mailbox-current-run-') @@ -485,6 +642,34 @@ describe('orchestration notification mailbox consistency', () => { restartedDb.close() }) + it('keeps mail retryable when the provider rejects Enter', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-mailbox-provider-refusal-') + const first = createRuntime(db) + const recordWrite = first.write as unknown as (id: string, payload: string) => unknown + const write = vi.fn((ptyId: string, data: string) => { + recordWrite(ptyId, data) + return data !== '\r' + }) + first.runtime.setPtyController({ + write, + kill: vi.fn(), + getForegroundProcess: async () => null + }) + const run = createBoundRun(db, 'Provider-refusal Run') + const message = insertDirectRunMessage(db, run.id, 'Retry after refusal') + + await driveToLiveIdle(first.runtime) + await vi.advanceTimersByTimeAsync(500) + expect(pointerCount(first.write)).toBe(1) + expect(db.getMessageById(message.id)?.delivered_at).toBeNull() + + const restarted = createRuntime(db) + await driveToLiveIdle(restarted.runtime) + expect(pointerCount(restarted.write)).toBe(1) + db.close() + }) + it('does not point newer Run mail behind an outstanding Delivery', async () => { vi.useFakeTimers() const db = createDatabase('orca-mailbox-outstanding-') @@ -551,11 +736,9 @@ describe('orchestration notification mailbox consistency', () => { delivered_at: expect.any(String) }) const internals = harness.runtime as unknown as { - lastPointedMessageSequenceByHandle: Map pointedMessageWatermarkOwnerByHandle: Map pointedMessageMailboxHandlesByPtyId: Map> } - expect(internals.lastPointedMessageSequenceByHandle.has(`dispatch:${dispatch.id}`)).toBe(false) expect(internals.pointedMessageWatermarkOwnerByHandle.has(`dispatch:${dispatch.id}`)).toBe( false ) @@ -599,6 +782,51 @@ describe('orchestration notification mailbox consistency', () => { db.close() }) + it('preserves an active Dispatch as owner of displaced direct mail', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-mailbox-displaced-dispatch-') + const harness = createRuntime(db) + const workerRun = db.createRun({ + objective: 'Worker Run', + coordinatorHandle: 'term_worker_coordinator', + coordinatorPaneKey: + '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' + }) + const task = db.createTask({ spec: 'Worker task', runId: workerRun.id }) + const dispatch = db.createDispatchContext( + task.id, + 'term_mailbox_before_remint', + `99999999-9999-4999-8999-999999999999:${LEAF_ID}` + ) + createBoundRun(db, 'Current coordinator Run') + const waiting = harness.runtime.waitForMessage(`dispatch:${dispatch.id}`, { + timeoutMs: 5_000 + }) + const message = db.insertMessage({ + from: 'term_worker_coordinator', + to: TERMINAL_HANDLE, + subject: 'Late worker instruction', + runId: workerRun.id + }) + + harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) + await expect(waiting).resolves.toBe('notified') + expect(pointerCount(harness.write)).toBe(0) + expect(db.getMessageById(message.id)?.to_handle).toBe(`dispatch:${dispatch.id}`) + + const currentRun = db.getCurrentRunForPane(PANE_KEY) + db.bindRun({ + runId: currentRun!.id, + coordinatorHandle: 'term_rebound_coordinator', + coordinatorPaneKey: + '77777777-7777-4777-8777-777777777777:88888888-8888-4888-8888-888888888888' + }) + const checked = await checkBoundMailbox(harness.runtime) + expect(checked).toMatchObject({ runId: workerRun.id, dispatchId: dispatch.id, count: 1 }) + expect(checked.messages).toEqual([expect.objectContaining({ id: message.id })]) + db.close() + }) + it('uses bounded ownership lookup for direct arrivals while the agent is working', async () => { vi.useFakeTimers() const db = createDatabase('orca-mailbox-bounded-') diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 7c3eed16c0c..0d4724dc6b1 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -90,6 +90,14 @@ function parseWorkerTerminalPriorOwnerIds(value: string): string[] | null { const MESSAGE_ID_UPDATE_BATCH_SIZE = 500 export const ORCHESTRATION_DELIVERY_BATCH_LIMIT = 50 +export type DirectMailboxRoutingPage = { + routedCount: number + hasMore: boolean +} + +export type ForeignDirectMailboxRoutingPage = DirectMailboxRoutingPage & { + mailboxes: { mailboxHandle: string; types: MessageType[] }[] +} // Why: indexable pre-filter for isEquivalentPaneKey — equal strings and equal leaves both share the // text after the first ':', so this narrows candidates without deciding equivalence itself. const RUN_PANE_KEY_MATCH_SUFFIX_SQL = @@ -2710,7 +2718,10 @@ export class OrchestrationDb { limit?: number wakeTypes?: MessageType[] }): { delivery: DeliveryRow; messages: MessageRow[]; replayed: boolean } | undefined { - const limit = Math.min(Math.max(params.limit ?? 50, 1), 50) + const limit = Math.min( + Math.max(params.limit ?? ORCHESTRATION_DELIVERY_BATCH_LIMIT, 1), + ORCHESTRATION_DELIVERY_BATCH_LIMIT + ) this.db.exec('BEGIN IMMEDIATE') try { this.requireCurrentConsumer(params.runId, params.consumerGeneration) @@ -3586,65 +3597,137 @@ export class OrchestrationDb { } // Why: change mailbox ownership without changing unread or acknowledgment state. - routeUnreadDirectMessagesToRunMailbox(runId: string, directHandle: string): number { - const result = this.db + private routeDirectMessagePage( + mailboxHandle: string, + runId: string, + directHandle: string + ): DirectMailboxRoutingPage { + const rows = this.db .prepare( - `UPDATE messages - SET to_handle = ? + `SELECT id FROM messages WHERE run_id = ? AND to_handle = ? AND read = 0 - AND delivery_contract = 'current_delivery'` + AND delivery_contract = 'current_delivery' + ORDER BY sequence LIMIT ?` ) - .run(`run:${runId}`, runId, directHandle) - return Number(result.changes) + .all(runId, directHandle, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1) as { id: string }[] + const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT) + if (page.length === 0) { + return { routedCount: 0, hasMore: false } + } + const placeholders = page.map(() => '?').join(',') + const result = this.db + .prepare(`UPDATE messages SET to_handle = ? WHERE id IN (${placeholders})`) + .run(mailboxHandle, ...page.map((row) => row.id)) + return { + routedCount: Number(result.changes), + hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT + } + } + + routeUnreadDirectMessagesToRunMailbox( + runId: string, + directHandle: string + ): DirectMailboxRoutingPage { + return this.routeDirectMessagePage(`run:${runId}`, runId, directHandle) } routeUnreadDirectMessagesToDispatchMailbox( dispatchId: string, runId: string, directHandle: string - ): number { - const result = this.db - .prepare( - `UPDATE messages - SET to_handle = ? - WHERE run_id = ? AND to_handle = ? AND read = 0 - AND delivery_contract = 'current_delivery'` - ) - .run(`dispatch:${dispatchId}`, runId, directHandle) - return Number(result.changes) + ): DirectMailboxRoutingPage { + return this.routeDirectMessagePage(`dispatch:${dispatchId}`, runId, directHandle) } - routeForeignDirectMessagesToRunMailboxes( + private findActiveDispatchForDirectMessageOwner( + runId: string, directHandle: string, - currentRunId: string - ): { runId: string; types: MessageType[] }[] { - const routedRows = this.db + paneKey?: string + ): DispatchContextRow | undefined { + const exact = this.db .prepare( - `SELECT DISTINCT messages.run_id, messages.type FROM messages - WHERE to_handle = ? AND run_id <> ? AND read = 0 AND delivered_at IS NULL - AND delivery_contract = 'current_delivery' - AND EXISTS (SELECT 1 FROM runs WHERE runs.id = messages.run_id AND runs.legacy = 0)` + `SELECT * FROM dispatch_contexts + WHERE run_id = ? AND assignee_handle = ? AND status IN ('pending', 'dispatched') + ORDER BY rowid DESC LIMIT 1` ) - .all(directHandle, currentRunId) as { run_id: string; type: MessageType }[] - if (routedRows.length === 0) { - return [] + .get(runId, directHandle) as DispatchContextRow | undefined + if (exact || !paneKey || !parsePaneKey(paneKey)) { + return exact } - this.db + return this.db .prepare( - `UPDATE messages - SET to_handle = 'run:' || run_id + `SELECT * FROM dispatch_contexts + WHERE run_id = ? AND assignee_pane_key IS NOT NULL + AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1 + AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ? + ORDER BY rowid DESC LIMIT 1` + ) + .get(runId, paneKeyMatchSuffix(paneKey)) as DispatchContextRow | undefined + } + + routeForeignDirectMessagesToOwnedMailboxes( + directHandle: string, + currentRunId: string, + paneKey?: string + ): ForeignDirectMailboxRoutingPage { + const rows = this.db + .prepare( + `SELECT id, run_id, type FROM messages WHERE to_handle = ? AND run_id <> ? AND read = 0 AND delivered_at IS NULL AND delivery_contract = 'current_delivery' - AND EXISTS (SELECT 1 FROM runs WHERE runs.id = messages.run_id AND runs.legacy = 0)` + AND EXISTS (SELECT 1 FROM runs WHERE runs.id = messages.run_id AND runs.legacy = 0) + ORDER BY sequence LIMIT ?` ) - .run(directHandle, currentRunId) - const byRun = new Map>() - for (const row of routedRows) { - const types = byRun.get(row.run_id) ?? new Set() + .all(directHandle, currentRunId, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1) as { + id: string + run_id: string + type: MessageType + }[] + const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT) + if (page.length === 0) { + return { routedCount: 0, hasMore: false, mailboxes: [] } + } + const runIds = [...new Set(page.map((row) => row.run_id))] + const dispatchByRun = new Map() + for (const runId of runIds) { + const dispatch = this.findActiveDispatchForDirectMessageOwner(runId, directHandle, paneKey) + if (dispatch) { + dispatchByRun.set(runId, dispatch) + } + } + const idsByMailbox = new Map() + const byMailbox = new Map>() + for (const row of page) { + const dispatch = dispatchByRun.get(row.run_id) + const mailboxHandle = dispatch ? `dispatch:${dispatch.id}` : `run:${row.run_id}` + const ids = idsByMailbox.get(mailboxHandle) ?? [] + ids.push(row.id) + idsByMailbox.set(mailboxHandle, ids) + const types = byMailbox.get(mailboxHandle) ?? new Set() types.add(row.type) - byRun.set(row.run_id, types) + byMailbox.set(mailboxHandle, types) + } + this.db.exec('BEGIN IMMEDIATE') + try { + for (const [mailboxHandle, ids] of idsByMailbox) { + const placeholders = ids.map(() => '?').join(',') + this.db + .prepare(`UPDATE messages SET to_handle = ? WHERE id IN (${placeholders})`) + .run(mailboxHandle, ...ids) + } + this.db.exec('COMMIT') + } catch (error) { + this.db.exec('ROLLBACK') + throw error + } + return { + routedCount: page.length, + hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT, + mailboxes: [...byMailbox].map(([mailboxHandle, types]) => ({ + mailboxHandle, + types: [...types] + })) } - return [...byRun].map(([runId, types]) => ({ runId, types: [...types] })) } areUndeliveredUnreadMessages(toHandle: string, ids: string[]): boolean { @@ -5114,18 +5197,14 @@ export class OrchestrationDb { } findActiveRemoteAttachmentForPane(paneKey: string): RemoteDispatchAttachmentRow | undefined { - const exact = this.db - .prepare( - `SELECT * FROM remote_dispatch_attachments - WHERE state IN ('starting', 'ready') AND pane_key = ? - ORDER BY rowid DESC LIMIT 1` - ) - .get(paneKey) as RemoteDispatchAttachmentRow | undefined - if (exact) { - return exact - } if (!parsePaneKey(paneKey)) { - return undefined + return this.db + .prepare( + `SELECT * FROM remote_dispatch_attachments + WHERE state IN ('starting', 'ready') AND pane_key = ? + ORDER BY rowid DESC LIMIT 1` + ) + .get(paneKey) as RemoteDispatchAttachmentRow | undefined } return this.db .prepare( diff --git a/src/main/runtime/orchestration/formatter.test.ts b/src/main/runtime/orchestration/formatter.test.ts index f7d9c14d880..62f46e03020 100644 --- a/src/main/runtime/orchestration/formatter.test.ts +++ b/src/main/runtime/orchestration/formatter.test.ts @@ -186,8 +186,8 @@ describe('formatMessagesForInjection', () => { describe('formatMessagePointer', () => { it('formats a singular pointer without message content', () => { - expect(formatMessagePointer(1)).toBe( - '\nYou have 1 orchestration message. Run `orca orchestration check`.\n' + expect(formatMessagePointer(1, 'run:run_1')).toBe( + '\nYou have 1 orchestration message. Run `orca orchestration check --run run_1`.\n' ) }) diff --git a/src/main/runtime/orchestration/formatter.ts b/src/main/runtime/orchestration/formatter.ts index 678c5e08499..c204cac18c0 100644 --- a/src/main/runtime/orchestration/formatter.ts +++ b/src/main/runtime/orchestration/formatter.ts @@ -108,7 +108,10 @@ export function formatMessagesForInjection(messages: MessageRow[]): string { return `\n--- Orchestration Messages (${messages.length}) ---\n${banners}\n---\n` } -export function formatMessagePointer(count: number): string { +export function formatMessagePointer(count: number, mailboxHandle?: string): string { const noun = count === 1 ? 'message' : 'messages' - return `\nYou have ${count} orchestration ${noun}. Run \`orca orchestration check\`.\n` + const runFlag = mailboxHandle?.startsWith('run:') + ? ` --run ${mailboxHandle.slice('run:'.length)}` + : '' + return `\nYou have ${count} orchestration ${noun}. Run \`orca orchestration check${runFlag}\`.\n` } diff --git a/src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts b/src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts index ef2c4011897..f43b87dce4a 100644 --- a/src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts +++ b/src/main/runtime/orchestration/orchestration-worker-dispatch-db.test.ts @@ -328,6 +328,9 @@ describe('OrchestrationDb worker Dispatch state', () => { expect(d.findActiveRemoteAttachmentForPane(`tab_reminted:${leafId}`)?.dispatch_id).toBe( 'ctx_valid_new' ) + expect(d.findActiveRemoteAttachmentForPane(`tab_old:${leafId}`)?.dispatch_id).toBe( + 'ctx_valid_new' + ) }) it('returns already-settled when completion wins before stop', () => { diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index 77dd8279ac9..f0bf56d1f7d 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -741,6 +741,14 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ const generation = run.consumer_generation const address = `run:${run.id}` runtime.ensureOrchestrationFederationRelay(run.id) + const routedDirect = db.routeUnreadDirectMessagesToRunMailbox(run.id, handle) + if ( + routedDirect.routedCount === 0 && + run.coordinator_handle && + run.coordinator_handle !== handle + ) { + db.routeUnreadDirectMessagesToRunMailbox(run.id, run.coordinator_handle) + } const acknowledged = params.ack ? db.acknowledgeRunDelivery({ @@ -909,6 +917,24 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ : undefined if (workerMailbox) { const address = `dispatch:${workerMailbox.dispatchId}` + if (activeDispatch) { + const routedDirect = db.routeUnreadDirectMessagesToDispatchMailbox( + activeDispatch.id, + activeDispatch.run_id, + handle + ) + if ( + routedDirect.routedCount === 0 && + activeDispatch.assignee_handle && + activeDispatch.assignee_handle !== handle + ) { + db.routeUnreadDirectMessagesToDispatchMailbox( + activeDispatch.id, + activeDispatch.run_id, + activeDispatch.assignee_handle + ) + } + } const showAll = params.all === true || (params.unread === false && params.peek !== true) const messages = showAll ? db.getAllMessagesForHandle(address, 100, typeFilter) diff --git a/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts b/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts index 62bec91360f..85c541cea2b 100644 --- a/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts +++ b/src/main/runtime/terminal-send-stale-leaf-liveness.test.ts @@ -607,20 +607,21 @@ describe('push-on-idle orchestration delivery absence gate', () => { const internals = runtime as unknown as { messageDeliveryFlightsByPtyId: Map parkedMessageRedeliveriesByPtyId: Map - lastPointedMessageSequenceByHandle: Map + pointedMessageWatermarkOwnerByHandle: Map } stub.insert('first') runtime.deliverPendingMessagesForHandle(handle) stub.insert('second') runtime.deliverPendingMessagesForHandle(handle) + await Promise.resolve() expect(internals.messageDeliveryFlightsByPtyId.size).toBe(1) expect(internals.parkedMessageRedeliveriesByPtyId.size).toBe(1) - expect(internals.lastPointedMessageSequenceByHandle.size).toBe(1) + expect(internals.pointedMessageWatermarkOwnerByHandle.size).toBe(1) runtime.onPtyExit(STALE_PTY_ID, 0) expect(internals.messageDeliveryFlightsByPtyId.size).toBe(0) expect(internals.parkedMessageRedeliveriesByPtyId.size).toBe(0) - expect(internals.lastPointedMessageSequenceByHandle.size).toBe(0) + expect(internals.pointedMessageWatermarkOwnerByHandle.size).toBe(0) await vi.advanceTimersByTimeAsync(500) expect(write.mock.calls.filter(([, data]) => data === '\r')).toHaveLength(0)