From be07b43a2b7377b69bed04e0f39bc4cbb70593fc Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Mon, 17 Aug 2026 01:50:17 -0700 Subject: [PATCH] fix(orchestration): enforce honest recipient routing (#14964) --- ...rchestration-send-receipt-warnings.test.ts | 87 +++ src/cli/handlers/orchestration.ts | 36 +- src/main/runtime/orca-runtime.test.ts | 18 + src/main/runtime/orca-runtime.ts | 17 + .../orchestration-cli-subprocess.test.ts | 9 +- src/main/runtime/orchestration/db.ts | 2 + .../db/dispatch-context/dispatch-lookup.ts | 45 ++ .../db/messages/message-insert.ts | 52 +- .../runs/legacy-coordinator-mail-takeover.ts | 8 +- .../orchestration/db/runs/run-lookup.ts | 46 ++ .../message-batch-atomicity.test.ts | 29 + .../orchestration-legacy-storage-db.test.ts | 17 + src/main/runtime/rpc/errors.ts | 3 + .../rpc/methods/orchestration-check.test.ts | 3 + .../orchestration-recipient-routing.test.ts | 582 ++++++++++++++++++ .../orchestration-recipient-routing.ts | 145 +++++ .../methods/orchestration-rpc-test-harness.ts | 3 + .../rpc/methods/orchestration-send.test.ts | 7 + src/main/runtime/rpc/methods/orchestration.ts | 131 +++- ...hestration-legacy-coordinator-race.test.ts | 16 +- ...stration-runtime-update-settlement.test.ts | 5 +- src/main/ssh/ssh-remote-orca-cli.test.ts | 23 +- 22 files changed, 1223 insertions(+), 61 deletions(-) create mode 100644 src/cli/handlers/orchestration-send-receipt-warnings.test.ts create mode 100644 src/main/runtime/rpc/methods/orchestration-recipient-routing.test.ts create mode 100644 src/main/runtime/rpc/methods/orchestration-recipient-routing.ts diff --git a/src/cli/handlers/orchestration-send-receipt-warnings.test.ts b/src/cli/handlers/orchestration-send-receipt-warnings.test.ts new file mode 100644 index 00000000000..8514b1d3b61 --- /dev/null +++ b/src/cli/handlers/orchestration-send-receipt-warnings.test.ts @@ -0,0 +1,87 @@ +import { expect, it, vi } from 'vitest' + +const callMock = vi.fn() +vi.mock('../format', () => ({ printResult: vi.fn() })) +vi.mock('../selectors', () => ({ getTerminalHandle: vi.fn() })) + +import { printResult } from '../format' +import { ORCHESTRATION_HANDLERS } from './orchestration' + +async function formatSend(result: unknown): Promise { + callMock.mockReset().mockResolvedValueOnce({ result }) + await ORCHESTRATION_HANDLERS['orchestration send']({ + flags: new Map([ + ['from', 'term_sender'], + ['to', 'term_recipient'], + ['subject', 'ping'] + ]), + client: { call: callMock }, + cwd: '/workspace', + json: false + } as never) + const printCall = vi.mocked(printResult).mock.calls.at(-1) + const formatter = printCall?.[2] as (value: unknown) => string + return formatter(result) +} + +it('prints a live terminal-only delivery limitation', async () => { + const line = await formatSend({ + message: { id: 'msg_1' }, + warnings: [ + { + code: 'legacy_terminal_recipient', + recipient: 'term_recipient', + message: 'term_recipient is live now, but its mailbox is not restart-durable.' + } + ] + }) + + expect(line).toBe( + 'Sent msg_1\nWarning: term_recipient is live now, but its mailbox is not restart-durable.' + ) +}) + +it('shows partial fan-out omissions without hiding delivered recipients', async () => { + const line = await formatSend({ + messages: [{ id: 'msg_1' }], + recipients: 1, + warnings: [ + { + code: 'recipient_unreachable', + recipient: 'term_gone', + message: 'Terminal term_gone has no live pane or durable mailbox.' + } + ] + }) + + expect(line).toBe( + 'Sent 1 messages to 1 recipients\nWarning: Terminal term_gone has no live pane or durable mailbox.' + ) +}) + +it('prints delivery limitations on relayed receipts', async () => { + const line = await formatSend({ + relay: { + messageId: 'relay_1', + sequence: 1, + dispatchId: 'ctx_remote', + destination: 'worker', + accepted: true + }, + warnings: [ + { + code: 'legacy_terminal_recipient', + recipient: 'term_remote', + message: 'term_remote is reachable through a compatibility address.' + } + ] + }) + + expect(line).toBe( + 'Queued relay_1 for worker Dispatch ctx_remote\nWarning: term_remote is reachable through a compatibility address.' + ) +}) + +it('leaves a canonical receipt unchanged', async () => { + await expect(formatSend({ message: { id: 'msg_1' } })).resolves.toBe('Sent msg_1') +}) diff --git a/src/cli/handlers/orchestration.ts b/src/cli/handlers/orchestration.ts index 5fe16b22181..e529fa62d11 100644 --- a/src/cli/handlers/orchestration.ts +++ b/src/cli/handlers/orchestration.ts @@ -100,9 +100,23 @@ type LifecycleSendResult = | { action: 'settled'; outcome: 'succeeded' | 'failed'; duplicate?: boolean } | { action: 'rejected'; code: string; reason: string } +type SendRecipientWarning = { + code: string + recipient: string + message: string +} + type OrchestrationSendResult = - | { message: { id: string; run_id?: string }; lifecycle?: LifecycleSendResult } - | { messages: { id: string }[]; recipients: number } + | { + message: { id: string; run_id?: string } + lifecycle?: LifecycleSendResult + warnings?: SendRecipientWarning[] + } + | { + messages: { id: string }[] + recipients: number + warnings?: SendRecipientWarning[] + } | { relay: { messageId: string @@ -112,6 +126,7 @@ type OrchestrationSendResult = accepted: true } lifecycle?: LifecycleSendResult + warnings?: SendRecipientWarning[] } function resolveCompatibilityCliCommand(): 'orca' | 'orca-ide' | 'orca-dev' { @@ -598,16 +613,25 @@ export const ORCHESTRATION_HANDLERS: Record = { throw new RuntimeClientError(result.result.lifecycle.code, result.result.lifecycle.reason) } printResult(result, json, (r) => { + const warnings = 'warnings' in r ? (r.warnings ?? []) : [] + const withWarnings = (line: string): string => + warnings.length > 0 + ? [line, ...warnings.map((warning) => `Warning: ${warning.message}`)].join('\n') + : line if ('message' in r) { - return `Sent ${r.message.id}` + return withWarnings(`Sent ${r.message.id}`) } if ('relay' in r) { if (r.relay.destination === 'worker') { - return `Queued ${r.relay.messageId} for worker Dispatch ${r.relay.dispatchId}` + return withWarnings( + `Queued ${r.relay.messageId} for worker Dispatch ${r.relay.dispatchId}` + ) } - return `Queued ${r.relay.messageId} for Run home (Dispatch ${r.relay.dispatchId})` + return withWarnings( + `Queued ${r.relay.messageId} for Run home (Dispatch ${r.relay.dispatchId})` + ) } - return `Sent ${r.messages.length} messages to ${r.recipients} recipients` + return withWarnings(`Sent ${r.messages.length} messages to ${r.recipients} recipients`) }) }, diff --git a/src/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts index 17117479f80..8c327526bf1 100644 --- a/src/main/runtime/orca-runtime.test.ts +++ b/src/main/runtime/orca-runtime.test.ts @@ -49884,6 +49884,7 @@ describe('OrcaRuntimeService', () => { expect(runtime.resolveLeafForHandle(handle)).toEqual({ ptyId: 'pty-b' }) // The guarded resolver surfaces the staleness so clients can re-derive. expect(() => runtime.resolveLiveLeafForHandle(handle)).toThrow('terminal_handle_stale') + expect(runtime.getLiveTerminalPaneKey(handle)).toBeNull() }) it('lets a handle issued before its first PTY adopt that PTY without erroring', async () => { @@ -49904,6 +49905,23 @@ describe('OrcaRuntimeService', () => { expect(runtime.resolveLiveLeafForHandle(handle)).toEqual({ ptyId: 'pty-a' }) }) + + it('keeps terminal cwd resolution fail-soft when the provider is unavailable', async () => { + const runtime = new OrcaRuntimeService(store) + runtime.attachWindow(1) + syncSingleTerminalGraph(runtime, 'pty-a') + const handle = issueLeafHandle(runtime, 'pty-a') + runtime.setPtyController({ + write: () => true, + kill: () => true, + getForegroundProcess: async () => null, + getCwd: async () => { + throw new Error('ssh disconnected') + } + }) + + await expect(runtime.resolveTerminalCwd(handle)).resolves.toBeNull() + }) }) describe('mobile terminal create resilience (#7718)', () => { diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index ed68896ecbe..50843abb927 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -17765,6 +17765,23 @@ export class OrcaRuntimeService { return this.getPaneKeyForTerminalHandle(handle) } + getLiveTerminalPaneKey(handle: string): string | null { + const runtimePty = this.getLivePtyForHandle(handle) + if (runtimePty) { + return runtimePty.pty.connected ? (runtimePty.pty.paneKey ?? null) : null + } + try { + const leaf = this.resolveLiveLeafForHandle(handle) + if (!leaf?.ptyId) { + return null + } + const pty = this.ptysById.get(leaf.ptyId) + return pty?.connected === false ? null : this.getPaneKeyForTerminalHandle(handle) + } catch { + return null + } + } + getTerminalWorktreeIdForPaneKey(paneKey: string): string | null { const parsed = parsePaneKey(paneKey) const leaf = parsed ? this.leaves.get(this.getLeafKey(parsed.tabId, parsed.leafId)) : null diff --git a/src/main/runtime/orchestration-cli-subprocess.test.ts b/src/main/runtime/orchestration-cli-subprocess.test.ts index b511c472931..d8306f9399d 100644 --- a/src/main/runtime/orchestration-cli-subprocess.test.ts +++ b/src/main/runtime/orchestration-cli-subprocess.test.ts @@ -200,7 +200,14 @@ describeIfBuilt('orca orchestration reset subprocess', () => { runtime.setOrchestrationDb(db) const coordinatorPaneKey = 'tab_cli:11111111-1111-4111-8111-111111111111' vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) => - handle === 'term_cli' ? coordinatorPaneKey : null + handle === 'term_cli' + ? coordinatorPaneKey + : handle === 'term_target' + ? 'tab_target:33333333-3333-4333-8333-333333333333' + : null + ) + vi.spyOn(runtime, 'getLiveTerminalPaneKey').mockImplementation((handle) => + runtime.getTerminalPaneKey(handle) ) db.createRun({ objective: 'CLI reset subprocess fixture', diff --git a/src/main/runtime/orchestration/db.ts b/src/main/runtime/orchestration/db.ts index 03a6ee36dfa..c6cab061700 100644 --- a/src/main/runtime/orchestration/db.ts +++ b/src/main/runtime/orchestration/db.ts @@ -11,6 +11,8 @@ export type { ForeignDirectMailboxRoutingPage, MailboxRoutingPage } from './db/messages/mailbox-routing-page' +export type { MessageInsert } from './db/messages/message-insert' +export type { LegacyAdoptedMailboxOwner } from './db/runs/run-lookup' export type { MessageType, diff --git a/src/main/runtime/orchestration/db/dispatch-context/dispatch-lookup.ts b/src/main/runtime/orchestration/db/dispatch-context/dispatch-lookup.ts index f5f760f8a8b..1fa95aa2882 100644 --- a/src/main/runtime/orchestration/db/dispatch-context/dispatch-lookup.ts +++ b/src/main/runtime/orchestration/db/dispatch-context/dispatch-lookup.ts @@ -37,6 +37,49 @@ export function getActiveDispatchForIdentity( return this.findActiveDispatchForAssignee(handle, paneKey) } +export function getActiveDispatchMailboxOwners( + this: OrchestrationDb, + handle: string, + paneKey?: string +): DispatchContextRow[] { + const byHandle = this.db + .prepare( + `SELECT * FROM dispatch_contexts + WHERE assignee_handle = ? AND status IN ('pending', 'dispatched') + ORDER BY rowid DESC` + ) + .all(handle) as DispatchContextRow[] + if (byHandle.length > 0 || !paneKey) { + return byHandle + } + + const byExactPane = this.db + .prepare( + `SELECT * FROM dispatch_contexts + WHERE assignee_pane_key = ? AND status IN ('pending', 'dispatched') + ORDER BY rowid DESC` + ) + .all(paneKey) as DispatchContextRow[] + if (byExactPane.length > 0 || !parsePaneKey(paneKey)) { + return byExactPane + } + return ( + this.db + .prepare( + `SELECT * FROM dispatch_contexts + WHERE assignee_pane_key IS NOT NULL + AND status IN ('pending', 'dispatched') AND instr(assignee_pane_key, ':') > 1 + AND ${DISPATCH_PANE_KEY_MATCH_SUFFIX_SQL} = ? + ORDER BY rowid DESC` + ) + .all(paneKeyMatchSuffix(paneKey)) as DispatchContextRow[] + ).filter( + (dispatch) => + dispatch.assignee_pane_key !== null && + isEquivalentPaneKey(dispatch.assignee_pane_key, paneKey) + ) +} + export function isDispatchMessageSender( this: OrchestrationDb, params: { @@ -125,6 +168,7 @@ export type DispatchLookupMethods = { getActiveDispatchForTerminal: typeof getActiveDispatchForTerminal hasAnyDispatchContexts: typeof hasAnyDispatchContexts getActiveDispatchForIdentity: typeof getActiveDispatchForIdentity + getActiveDispatchMailboxOwners: typeof getActiveDispatchMailboxOwners isDispatchMessageSender: typeof isDispatchMessageSender findActiveDispatchForAssignee: typeof findActiveDispatchForAssignee getLatestDispatchForTerminal: typeof getLatestDispatchForTerminal @@ -135,6 +179,7 @@ export function attachDispatchLookup(ctor: { prototype: object }): void { getActiveDispatchForTerminal, hasAnyDispatchContexts, getActiveDispatchForIdentity, + getActiveDispatchMailboxOwners, isDispatchMessageSender, findActiveDispatchForAssignee, getLatestDispatchForTerminal diff --git a/src/main/runtime/orchestration/db/messages/message-insert.ts b/src/main/runtime/orchestration/db/messages/message-insert.ts index d1a1b2f5611..f1a3a83dbbd 100644 --- a/src/main/runtime/orchestration/db/messages/message-insert.ts +++ b/src/main/runtime/orchestration/db/messages/message-insert.ts @@ -6,23 +6,24 @@ import type { OrchestrationDb } from '../orchestration-db' // ── Messages ── -export function insertMessage( - this: OrchestrationDb, - msg: { - id?: string - from: string - to: string - subject: string - body?: string - type?: MessageType - priority?: MessagePriority - threadId?: string - payload?: string - senderPaneKey?: string - runId?: string - deliveryContract?: MessageDeliveryContract - } -): MessageRow { +const MESSAGE_INSERT_SAVEPOINT = 'message_insert_batch' + +export type MessageInsert = { + id?: string + from: string + to: string + subject: string + body?: string + type?: MessageType + priority?: MessagePriority + threadId?: string + payload?: string + senderPaneKey?: string + runId?: string + deliveryContract?: MessageDeliveryContract +} + +export function insertMessage(this: OrchestrationDb, msg: MessageInsert): MessageRow { const runId = msg.runId ?? LEGACY_RUN_ID const deliveryContract = msg.deliveryContract ?? 'current_delivery' this.requireRun(runId) @@ -53,12 +54,27 @@ export function insertMessage( ) } +export function insertMessages(this: OrchestrationDb, messages: MessageInsert[]): MessageRow[] { + this.db.exec(`SAVEPOINT ${MESSAGE_INSERT_SAVEPOINT}`) + try { + const inserted = messages.map((message) => this.insertMessage(message)) + this.db.exec(`RELEASE ${MESSAGE_INSERT_SAVEPOINT}`) + return inserted + } catch (error) { + this.db.exec(`ROLLBACK TO ${MESSAGE_INSERT_SAVEPOINT}`) + this.db.exec(`RELEASE ${MESSAGE_INSERT_SAVEPOINT}`) + throw error + } +} + export type MessageInsertMethods = { insertMessage: typeof insertMessage + insertMessages: typeof insertMessages } export function attachMessageInsert(ctor: { prototype: object }): void { Object.assign(ctor.prototype, { - insertMessage + insertMessage, + insertMessages }) } diff --git a/src/main/runtime/orchestration/db/runs/legacy-coordinator-mail-takeover.ts b/src/main/runtime/orchestration/db/runs/legacy-coordinator-mail-takeover.ts index f2892b0880b..972b3b83228 100644 --- a/src/main/runtime/orchestration/db/runs/legacy-coordinator-mail-takeover.ts +++ b/src/main/runtime/orchestration/db/runs/legacy-coordinator-mail-takeover.ts @@ -83,15 +83,14 @@ export function getUniqueLegacyCoordinatorHandle( .prepare( `SELECT DISTINCT assignee_handle AS handle FROM dispatch_contexts - WHERE run_id = ? AND contract_version = ? - AND assignee_handle IS NOT NULL + WHERE run_id = ? AND assignee_handle IS NOT NULL UNION SELECT DISTINCT terminal_handle AS handle FROM legacy_compatibility_principals WHERE run_id = ? AND role = 'worker' AND status IN ('committed', 'settled')` ) - .all(runId, LEGACY_CONTRACT_VERSION, runId) as { handle: string }[] + .all(runId, runId) as { handle: string }[] ).map((row) => row.handle) ) const durableRows = this.db @@ -107,10 +106,9 @@ export function getUniqueLegacyCoordinatorHandle( AND EXISTS( SELECT 1 FROM dispatch_contexts d WHERE d.task_id = t.id AND d.run_id = t.run_id - AND d.contract_version = ? )` ) - .all(adoption.adopted_at, runId, adoption.adopted_at, LEGACY_CONTRACT_VERSION) as { + .all(adoption.adopted_at, runId, adoption.adopted_at) as { handle: string }[] if (durableRows.some((row) => workerHandles.has(row.handle))) { diff --git a/src/main/runtime/orchestration/db/runs/run-lookup.ts b/src/main/runtime/orchestration/db/runs/run-lookup.ts index dd09d1f832b..061a7b39497 100644 --- a/src/main/runtime/orchestration/db/runs/run-lookup.ts +++ b/src/main/runtime/orchestration/db/runs/run-lookup.ts @@ -10,11 +10,53 @@ import { encodeRunListCursor, decodeRunListCursor } from '../run-list-cursor' import type { RunListPage } from '../run-list-page' import type { OrchestrationDb } from '../orchestration-db' +export type LegacyAdoptedMailboxOwner = { + runId: string + terminalHandle: string +} + export function getRun(this: OrchestrationDb, id: string): RunRow | undefined { const run = this.getRunRaw(id) return run ? exposeRunTimestamps(run) : undefined } +export function getLegacyAdoptedRunMailboxOwner( + this: OrchestrationDb +): LegacyAdoptedMailboxOwner | null { + const adoption = this.getLegacyAdoption() + if (!adoption) { + return null + } + const terminalHandle = this.getUniqueLegacyCoordinatorHandle(adoption.adopted_run_id) + return terminalHandle ? { runId: adoption.adopted_run_id, terminalHandle } : null +} + +export function getRunMailboxOwnerIdsForHandle( + this: OrchestrationDb, + terminalHandle: string, + legacyAdoptedMailboxOwner?: LegacyAdoptedMailboxOwner | null +): string[] { + const runIds = ( + this.db + .prepare( + `SELECT coordinator.run_id + FROM run_coordinator_handles AS coordinator + JOIN runs ON runs.id = coordinator.run_id AND runs.legacy = 0 + WHERE coordinator.terminal_handle = ? + ORDER BY coordinator.run_id` + ) + .all(terminalHandle) as { run_id: string }[] + ).map((row) => row.run_id) + const adoptedOwner = + legacyAdoptedMailboxOwner === undefined + ? this.getLegacyAdoptedRunMailboxOwner() + : legacyAdoptedMailboxOwner + if (adoptedOwner?.terminalHandle === terminalHandle) { + runIds.push(adoptedOwner.runId) + } + return [...new Set(runIds)].sort() +} + export function listRuns( this: OrchestrationDb, params: { limit?: number; cursor?: string } = {} @@ -117,6 +159,8 @@ export function fenceOutstandingDelivery(this: OrchestrationDb, runId: string): export type RunLookupMethods = { getRun: typeof getRun + getLegacyAdoptedRunMailboxOwner: typeof getLegacyAdoptedRunMailboxOwner + getRunMailboxOwnerIdsForHandle: typeof getRunMailboxOwnerIdsForHandle listRuns: typeof listRuns getCurrentRunForPane: typeof getCurrentRunForPane runsBoundToPane: typeof runsBoundToPane @@ -129,6 +173,8 @@ export type RunLookupMethods = { export function attachRunLookup(ctor: { prototype: object }): void { Object.assign(ctor.prototype, { getRun, + getLegacyAdoptedRunMailboxOwner, + getRunMailboxOwnerIdsForHandle, listRuns, getCurrentRunForPane, runsBoundToPane, diff --git a/src/main/runtime/orchestration/message-batch-atomicity.test.ts b/src/main/runtime/orchestration/message-batch-atomicity.test.ts index a9bf6237f55..3c35465a3b4 100644 --- a/src/main/runtime/orchestration/message-batch-atomicity.test.ts +++ b/src/main/runtime/orchestration/message-batch-atomicity.test.ts @@ -91,4 +91,33 @@ describe('message batch atomicity', () => { } expect(first).toEqual({ subject: 'outer change', read: 0 }) }) + + it('preserves an outer transaction when a message insert batch rolls back', () => { + db = new OrchestrationDb(':memory:') + const sqlite = (db as unknown as { db: Database.Database }).db + sqlite.exec(` + CREATE TRIGGER reject_second_message_insert + BEFORE INSERT ON messages WHEN NEW.id = 'inner_second' + BEGIN + SELECT RAISE(ABORT, 'blocked'); + END; + BEGIN IMMEDIATE; + INSERT INTO messages (id, from_handle, to_handle, subject) + VALUES ('outer', 'sender', 'recipient', 'outer change'); + `) + + expect(() => + db?.insertMessages([ + { id: 'inner_first', from: 'sender', to: 'recipient', subject: 'first' }, + { id: 'inner_second', from: 'sender', to: 'recipient', subject: 'second' } + ]) + ).toThrow('blocked') + sqlite.exec('COMMIT') + + expect( + sqlite + .prepare("SELECT id FROM messages WHERE id IN ('outer', 'inner_first') ORDER BY id") + .all() + ).toEqual([{ id: 'outer' }]) + }) }) diff --git a/src/main/runtime/orchestration/orchestration-legacy-storage-db.test.ts b/src/main/runtime/orchestration/orchestration-legacy-storage-db.test.ts index b6889d20c4e..6efdf16f1ff 100644 --- a/src/main/runtime/orchestration/orchestration-legacy-storage-db.test.ts +++ b/src/main/runtime/orchestration/orchestration-legacy-storage-db.test.ts @@ -79,6 +79,8 @@ describe('OrchestrationDb legacy contract storage', () => { scheduler_state_lost: 1 }) expect(db.getRun(adoptedRunId)).toMatchObject({ legacy: 0, consumer_generation: 0 }) + expect(db.getRunMailboxOwnerIdsForHandle('term_legacy_coord')).toEqual([adoptedRunId]) + expect(db.getRunMailboxOwnerIdsForHandle('term_invented')).toEqual([]) expect(db.listTasks({ runId: LEGACY_RUN_ID })).toEqual([]) expect(db.getDispatchContextById(fixture.legacyDispatchId)).toMatchObject({ run_id: adoptedRunId, @@ -146,6 +148,21 @@ describe('OrchestrationDb legacy contract storage', () => { ) }) + it('fails closed when an adopted coordinator handle becomes a current-contract worker', () => { + const state = openAdoptedFixture() + expect(db!.getRunMailboxOwnerIdsForHandle('term_legacy_coord')).toEqual([state.adoptedRunId]) + const task = db!.createTask({ + runId: state.adoptedRunId, + spec: 'mixed contract worker identity', + createdByTerminalHandle: 'term_legacy_coord' + }) + + const dispatch = db!.createDispatchContext(task.id, 'term_legacy_coord', 'tab_mixed:leaf_mixed') + + expect(dispatch.contract_version).toBe(CURRENT_CONTRACT_VERSION) + expect(db!.getRunMailboxOwnerIdsForHandle('term_legacy_coord')).toEqual([]) + }) + it('does not synthesize an adopted Run or compatibility authority for a fresh database', () => { db = new OrchestrationDb(':memory:') diff --git a/src/main/runtime/rpc/errors.ts b/src/main/runtime/rpc/errors.ts index 5d07ace973d..0b083a64d61 100644 --- a/src/main/runtime/rpc/errors.ts +++ b/src/main/runtime/rpc/errors.ts @@ -86,6 +86,9 @@ const STRUCTURED_RUNTIME_PASSTHROUGH_CODES: ReadonlySet = new Set([ 'task_not_startable', 'dispatch_not_found', 'dispatch_run_mismatch', + 'terminal_not_found', + 'recipient_ambiguous', + 'recipient_run_mismatch', 'dispatch_inactive', 'worker_identity_changed', 'cursor_invalid', diff --git a/src/main/runtime/rpc/methods/orchestration-check.test.ts b/src/main/runtime/rpc/methods/orchestration-check.test.ts index ce45a2cfec4..b933bbb0831 100644 --- a/src/main/runtime/rpc/methods/orchestration-check.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-check.test.ts @@ -659,6 +659,9 @@ describe('orchestration RPC methods', () => { it('keeps waiting for requested types when an unrelated status arrives', async () => { setup() + vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) => + handle === 'coord' ? 'tab_wait:leaf_wait' : null + ) const waitPromise = call('orchestration.check', { terminal: 'coord', diff --git a/src/main/runtime/rpc/methods/orchestration-recipient-routing.test.ts b/src/main/runtime/rpc/methods/orchestration-recipient-routing.test.ts new file mode 100644 index 00000000000..99a4c238807 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration-recipient-routing.test.ts @@ -0,0 +1,582 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { ORCHESTRATION_CONTRACT_VERSION } from '../../../../shared/protocol-version' +import type { RuntimeTerminalSummary } from '../../../../shared/runtime-types' +import type { OrchestrationDb } from '../../orchestration/db' +import type { OrcaRuntimeService } from '../../orca-runtime' +import type { RpcContext, RpcRequest } from '../core' +import { RpcDispatcher } from '../dispatcher' +import { ORCHESTRATION_METHODS } from './orchestration' +import { createOrchestrationRpcHarness } from './orchestration-rpc-test-harness' + +type SendWarning = { code: string; recipient: string; message: string } +type SendResult = { + message: { id: string; run_id: string; to_handle: string } + warnings?: SendWarning[] +} +type GroupSendResult = { + messages: { id: string; run_id: string; to_handle: string }[] + recipients: number + warnings?: SendWarning[] +} + +describe('orchestration recipient routing oracle', () => { + const harness = createOrchestrationRpcHarness() + let db: OrchestrationDb + let runtime: OrcaRuntimeService + let ctx: RpcContext + let senderRunId: string + + function setup(): void { + const state = harness.setup() + db = state.db + runtime = state.runtime + ctx = state.ctx + senderRunId = state.activeRunId! + } + + async function call(params: Record): Promise { + return harness.call('orchestration.send', params, ctx) + } + + function mockTerminalPaneKeys(resolve: (handle: string) => string | null): void { + vi.mocked(runtime.getTerminalPaneKey).mockImplementation(resolve) + vi.mocked(runtime.getLiveTerminalPaneKey).mockImplementation(resolve) + } + + afterEach(() => harness.cleanup()) + + it('rejects an unknown terminal without creating an unread row', async () => { + setup() + + await expect( + call({ from: 'term_coord', to: 'term_invented', subject: 'unreachable' }) + ).rejects.toMatchObject({ code: 'terminal_not_found' }) + expect(db.getInbox(100)).toEqual([]) + }) + + it('rejects a closed terminal that has no durable mailbox owner', async () => { + setup() + vi.mocked(runtime.getTerminalProcessIncarnation).mockReturnValue('retained-pty-record') + + await expect( + call({ from: 'term_coord', to: 'term_closed', subject: 'closed' }) + ).rejects.toMatchObject({ code: 'terminal_not_found' }) + expect(db.getInbox(100)).toEqual([]) + }) + + it('keeps a live terminal-only recipient readable and reports its delivery limitation', async () => { + setup() + mockTerminalPaneKeys((handle) => + handle === 'term_coord' + ? harness.coordinatorPaneKey + : handle === 'term_live' + ? 'tab_live:leaf_live' + : null + ) + + const result = (await call({ + from: 'term_coord', + to: 'term_live', + subject: 'compatibility' + })) as SendResult + + expect(result).toMatchObject({ + message: { run_id: senderRunId, to_handle: 'term_live' }, + warnings: [{ code: 'legacy_terminal_recipient', recipient: 'term_live' }] + }) + const check = (await harness.call( + 'orchestration.check', + { terminal: 'term_live', peek: true }, + ctx + )) as { + messages: { id: string }[] + } + expect(check.messages.map((message) => message.id)).toEqual([result.message.id]) + }) + + it('normalizes a cross-Run coordinator handle to the recipient Run mailbox', async () => { + setup() + const recipientPane = 'tab_recipient:leaf_recipient' + const recipientRun = db.createRun({ + objective: 'Recipient Run', + coordinatorHandle: 'term_recipient', + coordinatorPaneKey: recipientPane + }) + mockTerminalPaneKeys((handle) => + handle === 'term_coord' + ? harness.coordinatorPaneKey + : handle === 'term_recipient' + ? recipientPane + : null + ) + + const result = (await call({ + from: 'term_coord', + to: 'term_recipient', + subject: 'cross-run' + })) as SendResult + + expect(result.message).toMatchObject({ + run_id: recipientRun.id, + to_handle: `run:${recipientRun.id}` + }) + const check = (await harness.call( + 'orchestration.check', + { terminal: 'term_recipient', peek: true }, + ctx + )) as { messages: { id: string }[] } + expect(check.messages.map((message) => message.id)).toEqual([result.message.id]) + }) + + it('never lets a stale leaf handle adopt its replacement pane Run', async () => { + setup() + const staleOwner = db.createRun({ + objective: 'Original pane owner', + coordinatorHandle: 'term_stale', + coordinatorPaneKey: 'tab_original:leaf_shared' + }) + const replacement = db.createRun({ + objective: 'Replacement pane owner', + coordinatorHandle: 'term_replacement', + coordinatorPaneKey: 'tab_replacement:leaf_shared' + }) + vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) => + handle === 'term_coord' + ? harness.coordinatorPaneKey + : handle === 'term_stale' || handle === 'term_unowned_stale' + ? 'tab_replacement:leaf_shared' + : null + ) + vi.mocked(runtime.getLiveTerminalPaneKey).mockImplementation((handle) => + handle === 'term_coord' ? harness.coordinatorPaneKey : null + ) + + const routed = (await call({ + from: 'term_coord', + to: 'term_stale', + subject: 'historical owner' + })) as SendResult + + expect(routed.message).toMatchObject({ + run_id: staleOwner.id, + to_handle: `run:${staleOwner.id}` + }) + expect(routed.message.run_id).not.toBe(replacement.id) + await expect( + call({ from: 'term_coord', to: 'term_unowned_stale', subject: 'no owner' }) + ).rejects.toMatchObject({ code: 'terminal_not_found' }) + }) + + it('normalizes an active Dispatch owner even when no pane is live', async () => { + setup() + const task = db.createTask({ spec: 'detached worker' }) + const dispatch = db.createDispatchContext(task.id, 'term_detached', 'tab_gone:leaf_gone') + + const result = (await call({ + from: 'term_coord', + to: 'term_detached', + subject: 'wait durably' + })) as SendResult + + expect(result.message).toMatchObject({ + run_id: senderRunId, + to_handle: `dispatch:${dispatch.id}` + }) + expect(result.warnings).toBeUndefined() + expect(db.getUnreadMessages(`dispatch:${dispatch.id}`)).toHaveLength(1) + }) + + it('reports an explicit Run mismatch for one detached Dispatch owner', async () => { + setup() + const foreignRun = db.createRun({ + objective: 'Foreign worker Run', + coordinatorHandle: 'term_foreign_coord', + coordinatorPaneKey: 'tab_foreign:leaf_coord' + }) + const task = db.createTask({ spec: 'detached foreign worker', runId: foreignRun.id }) + db.createDispatchContext(task.id, 'term_detached_foreign', 'tab_gone:leaf_gone') + + await expect( + call({ + from: 'term_coord', + to: 'term_detached_foreign', + run: senderRunId, + subject: 'wrong Run' + }) + ).rejects.toMatchObject({ code: 'recipient_run_mismatch' }) + expect(db.getInbox(100)).toEqual([]) + }) + + it('matches check precedence when a live Run coordinator pane overlaps a Dispatch', async () => { + setup() + const overlapPane = 'tab_overlap:leaf_overlap' + const task = db.createTask({ spec: 'overlapped worker' }) + db.createDispatchContext(task.id, 'term_overlap', overlapPane) + const recipientRun = db.createRun({ + objective: 'Overlapping coordinator', + coordinatorHandle: 'term_overlap', + coordinatorPaneKey: overlapPane + }) + mockTerminalPaneKeys((handle) => + handle === 'term_coord' + ? harness.coordinatorPaneKey + : handle === 'term_overlap' + ? overlapPane + : null + ) + + const result = (await call({ + from: 'term_coord', + to: 'term_overlap', + subject: 'same read path' + })) as SendResult + + expect(result.message).toMatchObject({ + run_id: recipientRun.id, + to_handle: `run:${recipientRun.id}` + }) + const check = (await harness.call( + 'orchestration.check', + { terminal: 'term_overlap', peek: true }, + ctx + )) as { messages: { id: string }[] } + expect(check.messages.map((message) => message.id)).toEqual([result.message.id]) + }) + + it('keeps same-Run historical coordinator routing from the canonical mailbox change', async () => { + setup() + db.bindRun({ + runId: senderRunId, + coordinatorHandle: 'term_current', + coordinatorPaneKey: 'tab_current:leaf_current' + }) + mockTerminalPaneKeys((handle) => + handle === 'term_current' ? 'tab_current:leaf_current' : null + ) + + const result = (await call({ + from: 'term_current', + to: 'term_coord', + subject: 'historical' + })) as SendResult + + expect(result.message.to_handle).toBe(`run:${senderRunId}`) + expect(result.warnings).toBeUndefined() + }) + + it('rejects a historical handle that names more than one foreign Run', async () => { + setup() + db.createRun({ + objective: 'First owner', + coordinatorHandle: 'term_ambiguous', + coordinatorPaneKey: 'tab_first:leaf_first' + }) + db.createRun({ + objective: 'Second owner', + coordinatorHandle: 'term_ambiguous', + coordinatorPaneKey: 'tab_second:leaf_second' + }) + + await expect( + call({ from: 'term_coord', to: 'term_ambiguous', subject: 'ambiguous' }) + ).rejects.toMatchObject({ code: 'recipient_ambiguous' }) + expect(db.getInbox(100)).toEqual([]) + }) + + it.each(['@all', '@worktree:wt_target'])( + 'partially delivers %s when a listed recipient disappears before routing', + async (address) => { + setup() + vi.spyOn(runtime, 'listTerminals').mockResolvedValue({ + terminals: [terminal('term_coord'), terminal('term_live'), terminal('term_disappeared')], + totalCount: 3, + truncated: false + }) + mockTerminalPaneKeys((handle) => + handle === 'term_coord' + ? harness.coordinatorPaneKey + : handle === 'term_live' + ? 'tab_live:leaf_live' + : null + ) + const adoptionLookup = vi.spyOn(db, 'getLegacyAdoptedRunMailboxOwner') + + const result = (await call({ + from: 'term_coord', + to: address, + subject: 'fan-out' + })) as GroupSendResult + + expect(result.messages).toHaveLength(1) + expect(result.messages[0]).toMatchObject({ to_handle: 'term_live' }) + expect(result.recipients).toBe(1) + expect(result.warnings?.map((warning) => warning.code).sort()).toEqual([ + 'legacy_terminal_recipient', + 'recipient_unreachable' + ]) + expect(db.getInbox(100)).toHaveLength(1) + expect(adoptionLookup).toHaveBeenCalledTimes(1) + } + ) + + it('fans out once when historical handles resolve to the same Run mailbox', async () => { + setup() + const foreignRun = db.createRun({ + objective: 'Foreign Run', + coordinatorHandle: 'term_foreign_first', + coordinatorPaneKey: 'tab_foreign_first:leaf_foreign_first' + }) + db.bindRun({ + runId: foreignRun.id, + coordinatorHandle: 'term_foreign_second', + coordinatorPaneKey: 'tab_foreign_second:leaf_foreign_second' + }) + vi.spyOn(runtime, 'listTerminals').mockResolvedValue({ + terminals: [ + terminal('term_coord'), + terminal('term_foreign_first'), + terminal('term_foreign_second') + ], + totalCount: 3, + truncated: false + }) + mockTerminalPaneKeys((handle) => (handle === 'term_coord' ? harness.coordinatorPaneKey : null)) + + const result = (await call({ + from: 'term_coord', + to: '@all', + subject: 'one mailbox' + })) as GroupSendResult + + expect(result.recipients).toBe(1) + expect(result.messages).toHaveLength(1) + expect(result.messages[0]).toMatchObject({ + run_id: foreignRun.id, + to_handle: `run:${foreignRun.id}` + }) + expect(db.getInbox(100)).toHaveLength(1) + }) + + it('excludes historical handles that resolve back to the sender mailbox', async () => { + setup() + db.bindRun({ + runId: senderRunId, + coordinatorHandle: 'term_middle', + coordinatorPaneKey: 'tab_middle:leaf_middle' + }) + db.bindRun({ + runId: senderRunId, + coordinatorHandle: 'term_sender', + coordinatorPaneKey: 'tab_sender:leaf_sender' + }) + vi.spyOn(runtime, 'listTerminals').mockResolvedValue({ + terminals: [ + terminal('term_sender'), + terminal('term_coord'), + terminal('term_middle'), + terminal('term_live') + ], + totalCount: 4, + truncated: false + }) + mockTerminalPaneKeys((handle) => + handle === 'term_sender' + ? 'tab_sender:leaf_sender' + : handle === 'term_live' + ? 'tab_live:leaf_live' + : null + ) + + const result = (await call({ + from: 'term_sender', + to: '@all', + subject: 'exclude self aliases' + })) as GroupSendResult + + expect(result.messages).toHaveLength(1) + expect(result.messages[0].to_handle).toBe('term_live') + expect(result.warnings).toMatchObject([{ code: 'legacy_terminal_recipient' }]) + }) + + it('replays one honest receipt and discards retry receipts for rejected recipients', async () => { + setup() + mockTerminalPaneKeys((handle) => + handle === 'term_coord' + ? harness.coordinatorPaneKey + : handle === 'term_live' + ? 'tab_live:leaf_live' + : null + ) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + const rejected = request('rpc_reject_1', 'retry_reject', 'term_missing') + + const firstRejected = await dispatcher.dispatch(rejected) + const retriedRejected = await dispatcher.dispatch({ ...rejected, id: 'rpc_reject_2' }) + expect(firstRejected).toMatchObject({ ok: false, error: { code: 'terminal_not_found' } }) + expect(retriedRejected).toMatchObject({ ok: false, error: { code: 'terminal_not_found' } }) + expect(db.getInbox(100)).toEqual([]) + const callerFingerprint = db.getOrCreateLocalMutationCallerFingerprint() + expect(db.getMutationReceipt(callerFingerprint, 'retry_reject')).toBeUndefined() + + const accepted = request('rpc_accept_1', 'retry_accept', 'term_live') + const firstAccepted = await dispatcher.dispatch(accepted) + const replayed = await dispatcher.dispatch({ ...accepted, id: 'rpc_accept_2' }) + expect(firstAccepted).toMatchObject({ + ok: true, + result: { + message: { id: expect.stringMatching(/^msg_/) }, + warnings: [{ code: 'legacy_terminal_recipient' }], + mutation: { requestId: 'retry_accept', replayed: false } + } + }) + expect(replayed).toMatchObject({ + ok: true, + result: { + message: { id: firstAccepted.ok ? (firstAccepted.result as SendResult).message.id : '' }, + warnings: [{ code: 'legacy_terminal_recipient' }], + mutation: { requestId: 'retry_accept', replayed: true } + } + }) + expect(db.getInbox(100)).toHaveLength(1) + expect(db.getMutationReceipt(callerFingerprint, 'retry_accept')).toMatchObject({ + state: 'completed', + receipt: expect.stringContaining('legacy_terminal_recipient') + }) + }) + + it('serializes ambiguity and explicit Run mismatch through the RPC boundary', async () => { + setup() + db.createRun({ + objective: 'First owner', + coordinatorHandle: 'term_ambiguous', + coordinatorPaneKey: 'tab_first:leaf_first' + }) + db.createRun({ + objective: 'Second owner', + coordinatorHandle: 'term_ambiguous', + coordinatorPaneKey: 'tab_second:leaf_second' + }) + const foreignRun = db.createRun({ + objective: 'Foreign owner', + coordinatorHandle: 'term_foreign', + coordinatorPaneKey: 'tab_foreign:leaf_foreign' + }) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + + const ambiguous = await dispatcher.dispatch( + request('rpc_ambiguous', 'retry_ambiguous', 'term_ambiguous') + ) + const mismatch = await dispatcher.dispatch( + request('rpc_mismatch', 'retry_mismatch', 'term_foreign', { run: senderRunId }) + ) + + expect(ambiguous).toMatchObject({ ok: false, error: { code: 'recipient_ambiguous' } }) + expect(mismatch).toMatchObject({ ok: false, error: { code: 'recipient_run_mismatch' } }) + expect(foreignRun.id).not.toBe(senderRunId) + expect(db.getInbox(100)).toEqual([]) + }) + + it('rolls back a partial group insert before an idempotent retry', async () => { + setup() + vi.spyOn(runtime, 'listTerminals').mockResolvedValue({ + terminals: [terminal('term_coord'), terminal('term_first'), terminal('term_second')], + totalCount: 3, + truncated: false + }) + mockTerminalPaneKeys((handle) => + handle === 'term_coord' + ? harness.coordinatorPaneKey + : handle === 'term_first' + ? 'tab_first:leaf_first' + : handle === 'term_second' + ? 'tab_second:leaf_second' + : null + ) + const insertMessage = db.insertMessage.bind(db) + vi.spyOn(db, 'insertMessage') + .mockImplementationOnce(insertMessage) + .mockImplementationOnce(() => { + throw new Error('injected second insert failure') + }) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + const group = request('rpc_group_partial', 'retry_group_partial', '@all') + + const failed = await dispatcher.dispatch(group) + expect(failed.ok).toBe(false) + expect(db.getInbox(100)).toEqual([]) + + const retried = await dispatcher.dispatch({ ...group, id: 'rpc_group_partial_retry' }) + expect(retried).toMatchObject({ + ok: true, + result: { messages: [{ id: expect.any(String) }, { id: expect.any(String) }] } + }) + expect(db.getInbox(100)).toHaveLength(2) + }) + + it('replays a completed group receipt when notification fails after durable insertion', async () => { + setup() + vi.spyOn(runtime, 'listTerminals').mockResolvedValue({ + terminals: [terminal('term_coord'), terminal('term_live')], + totalCount: 2, + truncated: false + }) + mockTerminalPaneKeys((handle) => + handle === 'term_coord' + ? harness.coordinatorPaneKey + : handle === 'term_live' + ? 'tab_live:leaf_live' + : null + ) + vi.spyOn(runtime, 'notifyMessageArrived').mockImplementationOnce(() => { + throw new Error('injected notification failure') + }) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + const group = request('rpc_group_notify', 'retry_group_notify', '@all') + + const failed = await dispatcher.dispatch(group) + const retried = await dispatcher.dispatch({ ...group, id: 'rpc_group_notify_retry' }) + + expect(failed.ok).toBe(false) + expect(retried).toMatchObject({ + ok: true, + result: { + messages: [{ id: expect.any(String) }], + mutation: { requestId: 'retry_group_notify', replayed: true } + } + }) + expect(db.getInbox(100)).toHaveLength(1) + }) +}) + +function terminal(handle: string): RuntimeTerminalSummary { + return { + handle, + ptyId: handle, + worktreeId: 'wt_target', + worktreePath: '/workspace', + branch: 'main', + tabId: `tab_${handle}`, + leafId: `leaf_${handle}`, + title: null, + connected: true, + writable: true, + lastOutputAt: null, + preview: '' + } +} + +function request( + id: string, + requestId: string, + to: string, + extraParams: Record = {} +): RpcRequest { + return { + id, + authToken: 'test-token', + method: 'orchestration.send', + params: { from: 'term_coord', to, subject: 'retry', ...extraParams }, + orchestrationContractVersion: ORCHESTRATION_CONTRACT_VERSION, + orchestrationRequestId: requestId + } +} diff --git a/src/main/runtime/rpc/methods/orchestration-recipient-routing.ts b/src/main/runtime/rpc/methods/orchestration-recipient-routing.ts new file mode 100644 index 00000000000..1de1151d8a2 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration-recipient-routing.ts @@ -0,0 +1,145 @@ +import type { LegacyAdoptedMailboxOwner, OrchestrationDb } from '../../orchestration/db' +import type { DispatchContextRow } from '../../orchestration/types' +import type { OrcaRuntimeService } from '../../orca-runtime' + +export type SendRecipientWarning = { + code: + | 'legacy_terminal_recipient' + | 'recipient_unreachable' + | 'recipient_ambiguous' + | 'recipient_run_mismatch' + recipient: string + message: string +} + +export type BareRecipientResolution = + | { + ok: true + to: string + runId: string | undefined + warning?: SendRecipientWarning + } + | { + ok: false + code: 'terminal_not_found' | 'recipient_ambiguous' | 'recipient_run_mismatch' + message: string + warning: SendRecipientWarning + } + +export function resolveBareOrchestrationRecipient(params: { + runtime: OrcaRuntimeService + db: OrchestrationDb + handle: string + senderRunId?: string + explicitRunId?: string + legacyAdoptedMailboxOwner?: LegacyAdoptedMailboxOwner | null +}): BareRecipientResolution { + const { runtime, db, handle } = params + const paneKey = runtime.getLiveTerminalPaneKey(handle) ?? undefined + const boundRun = paneKey ? db.getCurrentRunForPane(paneKey) : undefined + if (boundRun) { + const mismatch = runMismatch(handle, boundRun.id, params.explicitRunId) + return mismatch ?? { ok: true, to: `run:${boundRun.id}`, runId: boundRun.id } + } + + const dispatches = db.getActiveDispatchMailboxOwners(handle, paneKey) + const dispatch = selectDispatch(dispatches, params.explicitRunId) + if (dispatches.length > 0 && !dispatch) { + return ambiguous( + handle, + dispatches.map((candidate) => `dispatch:${candidate.id}`) + ) + } + if (dispatch) { + const mismatch = runMismatch(handle, dispatch.run_id, params.explicitRunId) + return mismatch ?? { ok: true, to: `dispatch:${dispatch.id}`, runId: dispatch.run_id } + } + + const ownerRunIds = db.getRunMailboxOwnerIdsForHandle(handle, params.legacyAdoptedMailboxOwner) + const selectedRunId = selectHistoricalRun(ownerRunIds, params) + if (ownerRunIds.length > 0 && !selectedRunId) { + return ambiguous( + handle, + ownerRunIds.map((runId) => `run:${runId}`) + ) + } + if (selectedRunId) { + const mismatch = runMismatch(handle, selectedRunId, params.explicitRunId) + return mismatch ?? { ok: true, to: `run:${selectedRunId}`, runId: selectedRunId } + } + + if (paneKey) { + return { + ok: true, + to: handle, + runId: params.senderRunId, + warning: { + code: 'legacy_terminal_recipient', + recipient: handle, + message: `${handle} is a live terminal-only mailbox. Delivery is not durable after that terminal closes; prefer run: or dispatch:.` + } + } + } + + const message = `Terminal ${handle} has no live pane or durable Run/Dispatch mailbox.` + return { + ok: false, + code: 'terminal_not_found', + message, + warning: { code: 'recipient_unreachable', recipient: handle, message } + } +} + +function selectDispatch( + dispatches: DispatchContextRow[], + explicitRunId: string | undefined +): DispatchContextRow | undefined { + if (dispatches.length === 1) { + return dispatches[0] + } + if (!explicitRunId) { + return undefined + } + const matches = dispatches.filter((dispatch) => dispatch.run_id === explicitRunId) + return matches.length === 1 ? matches[0] : undefined +} + +function selectHistoricalRun( + ownerRunIds: string[], + params: { senderRunId?: string; explicitRunId?: string } +): string | undefined { + if (params.explicitRunId && ownerRunIds.includes(params.explicitRunId)) { + return params.explicitRunId + } + if (params.senderRunId && ownerRunIds.includes(params.senderRunId)) { + return params.senderRunId + } + return ownerRunIds.length === 1 ? ownerRunIds[0] : undefined +} + +function ambiguous(handle: string, addresses: string[]): BareRecipientResolution { + const message = `${handle} resolves to multiple durable mailboxes (${addresses.join(', ')}). Use an explicit canonical address.` + return { + ok: false, + code: 'recipient_ambiguous', + message, + warning: { code: 'recipient_ambiguous', recipient: handle, message } + } +} + +function runMismatch( + handle: string, + resolvedRunId: string, + explicitRunId: string | undefined +): BareRecipientResolution | undefined { + if (!explicitRunId || explicitRunId === resolvedRunId) { + return undefined + } + const message = `${handle} belongs to Run ${resolvedRunId}, not explicitly requested Run ${explicitRunId}.` + return { + ok: false, + code: 'recipient_run_mismatch', + message, + warning: { code: 'recipient_run_mismatch', recipient: handle, message } + } +} diff --git a/src/main/runtime/rpc/methods/orchestration-rpc-test-harness.ts b/src/main/runtime/rpc/methods/orchestration-rpc-test-harness.ts index c08a335548a..b0a77bfa29f 100644 --- a/src/main/runtime/rpc/methods/orchestration-rpc-test-harness.ts +++ b/src/main/runtime/rpc/methods/orchestration-rpc-test-harness.ts @@ -30,6 +30,9 @@ export function createOrchestrationRpcHarness() { vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) => handle === 'term_coord' ? coordinatorPaneKey : null ) + vi.spyOn(runtime, 'getLiveTerminalPaneKey').mockImplementation((handle) => + runtime.getTerminalPaneKey(handle) + ) vi.spyOn(runtime, 'getTerminalProcessIncarnation').mockImplementation((handle) => handle.startsWith('term_') ? `runtime_test:${handle}:1` : null ) diff --git a/src/main/runtime/rpc/methods/orchestration-send.test.ts b/src/main/runtime/rpc/methods/orchestration-send.test.ts index 1169e7b4e1c..dafcbcc4327 100644 --- a/src/main/runtime/rpc/methods/orchestration-send.test.ts +++ b/src/main/runtime/rpc/methods/orchestration-send.test.ts @@ -601,6 +601,13 @@ describe('orchestration RPC methods', () => { totalCount: terminals.length, truncated: false }) + vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) => { + if (handle === 'term_coord') { + return coordinatorPaneKey + } + const terminal = terminals.find((candidate) => candidate.handle === handle) + return terminal ? `${terminal.tabId}:${terminal.leafId}` : null + }) vi.spyOn(runtime, 'getAgentStatusForHandle').mockImplementation( (handle: string) => agentStatuses?.[handle] ?? null ) diff --git a/src/main/runtime/rpc/methods/orchestration.ts b/src/main/runtime/rpc/methods/orchestration.ts index 908673ec7a5..30e0b516129 100644 --- a/src/main/runtime/rpc/methods/orchestration.ts +++ b/src/main/runtime/rpc/methods/orchestration.ts @@ -23,6 +23,10 @@ import { } from '../../../../shared/orchestration-rpc-contract' import { clampOrchestrationAskTimeoutMs } from '../../../../shared/orchestration-ask-timeout' import { ORCHESTRATION_GATE_METHODS } from './orchestration-gates' +import { + resolveBareOrchestrationRecipient, + type SendRecipientWarning +} from './orchestration-recipient-routing' import { resolveRunScope } from './orchestration-run-scope' import { ORCHESTRATION_RUN_METHODS } from './orchestration-runs' import { ORCHESTRATION_WORKER_METHODS } from './orchestration-worker-methods' @@ -445,6 +449,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ legacyCoordinatorRunId, revalidateLegacyCoordinator, orchestrationCompatibilityCallerAuthority, + recordMutationReceipt, signal } ) => { @@ -592,8 +597,35 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ ) } + const sendWarnings: SendRecipientWarning[] = [] + let messageRunId = routing.run?.id + if (!isGroupAddress(to) && !to.startsWith('run:') && !to.startsWith('dispatch:')) { + const recipient = resolveBareOrchestrationRecipient({ + runtime, + db, + handle: to, + senderRunId: routing.run?.id, + explicitRunId: params.run + }) + if (!recipient.ok) { + throw new OrchestrationError(recipient.code, recipient.message) + } + to = recipient.to + messageRunId = recipient.runId + if (recipient.warning) { + sendWarnings.push(recipient.warning) + } + } + const withSendWarnings = ( + receipt: T + ): T & { + warnings?: SendRecipientWarning[] + } => (sendWarnings.length > 0 ? { ...receipt, warnings: sendWarnings } : receipt) + if (!isGroupAddress(to)) { - const federatedDispatchId = routing.dispatchId + const federatedDispatchId = to.startsWith('dispatch:') + ? to.slice('dispatch:'.length) + : undefined const federatedTarget = federatedDispatchId && to === `dispatch:${federatedDispatchId}` ? db.getFederatedDispatch(federatedDispatchId) @@ -636,8 +668,8 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ payload: params.payload ?? null }) }) - runtime.ensureOrchestrationFederationRelay(routing.run?.id) - return { + runtime.ensureOrchestrationFederationRelay(messageRunId) + return withSendWarnings({ relay: { messageId: relay.message_id, sequence: relay.sequence, @@ -645,7 +677,7 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ destination: 'worker', accepted: true } - } + }) } // Point-to-point — existing single-recipient behavior revalidateLegacyCoordinator?.() @@ -665,10 +697,10 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ ? bindCoordinatorMutationPayload(messageType, params.payload, dispatch.id) : params.payload, senderPaneKey, - runId: routing.run?.id, + runId: messageRunId, deliveryContract: legacyWorkerDeliveryContract( runtime, - routing.run?.id ?? legacyCoordinatorRunId, + messageRunId ?? legacyCoordinatorRunId, to ) }) @@ -738,14 +770,14 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ const rejection = db.convertLifecycleMessageToRejection(msg.id, code, authority.reason) ?? msg runtime.notifyMessageArrived(rejection.to_handle, rejection.type) - return { + return withSendWarnings({ message: rejection, lifecycle: { action: 'rejected', code, reason: authority.reason } - } + }) } } // Why: reconcile releases the dispatch lock before waking recipients, else a woken coordinator re-dispatches while the lock is still held. @@ -753,20 +785,20 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ const reconciled = reconcileLifecycleMessage(db, msg) // Why: a suppressed message is already read, so skip the notify that would wake a check --wait waiter to an empty result. if (reconciled.action === 'suppressed') { - return { message: msg } + return withSendWarnings({ message: msg }) } if (reconciled.action === 'rejected') { const rejection = db.getMessageById(msg.id) ?? msg runtime.notifyMessageArrived(rejection.to_handle, rejection.type) - return { message: rejection, lifecycle: reconciled } + return withSendWarnings({ message: rejection, lifecycle: reconciled }) } runtime.notifyMessageArrived(msg.to_handle, msg.type) - return msg.type === 'worker_done' - ? { message: msg, lifecycle: reconciled } - : { message: msg } + return withSendWarnings( + msg.type === 'worker_done' ? { message: msg, lifecycle: reconciled } : { message: msg } + ) } runtime.notifyMessageArrived(msg.to_handle, msg.type) - return { message: msg } + return withSendWarnings({ message: msg }) } // Why: fan out one message per recipient (independent read-tracking) but share a thread_id for correlation (Section 4.5). @@ -781,12 +813,57 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ throw new Error(`No recipients resolved for group address: ${to}`) } + const legacyAdoptedMailboxOwner = db.getLegacyAdoptedRunMailboxOwner() + const resolvedRecipients = handles.map((handle) => ({ + handle, + resolution: resolveBareOrchestrationRecipient({ + runtime, + db, + handle, + senderRunId: routing.run?.id, + explicitRunId: params.run, + legacyAdoptedMailboxOwner + }) + })) + const deliverableRecipients = resolvedRecipients.filter( + ( + recipient + ): recipient is typeof recipient & { + resolution: { ok: true; to: string; runId?: string; warning?: SendRecipientWarning } + } => recipient.resolution.ok + ) + const senderRecipient = resolveBareOrchestrationRecipient({ + runtime, + db, + handle: from, + senderRunId: routing.run?.id, + legacyAdoptedMailboxOwner + }) + const senderMailboxKey = senderRecipient.ok + ? `${senderRecipient.runId ?? ''}\u0000${senderRecipient.to}` + : undefined + const seenMailboxes = new Set() + const uniqueRecipients = deliverableRecipients.filter(({ resolution }) => { + const mailboxKey = `${resolution.runId ?? ''}\u0000${resolution.to}` + if (mailboxKey === senderMailboxKey || seenMailboxes.has(mailboxKey)) { + return false + } + seenMailboxes.add(mailboxKey) + return true + }) + if (uniqueRecipients.length === 0) { + throw new OrchestrationError( + 'terminal_not_found', + `No recipient of ${to} resolved to a live terminal or durable Run/Dispatch mailbox.` + ) + } + revalidateLegacyCoordinator?.() const threadId = params.threadId ?? `thread_${Date.now()}` - const messages = handles.map((handle) => - db.insertMessage({ + const messages = db.insertMessages( + uniqueRecipients.map(({ resolution }) => ({ from, - to: handle, + to: resolution.to, subject: params.subject, body: params.body, type: params.type as MessageType, @@ -794,19 +871,27 @@ export const ORCHESTRATION_METHODS: RpcMethod[] = [ threadId, payload: params.payload, senderPaneKey, - runId: routing.run?.id, + runId: resolution.runId, deliveryContract: legacyWorkerDeliveryContract( runtime, - routing.run?.id ?? legacyCoordinatorRunId, - handle + resolution.runId ?? legacyCoordinatorRunId, + resolution.to ) - }) + })) ) + const groupWarnings = resolvedRecipients.flatMap(({ resolution }) => + resolution.ok ? (resolution.warning ? [resolution.warning] : []) : [resolution.warning] + ) + const receipt = { + messages, + recipients: messages.length, + ...(groupWarnings.length > 0 ? { warnings: groupWarnings } : {}) + } + recordMutationReceipt?.(receipt) for (const message of messages) { runtime.notifyMessageArrived(message.to_handle, message.type) } - - return { messages, recipients: handles.length } + return receipt } }), diff --git a/src/main/runtime/rpc/orchestration-legacy-coordinator-race.test.ts b/src/main/runtime/rpc/orchestration-legacy-coordinator-race.test.ts index 468f261b725..4b8bb388226 100644 --- a/src/main/runtime/rpc/orchestration-legacy-coordinator-race.test.ts +++ b/src/main/runtime/rpc/orchestration-legacy-coordinator-race.test.ts @@ -68,6 +68,9 @@ function createHarness(): Harness { vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) => handle === COORDINATOR_HANDLE ? COORDINATOR_PANE : handle === WORKER_HANDLE ? WORKER_PANE : null ) + vi.spyOn(runtime, 'getLiveTerminalPaneKey').mockImplementation((handle) => + runtime.getTerminalPaneKey(handle) + ) vi.spyOn(runtime, 'verifyOrchestrationCompatibilityCaller').mockImplementation( (compatibilityEvidence) => { const coordinator = @@ -199,7 +202,7 @@ describe('legacy coordinator takeover races', () => { result: { message: { run_id: harness.adoptedRunId, - to_handle: WORKER_HANDLE, + to_handle: `dispatch:${harness.dispatchId}`, delivery_contract: 'legacy_direct' } } @@ -310,6 +313,15 @@ describe('legacy coordinator takeover races', () => { it('partitions a coordinator group send by legacy recipient contract', async () => { const harness = createHarness() + vi.mocked(harness.runtime.getTerminalPaneKey).mockImplementation((handle) => + handle === COORDINATOR_HANDLE + ? COORDINATOR_PANE + : handle === WORKER_HANDLE + ? WORKER_PANE + : handle === 'term_current_worker' + ? 'tab_current_worker:leaf_current_worker' + : null + ) vi.spyOn(harness.runtime, 'listTerminals').mockResolvedValue({ terminals: [ { handle: COORDINATOR_HANDLE }, @@ -336,7 +348,7 @@ describe('legacy coordinator takeover races', () => { expect.arrayContaining([ expect.objectContaining({ run_id: harness.adoptedRunId, - to_handle: WORKER_HANDLE, + to_handle: `dispatch:${harness.dispatchId}`, delivery_contract: 'legacy_direct' }), expect.objectContaining({ diff --git a/src/main/runtime/rpc/orchestration-runtime-update-settlement.test.ts b/src/main/runtime/rpc/orchestration-runtime-update-settlement.test.ts index 57ac1b014ff..4dc8cd892b1 100644 --- a/src/main/runtime/rpc/orchestration-runtime-update-settlement.test.ts +++ b/src/main/runtime/rpc/orchestration-runtime-update-settlement.test.ts @@ -192,7 +192,7 @@ function entityCounts(db: OrchestrationDb): Record { } function resultOf(response: RpcResponse): Record { - expect(response.ok).toBe(true) + expect(response.ok, JSON.stringify(response)).toBe(true) if (!response.ok) { throw new Error(response.error.message) } @@ -352,6 +352,9 @@ describe('orchestration runtime update settlement', () => { it('routes ordinary mail with the same attested authority without settling work', async () => { const harness = createUpdateHarness() + expect(harness.db.getRunMailboxOwnerIdsForHandle(COORDINATOR_HANDLE)).toEqual([ + harness.adoptedRunId + ]) const response = await harness.createDispatcher().dispatch( request( 'orchestration.send', diff --git a/src/main/ssh/ssh-remote-orca-cli.test.ts b/src/main/ssh/ssh-remote-orca-cli.test.ts index 27947c1d052..e9cb35e352b 100644 --- a/src/main/ssh/ssh-remote-orca-cli.test.ts +++ b/src/main/ssh/ssh-remote-orca-cli.test.ts @@ -81,7 +81,9 @@ describe('runRemoteOrcaCli', () => { }), getLegacyAdoption: vi.fn(() => undefined), getActiveDispatchForIdentity: vi.fn(() => undefined), + getActiveDispatchMailboxOwners: vi.fn(() => []), getCurrentRunForPane: vi.fn(() => undefined), + getRunMailboxOwnerIdsForHandle: vi.fn(() => []), findActiveRemoteAttachmentForPane: vi.fn(() => undefined) } const runtime = { @@ -96,6 +98,8 @@ describe('runRemoteOrcaCli', () => { }), getOrchestrationDb: () => db, getTerminalPaneKey: () => null, + getLiveTerminalPaneKey: (handle: string) => + handle === 'term_windows' ? 'tab_windows:leaf_windows' : null, deliverPendingMessagesForHandle: vi.fn(), notifyMessageArrived: vi.fn(), linearIssueContext: vi.fn(async (request: unknown) => ({ @@ -188,9 +192,18 @@ describe('runRemoteOrcaCli', () => { LEGACY_FALLBACK_OPTIONS ) - expect(result.exitCode).toBe(0) - const payload = JSON.parse(result.stdout) as { ok: boolean } - expect(payload.ok).toBe(true) + expect(result.exitCode, result.stdout).toBe(0) + expect(JSON.parse(result.stdout)).toMatchObject({ + ok: true, + result: { + warnings: [ + { + code: 'legacy_terminal_recipient', + recipient: 'term_windows' + } + ] + } + }) expect(db.getUnreadMessages('term_windows')[0]?.from_handle).toBe('term_ssh') }) @@ -210,7 +223,7 @@ describe('runRemoteOrcaCli', () => { LEGACY_FALLBACK_OPTIONS ) - expect(result.exitCode).toBe(0) + expect(result.exitCode, result.stdout).toBe(0) expect(db.insertMessage).toHaveBeenCalledWith( expect.objectContaining({ senderPaneKey: undefined }) ) @@ -481,7 +494,7 @@ describe('runRemoteOrcaCli', () => { LEGACY_FALLBACK_OPTIONS ) - expect(result.exitCode).toBe(0) + expect(result.exitCode, result.stdout).toBe(0) const payload = JSON.parse(result.stdout) as { ok: boolean } expect(payload.ok).toBe(true) const message = db.getUnreadMessages('term_windows')[0]