From 481d8bbe84da4b729bdbab8386ae7fb32505c02a Mon Sep 17 00:00:00 2001 From: Jinwoo-H Date: Fri, 14 Aug 2026 17:59:23 -0400 Subject: [PATCH] fix(orchestration): preserve mailbox delivery identity --- config/reliability-gates.jsonc | 98 +++- src/main/providers/ssh-pty-write.test.ts | 47 ++ src/main/providers/ssh-pty-write.ts | 20 +- src/main/runtime/orca-runtime.ts | 17 +- ...estration-mailbox-detached-routing.test.ts | 424 +++++++++++++++++ ...n-mailbox-notification-consistency.test.ts | 233 +++------ ...ation-mailbox-notification-test-harness.ts | 6 +- ...stration-message-delivery-identity.test.ts | 127 ++++- src/main/runtime/orchestration/db.ts | 444 +++++++++++++----- ...ederation-acknowledgment-migration.test.ts | 4 +- .../orchestration/federation-sync.test.ts | 91 ++++ .../runtime/orchestration/federation-sync.ts | 4 +- .../mailbox-notification-coordinator.ts | 39 +- .../runtime/orchestration/mailbox-owner.ts | 38 ++ .../mutation-receipt-capacity.test.ts | 2 +- ...hestration-db-retention-pagination.test.ts | 8 +- .../orchestration-reset-db.test.ts | 3 + .../orchestration-run-delivery-db.test.ts | 24 + .../run-coordinator-handle-migration.test.ts | 138 ++++++ .../runtime/rpc/methods/orchestration-runs.ts | 2 + .../runtime/rpc/methods/orchestration.test.ts | 103 ++++ src/main/runtime/rpc/methods/orchestration.ts | 65 ++- .../runtime/rpc/orchestration-legacy-mail.ts | 2 +- 23 files changed, 1605 insertions(+), 334 deletions(-) create mode 100644 src/main/runtime/orchestration-mailbox-detached-routing.test.ts create mode 100644 src/main/runtime/orchestration/run-coordinator-handle-migration.test.ts diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 8c29c88336f..65b741c4334 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -883,7 +883,7 @@ ], "evidenceRuns": [ { - "date": "2026-08-13", + "date": "2026-08-14", "runner": "local", "platform": "macos", "command": "pnpm --dir mobile exec vitest run --root .. mobile/src/transport/client-context.test.ts mobile/src/transport/host-open-recovery.test.tsx mobile/src/transport/host-open-retry-scheduler.test.ts mobile/src/transport/rpc-session-liveness-watchdog.test.ts mobile/src/transport/rpc-session-liveness-integration.test.ts mobile/src/transport/mobile-relay-rpc-session.test.ts mobile/src/transport/mobile-relay-rpc-session-liveness.test.ts mobile/src/transport/mobile-endpoint-nudge-router.test.ts mobile/src/transport/mobile-relay-e2ee-link.test.ts mobile/src/transport/stable-logical-rpc-client.test.ts", @@ -8805,23 +8805,27 @@ "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. The STA-4325 oracle separately fixes one status, dispatch, and worker_done message across direct-handle and canonical Run recipients, then proves notification count, concurrent Delivery identity, unread/peek/all/type filters, explicit acknowledgment, restart replay, and filtered-waiter wakeup against the same isolated SQLite rows. 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.", + "coverageNotes": "A deterministic isolated runtime/SQLite/RPC harness covers Run and Dispatch ownership, exact handle and remint-stable pane recipients, fixed Delivery replay and acknowledgment, restart, concurrent pollers, filtered waiters, and built-CLI parity with direct DB state. Schema v28 durably records every coordinator handle through SQLite triggers, backfills current handles on open, recreates routing triggers after old table rebuilds, and atomically canonicalizes coordinator-addressed inserts to run:. A 125-row oracle and 151-row reconciliation suite cover paged old-handle backlogs, commit-without-notify recovery, duplicate Dispatch owners, cancellation, and lifecycle fencing. Federation tests use real databases, dropped acknowledgments, restart/reconnect, and protocol v1/v2 peers; duplicate read rows do not wake again. Daemon and SSH pointer settlement is deterministic, with SSH bounded at 60 seconds beyond the 20-second mux health window. No RPC parameter, stream opcode, or published payload changed. Live headed/headless paired-runtime, physical SSH, WSL, Linux, and Windows acceptance remain gaps.", "motivatingLinks": [ "https://linear.app/stably/issue/STA-4325", "https://github.com/stablyai/orca/issues/14522", "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. 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.", + "invariant": "Each message has one stable row ID and authoritative recipient; coordinator-addressed current-delivery inserts are atomically owned by run:. Pointer staging may set delivered_at but never consumes mail. Each Run consumer generation has at most one outstanding Delivery with a fixed ID and fixed message IDs; ordinary checks replay it until an explicit matching acknowledgment marks exactly those rows read. Rebinding fences the old generation, notification types/counts correspond to unread rows retrievable under the same authority, and federation replay imports each stable message identity once without re-waking an already-read duplicate.", + "oracle": "Seed status, dispatch, and worker_done rows across direct-handle and canonical Run recipients in an isolated DB. Compare pointer count, RPC and built-CLI check output, direct SQLite rows, unread/peek/all/type filters, concurrent pollers, fixed Delivery IDs, explicit acknowledgment, restart, filtered check --wait, and coordinator remint. Route a 125-row old-handle backlog, inject a commit without notification, and require startup repair. Exercise duplicate Run/Dispatch owners, stale panes, 50-row pages, cancellation, lifecycle fencing, and absent PTYs. Drop a federation ACK, reconnect/restart v1/v2 peers, and require stable import plus no duplicate read-row wake. Hold a healthy SSH write past five seconds but below the 60-second settlement deadline, then separately exceed the bound and require retryable undelivered state.", "commands": [ - "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-message-delivery-identity.test.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" + "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-message-delivery-identity.test.ts --reporter=dot --testTimeout=5000", + "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-detached-routing.test.ts src/main/runtime/orchestration-mailbox-transport-settlement.test.ts src/main/runtime/orchestration/run-coordinator-handle-migration.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", + "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/federation-sync.test.ts src/main/runtime/rpc/methods/orchestration-federation.test.ts src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts --reporter=dot" ], "testFiles": [ "src/main/runtime/orchestration-message-delivery-identity.test.ts", "src/main/runtime/orchestration-mailbox-notification-consistency.test.ts", + "src/main/runtime/orchestration-mailbox-detached-routing.test.ts", "src/main/runtime/orchestration-mailbox-routing-races.test.ts", "src/main/runtime/orchestration-mailbox-transport-settlement.test.ts", + "src/main/runtime/orchestration/run-coordinator-handle-migration.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", @@ -8832,7 +8836,10 @@ "src/main/daemon/degraded-daemon-pty-provider.test.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" + "src/main/runtime/rpc/methods/orchestration.test.ts", + "src/main/runtime/orchestration/federation-sync.test.ts", + "src/main/runtime/rpc/methods/orchestration-federation.test.ts", + "src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts" ], "assertionRefs": [ { @@ -8842,7 +8849,10 @@ "notification count equals the concurrently replayed fixed Delivery batch", "unread, peek, all, and type filters preserve explicit acknowledgment semantics", "restart replays one outstanding Delivery ID without inserting another Delivery row", - "a filtered check --wait wakes once without a duplicate terminal pointer" + "a filtered check --wait wakes once without a duplicate terminal pointer", + "a 125-row old-handle backlog is atomically rebound before coordinator identity changes", + "committed coordinator mail is canonicalized without relying on notification delivery", + "the built CLI returns the same fixed Delivery and count as isolated SQLite" ] }, { @@ -8863,12 +8873,47 @@ "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", + "same-owner filtered checks drain later 50-row pages before waiting" + ] + }, + { + "file": "src/main/runtime/orchestration-mailbox-detached-routing.test.ts", + "assertions": [ "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", - "filtered waiters remain single-consumer when direct rows route into Run or Dispatch mailboxes", - "same-owner filtered checks drain later 50-row pages before waiting", - "working-agent arrival uses bounded ownership lookup without a full mailbox scan" + "bound and filtered waiters wake only for residual mailbox types", + "stale-leaf arrivals retain Run or Dispatch pane identity", + "unowned detached-recipient mail remains directly addressable", + "duplicate Dispatch ownership is deduplicated before the 51-row branch cap" + ] + }, + { + "file": "src/main/runtime/orchestration/run-coordinator-handle-migration.test.ts", + "assertions": [ + "v27 databases backfill current coordinator handles during v28 migration", + "v28 triggers retain A-to-B-to-C coordinator history across old-runtime writes", + "message-table rebuild migration recreates atomic coordinator-mail canonicalization" + ] + }, + { + "file": "src/main/runtime/orchestration/federation-sync.test.ts", + "assertions": [ + "an acknowledged duplicate replay does not wake a waiter", + "overlapping syncs preserve one durable forward-only checkpoint" + ] + }, + { + "file": "src/main/runtime/rpc/methods/orchestration-federation.test.ts", + "assertions": [ + "a lost relay acknowledgment retries without duplicating the home message", + "a reordered relay gap converges without loss or duplication" + ] + }, + { + "file": "src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts", + "assertions": [ + "protocol v1 and v2 completion acknowledgments replay after Run-home restart", + "terminal settlement remains replayable until the worker durably acknowledges it" ] }, { @@ -8909,10 +8954,19 @@ "date": "2026-08-13", "runner": "local", "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", + "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-detached-routing.test.ts src/main/runtime/orchestration-mailbox-transport-settlement.test.ts src/main/runtime/orchestration/run-coordinator-handle-migration.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": 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." + "durationSeconds": 9.96, + "summary": "241 tests passed across mailbox identity, durable coordinator-handle migration, insertion-time canonicalization, duplicate-free 51-row ownership branch caps, direct and Dispatch pointer suppression, persisted reconciliation, 50-row paging and filtered waits, cross-PTY serialization, lifecycle fencing, bounded daemon and SSH transport settlement, outstanding Deliveries, reminted Dispatch ownership, acknowledgment, cancellation, and bounded pane lookup." + }, + { + "date": "2026-08-14", + "runner": "local", + "platform": "macos", + "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration/federation-sync.test.ts src/main/runtime/rpc/methods/orchestration-federation.test.ts src/main/runtime/rpc/methods/orchestration-federation-lifecycle-settlement.test.ts --reporter=dot", + "result": "passed", + "durationSeconds": 8.42, + "summary": "52 tests passed with real OrchestrationDb rows, a deliberately dropped federation acknowledgment, reconnect/restart, forward-only checkpoints, duplicate read-row wake suppression, and protocol v1/v2 lifecycle settlement replay. The broader final federation/cross-version set passed 77/77." }, { "date": "2026-08-14", @@ -8920,17 +8974,17 @@ "platform": "macos", "command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/orchestration-message-delivery-identity.test.ts --reporter=dot --testTimeout=5000", "result": "passed", - "durationSeconds": 5.58, - "summary": "The byte-identical two-case STA-4325 oracle failed 2/2 on reported v1.4.180 and latest origin/main 9bb8836bb6, passed 2/2 on the rebased candidate, and failed 2/2 with all candidate production changes disabled. The red path announced two direct rows while the bound check exposed only one canonical row; the second direct-only row was invisible to check before and after restart. The green path returned all three fixed message IDs in one Delivery, replayed that Delivery ID under concurrent pollers and restart, acknowledged only that batch, and woke one filtered worker_done waiter." + "durationSeconds": 10.45, + "summary": "The byte-identical five-case STA-4325 oracle (SHA-256 76a2d672d27e04e5b1db2f52df3b303468bdf006be7cfd003bf443618a856c35) failed 5/5 on reported v1.4.180, latest main, and production-reverted candidate, then passed 5/5 on the final candidate. It aligns pointer counts with one fixed Delivery across concurrent pollers and filters, replays that Delivery through restart, routes a 125-row old-handle backlog before rebind, repairs committed mail without relying on a notification callback, and matches built-CLI output to direct isolated SQLite state." }, { - "date": "2026-08-13", + "date": "2026-08-14", "runner": "local", "platform": "macos", "command": "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", "result": "passed", - "durationSeconds": 10.81, - "summary": "1,275 tests passed and 1 was skipped across Run-bound pointer delivery, PTY retirement and respawn, stale-leaf liveness, direct-mail routing, filtered waiter ownership, and orchestration RPC behavior." + "durationSeconds": 17.4, + "summary": "1,293 tests passed and 1 was skipped across Run-bound pointer delivery, PTY retirement and respawn, stale-leaf liveness, direct-mail routing, filtered waiter ownership, canonical stored-recipient notification, and orchestration RPC behavior." } ], "runtimeBudget": { @@ -8943,11 +8997,11 @@ }, "redGreenEvidence": { "status": "complete", - "evidence": "The byte-identical STA-4325 identity oracle failed 2/2 on v1.4.180 and latest origin/main 9bb8836bb6, passed 2/2 on the candidate, and failed 2/2 with the candidate production changes disabled. Added restart-upgrade, reminted-Dispatch, pointer-pinning, refusal, and bounded-backlog scenarios are candidate regressions; live topology coverage remains a separate promotion gap." + "evidence": "The byte-identical five-test oracle (SHA-256 76a2d672d27e04e5b1db2f52df3b303468bdf006be7cfd003bf443618a856c35) failed 5/5 on reported v1.4.180 commit 0b62333cf179, failed 5/5 on latest origin/main 9bb8836bb6, passed 5/5 on the final candidate, and failed 5/5 with candidate production changes reverted to 9bb8836bb6. Baselines announced two messages while check exposed one, returned no direct-only row after restart, left all 125 old-handle rows direct, failed commit-without-notify canonicalization, and made built CLI report one of two rows. The candidate aligned every count, stable message ID, fixed Delivery, acknowledgment, restart, backlog, and CLI/SQLite assertion. Live topology coverage remains a separate promotion gap." }, "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 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." + "evidence": "Coordinator inserts use one SQLite trigger in the insertion transaction; current handles are backfilled and historical A-to-B-to-C bindings are trigger-recorded. Bulk rebind is one indexed O(backlog) UPDATE. Detached routing is an ownership-driven UNION whose coordinator, exact-Dispatch, and pane-Dispatch branches each cap at 51 duplicate-free rows before the final merge; full drains measured 40/77/187/1,040 ms for 5k/10k/20k/100k rows versus the prior 96/429/1,908 ms at 5k/10k/20k. Each transaction routes at most 50 rows and yields between pages. Typed partial indexes keep filtered Run peek at 0.08-0.13 ms over 100k rows. Trigger overhead measured about 0.6-1.0 microseconds per message. SSH pointer writes use one cleared, unrefed 60-second timer per pending write, beyond the 20-second mux health window. No subprocess, provider fanout, polling, persistent listener, renderer work, or wire payload was added." }, "promotionCriteria": [ "Add isolated live restart and SSH or paired-runtime evidence.", diff --git a/src/main/providers/ssh-pty-write.test.ts b/src/main/providers/ssh-pty-write.test.ts index 7b0266fc939..e76bbafd2e0 100644 --- a/src/main/providers/ssh-pty-write.test.ts +++ b/src/main/providers/ssh-pty-write.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from 'vitest' import { SshPtyProvider } from './ssh-pty-provider' +import { SSH_PTY_WRITE_SETTLEMENT_TIMEOUT_MS } from './ssh-pty-write' describe('SSH PTY writes', () => { it('rejects writes synchronously after the transport is disposed', () => { @@ -36,4 +37,50 @@ describe('SSH PTY writes', () => { await expect(pending).resolves.toBe(false) }) + + it('disconnects a transport whose write settlement never arrives', async () => { + vi.useFakeTimers() + const mux = { + isDisposed: vi.fn().mockReturnValue(false), + notify: vi.fn(), + notifyWithSettlement: vi.fn(), + onNotification: vi.fn(), + dispose: vi.fn() + } + const provider = new SshPtyProvider('conn-1', mux as never) + const pending = provider.writeWithSettlement('ssh:conn-1@@pty-1', 'pointer') + + await vi.advanceTimersByTimeAsync(SSH_PTY_WRITE_SETTLEMENT_TIMEOUT_MS - 1) + expect(mux.dispose).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(1) + + await expect(pending).resolves.toBe(false) + expect(mux.dispose).toHaveBeenCalledWith('connection_lost') + vi.useRealTimers() + }) + + it('accepts a healthy settlement after the mux health window', async () => { + vi.useFakeTimers() + let settle: ((result: { ok: true }) => void) | undefined + const mux = { + isDisposed: vi.fn().mockReturnValue(false), + notify: vi.fn(), + notifyWithSettlement: vi.fn( + (_method: string, _params: unknown, callback: (result: { ok: true }) => void) => { + settle = callback + } + ), + onNotification: vi.fn(), + dispose: vi.fn() + } + const provider = new SshPtyProvider('conn-1', mux as never) + const pending = provider.writeWithSettlement('ssh:conn-1@@pty-1', 'pointer') + + await vi.advanceTimersByTimeAsync(SSH_PTY_WRITE_SETTLEMENT_TIMEOUT_MS - 1) + settle?.({ ok: true }) + + await expect(pending).resolves.toBe(true) + expect(mux.dispose).not.toHaveBeenCalled() + vi.useRealTimers() + }) }) diff --git a/src/main/providers/ssh-pty-write.ts b/src/main/providers/ssh-pty-write.ts index 5252109789e..80d43c610a3 100644 --- a/src/main/providers/ssh-pty-write.ts +++ b/src/main/providers/ssh-pty-write.ts @@ -1,4 +1,8 @@ import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer' +import { TIMEOUT_MS } from '../ssh/relay-protocol' + +// Allow ordinary-lane backpressure to clear well beyond the mux health window. +export const SSH_PTY_WRITE_SETTLEMENT_TIMEOUT_MS = TIMEOUT_MS * 3 export function writeToSshPty( mux: SshChannelMultiplexer, @@ -18,6 +22,20 @@ export function writeToSshPtyWithSettlement( data: string ): Promise { return new Promise((resolve) => { - mux.notifyWithSettlement('pty.data', { id: relayPtyId, data }, (result) => resolve(result.ok)) + let settled = false + const finish = (accepted: boolean): void => { + if (settled) { + return + } + settled = true + clearTimeout(timer) + resolve(accepted) + } + const timer = setTimeout(() => { + mux.dispose('connection_lost') + finish(false) + }, SSH_PTY_WRITE_SETTLEMENT_TIMEOUT_MS) + timer.unref?.() + mux.notifyWithSettlement('pty.data', { id: relayPtyId, data }, (result) => finish(result.ok)) }) } diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 3d6ab96a1df..4f15f89fe08 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -2962,6 +2962,10 @@ export class OrcaRuntimeService { pointerDelivery: this.orchestrationMailboxPointerDelivery, getDb: () => this._orchestrationDb, getLiveLeafForHandle: (handle) => this.getLiveLeafForHandle(handle).leaf, + getPaneKeyForHandle: (handle) => { + const record = this.handles.get(handle) + return record ? `${record.tabId}:${record.leafId}` : undefined + }, getMessageWaiters: (mailboxHandle) => this.messageWaitersByHandle.get(mailboxHandle), hasTerminalHandle: (handle) => this.handles.has(handle), deliverForHandle: (handle, reservedTypes) => @@ -32402,7 +32406,18 @@ export class OrcaRuntimeService { private scheduleRestoredMessageRepoints(): void { const handles = this._orchestrationDb?.getUndeliveredUnreadMailboxHandles?.() ?? [] for (const handle of handles) { - if (!handle.startsWith('dispatch:')) { + if (handle.startsWith('dispatch:')) { + continue + } + if (handle.startsWith('run:')) { + this.mailPointerRepointScheduler.schedule(handle) + continue + } + const routed = this.orchestrationMailboxOwner.routeDetachedDirectMessages(handle) + for (const mailbox of routed.mailboxes) { + this.mailPointerRepointScheduler.schedule(mailbox.mailboxHandle) + } + if (!routed.hasMore) { this.mailPointerRepointScheduler.schedule(handle) } } diff --git a/src/main/runtime/orchestration-mailbox-detached-routing.test.ts b/src/main/runtime/orchestration-mailbox-detached-routing.test.ts new file mode 100644 index 00000000000..74b4f02c7da --- /dev/null +++ b/src/main/runtime/orchestration-mailbox-detached-routing.test.ts @@ -0,0 +1,424 @@ +import { rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + checkBoundMailbox, + createBoundRun, + createDatabase, + createRuntime, + driveToLiveIdle, + insertDirectRunMessage, + LEAF_ID, + PANE_KEY, + pointerCount, + PTY_ID, + sqliteFor, + temporaryDirectories, + TERMINAL_HANDLE +} 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) } +})) + +describe('orchestration detached mailbox routing', () => { + afterEach(() => { + vi.useRealTimers() + for (const directory of temporaryDirectories.splice(0)) { + rmSync(directory, { recursive: true, force: true }) + } + }) + + it('routes active worker direct mail without injecting an unpinned Dispatch pointer', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-mailbox-dispatch-') + const harness = createRuntime(db) + const run = db.createRun({ + objective: 'Worker Run', + coordinatorHandle: 'term_coordinator', + coordinatorPaneKey: + '33333333-3333-4333-8333-333333333333:44444444-4444-4444-8444-444444444444' + }) + const task = db.createTask({ spec: 'Worker task', runId: run.id }) + const dispatch = db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY) + await driveToLiveIdle(harness.runtime) + const message = db.insertMessage({ + from: 'term_coordinator', + to: TERMINAL_HANDLE, + subject: 'Worker status', + runId: run.id + }) + + harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) + await Promise.resolve() + await vi.advanceTimersByTimeAsync(500) + const checked = await checkBoundMailbox(harness.runtime) + + expect(pointerCount(harness.write)).toBe(0) + 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 + }) + db.close() + }) + + it('normalizes direct mail that arrives while its Run is displaced', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-mailbox-late-rebind-') + const harness = createRuntime(db) + const runA = createBoundRun(db, 'Run A') + const runB = createBoundRun(db, 'Run B') + const rebound = db.bindRun({ + runId: runA.id, + coordinatorHandle: 'term_new_coordinator', + coordinatorPaneKey: + '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' + }) + const waiting = harness.runtime.waitForMessage(`run:${runA.id}`, { timeoutMs: 5_000 }) + const message = db.insertMessage({ + from: 'term_worker', + to: TERMINAL_HANDLE, + subject: 'Late Run A status', + runId: runA.id + }) + sqliteFor(db) + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(TERMINAL_HANDLE, message.id) + harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) + await Promise.resolve() + + await expect(waiting).resolves.toBe('notified') + expect(pointerCount(harness.write)).toBe(0) + expect(db.getMessageById(message.id)?.to_handle).toBe(`run:${runA.id}`) + expect(db.getCurrentRunForPane(PANE_KEY)?.id).toBe(runB.id) + + const delivery = db.getOrCreateRunDelivery({ + runId: runA.id, + consumerGeneration: rebound!.consumer_generation + }) + + expect(delivery?.messages).toEqual([expect.objectContaining({ id: message.id })]) + 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-') + const harness = createRuntime(db) + const run = createBoundRun(db, 'Working Run') + await harness.runtime.listTerminals() + harness.runtime.onPtyData(PTY_ID, '\x1b]0;Codex working\x07', 1) + const fullMailboxScan = vi.spyOn(db, 'getUndeliveredUnreadMessages') + const message = insertDirectRunMessage(db, run.id, 'Direct coordinator status') + + harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) + await Promise.resolve() + + expect(fullMailboxScan).not.toHaveBeenCalled() + expect(pointerCount(harness.write)).toBe(0) + expect(db.getMessageById(message.id)?.to_handle).toBe(`run:${run.id}`) + db.close() + }) + + it('wakes a bound Run waiter when same-Run direct mail arrives', async () => { + vi.useFakeTimers() + const db = createDatabase('orca-mailbox-waiter-') + const harness = createRuntime(db) + const run = createBoundRun(db, 'Waiting Run') + await driveToLiveIdle(harness.runtime) + + const waiting = checkBoundMailbox(harness.runtime, { wait: true }) + const internals = harness.runtime as unknown as { + messageWaitersByHandle: Map> + } + for (let attempt = 0; attempt < 20; attempt += 1) { + if (internals.messageWaitersByHandle.has(`run:${run.id}`)) { + break + } + await Promise.resolve() + } + expect(internals.messageWaitersByHandle.has(`run:${run.id}`)).toBe(true) + + const message = insertDirectRunMessage(db, run.id, 'Wake the Run waiter') + harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) + await vi.advanceTimersByTimeAsync(5_000) + + await expect(waiting).resolves.toMatchObject({ + runId: run.id, + count: 1, + messages: [expect.objectContaining({ id: message.id })] + }) + expect(pointerCount(harness.write)).toBe(0) + db.close() + }) + + it('routes a stale-leaf arrival to its bound Run waiter', async () => { + const db = createDatabase('orca-mailbox-stale-leaf-waiter-') + const harness = createRuntime(db) + const run = createBoundRun(db, 'Stale leaf waiter') + const waiting = checkBoundMailbox(harness.runtime, { wait: true }) + const waiterInternals = harness.runtime as unknown as { + messageWaitersByHandle: Map> + } + await vi.waitFor(() => { + expect(waiterInternals.messageWaitersByHandle.has(`run:${run.id}`)).toBe(true) + }) + const internals = harness.runtime as unknown as { leaves: Map } + internals.leaves.clear() + const message = insertDirectRunMessage(db, run.id, 'Arrived during graph remount') + + harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) + + await expect(waiting).resolves.toMatchObject({ + runId: run.id, + count: 1, + messages: [expect.objectContaining({ id: message.id })] + }) + expect(db.getMessageById(message.id)?.to_handle).toBe(`run:${run.id}`) + db.close() + }) + + it('keeps an unowned detached-recipient message in its direct mailbox', async () => { + const db = createDatabase('orca-mailbox-detached-direct-') + const harness = createRuntime(db) + const run = createBoundRun(db, 'Detached direct recipient') + const detachedHandle = 'term_detached_worker' + const waiting = harness.runtime.waitForMessage(detachedHandle, { timeoutMs: 5_000 }) + const message = db.insertMessage({ + from: TERMINAL_HANDLE, + to: detachedHandle, + subject: 'Instruction before Dispatch ownership exists', + type: 'dispatch', + runId: run.id, + deliveryContract: 'current_delivery' + }) + + harness.runtime.notifyMessageArrived(detachedHandle, message.type) + + await expect(waiting).resolves.toBe('notified') + expect(db.getMessageById(message.id)?.to_handle).toBe(detachedHandle) + db.close() + }) + + it('wakes canonical and direct waiters for mixed detached ownership', async () => { + const db = createDatabase('orca-mailbox-detached-mixed-') + const harness = createRuntime(db) + const detachedHandle = 'term_retired_coordinator' + const ownedRun = db.createRun({ + objective: 'Retired coordinator ownership', + coordinatorHandle: detachedHandle, + coordinatorPaneKey: + '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' + }) + const directRun = createBoundRun(db, 'Unowned detached recipient') + const canonicalWait = harness.runtime.waitForMessage(`run:${ownedRun.id}`, { + timeoutMs: 5_000 + }) + const directWait = harness.runtime.waitForMessage(detachedHandle, { timeoutMs: 5_000 }) + const owned = db.insertMessage({ + from: 'term_worker', + to: detachedHandle, + subject: 'Owned old coordinator mail', + type: 'status', + runId: ownedRun.id, + deliveryContract: 'current_delivery' + }) + sqliteFor(db) + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(detachedHandle, owned.id) + const direct = db.insertMessage({ + from: TERMINAL_HANDLE, + to: detachedHandle, + subject: 'Unowned direct mail', + type: 'status', + runId: directRun.id, + deliveryContract: 'current_delivery' + }) + + harness.runtime.notifyMessageArrived(detachedHandle, 'status') + + await expect(Promise.all([canonicalWait, directWait])).resolves.toEqual([ + 'notified', + 'notified' + ]) + expect(db.getMessageById(owned.id)?.to_handle).toBe(`run:${ownedRun.id}`) + expect(db.getMessageById(direct.id)?.to_handle).toBe(detachedHandle) + db.close() + }) + + it('does not wake a filtered direct waiter for a routed type', async () => { + const db = createDatabase('orca-mailbox-detached-filtered-direct-') + const harness = createRuntime(db) + const detachedHandle = 'term_detached_filtered' + const ownedRun = db.createRun({ + objective: 'Owned detached mail', + coordinatorHandle: detachedHandle, + coordinatorPaneKey: + '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' + }) + const unownedRun = createBoundRun(db, 'Unowned detached mail') + const waiting = harness.runtime.waitForMessage(detachedHandle, { + typeFilter: ['worker_done'], + timeoutMs: 5_000 + }) + let settled = false + void waiting.then(() => { + settled = true + }) + db.insertMessage({ + from: TERMINAL_HANDLE, + to: detachedHandle, + subject: 'Unowned status', + type: 'status', + runId: unownedRun.id, + deliveryContract: 'current_delivery' + }) + const owned = db.insertMessage({ + from: 'term_worker', + to: detachedHandle, + subject: 'Owned completion', + type: 'worker_done', + runId: ownedRun.id, + deliveryContract: 'current_delivery' + }) + sqliteFor(db) + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(detachedHandle, owned.id) + + harness.runtime.notifyMessageArrived(detachedHandle, owned.type) + await Promise.resolve() + + expect(settled).toBe(false) + expect(db.getMessageById(owned.id)?.to_handle).toBe(`run:${ownedRun.id}`) + harness.runtime.cancelMessageWaiters(detachedHandle) + await expect(waiting).resolves.toBe('cancelled') + db.close() + }) + + it('routes a stale-leaf reminted handle to its Dispatch waiter', async () => { + const db = createDatabase('orca-mailbox-stale-leaf-reminted-dispatch-') + const harness = createRuntime(db) + const run = db.createRun({ + objective: 'Reminted Dispatch', + coordinatorHandle: 'term_coordinator', + coordinatorPaneKey: + '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' + }) + const task = db.createTask({ spec: 'Reminted worker', runId: run.id }) + const dispatch = db.createDispatchContext(task.id, 'term_before_remint', PANE_KEY) + const waiting = harness.runtime.waitForMessage(`dispatch:${dispatch.id}`, { + typeFilter: ['dispatch'], + timeoutMs: 5_000 + }) + const internals = harness.runtime as unknown as { leaves: Map } + internals.leaves.clear() + const message = db.insertMessage({ + from: 'term_coordinator', + to: TERMINAL_HANDLE, + subject: 'Instruction after remint', + type: 'dispatch', + runId: run.id, + deliveryContract: 'current_delivery' + }) + + harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) + + await expect(waiting).resolves.toBe('notified') + expect(db.getMessageById(message.id)?.to_handle).toBe(`dispatch:${dispatch.id}`) + db.close() + }) + + it('deduplicates Dispatch ownership before limiting a routing page', () => { + const db = createDatabase('orca-mailbox-duplicate-dispatch-owners-') + const run = db.createRun({ + objective: 'Duplicate Dispatch ownership', + coordinatorHandle: 'term_coordinator', + coordinatorPaneKey: + '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' + }) + const task = db.createTask({ spec: 'Duplicate worker ownership', runId: run.id }) + const detachedHandle = 'term_duplicate_dispatch_owner' + const sqlite = sqliteFor(db) + const insertDispatch = sqlite.prepare( + `INSERT INTO dispatch_contexts (id, run_id, task_id, assignee_handle, status) + VALUES (?, ?, ?, ?, 'dispatched')` + ) + for (let index = 0; index < 51; index += 1) { + insertDispatch.run(`ctx_duplicate_${index}`, run.id, task.id, detachedHandle) + } + const messages = Array.from({ length: 52 }, (_, index) => + db.insertMessage({ + from: 'term_coordinator', + to: detachedHandle, + subject: `Dispatch message ${index}`, + type: 'dispatch', + runId: run.id, + deliveryContract: 'current_delivery' + }) + ) + + expect(db.routeForeignDirectMessagesToOwnedMailboxes(detachedHandle)).toMatchObject({ + routedCount: 50, + hasMore: true + }) + expect(db.routeForeignDirectMessagesToOwnedMailboxes(detachedHandle)).toMatchObject({ + routedCount: 2, + hasMore: false + }) + expect( + messages.every((message) => db.getMessageById(message.id)?.to_handle.startsWith('dispatch:')) + ).toBe(true) + db.close() + }) +}) diff --git a/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts b/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts index 5cb58a6f8fc..38f574d9491 100644 --- a/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts +++ b/src/main/runtime/orchestration-mailbox-notification-consistency.test.ts @@ -218,6 +218,65 @@ describe('orchestration notification mailbox consistency', () => { TERMINAL_HANDLE ) ).toContain('idx_messages_undelivered_direct_run') + expect( + plan( + `SELECT * FROM messages INDEXED BY idx_messages_unread_current_run_type + WHERE run_id = ? AND to_handle = ? AND read = 0 + AND delivery_contract = 'current_delivery' AND type IN (?) + ORDER BY sequence LIMIT 100`, + 'run_indexed', + 'run:run_indexed', + 'worker_done' + ) + ).toContain('idx_messages_unread_current_run_type') + db.close() + }) + + it('finds owned mail without scanning an older unowned direct backlog', () => { + const db = createDatabase('orca-mailbox-owned-routing-plan-') + const detachedHandle = 'term_backlogged_detached' + const ownedRun = db.createRun({ + objective: 'Owned after backlog', + coordinatorHandle: detachedHandle, + coordinatorPaneKey: + '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' + }) + const unownedRun = createBoundRun(db, 'Unowned backlog') + for (let index = 0; index < 500; index += 1) { + db.insertMessage({ + from: TERMINAL_HANDLE, + to: detachedHandle, + subject: `Unowned ${index}`, + type: 'status', + runId: unownedRun.id, + deliveryContract: 'current_delivery' + }) + } + const owned = db.insertMessage({ + from: 'term_worker', + to: detachedHandle, + subject: 'Owned after unowned backlog', + type: 'worker_done', + runId: ownedRun.id, + deliveryContract: 'current_delivery' + }) + sqliteFor(db) + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(detachedHandle, owned.id) + const prepare = vi.spyOn(sqliteFor(db), 'prepare') + + expect(db.routeForeignDirectMessagesToOwnedMailboxes(detachedHandle)).toMatchObject({ + routedCount: 1, + hasMore: false + }) + + const routingSql = prepare.mock.calls + .map(([sql]) => sql) + .find((sql) => sql.includes('FROM run_coordinator_handles AS coordinator')) + expect(routingSql).toContain('INDEXED BY idx_messages_undelivered_direct_run') + expect(routingSql).not.toContain('NOT EXISTS') + expect(db.getMessageById(owned.id)?.to_handle).toBe(`run:${ownedRun.id}`) + expect(db.getUnreadDirectMessageTypes(detachedHandle)).toEqual(['status']) db.close() }) @@ -603,174 +662,6 @@ describe('orchestration notification mailbox consistency', () => { db.close() }) - it('routes active worker direct mail without injecting an unpinned Dispatch pointer', async () => { - vi.useFakeTimers() - const db = createDatabase('orca-mailbox-dispatch-') - const harness = createRuntime(db) - const run = db.createRun({ - objective: 'Worker Run', - coordinatorHandle: 'term_coordinator', - coordinatorPaneKey: - '33333333-3333-4333-8333-333333333333:44444444-4444-4444-8444-444444444444' - }) - const task = db.createTask({ spec: 'Worker task', runId: run.id }) - const dispatch = db.createDispatchContext(task.id, TERMINAL_HANDLE, PANE_KEY) - await driveToLiveIdle(harness.runtime) - const message = db.insertMessage({ - from: 'term_coordinator', - to: TERMINAL_HANDLE, - subject: 'Worker status', - runId: run.id - }) - - harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) - await Promise.resolve() - await vi.advanceTimersByTimeAsync(500) - const checked = await checkBoundMailbox(harness.runtime) - - expect(pointerCount(harness.write)).toBe(0) - 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 - }) - db.close() - }) - - it('normalizes direct mail that arrives while its Run is displaced', async () => { - vi.useFakeTimers() - const db = createDatabase('orca-mailbox-late-rebind-') - const harness = createRuntime(db) - const runA = createBoundRun(db, 'Run A') - const runB = createBoundRun(db, 'Run B') - const rebound = db.bindRun({ - runId: runA.id, - coordinatorHandle: 'term_new_coordinator', - coordinatorPaneKey: - '55555555-5555-4555-8555-555555555555:66666666-6666-4666-8666-666666666666' - }) - const waiting = harness.runtime.waitForMessage(`run:${runA.id}`, { timeoutMs: 5_000 }) - const message = db.insertMessage({ - from: 'term_worker', - to: TERMINAL_HANDLE, - subject: 'Late Run A status', - runId: runA.id - }) - harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) - await Promise.resolve() - - await expect(waiting).resolves.toBe('notified') - expect(pointerCount(harness.write)).toBe(0) - expect(db.getMessageById(message.id)?.to_handle).toBe(`run:${runA.id}`) - expect(db.getCurrentRunForPane(PANE_KEY)?.id).toBe(runB.id) - - const delivery = db.getOrCreateRunDelivery({ - runId: runA.id, - consumerGeneration: rebound!.consumer_generation - }) - - expect(delivery?.messages).toEqual([expect.objectContaining({ id: message.id })]) - 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-') - const harness = createRuntime(db) - const run = createBoundRun(db, 'Working Run') - await harness.runtime.listTerminals() - harness.runtime.onPtyData(PTY_ID, '\x1b]0;Codex working\x07', 1) - const fullMailboxScan = vi.spyOn(db, 'getUndeliveredUnreadMessages') - const message = insertDirectRunMessage(db, run.id, 'Direct coordinator status') - - harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) - await Promise.resolve() - - expect(fullMailboxScan).not.toHaveBeenCalled() - expect(pointerCount(harness.write)).toBe(0) - expect(db.getMessageById(message.id)?.to_handle).toBe(`run:${run.id}`) - db.close() - }) - - it('wakes a bound Run waiter when same-Run direct mail arrives', async () => { - vi.useFakeTimers() - const db = createDatabase('orca-mailbox-waiter-') - const harness = createRuntime(db) - const run = createBoundRun(db, 'Waiting Run') - await driveToLiveIdle(harness.runtime) - - const waiting = checkBoundMailbox(harness.runtime, { wait: true }) - const internals = harness.runtime as unknown as { - messageWaitersByHandle: Map> - } - for (let attempt = 0; attempt < 20; attempt += 1) { - if (internals.messageWaitersByHandle.has(`run:${run.id}`)) { - break - } - await Promise.resolve() - } - expect(internals.messageWaitersByHandle.has(`run:${run.id}`)).toBe(true) - - const message = insertDirectRunMessage(db, run.id, 'Wake the Run waiter') - harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, message.type) - await vi.advanceTimersByTimeAsync(5_000) - - await expect(waiting).resolves.toMatchObject({ - runId: run.id, - count: 1, - messages: [expect.objectContaining({ id: message.id })] - }) - expect(pointerCount(harness.write)).toBe(0) - db.close() - }) - it('drains persisted Run pages before installing a filtered waiter', async () => { const db = createDatabase('orca-mailbox-filtered-run-backlog-') const harness = createRuntime(db) @@ -785,6 +676,9 @@ describe('orchestration notification mailbox consistency', () => { type: 'question', runId: run.id }) + sqliteFor(db) + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(TERMINAL_HANDLE, question.id) const checked = await checkBoundMailbox(harness.runtime, { wait: true, @@ -825,6 +719,9 @@ describe('orchestration notification mailbox consistency', () => { type: 'question', runId: run.id }) + sqliteFor(db) + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(TERMINAL_HANDLE, question.id) const arrivingStatus = insertDirectRunMessage(db, run.id, 'Status arrival trigger') harness.runtime.notifyMessageArrived(TERMINAL_HANDLE, arrivingStatus.type) diff --git a/src/main/runtime/orchestration-mailbox-notification-test-harness.ts b/src/main/runtime/orchestration-mailbox-notification-test-harness.ts index cc476f1f278..97426993a53 100644 --- a/src/main/runtime/orchestration-mailbox-notification-test-harness.ts +++ b/src/main/runtime/orchestration-mailbox-notification-test-harness.ts @@ -43,7 +43,7 @@ export function createBoundRun(db: OrchestrationDb, objective: string) { } export function insertDirectRunMessage(db: OrchestrationDb, runId: string, subject: string) { - return db.insertMessage({ + const message = db.insertMessage({ from: 'term_worker', to: TERMINAL_HANDLE, subject, @@ -51,6 +51,10 @@ export function insertDirectRunMessage(db: OrchestrationDb, runId: string, subje runId, deliveryContract: 'current_delivery' }) + sqliteFor(db) + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(TERMINAL_HANDLE, message.id) + return db.getMessageById(message.id)! } export type MailboxNotificationHarness = { diff --git a/src/main/runtime/orchestration-message-delivery-identity.test.ts b/src/main/runtime/orchestration-message-delivery-identity.test.ts index cb96071addc..c9fba128cf2 100644 --- a/src/main/runtime/orchestration-message-delivery-identity.test.ts +++ b/src/main/runtime/orchestration-message-delivery-identity.test.ts @@ -22,6 +22,7 @@ const LEAF_ID = '22222222-2222-4222-8222-222222222222' const PANE_KEY = `${TAB_ID}:${LEAF_ID}` const PTY_ID = 'pty-sta-4325' const TERMINAL_HANDLE = 'term_sta_4325' +const REMINTED_TERMINAL_HANDLE = 'term_sta_4325_reminted' const WORKTREE_ID = 'repo-sta-4325::/tmp/sta-4325' const LAUNCH_TOKEN = 'sta-4325-launch' const temporaryDirectories: string[] = [] @@ -40,6 +41,7 @@ type CheckResult = { type Sqlite = { prepare: (sql: string) => { all: (...params: unknown[]) => unknown[] + run: (...params: unknown[]) => unknown } } @@ -54,7 +56,10 @@ function createDatabase(prefix: string): { db: OrchestrationDb; path: string } { return { db: new OrchestrationDb(path), path } } -function createRuntime(db: OrchestrationDb): { +function createRuntime( + db: OrchestrationDb, + terminalHandle = TERMINAL_HANDLE +): { runtime: OrcaRuntimeService write: ReturnType } { @@ -71,7 +76,7 @@ function createRuntime(db: OrchestrationDb): { incarnationId: 'sta-4325-incarnation', agentLaunchAuthority: { launchToken: LAUNCH_TOKEN, launchAgent: 'codex' } }) - runtime.registerPreAllocatedHandleForPty(PTY_ID, TERMINAL_HANDLE) + runtime.registerPreAllocatedHandleForPty(PTY_ID, terminalHandle) runtime.attachWindow(1) runtime.syncWindowGraph(1, { tabs: [ @@ -107,17 +112,18 @@ async function check( runtime: OrcaRuntimeService, params: Record = {} ): Promise { + const terminal = typeof params.terminal === 'string' ? params.terminal : TERMINAL_HANDLE const response = await new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }).dispatch({ id: `req-sta-4325-${Math.random()}`, authToken: 'test-auth-token', method: 'orchestration.check', orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, orchestrationCompatibilityEvidence: { - terminalHandle: TERMINAL_HANDLE, + terminalHandle: terminal, paneKey: PANE_KEY, launchToken: LAUNCH_TOKEN }, - params: { terminal: TERMINAL_HANDLE, ...params } + params: { terminal, ...params } }) expect(response.ok).toBe(true) if (!response.ok) { @@ -277,17 +283,25 @@ describe('STA-4325 message and delivery identity', () => { fixture.db.close() const reopened = new OrchestrationDb(fixture.path) - const restarted = createRuntime(reopened) + const restarted = createRuntime(reopened, REMINTED_TERMINAL_HANDLE) await driveToLiveIdle(restarted.runtime) - const afterRestart = await check(restarted.runtime) + const afterRestart = await check(restarted.runtime, { terminal: REMINTED_TERMINAL_HANDLE }) expect(afterRestart).toMatchObject({ deliveryId: beforeRestart.deliveryId, replayed: true }) expect(afterRestart.messages.map((message) => message.id)).toEqual([status.id]) expect(sqliteFor(reopened).prepare('SELECT id FROM deliveries').all()).toEqual([ { id: beforeRestart.deliveryId } ]) - await check(restarted.runtime, { ack: afterRestart.deliveryId }) - const waiting = check(restarted.runtime, { wait: true, types: 'worker_done', timeoutMs: 5_000 }) + await check(restarted.runtime, { + terminal: REMINTED_TERMINAL_HANDLE, + ack: afterRestart.deliveryId + }) + const waiting = check(restarted.runtime, { + terminal: REMINTED_TERMINAL_HANDLE, + wait: true, + types: 'worker_done', + timeoutMs: 5_000 + }) const internals = restarted.runtime as unknown as { messageWaitersByHandle: Map> } @@ -307,7 +321,7 @@ describe('STA-4325 message and delivery identity', () => { runId: run.id, deliveryContract: 'current_delivery' }) - restarted.runtime.notifyMessageArrived(TERMINAL_HANDLE, done.type) + restarted.runtime.notifyMessageArrived(done.to_handle, done.type) await vi.advanceTimersByTimeAsync(5_000) await expect(waiting).resolves.toMatchObject({ @@ -319,6 +333,101 @@ describe('STA-4325 message and delivery identity', () => { reopened.close() }) + it('routes the complete direct backlog before rebinding forgets its old handle', () => { + const fixture = createDatabase('orca-sta-4325-rebind-backlog-') + const first = fixture.db.createRun({ + objective: 'Old coordinator', + coordinatorHandle: TERMINAL_HANDLE, + coordinatorPaneKey: PANE_KEY + }) + for (let index = 0; index < 125; index += 1) { + const message = fixture.db.insertMessage({ + from: 'term_worker', + to: TERMINAL_HANDLE, + subject: `backlog ${index}`, + type: 'status', + runId: first.id, + deliveryContract: 'current_delivery' + }) + sqliteFor(fixture.db) + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run(TERMINAL_HANDLE, message.id) + } + const plan = sqliteFor(fixture.db) + .prepare( + `EXPLAIN QUERY PLAN UPDATE messages SET to_handle = ? + WHERE run_id = ? AND to_handle = ? AND read = 0 + AND delivery_contract = 'current_delivery'` + ) + .all(`run:${first.id}`, first.id, TERMINAL_HANDLE) as { detail: string }[] + expect(plan.map((row) => row.detail).join(' ')).toMatch( + /SEARCH messages USING INDEX (idx_messages_delivery_contract|idx_messages_unread_current_inbox)/ + ) + fixture.db.createRun({ + objective: 'Replacement coordinator', + coordinatorHandle: TERMINAL_HANDLE, + coordinatorPaneKey: PANE_KEY + }) + expect(fixture.db.getRun(first.id)?.coordinator_handle).toBeNull() + fixture.db.close() + + const reopened = new OrchestrationDb(fixture.path) + expect( + sqliteFor(reopened) + .prepare( + `SELECT to_handle, COUNT(*) AS count FROM messages + WHERE run_id = ? GROUP BY to_handle ORDER BY to_handle` + ) + .all(first.id) + ).toEqual([{ to_handle: `run:${first.id}`, count: 125 }]) + reopened.close() + }) + + it('repairs committed mail sent to a forgotten handle after restart', async () => { + const fixture = createDatabase('orca-sta-4325-late-old-handle-') + const run = fixture.db.createRun({ + objective: 'Late old-handle arrival', + coordinatorHandle: TERMINAL_HANDLE, + coordinatorPaneKey: PANE_KEY + }) + fixture.db.bindRun({ + runId: run.id, + coordinatorHandle: REMINTED_TERMINAL_HANDLE, + coordinatorPaneKey: PANE_KEY + }) + const done = fixture.db.insertMessage({ + from: 'term_worker', + to: TERMINAL_HANDLE, + subject: 'arrived after rebind', + type: 'worker_done', + runId: run.id, + deliveryContract: 'current_delivery' + }) + expect(fixture.db.getMessageById(done.id)?.to_handle).toBe(`run:${run.id}`) + expect( + fixture.db + .getOrCreateRunDelivery({ + runId: run.id, + consumerGeneration: fixture.db.getRun(run.id)!.consumer_generation + }) + ?.messages.map((message) => message.id) + ).toEqual([done.id]) + fixture.db.close() + + const reopened = new OrchestrationDb(fixture.path) + const restarted = createRuntime(reopened, REMINTED_TERMINAL_HANDLE) + await driveToLiveIdle(restarted.runtime) + const checked = await check(restarted.runtime, { terminal: REMINTED_TERMINAL_HANDLE }) + + expect(checked).toMatchObject({ + runId: run.id, + count: 1, + messages: [expect.objectContaining({ id: done.id })] + }) + expect(reopened.getMessageById(done.id)?.to_handle).toBe(`run:${run.id}`) + reopened.close() + }) + it('keeps the built CLI count and Delivery output aligned with isolated SQLite state', async () => { expect(existsSync(CLI_PATH)).toBe(true) const userDataPath = mkdtempSync(join(tmpdir(), 'orca-sta-4325-cli-')) diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index aab8fc6614d..97df9635b89 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -310,8 +310,8 @@ type RunListCursor = { id: string } -// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments. -const SCHEMA_VERSION = 27 +// Schema versions: v2 'heartbeat'+last_heartbeat_at, v3 delivered_at, v4 task-creator terminal, v5 task_title/display_name, v6 pane identity, v7 lightweight Runs, v8 crash-safe Run deliveries, v9 durable question threads, v10 Dispatch capabilities, v11 durable mutation receipts, v12 composed worker state, v18 post-v6 version-skew repair, v19 adopted legacy Runs and compatibility receipts, v20 legacy question backfill, v21 legacy scheduler-loss provenance, v22 dispatch assignee lookup, v23 worker terminal resource ownership, v24 creator-incarnation authority, v25 active Dispatch handle lookup, v26 indexed mutation receipt capacity, v27 durable federation acknowledgments, v28 Run coordinator handle history. +const SCHEMA_VERSION = 28 function hardenOrchestrationDatabaseFiles(dbPath: (string & {}) | ':memory:'): void { if (dbPath === ':memory:' || process.platform === 'win32') { @@ -343,6 +343,8 @@ export class OrchestrationDb { this.db.pragma('busy_timeout = 5000') this.createTables() this.migrate() + this.createCoordinatorMailRoutingTrigger() + this.rememberCurrentRunCoordinatorHandles() hardenOrchestrationDatabaseFiles(dbPath) } @@ -389,6 +391,38 @@ export class OrchestrationDb { CREATE INDEX IF NOT EXISTS idx_inbox ON messages(to_handle, read); CREATE INDEX IF NOT EXISTS idx_thread ON messages(thread_id); + CREATE TABLE IF NOT EXISTS run_coordinator_handles ( + run_id TEXT NOT NULL, + terminal_handle TEXT NOT NULL, + first_bound_at TEXT NOT NULL DEFAULT (datetime('now')), + PRIMARY KEY (run_id, terminal_handle) + ); + + CREATE INDEX IF NOT EXISTS idx_run_coordinator_handles_handle + ON run_coordinator_handles(terminal_handle, run_id); + + CREATE TRIGGER IF NOT EXISTS trg_runs_remember_coordinator_insert + AFTER INSERT ON runs + WHEN NEW.legacy = 0 AND NEW.coordinator_handle IS NOT NULL + BEGIN + INSERT OR IGNORE INTO run_coordinator_handles (run_id, terminal_handle) + VALUES (NEW.id, NEW.coordinator_handle); + END; + + CREATE TRIGGER IF NOT EXISTS trg_runs_remember_coordinator_update + AFTER UPDATE OF coordinator_handle ON runs + WHEN NEW.legacy = 0 AND NEW.coordinator_handle IS NOT NULL + BEGIN + INSERT OR IGNORE INTO run_coordinator_handles (run_id, terminal_handle) + VALUES (NEW.id, NEW.coordinator_handle); + END; + + CREATE TRIGGER IF NOT EXISTS trg_runs_forget_coordinator_handles + AFTER DELETE ON runs + BEGIN + DELETE FROM run_coordinator_handles WHERE run_id = OLD.id; + END; + CREATE TABLE IF NOT EXISTS deliveries ( id TEXT PRIMARY KEY, run_id TEXT NOT NULL, @@ -1031,6 +1065,13 @@ export class OrchestrationDb { 'ALTER TABLE federated_dispatches ADD COLUMN to_home_acknowledged_sequence INTEGER NOT NULL DEFAULT 0' ) } + if (current < 28) { + this.db.exec(` + INSERT OR IGNORE INTO run_coordinator_handles (run_id, terminal_handle) + SELECT id, coordinator_handle FROM runs + WHERE legacy = 0 AND coordinator_handle IS NOT NULL + `) + } this.db.exec(` CREATE INDEX IF NOT EXISTS idx_dispatch_assignee_pane_leaf ON dispatch_contexts(${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL}) @@ -1452,6 +1493,12 @@ export class OrchestrationDb { CREATE INDEX IF NOT EXISTS idx_messages_unread_current_inbox ON messages(to_handle, sequence) WHERE read = 0 AND delivery_contract = 'current_delivery'; + CREATE INDEX IF NOT EXISTS idx_messages_unread_current_inbox_type + ON messages(to_handle, type, sequence) + WHERE read = 0 AND delivery_contract = 'current_delivery'; + CREATE INDEX IF NOT EXISTS idx_messages_unread_current_run_type + ON messages(run_id, to_handle, type, sequence) + WHERE read = 0 AND delivery_contract = 'current_delivery'; `) } @@ -2296,6 +2343,7 @@ export class OrchestrationDb { ) VALUES (?, ?, ?, ?, 1, 0)` ) .run(id, params.objective, params.coordinatorHandle, params.coordinatorPaneKey) + this.rememberRunCoordinatorHandle(id, params.coordinatorHandle) this.db.exec('COMMIT') } catch (error) { this.db.exec('ROLLBACK') @@ -2417,7 +2465,8 @@ export class OrchestrationDb { Boolean(value) ) )) { - this.routeUnreadDirectMessagesToRunMailbox(params.runId, handle) + this.rememberRunCoordinatorHandle(params.runId, handle) + this.routeAllUnreadDirectMessagesToRunMailbox(params.runId, handle) } if ( (params.takeoverLegacy && !takeoverAlreadyApplied) || @@ -2526,7 +2575,7 @@ export class OrchestrationDb { for (const run of this.runsBoundToPane(paneKey)) { if (run.id !== exceptRunId) { if (run.coordinator_handle) { - this.routeUnreadDirectMessagesToRunMailbox(run.id, run.coordinator_handle) + this.routeAllUnreadDirectMessagesToRunMailbox(run.id, run.coordinator_handle) } this.db .prepare( @@ -2548,6 +2597,47 @@ export class OrchestrationDb { } } + private rememberRunCoordinatorHandle(runId: string, terminalHandle: string): void { + this.db + .prepare( + `INSERT OR IGNORE INTO run_coordinator_handles (run_id, terminal_handle) VALUES (?, ?)` + ) + .run(runId, terminalHandle) + } + + private rememberCurrentRunCoordinatorHandles(): void { + this.db.exec(` + INSERT OR IGNORE INTO run_coordinator_handles (run_id, terminal_handle) + SELECT id, coordinator_handle FROM runs + WHERE legacy = 0 AND coordinator_handle IS NOT NULL + `) + } + + private createCoordinatorMailRoutingTrigger(): void { + this.db.exec(` + DROP TRIGGER IF EXISTS trg_messages_route_coordinator_mail; + CREATE TRIGGER trg_messages_route_coordinator_mail + AFTER INSERT ON messages + WHEN NEW.read = 0 AND NEW.delivery_contract = 'current_delivery' + AND EXISTS ( + SELECT 1 FROM runs + WHERE runs.id = NEW.run_id AND runs.legacy = 0 + ) + AND EXISTS ( + SELECT 1 FROM run_coordinator_handles + WHERE run_id = NEW.run_id AND terminal_handle = NEW.to_handle + ) + AND NOT EXISTS ( + SELECT 1 FROM dispatch_contexts + WHERE run_id = NEW.run_id AND assignee_handle = NEW.to_handle + AND status IN ('pending', 'dispatched') + ) + BEGIN + UPDATE messages SET to_handle = 'run:' || NEW.run_id WHERE sequence = NEW.sequence; + END; + `) + } + private fenceOutstandingDelivery(runId: string): void { this.db .prepare( @@ -2893,6 +2983,31 @@ export class OrchestrationDb { ) } + getUnreadRunMailbox(runId: string, limit = 100, types?: MessageType[]): MessageRow[] { + const address = `run:${runId}` + const conditions = [ + 'run_id = ?', + 'to_handle = ?', + 'read = 0', + "delivery_contract = 'current_delivery'" + ] + const params: (string | number)[] = [runId, address] + if (types?.length) { + conditions.push(`type IN (${types.map(() => '?').join(',')})`) + params.push(...types) + } + const indexClause = types?.length ? ' INDEXED BY idx_messages_unread_current_run_type' : '' + params.push(Math.max(1, Math.floor(limit))) + return exposeMessageListTimestamps( + this.db + .prepare( + `SELECT * FROM messages${indexClause} WHERE ${conditions.join(' AND ')} + ORDER BY sequence ASC LIMIT ?` + ) + .all(...params) as MessageRow[] + ) + } + // ── Messages ── insertMessage(msg: { @@ -3645,36 +3760,58 @@ export class OrchestrationDb { directHandle: string, throughSequence?: number ): MailboxRoutingPage { - const throughClause = throughSequence === undefined ? '' : ' AND sequence <= ?' - const params: (string | number)[] = [runId, directHandle] - if (throughSequence !== undefined) { - params.push(throughSequence) + this.db.exec('BEGIN IMMEDIATE') + try { + const throughClause = throughSequence === undefined ? '' : ' AND sequence <= ?' + const params: (string | number)[] = [runId, directHandle] + if (throughSequence !== undefined) { + params.push(throughSequence) + } + params.push(ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1) + const rows = this.db + .prepare( + `SELECT id, type FROM messages + WHERE run_id = ? AND to_handle = ? AND read = 0 + AND delivery_contract = 'current_delivery'${throughClause} + ORDER BY sequence LIMIT ?` + ) + .all(...params) as { + id: string + type: MessageType + }[] + const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT) + if (page.length === 0) { + this.db.exec('COMMIT') + return { routedCount: 0, hasMore: false, types: [] } + } + const placeholders = page.map(() => '?').join(',') + const result = this.db + .prepare( + `UPDATE messages SET to_handle = ? + WHERE run_id = ? AND to_handle = ? AND read = 0 + AND delivery_contract = 'current_delivery' AND id IN (${placeholders})` + ) + .run(mailboxHandle, runId, directHandle, ...page.map((row) => row.id)) + this.db.exec('COMMIT') + return { + routedCount: Number(result.changes), + hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT, + types: [...new Set(page.map((row) => row.type))] + } + } catch (error) { + this.db.exec('ROLLBACK') + throw error } - params.push(ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1) - const rows = this.db + } + + private routeAllUnreadDirectMessagesToRunMailbox(runId: string, directHandle: string): void { + this.db .prepare( - `SELECT id, type FROM messages + `UPDATE messages SET to_handle = ? WHERE run_id = ? AND to_handle = ? AND read = 0 - AND delivery_contract = 'current_delivery'${throughClause} - ORDER BY sequence LIMIT ?` + AND delivery_contract = 'current_delivery'` ) - .all(...params) as { - id: string - type: MessageType - }[] - const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT) - if (page.length === 0) { - return { routedCount: 0, hasMore: false, types: [] } - } - 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, - types: [...new Set(page.map((row) => row.type))] - } + .run(`run:${runId}`, runId, directHandle) } routeUnreadDirectMessagesToRunMailbox( @@ -3705,34 +3842,48 @@ export class OrchestrationDb { 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'${throughClause} - ORDER BY sequence LIMIT ?` - ) - .all(...params) as { - id: string - type: MessageType - }[] - const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT) - if (page.length === 0) { - return { routedCount: 0, hasMore: false, types: [] } - } - const placeholders = page.map(() => '?').join(',') - const result = this.db - .prepare(`UPDATE messages SET to_handle = ? WHERE id IN (${placeholders})`) - .run(`run:${runId}`, ...page.map((row) => row.id)) - return { - routedCount: Number(result.changes), - hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT, - types: [...new Set(page.map((row) => row.type))] + this.db.exec('BEGIN IMMEDIATE') + try { + const throughClause = throughSequence === undefined ? '' : ' AND sequence <= ?' + const params: (string | number)[] = [runId, 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 run_id = ? AND to_handle = ? AND read = 0 + AND delivery_contract = 'current_delivery'${throughClause} + ORDER BY sequence LIMIT ?` + ) + .all(...params) as { + id: string + type: MessageType + }[] + const page = rows.slice(0, ORCHESTRATION_DELIVERY_BATCH_LIMIT) + if (page.length === 0) { + this.db.exec('COMMIT') + return { routedCount: 0, hasMore: false, types: [] } + } + const placeholders = page.map(() => '?').join(',') + const result = this.db + .prepare( + `UPDATE messages SET to_handle = ? + WHERE run_id = ? AND to_handle = ? AND read = 0 + AND delivery_contract = 'current_delivery' + AND id IN (${placeholders})` + ) + .run(`run:${runId}`, runId, dispatchMailbox, ...page.map((row) => row.id)) + this.db.exec('COMMIT') + return { + routedCount: Number(result.changes), + hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT, + types: [...new Set(page.map((row) => row.type))] + } + } catch (error) { + this.db.exec('ROLLBACK') + throw error } } @@ -3775,68 +3926,146 @@ export class OrchestrationDb { routeForeignDirectMessagesToOwnedMailboxes( directHandle: string, - currentRunId: 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) - ORDER BY sequence LIMIT ?` - ) - .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, types: [], 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) - byMailbox.set(mailboxHandle, types) - } this.db.exec('BEGIN IMMEDIATE') try { + const runExclusion = currentRunId === undefined ? '' : ' AND candidate.run_id <> ?' + const paneSuffix = paneKey && parsePaneKey(paneKey) ? paneKeyMatchSuffix(paneKey) : undefined + const exclusionParams = currentRunId === undefined ? [] : [currentRunId] + const branches = [ + `SELECT candidate.id, candidate.run_id, candidate.type, candidate.sequence + FROM run_coordinator_handles AS coordinator + JOIN runs AS owner_run ON owner_run.id = coordinator.run_id AND owner_run.legacy = 0 + JOIN messages AS candidate INDEXED BY idx_messages_undelivered_direct_run + ON candidate.run_id = coordinator.run_id + AND candidate.to_handle = coordinator.terminal_handle + WHERE coordinator.terminal_handle = ?${runExclusion} + AND candidate.read = 0 AND candidate.delivered_at IS NULL + AND candidate.delivery_contract = 'current_delivery'`, + `SELECT candidate.id, candidate.run_id, candidate.type, candidate.sequence + FROM ( + SELECT direct_dispatch.run_id + FROM dispatch_contexts AS direct_dispatch + INDEXED BY idx_dispatch_active_assignee_handle + JOIN runs AS owner_run + ON owner_run.id = direct_dispatch.run_id AND owner_run.legacy = 0 + WHERE direct_dispatch.assignee_handle = ? + AND direct_dispatch.status IN ('pending', 'dispatched') + GROUP BY direct_dispatch.run_id + ) AS dispatch_owner + JOIN messages AS candidate INDEXED BY idx_messages_undelivered_direct_run + ON candidate.run_id = dispatch_owner.run_id AND candidate.to_handle = ? + WHERE 1 = 1${runExclusion} + AND candidate.read = 0 AND candidate.delivered_at IS NULL + AND candidate.delivery_contract = 'current_delivery'` + ] + const branchParams: (string | number)[][] = [ + [directHandle, ...exclusionParams, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1], + [directHandle, directHandle, ...exclusionParams, ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1] + ] + if (paneSuffix !== undefined) { + branches.push( + `SELECT candidate.id, candidate.run_id, candidate.type, candidate.sequence + FROM ( + SELECT pane_dispatch.run_id + FROM dispatch_contexts AS pane_dispatch INDEXED BY idx_dispatch_assignee_pane_leaf + JOIN runs AS owner_run + ON owner_run.id = pane_dispatch.run_id AND owner_run.legacy = 0 + WHERE pane_dispatch.assignee_pane_key IS NOT NULL + AND pane_dispatch.status IN ('pending', 'dispatched') + AND instr(pane_dispatch.assignee_pane_key, ':') > 1 + AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ? + GROUP BY pane_dispatch.run_id + ) AS pane_owner + JOIN messages AS candidate INDEXED BY idx_messages_undelivered_direct_run + ON candidate.run_id = pane_owner.run_id AND candidate.to_handle = ? + WHERE 1 = 1${runExclusion} + AND candidate.read = 0 AND candidate.delivered_at IS NULL + AND candidate.delivery_contract = 'current_delivery'` + ) + branchParams.push([ + paneSuffix, + directHandle, + ...exclusionParams, + ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1 + ]) + } + const limitedBranches = branches.map( + (branch) => `SELECT * FROM (${branch} ORDER BY candidate.sequence LIMIT ?)` + ) + const rows = this.db + .prepare( + `SELECT id, run_id, type FROM (${limitedBranches.join(' UNION ')}) + ORDER BY sequence LIMIT ?` + ) + .all(...branchParams.flat(), 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) { + this.db.exec('COMMIT') + return { routedCount: 0, hasMore: false, types: [], 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) + byMailbox.set(mailboxHandle, types) + } 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) + .prepare( + `UPDATE messages SET to_handle = ? + WHERE to_handle = ? AND read = 0 AND delivered_at IS NULL + AND delivery_contract = 'current_delivery' AND id IN (${placeholders})` + ) + .run(mailboxHandle, directHandle, ...ids) } this.db.exec('COMMIT') + return { + routedCount: page.length, + hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT, + types: [...new Set(page.map((row) => row.type))], + mailboxes: [...byMailbox].map(([mailboxHandle, types]) => ({ + mailboxHandle, + types: [...types] + })) + } } catch (error) { this.db.exec('ROLLBACK') throw error } - return { - routedCount: page.length, - hasMore: rows.length > ORCHESTRATION_DELIVERY_BATCH_LIMIT, - types: [...new Set(page.map((row) => row.type))], - mailboxes: [...byMailbox].map(([mailboxHandle, types]) => ({ - mailboxHandle, - types: [...types] - })) - } + } + + getUnreadDirectMessageTypes(directHandle: string): MessageType[] { + return ( + this.db + .prepare( + `SELECT DISTINCT type FROM messages INDEXED BY idx_messages_unread_current_inbox_type + WHERE to_handle = ? AND read = 0 AND delivery_contract = 'current_delivery' + ORDER BY type` + ) + .all(directHandle) as { type: MessageType }[] + ).map((row) => row.type) } areUndeliveredUnreadMessages(toHandle: string, ids: string[]): boolean { @@ -7433,6 +7662,7 @@ export class OrchestrationDb { DELETE FROM dispatch_contexts; DELETE FROM tasks; DELETE FROM messages; + DELETE FROM run_coordinator_handles; DELETE FROM runs; INSERT INTO runs (id, objective, home_database, consumer_generation, legacy) VALUES ('${LEGACY_RUN_ID}', 'Legacy orchestration state (inspect only)', 'this_database', 0, 1); diff --git a/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts b/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts index 80959d3df48..f216838b261 100644 --- a/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts +++ b/src/main/runtime/orchestration/federation-acknowledgment-migration.test.ts @@ -27,7 +27,7 @@ describe('federation acknowledgment migration', () => { const oldDb = new Database(dbPath) oldDb.exec('ALTER TABLE federated_dispatches DROP COLUMN to_home_acknowledged_sequence') oldDb.pragma('user_version = 26') - expect(resolveOrchestrationMigrationStartVersion(oldDb, 26, 27)).toBe(26) + expect(resolveOrchestrationMigrationStartVersion(oldDb, 26, 28)).toBe(26) oldDb .prepare( `INSERT INTO federated_dispatches ( @@ -41,7 +41,7 @@ describe('federation acknowledgment migration', () => { db = new OrchestrationDb(dbPath) const sqlite = (db as unknown as { db: Database.Database }).db - expect(sqlite.pragma('user_version', { simple: true })).toBe(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) expect(db.getFederatedDispatch('ctx_migrated')).toMatchObject({ to_home_imported_sequence: 2, to_home_acknowledged_sequence: 0 diff --git a/src/main/runtime/orchestration/federation-sync.test.ts b/src/main/runtime/orchestration/federation-sync.test.ts index af4d2782b6f..76e14ac3200 100644 --- a/src/main/runtime/orchestration/federation-sync.test.ts +++ b/src/main/runtime/orchestration/federation-sync.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from 'vitest' import { OrcaRuntimeService } from '../orca-runtime' +import { OrchestrationDb } from './db' import { acquireFederationAckLease, clearFederationAckCheckpoints, @@ -118,6 +119,96 @@ describe('federation relay parsing', () => { }) describe('federation relay acknowledgments', () => { + it('does not wake a waiter for an acknowledged duplicate replay', async () => { + const db = new OrchestrationDb(':memory:') + const run = db.createRun({ + objective: 'Federation replay wake', + coordinatorHandle: 'term_coordinator', + coordinatorPaneKey: 'tab_coordinator:11111111-1111-4111-8111-111111111111' + }) + const task = db.createTask({ spec: 'Remote work', runId: run.id }) + const { dispatch } = db.createStartingWorkerDispatch({ + taskId: task.id, + startOptions: {}, + federation: { + environmentId: 'environment_windows', + environmentName: 'windows', + peerFingerprint: 'windows_peer_fingerprint', + protocolVersion: 3 + } + }) + db.recordWorkerStage({ dispatchId: dispatch.id, stage: 'ready', state: 'ready' }) + const relayItem = (sequence: number) => ({ + dispatch_id: dispatch.id, + direction: 'to_home' as const, + sequence, + message_id: `msg_federated_${sequence}`, + kind: 'status', + payload: JSON.stringify({ + subject: `Remote status ${sequence}`, + body: `Update ${sequence}`, + type: 'status' + }) + }) + let pulled = [relayItem(1)] + let rejectAck = true + const runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb(db) + vi.spyOn(runtime, 'resolveOrchestrationWorkerServer').mockReturnValue({ + peerFingerprint: 'windows_peer_fingerprint' + } as never) + vi.spyOn(runtime, 'callOrchestrationWorkerServer').mockImplementation( + async (_environmentId, method) => { + if (method === 'orchestration.federationPull') { + return { runtimeEpoch: 'remote_epoch_1', items: pulled } + } + if (method === 'orchestration.federationAck') { + if (rejectAck) { + rejectAck = false + throw new Error('ack response lost before remote mutation') + } + return { acknowledgedThrough: pulled.at(-1)?.sequence ?? 0 } + } + throw new Error(`Unexpected method ${method}`) + } + ) + + await expect(syncFederatedDispatch(runtime, dispatch.id)).rejects.toThrow( + 'ack response lost before remote mutation' + ) + const first = db.getOrCreateRunDelivery({ + runId: run.id, + consumerGeneration: run.consumer_generation + }) + expect(first?.messages.map((message) => message.id)).toEqual(['msg_federated_1']) + db.acknowledgeRunDelivery({ + runId: run.id, + consumerGeneration: run.consumer_generation, + deliveryId: first!.delivery.id + }) + const waiting = runtime.waitForMessage(`run:${run.id}`, { + typeFilter: ['status'], + timeoutMs: 5_000 + }) + let settled = false + void waiting.then(() => { + settled = true + }) + + await syncFederatedDispatch(runtime, dispatch.id) + await Promise.resolve() + expect(settled).toBe(false) + expect(db.getMessageById('msg_federated_1')?.read).toBe(1) + + pulled = [relayItem(2)] + await syncFederatedDispatch(runtime, dispatch.id) + await expect(waiting).resolves.toBe('notified') + expect(db.getUnreadMessages(`run:${run.id}`).map((message) => message.id)).toEqual([ + 'msg_federated_2' + ]) + db.close() + }) + it('drains a terminal retry from the page after the first terminal report', async () => { const pending = Array.from({ length: 51 }, (_, index) => { const sequence = index + 1 diff --git a/src/main/runtime/orchestration/federation-sync.ts b/src/main/runtime/orchestration/federation-sync.ts index f73c3757582..3bc7f0f5180 100644 --- a/src/main/runtime/orchestration/federation-sync.ts +++ b/src/main/runtime/orchestration/federation-sync.ts @@ -129,7 +129,9 @@ async function syncFederatedDispatchPages( }) } cursor = item.sequence - runtime.notifyMessageArrived(stored.message.to_handle, stored.message.type) + if (stored.message.read === 0) { + runtime.notifyMessageArrived(stored.message.to_handle, stored.message.type) + } imported += stored.duplicate ? 0 : 1 } diff --git a/src/main/runtime/orchestration/mailbox-notification-coordinator.ts b/src/main/runtime/orchestration/mailbox-notification-coordinator.ts index 0e6a8711b70..4856a08843b 100644 --- a/src/main/runtime/orchestration/mailbox-notification-coordinator.ts +++ b/src/main/runtime/orchestration/mailbox-notification-coordinator.ts @@ -14,6 +14,7 @@ type NotificationCoordinatorDependencies getDb: () => OrchestrationDb | null getLiveLeafForHandle: (handle: string) => OrchestrationMailboxLeaf + getPaneKeyForHandle: (handle: string) => string | undefined getMessageWaiters: (mailboxHandle: string) => ReadonlySet | undefined hasTerminalHandle: (handle: string) => boolean deliverForHandle: (handle: string, reservedTypes?: ReadonlySet) => void @@ -39,15 +40,19 @@ export class OrchestrationMailboxNotificationCoordinator< } notifyMessageArrived(handle: string, messageType?: string): void { - const { mailboxHandle, forwarded } = this.resolveArrivedMailboxes(handle) + const { mailboxHandle, forwarded, directTypes } = this.resolveArrivedMailboxes(handle) this.notifyForwarded(forwarded) if (!mailboxHandle) { return } const waiters = this.deps.getMessageWaiters(mailboxHandle) + const arrivedTypes = directTypes ?? (messageType ? [messageType] : undefined) const consumers = waiters ? [...waiters].filter( - (waiter) => !messageType || !waiter.typeFilter || waiter.typeFilter.includes(messageType) + (waiter) => + !arrivedTypes || + !waiter.typeFilter || + arrivedTypes.some((type) => waiter.typeFilter?.includes(type)) ) : [] if (consumers.length === 0) { @@ -89,10 +94,17 @@ export class OrchestrationMailboxNotificationCoordinator< private resolveArrivedMailboxes(handle: string): { mailboxHandle: string | null forwarded: RoutedOrchestrationMailbox[] + directTypes?: string[] } { - if (!this.deps.hasTerminalHandle(handle) || !this.deps.getDb()) { + if (handle.startsWith('run:') || handle.startsWith('dispatch:')) { return { mailboxHandle: handle, forwarded: [] } } + if (!this.deps.getDb()) { + return { mailboxHandle: handle, forwarded: [] } + } + if (!this.deps.hasTerminalHandle(handle)) { + return this.resolveDetachedMailbox(handle) + } try { const leaf = this.deps.getLiveLeafForHandle(handle) return { @@ -102,7 +114,26 @@ export class OrchestrationMailboxNotificationCoordinator< forwarded: this.deps.mailboxOwner.routeForeignDirectMessages(leaf) } } catch { - return { mailboxHandle: null, forwarded: [] } + return this.resolveDetachedMailbox(handle) + } + } + + private resolveDetachedMailbox(handle: string): { + mailboxHandle: string | null + forwarded: RoutedOrchestrationMailbox[] + directTypes?: string[] + } { + const routed = this.deps.mailboxOwner.routeDetachedDirectMessages( + handle, + this.deps.getPaneKeyForHandle(handle) + ) + const directTypes = routed.hasMore + ? [] + : (this.deps.getDb()?.getUnreadDirectMessageTypes(handle) ?? []) + return { + mailboxHandle: directTypes.length > 0 ? handle : null, + forwarded: routed.mailboxes, + directTypes } } diff --git a/src/main/runtime/orchestration/mailbox-owner.ts b/src/main/runtime/orchestration/mailbox-owner.ts index 3915e9be285..56f79d78d9a 100644 --- a/src/main/runtime/orchestration/mailbox-owner.ts +++ b/src/main/runtime/orchestration/mailbox-owner.ts @@ -17,6 +17,11 @@ export type RoutedOrchestrationMailbox = { types: string[] } +export type DetachedMailboxRoutingPage = { + mailboxes: RoutedOrchestrationMailbox[] + hasMore: boolean +} + type OrchestrationMailboxOwnerDependencies = { getDb: () => OrchestrationDb | null getLeaf: (leafKey: string) => OrchestrationMailboxLeaf | undefined @@ -29,6 +34,7 @@ type OrchestrationMailboxOwnerDependencies = { export class OrchestrationMailboxOwner { private readonly pendingDirectReconciliations = new Set() + private readonly pendingDetachedDirectReconciliations = new Set() constructor(private readonly deps: OrchestrationMailboxOwnerDependencies) {} @@ -113,6 +119,16 @@ export class OrchestrationMailboxOwner { return routed?.mailboxes ?? [] } + routeDetachedDirectMessages(directHandle: string, paneKey?: string): DetachedMailboxRoutingPage { + const routed = this.deps + .getDb() + ?.routeForeignDirectMessagesToOwnedMailboxes?.(directHandle, undefined, paneKey) + if (routed?.hasMore) { + this.scheduleDetachedDirectReconciliation(directHandle, paneKey) + } + return { mailboxes: routed?.mailboxes ?? [], hasMore: routed?.hasMore ?? false } + } + private resolveRunMailbox( db: OrchestrationDb, leaf: OrchestrationMailboxLeaf, @@ -204,4 +220,26 @@ export class OrchestrationMailboxOwner { this.resolve(currentLeaf) }) } + + private scheduleDetachedDirectReconciliation(directHandle: string, paneKey?: string): void { + if (this.pendingDetachedDirectReconciliations.has(directHandle)) { + return + } + this.pendingDetachedDirectReconciliations.add(directHandle) + setImmediate(() => { + this.pendingDetachedDirectReconciliations.delete(directHandle) + const routed = this.routeDetachedDirectMessages(directHandle, paneKey) + for (const mailbox of routed.mailboxes) { + for (const messageType of mailbox.types) { + this.deps.onForeignMailboxRouted(mailbox.mailboxHandle, messageType) + } + } + if (!routed.hasMore) { + for (const messageType of this.deps.getDb()?.getUnreadDirectMessageTypes(directHandle) ?? + []) { + this.deps.onForeignMailboxRouted(directHandle, messageType) + } + } + }) + } } diff --git a/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts b/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts index 327caf5f922..547117c1ed8 100644 --- a/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts +++ b/src/main/runtime/orchestration/mutation-receipt-capacity.test.ts @@ -103,7 +103,7 @@ describe('mutation receipt capacity schema', () => { db = new OrchestrationDb(dbPath) const sqlite = sqliteFor(db) - expect(sqlite.pragma('user_version', { simple: true })).toBe(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) expect(sqlite.prepare('SELECT receipt_count FROM mutation_receipt_ledger').get()).toEqual({ receipt_count: 20 }) diff --git a/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts b/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts index 4b8095da873..9a3afab1d81 100644 --- a/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts +++ b/src/main/runtime/orchestration/orchestration-db-retention-pagination.test.ts @@ -240,7 +240,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db = new OrchestrationDb(dbPath) const sqlite = sqliteFor(db) - expect(sqlite.pragma('user_version', { simple: true })).toBe(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) expect(db.getDispatchContextById(dispatch.id)).toMatchObject({ assignee_handle: 'term_worker' }) expect(db.getTask(task.id)).toMatchObject({ created_by_pane_key: null, @@ -277,7 +277,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db.close() db = new OrchestrationDb(dbPath) - expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(27) + expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(28) expect(db.getDispatchContextById(dispatch.id)).toBeDefined() }) @@ -309,7 +309,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db = new OrchestrationDb(dbPath) const sqlite = sqliteFor(db) - expect(sqlite.pragma('user_version', { simple: true })).toBe(27) + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) expect(db.getTask(task.id)).toMatchObject({ created_by_pane_key: 'tab_creator:leaf_creator', created_by_process_incarnation: 'pty_creator:incarnation-a', @@ -328,7 +328,7 @@ describe('OrchestrationDb dispatch assignee index migration', () => { db.close() db = new OrchestrationDb(dbPath) - expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(27) + expect(sqliteFor(db).pragma('user_version', { simple: true })).toBe(28) expect(db.getTask(task.id)?.created_by_process_incarnation).toBe('pty_creator:incarnation-a') }) }) diff --git a/src/main/runtime/orchestration/orchestration-reset-db.test.ts b/src/main/runtime/orchestration/orchestration-reset-db.test.ts index 1e0c62e5306..917260e7b2c 100644 --- a/src/main/runtime/orchestration/orchestration-reset-db.test.ts +++ b/src/main/runtime/orchestration/orchestration-reset-db.test.ts @@ -57,6 +57,9 @@ describe('OrchestrationDb reset scopes', () => { expect(db!.getTask(state.task.id)).toBeUndefined() expect(db!.getWorkerDispatch(state.started.dispatch.id)).toBeUndefined() expect(db!.getFederatedDispatch(state.started.dispatch.id)).toBeUndefined() + const sqlite = (db as unknown as { db: { prepare: (sql: string) => { all: () => unknown[] } } }) + .db + expect(sqlite.prepare('SELECT * FROM run_coordinator_handles').all()).toEqual([]) // The ledger survives so a lost reset response cannot replay as a new mutation. expect(db!.getMutationReceipt('caller_1', 'request_1')).toBeDefined() expect(db!.getInbox()).toEqual([]) diff --git a/src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts b/src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts index acd6e936f47..1d7a3ce45d5 100644 --- a/src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts +++ b/src/main/runtime/orchestration/orchestration-run-delivery-db.test.ts @@ -163,6 +163,30 @@ describe('OrchestrationDb Run state', () => { expect(replacement?.messages.map((message) => message.subject)).toEqual(['one']) }) + it('does not move a mismatched Run through another Run Dispatch mailbox', () => { + const d = createDb() + const runA = createBoundRun(d) + const runB = d.createRun({ + objective: 'Dispatch owner', + coordinatorHandle: 'term_other', + coordinatorPaneKey: 'tab_other:22222222-2222-4222-9222-222222222222' + }) + const task = d.createTask({ spec: 'work', runId: runB.id }) + const dispatch = d.createDispatchContext(task.id, 'term_worker') + const mismatched = d.insertMessage({ + from: 'worker', + to: `dispatch:${dispatch.id}`, + subject: 'wrong Run', + runId: runA.id + }) + + expect(d.routeUnreadDispatchMailboxToRunMailbox(dispatch.id, runB.id)).toMatchObject({ + routedCount: 0, + hasMore: false + }) + expect(d.getMessageById(mismatched.id)?.to_handle).toBe(`dispatch:${dispatch.id}`) + }) + it('replays an outstanding batch after reopening the database', () => { const dir = mkdtempSync(join(tmpdir(), 'orca-delivery-')) const dbPath = join(dir, 'orchestration.db') diff --git a/src/main/runtime/orchestration/run-coordinator-handle-migration.test.ts b/src/main/runtime/orchestration/run-coordinator-handle-migration.test.ts new file mode 100644 index 00000000000..471ccc17f76 --- /dev/null +++ b/src/main/runtime/orchestration/run-coordinator-handle-migration.test.ts @@ -0,0 +1,138 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' +import Database from '../../sqlite/sync-database' +import { OrchestrationDb } from './db' + +describe('Run coordinator handle history migration', () => { + let db: OrchestrationDb | undefined + let tempDir: string | undefined + + afterEach(() => { + db?.close() + if (tempDir) { + rmSync(tempDir, { recursive: true, force: true }) + } + }) + + it('preserves v27 coordinator authority for mail arriving after a rebind', () => { + tempDir = mkdtempSync(join(tmpdir(), 'orca-run-coordinator-handle-migration-')) + const dbPath = join(tempDir, 'orchestration.db') + db = new OrchestrationDb(dbPath) + const run = db.createRun({ + objective: 'migration authority', + coordinatorHandle: 'term_old', + coordinatorPaneKey: 'tab:leaf' + }) + db.close() + db = undefined + + const oldDb = new Database(dbPath) + oldDb.exec('DROP TABLE run_coordinator_handles') + oldDb.pragma('user_version = 27') + oldDb.close() + + db = new OrchestrationDb(dbPath) + const sqlite = (db as unknown as { db: Database.Database }).db + expect(sqlite.pragma('user_version', { simple: true })).toBe(28) + expect( + sqlite + .prepare('SELECT run_id, terminal_handle FROM run_coordinator_handles WHERE run_id = ?') + .all(run.id) + ).toEqual([{ run_id: run.id, terminal_handle: 'term_old' }]) + + db.bindRun({ + runId: run.id, + coordinatorHandle: 'term_new', + coordinatorPaneKey: 'tab:leaf' + }) + const late = db.insertMessage({ + runId: run.id, + from: 'worker', + to: 'term_old', + subject: 'late completion', + type: 'worker_done' + }) + sqlite.prepare('UPDATE messages SET to_handle = ? WHERE id = ?').run('term_old', late.id) + + expect(db.routeForeignDirectMessagesToOwnedMailboxes('term_old')).toMatchObject({ + routedCount: 1, + mailboxes: [{ mailboxHandle: `run:${run.id}`, types: ['worker_done'] }] + }) + expect(db.getMessageById(late.id)?.to_handle).toBe(`run:${run.id}`) + }) + + it('records every coordinator rebind performed by an older runtime', () => { + tempDir = mkdtempSync(join(tmpdir(), 'orca-run-coordinator-old-runtime-')) + const dbPath = join(tempDir, 'orchestration.db') + db = new OrchestrationDb(dbPath) + const run = db.createRun({ + objective: 'mixed-version authority', + coordinatorHandle: 'term_v28', + coordinatorPaneKey: 'tab:leaf' + }) + db.close() + db = undefined + + const oldRuntimeDb = new Database(dbPath) + oldRuntimeDb + .prepare('UPDATE runs SET coordinator_handle = ? WHERE id = ?') + .run('term_v27_first', run.id) + oldRuntimeDb + .prepare('UPDATE runs SET coordinator_handle = ? WHERE id = ?') + .run('term_v27_second', run.id) + expect(oldRuntimeDb.pragma('user_version', { simple: true })).toBe(28) + oldRuntimeDb.close() + + db = new OrchestrationDb(dbPath) + const sqlite = (db as unknown as { db: Database.Database }).db + expect( + sqlite + .prepare( + 'SELECT terminal_handle FROM run_coordinator_handles WHERE run_id = ? ORDER BY terminal_handle' + ) + .all(run.id) + ).toEqual([ + { terminal_handle: 'term_v27_first' }, + { terminal_handle: 'term_v27_second' }, + { terminal_handle: 'term_v28' } + ]) + const late = db.insertMessage({ + runId: run.id, + from: 'worker', + to: 'term_v27_first', + subject: 'late intermediate completion', + type: 'worker_done' + }) + expect(db.getMessageById(late.id)?.to_handle).toBe(`run:${run.id}`) + }) + + it('reinstalls coordinator routing after a migration removes the trigger', () => { + tempDir = mkdtempSync(join(tmpdir(), 'orca-run-coordinator-trigger-migration-')) + const dbPath = join(tempDir, 'orchestration.db') + db = new OrchestrationDb(dbPath) + const run = db.createRun({ + objective: 'trigger migration', + coordinatorHandle: 'term_migrated', + coordinatorPaneKey: 'tab:leaf' + }) + db.close() + db = undefined + + const oldDb = new Database(dbPath) + oldDb.exec('DROP TRIGGER trg_messages_route_coordinator_mail') + oldDb.pragma('user_version = 27') + oldDb.close() + + db = new OrchestrationDb(dbPath) + const message = db.insertMessage({ + runId: run.id, + from: 'worker', + to: 'term_migrated', + subject: 'first post-migration message' + }) + + expect(db.getMessageById(message.id)?.to_handle).toBe(`run:${run.id}`) + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration-runs.ts b/src/main/runtime/rpc/methods/orchestration-runs.ts index 36ddcb1201f..aa902171303 100644 --- a/src/main/runtime/rpc/methods/orchestration-runs.ts +++ b/src/main/runtime/rpc/methods/orchestration-runs.ts @@ -59,6 +59,7 @@ export const ORCHESTRATION_RUN_METHODS: RpcMethod[] = [ coordinatorHandle: params.from, coordinatorPaneKey: paneKey }) + runtime.cancelMessageWaiters(params.from) if (priorRun) { runtime.cancelMessageWaiters(`run:${priorRun.id}`) } @@ -104,6 +105,7 @@ export const ORCHESTRATION_RUN_METHODS: RpcMethod[] = [ `Run ${params.id} was not found or is inspect-only.` ) } + runtime.cancelMessageWaiters(params.from) runtime.cancelMessageWaiters(`run:${params.id}`) if (priorRun && priorRun.id !== params.id) { runtime.cancelMessageWaiters(`run:${priorRun.id}`) diff --git a/src/main/runtime/rpc/methods/orchestration.test.ts b/src/main/runtime/rpc/methods/orchestration.test.ts index 12487c253c1..6cb594d1479 100644 --- a/src/main/runtime/rpc/methods/orchestration.test.ts +++ b/src/main/runtime/rpc/methods/orchestration.test.ts @@ -294,6 +294,25 @@ describe('orchestration RPC methods', () => { await fenced }) + + it('fences an unbound direct waiter when its pane creates a Run', async () => { + setup(false) + vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue(coordinatorPaneKey) + const directWait = call('orchestration.check', { + terminal: 'term_coord', + wait: true, + timeoutMs: 5_000 + }) + const fenced = expect(directWait).rejects.toMatchObject({ code: 'consumer_fenced' }) + await Promise.resolve() + + await call('orchestration.runCreate', { + objective: 'Claim the direct mailbox', + from: 'term_coord' + }) + + await fenced + }) }) describe('orchestration.send', () => { @@ -314,6 +333,33 @@ describe('orchestration RPC methods', () => { expect(runtime.deliverPendingMessagesForHandle).toHaveBeenCalled() }) + it('wakes the Run mailbox after canonicalizing an old coordinator recipient', async () => { + setup() + const remintedPaneKey = 'tab_reminted:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb' + vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) => + handle === 'term_reminted' + ? remintedPaneKey + : handle === 'term_coord' + ? coordinatorPaneKey + : null + ) + db.bindRun({ + runId: activeRunId!, + coordinatorHandle: 'term_reminted', + coordinatorPaneKey: remintedPaneKey + }) + const waiting = runtime.waitForMessage(`run:${activeRunId}`, { timeoutMs: 5_000 }) + + const result = (await call('orchestration.send', { + from: 'term_reminted', + to: 'term_coord', + subject: 'late completion' + })) as { message: { to_handle: string } } + + expect(result.message.to_handle).toBe(`run:${activeRunId}`) + await expect(waiting).resolves.toBe('notified') + }) + it('routes exact Dispatch mail independently of terminal handles', async () => { setup() const task = db.createTask({ spec: 'controlled worker' }) @@ -1225,6 +1271,63 @@ describe('orchestration RPC methods', () => { expect(db.getUnreadMessages(`run:${activeRunId}`)).toHaveLength(0) }) + it('peeks an old unread Run row beyond the newest history page', async () => { + setup() + const unread = db.insertMessage({ + from: 'worker', + to: `run:${activeRunId}`, + subject: 'old unread', + runId: activeRunId + }) + for (let index = 0; index < 100; index += 1) { + const read = db.insertMessage({ + from: 'worker', + to: `run:${activeRunId}`, + subject: `new read ${index}`, + runId: activeRunId + }) + db.markAsRead([read.id]) + } + + const peeked = (await call('orchestration.check', { + terminal: 'term_coord', + peek: true + })) as { messages: { id: string }[]; count: number } + const delivered = (await call('orchestration.check', { + terminal: 'term_coord' + })) as { messages: { id: string }[]; count: number } + + expect(peeked).toMatchObject({ count: 1, messages: [{ id: unread.id }] }) + expect(delivered).toMatchObject({ count: 1, messages: [{ id: unread.id }] }) + }) + + it('waits for a filtered Run peek without consuming the arrival', async () => { + setup() + let arrivedId = '' + vi.spyOn(runtime, 'waitForMessage').mockImplementation(async () => { + arrivedId = db.insertMessage({ + from: 'worker', + to: `run:${activeRunId}`, + subject: 'peeked completion', + type: 'worker_done', + runId: activeRunId + }).id + return 'notified' + }) + + const peeked = (await call('orchestration.check', { + terminal: 'term_coord', + peek: true, + wait: true, + types: 'worker_done', + timeoutMs: 100 + })) as { messages: { id: string }[]; count: number } + + expect(peeked).toMatchObject({ count: 1, messages: [{ id: arrivedId }] }) + expect(db.getMessageById(arrivedId)?.read).toBe(0) + expect(db.getUnreadMessages(`run:${activeRunId}`, ['worker_done'])).toHaveLength(1) + }) + it('reconciles worker_done returned by a waiting manual check', async () => { setup() const { task, dispatch } = createDispatchedTask() diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index 831aa61e6c5..f52ee65da19 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -5,6 +5,7 @@ import { defineMethod, type RpcMethod } from '../core' import { OptionalFiniteNumber, OptionalString, OptionalBoolean, requiredString } from '../schemas' import { LEGACY_CONTRACT_VERSION, + type MessageRow, type MessageType, type MessagePriority, type TaskStatus @@ -655,7 +656,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ 'dispatch_capability_invalid', authority.reason ) ?? msg - runtime.notifyMessageArrived(to, rejection.type) + runtime.notifyMessageArrived(rejection.to_handle, rejection.type) return { message: rejection, lifecycle: { @@ -675,15 +676,15 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ } if (reconciled.action === 'rejected') { const rejection = db.getMessageById(msg.id) ?? msg - runtime.notifyMessageArrived(to, rejection.type) + runtime.notifyMessageArrived(rejection.to_handle, rejection.type) return { message: rejection, lifecycle: reconciled } } - runtime.notifyMessageArrived(to, msg.type) + runtime.notifyMessageArrived(msg.to_handle, msg.type) return msg.type === 'worker_done' ? { message: msg, lifecycle: reconciled } : { message: msg } } - runtime.notifyMessageArrived(to, msg.type) + runtime.notifyMessageArrived(msg.to_handle, msg.type) return { message: msg } } @@ -808,12 +809,9 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'outcome_unknown') ) } - if (params.peek || params.all || params.unread === false) { + if (params.all || (params.unread === false && !params.peek)) { const history = db.getRunMailboxHistory(run.id, 100, typeFilter) - const messages = - params.all || (params.unread === false && !params.peek) - ? history - : history.filter((message) => message.read === 0) + const messages = history const result = { messages, count: messages.length, @@ -829,13 +827,27 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ return { ...result, runId: run.id } } + const peekResult = (messages: MessageRow[]) => ({ + runId: run.id, + messages, + count: messages.length, + acknowledged: acknowledged?.delivery.id ?? null, + ...(params.format || params.inject + ? { formatted: messages.map(formatMessageBanner).join('\n\n') } + : {}) + }) + const readPeek = () => db.getUnreadRunMailbox(run.id, 100, typeFilter) const readDelivery = (wakeTypes?: MessageType[]) => db.getOrCreateRunDelivery({ runId: run.id, consumerGeneration: generation, wakeTypes }) - let current = readDelivery(params.wait ? typeFilter : undefined) + let peeked = params.peek ? readPeek() : [] + if (params.peek && peeked.length > 0) { + return peekResult(peeked) + } + let current = params.peek ? undefined : readDelivery(params.wait ? typeFilter : undefined) if (current) { return { runId: run.id, @@ -853,6 +865,9 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ } } if (!params.wait) { + if (params.peek) { + return peekResult([]) + } return { runId: run.id, deliveryId: null, @@ -899,6 +914,9 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ ) } if (waitResult === 'timed_out') { + if (params.peek) { + return { ...peekResult([]), timedOut: true, cancelled: false, connectionLost: false } + } return { runId: run.id, deliveryId: null, @@ -911,6 +929,14 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ } } if (waitResult === 'cancelled') { + if (params.peek) { + return { + ...peekResult([]), + timedOut: false, + cancelled: true, + connectionLost: signal?.aborted === true + } + } return { runId: run.id, deliveryId: null, @@ -923,6 +949,15 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ } } + if (params.peek) { + peeked = readPeek() + return { + ...peekResult(peeked), + timedOut: false, + cancelled: false, + connectionLost: false + } + } current = readDelivery(typeFilter) return { runId: run.id, @@ -1150,7 +1185,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ } // Why: signal aborts this waiter when the client socket closes, freeing the long-poll slot immediately rather than after timeoutMs (design doc §3.1). - await runtime.waitForMessage(handle, { + const waitResult = await runtime.waitForMessage(handle, { typeFilter: typeFilter as string[] | undefined, timeoutMs: params.timeoutMs ?? undefined, signal @@ -1158,6 +1193,12 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ if (signal?.aborted) { return { messages: [], count: 0 } } + if (waitResult === 'cancelled') { + throw new OrchestrationError( + 'consumer_fenced', + 'This direct mailbox became owned by a Run while the check was waiting.' + ) + } return readAndReturn() } }), @@ -1246,7 +1287,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ runId: original.run_id }) - runtime.notifyMessageArrived(original.from_handle, reply.type) + runtime.notifyMessageArrived(reply.to_handle, reply.type) return { message: reply } } }), diff --git a/src/main/runtime/rpc/orchestration-legacy-mail.ts b/src/main/runtime/rpc/orchestration-legacy-mail.ts index c83d7ffdf2f..0af41aebd88 100644 --- a/src/main/runtime/rpc/orchestration-legacy-mail.ts +++ b/src/main/runtime/rpc/orchestration-legacy-mail.ts @@ -162,7 +162,7 @@ export async function handleLegacyReply(args: { body: params.body }) if (!committed.duplicate) { - runtime.notifyMessageArrived(committed.question.asker_handle, 'status') + runtime.notifyMessageArrived(committed.message.to_handle, committed.message.type) } return { message: committed.message,