diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-methods.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-methods.ts index ee52efed6b9..b9aa437e179 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/check-methods.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-methods.ts @@ -89,7 +89,8 @@ export const ORCHESTRATION_CHECK_METHODS = [ typeFilter, signal, activeDispatch, - remoteAttachment + remoteAttachment, + recordMutationReceipt }) } const consumingCheck = params.peek !== true && params.all !== true && params.unread !== false diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-run-pending-mail.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-run-pending-mail.ts new file mode 100644 index 00000000000..5d453616f35 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-run-pending-mail.ts @@ -0,0 +1,140 @@ +import type { MessageType, OrchestrationDb, RunRow } from '../../../../orchestration/db' +import type { OrcaRuntimeService } from '../../../../orca-runtime' +import { interruptedAcknowledgedCheck } from '../routing' +import { checkWorkerMailbox } from './check-worker' +import { currentDispatchAssigneeRun } from './recipient-routing' +import { callerHoldsDispatchPane, dispatchFenced } from './dispatch-mailbox-fence' +import { orchestrationCallerIdentity } from '../runs/run-scope' +import type { OrchestrationSessionCaller } from '../../../../orchestration/orchestration-caller-identity' +import type { CheckParams } from '../schemas' +import type { z } from 'zod' + +export async function checkRunPendingMail(args: { + params: z.infer + runtime: OrcaRuntimeService + db: OrchestrationDb + run: RunRow + handle: string + paneKey: string | undefined + callerSession: OrchestrationSessionCaller | undefined + typeFilter: MessageType[] | undefined + signal: AbortSignal | undefined + revalidateConsumer: () => void + recordMutationReceipt: ((receipt: unknown) => void) | undefined +}): Promise<{ acknowledged?: string; result?: unknown }> { + const { + params, + runtime, + db, + run, + handle, + paneKey, + typeFilter, + signal, + revalidateConsumer, + recordMutationReceipt + } = args + const generation = run.consumer_generation + const address = `run:${run.id}` + // Drain pre-bind mail through its original Dispatch owner, never by changing its Run. + const residual = !params.run ? db.getActiveDispatchForIdentity(handle, paneKey) : undefined + const caller = orchestrationCallerIdentity(runtime, { + handle, + paneKey, + session: args.callerSession + }) + const residualAck = + params.ack && + residual && + db.getDeliveryRaw(params.ack)?.mailbox_handle === `dispatch:${residual.id}` + ? params.ack + : undefined + const acknowledgeRun = () => + params.ack + ? db.acknowledgeRunDelivery({ + runId: run.id, + consumerGeneration: generation, + deliveryId: params.ack + }) + : undefined + let acknowledged: { delivery: { id: string } } | undefined = residualAck + ? undefined + : acknowledgeRun() + const recordAcknowledged = () => { + if (acknowledged) { + recordMutationReceipt?.( + interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'outcome_unknown') + ) + } + } + recordAcknowledged() + if ( + residual && + residual.run_id !== run.id && + (residual.assignee_orca_session_id === null || + residual.assignee_orca_session_id === caller.orcaSessionId) && + callerHoldsDispatchPane(residual, paneKey) && + currentDispatchAssigneeRun(runtime, db, residual)?.id === run.id && + (residualAck || !db.hasOutstandingMailboxDelivery(address)) + ) { + const result = await checkWorkerMailbox({ + params: { ...params, ack: residualAck, wait: false }, + runtime, + db, + handle, + paneKey, + typeFilter, + signal, + activeDispatch: residual, + remoteAttachment: undefined, + wakeTypes: params.wait ? typeFilter : undefined, + // Accept the original owner's ack without creating a batch ahead of Run replay. + deferDelivery: () => db.hasOutstandingMailboxDelivery(address), + revalidateConsumer: () => { + revalidateConsumer() + const current = db.getActiveDispatchForIdentity(handle, paneKey) + if ( + !current || + current.id !== residual.id || + currentDispatchAssigneeRun(runtime, db, current)?.id !== run.id + ) { + throw dispatchFenced() + } + }, + recordMutationReceipt + }) + if (result?.acknowledged) { + acknowledged = { delivery: { id: result.acknowledged } } + } + recordAcknowledged() + try { + revalidateConsumer() + } catch (error) { + if (acknowledged) { + return { + acknowledged: acknowledged.delivery.id, + result: interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'consumer_fenced') + } + } + throw error + } + const inspectingHistory = + params.all === true || (params.unread === false && params.peek !== true) + if ( + result && + result.count > 0 && + (!inspectingHistory || db.getUnreadMessages(`dispatch:${residual.id}`).length > 0) + ) { + return { + acknowledged: acknowledged?.delivery.id, + result: { ...result, acknowledged: acknowledged?.delivery.id ?? null } + } + } + } + + // A supplied Delivery outside this caller's current Dispatch must still fail acknowledgment. + if (params.ack && !acknowledged) { + acknowledgeRun() + } + return { acknowledged: acknowledged?.delivery.id } +} diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-run.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-run.ts index 248d79ea68e..c48141cf7dd 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/check-run.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-run.ts @@ -10,6 +10,7 @@ import { resolveRunScope } from '../runs/run-scope' import type { CheckParams } from '../schemas' import type { z } from 'zod' import type { OrchestrationSessionCaller } from '../../../../orchestration/orchestration-caller-identity' +import { checkRunPendingMail } from './check-run-pending-mail' type CheckParamsInput = z.infer @@ -38,8 +39,7 @@ export async function checkRunMailbox(args: { signal, legacyCoordinatorRunId, revalidateLegacyCoordinator, - orchestrationCompatibilityEvidence, - recordMutationReceipt + orchestrationCompatibilityEvidence } = args const routeDirectSnapshot = async ( runId: string, @@ -72,41 +72,45 @@ export async function checkRunMailbox(args: { db.routeUnreadDirectMessagesToRunMailbox(run.id, coordinatorHandle, throughSequence) ) } - revalidateLegacyCoordinator?.() - const currentRun = resolveRunScope(runtime, { - runId: run.id, - callerTerminalHandle: handle, - callerPaneKey: paneKey, - callerSession, - requireCurrentConsumer: true, - legacyCoordinatorRunId, - callerEvidence: orchestrationCompatibilityEvidence - }) - if (currentRun.consumer_generation !== generation) { - throw new OrchestrationError( - 'consumer_fenced', - 'This mailbox consumer was replaced while routing pending mail.' - ) + const revalidateConsumer = (): void => { + revalidateLegacyCoordinator?.() + const currentRun = resolveRunScope(runtime, { + runId: run.id, + callerTerminalHandle: handle, + callerPaneKey: paneKey, + callerSession, + requireCurrentConsumer: true, + legacyCoordinatorRunId, + callerEvidence: orchestrationCompatibilityEvidence + }) + if (currentRun.consumer_generation !== generation) { + throw new OrchestrationError( + 'consumer_fenced', + 'This mailbox consumer was replaced while routing pending mail.' + ) + } } + revalidateConsumer() - const acknowledged = params.ack - ? db.acknowledgeRunDelivery({ - runId: run.id, - consumerGeneration: generation, - deliveryId: params.ack - }) - : undefined - if (acknowledged) { - recordMutationReceipt?.( - interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'outcome_unknown') - ) + const pending = await checkRunPendingMail({ ...args, run, revalidateConsumer }) + try { + revalidateConsumer() + } catch (error) { + if (pending.acknowledged) { + return interruptedAcknowledgedCheck(run.id, pending.acknowledged, 'consumer_fenced') + } + throw error } + if (pending.result) { + return pending.result + } + const acknowledged = pending.acknowledged if (params.all || (params.unread === false && !params.peek)) { const messages = db.getRunMailboxHistory(run.id, 100, typeFilter) const result = { messages: exposeMessages(messages), count: messages.length, - acknowledged: acknowledged?.delivery.id ?? null + acknowledged: acknowledged ?? null } if (params.format || params.inject) { return { @@ -122,7 +126,7 @@ export async function checkRunMailbox(args: { runId: run.id, messages: exposeMessages(messages), count: messages.length, - acknowledged: acknowledged?.delivery.id ?? null, + acknowledged: acknowledged ?? null, ...(params.format || params.inject ? { formatted: messages.map(formatMessageBanner).join('\n\n') } : {}) @@ -142,7 +146,7 @@ export async function checkRunMailbox(args: { messages: exposeMessages(current.messages), count: current.messages.length, replayed: current.replayed, - acknowledged: acknowledged?.delivery.id ?? null, + acknowledged: acknowledged ?? null, timedOut: false, cancelled: false, connectionLost: false, @@ -160,7 +164,7 @@ export async function checkRunMailbox(args: { deliveryId: null, messages: [], count: 0, - acknowledged: acknowledged?.delivery.id ?? null, + acknowledged: acknowledged ?? null, timedOut: false, cancelled: false, connectionLost: false @@ -174,17 +178,17 @@ export async function checkRunMailbox(args: { exclusive: true }) try { - revalidateLegacyCoordinator?.() + revalidateConsumer() } catch (error) { if (!acknowledged) { throw error } - return interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'consumer_fenced') + return interruptedAcknowledgedCheck(run.id, acknowledged, 'consumer_fenced') } const latestRun = db.getRun(run.id) if (!latestRun || latestRun.consumer_generation !== generation) { if (acknowledged) { - return interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'consumer_fenced') + return interruptedAcknowledgedCheck(run.id, acknowledged, 'consumer_fenced') } throw new OrchestrationError( 'consumer_fenced', @@ -193,7 +197,7 @@ export async function checkRunMailbox(args: { } if (waitResult === 'waiter_exists') { if (acknowledged) { - return interruptedAcknowledgedCheck(run.id, acknowledged.delivery.id, 'waiter_exists') + return interruptedAcknowledgedCheck(run.id, acknowledged, 'waiter_exists') } throw new OrchestrationError( 'waiter_exists', @@ -209,7 +213,7 @@ export async function checkRunMailbox(args: { deliveryId: null, messages: [], count: 0, - acknowledged: acknowledged?.delivery.id ?? null, + acknowledged: acknowledged ?? null, timedOut: true, cancelled: false, connectionLost: false @@ -229,7 +233,7 @@ export async function checkRunMailbox(args: { deliveryId: null, messages: [], count: 0, - acknowledged: acknowledged?.delivery.id ?? null, + acknowledged: acknowledged ?? null, timedOut: false, cancelled: true, connectionLost: signal?.aborted === true @@ -246,7 +250,7 @@ export async function checkRunMailbox(args: { messages: exposeMessages(current?.messages ?? []), count: current?.messages.length ?? 0, replayed: current?.replayed ?? false, - acknowledged: acknowledged?.delivery.id ?? null, + acknowledged: acknowledged ?? null, timedOut: false, cancelled: false, connectionLost: false, diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/check-worker.ts b/src/main/runtime/rpc/methods/orchestration/messaging/check-worker.ts index fc9fa6b04e9..7616f3e5cc1 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/check-worker.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/check-worker.ts @@ -5,6 +5,8 @@ import { formatMessageBanner } from '../../../../orchestration/formatter' import { exposeMessages } from './mailbox-message-receipt' import { routeAllMailboxPages } from '../schemas' import { asDispatchFence, callerHoldsDispatchPane, dispatchFenced } from './dispatch-mailbox-fence' +import { interruptedAcknowledgedCheck } from '../routing' +import { currentDispatchAssigneeRun } from './recipient-routing' import type { CheckParams } from '../schemas' import type { z } from 'zod' @@ -24,7 +26,11 @@ export async function checkWorkerMailbox(args: { signal: AbortSignal | undefined activeDispatch: ActiveDispatch | undefined remoteAttachment: RemoteAttachment | undefined -}): Promise { + wakeTypes?: MessageType[] + revalidateConsumer?: () => void + deferDelivery?: () => boolean + recordMutationReceipt?: (receipt: unknown) => void +}) { const { params, runtime, @@ -54,6 +60,7 @@ export async function checkWorkerMailbox(args: { } const deliveryRunId = workerMailbox.runId db.requireRun(deliveryRunId) + const mailboxIdentity = { runId: deliveryRunId, dispatchId: workerMailbox.dispatchId } const address = `dispatch:${workerMailbox.dispatchId}` // Why: a federated worker host has no dispatch_contexts row, so its generation lives on the // remote_dispatch_attachments row instead. @@ -165,6 +172,7 @@ export async function checkWorkerMailbox(args: { } } await revalidateWorkerMailbox() + args.revalidateConsumer?.() let acknowledged try { acknowledged = params.ack @@ -179,9 +187,17 @@ export async function checkWorkerMailbox(args: { } catch (error) { throw asDispatchFence(error) } + if (acknowledged) { + args.recordMutationReceipt?.( + interruptedAcknowledgedCheck(deliveryRunId, acknowledged.delivery.id, 'outcome_unknown') + ) + } const showAll = params.all === true || (params.unread === false && params.peek !== true) const readPeek = () => db.getUnreadMessages(address, typeFilter) const readDelivery = (wakeTypes?: MessageType[]) => { + if (args.deferDelivery?.()) { + return undefined + } try { return db.getOrCreateMailboxDelivery({ runId: deliveryRunId, @@ -197,8 +213,7 @@ export async function checkWorkerMailbox(args: { if (showAll) { const messages = db.getAllMessagesForHandle(address, 100, typeFilter) return { - ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}), - dispatchId: workerMailbox.dispatchId, + ...mailboxIdentity, messages: exposeMessages(messages), count: messages.length, acknowledged: acknowledged?.delivery.id ?? null, @@ -211,8 +226,7 @@ export async function checkWorkerMailbox(args: { const messages = readPeek() if (messages.length > 0 || !params.wait) { return { - ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}), - dispatchId: workerMailbox.dispatchId, + ...mailboxIdentity, messages: exposeMessages(messages), count: messages.length, acknowledged: acknowledged?.delivery.id ?? null, @@ -222,11 +236,10 @@ export async function checkWorkerMailbox(args: { } } } else { - const current = readDelivery(params.wait ? typeFilter : undefined) + const current = readDelivery(params.wait ? typeFilter : args.wakeTypes) if (current || !params.wait) { return { - ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}), - dispatchId: workerMailbox.dispatchId, + ...mailboxIdentity, deliveryId: current?.delivery.id ?? null, messages: exposeMessages(current?.messages ?? []), count: current?.messages.length ?? 0, @@ -241,19 +254,22 @@ export async function checkWorkerMailbox(args: { } } } - const waitResult = await runtime.waitForMessage(address, { - typeFilter: typeFilter as string[] | undefined, - timeoutMs: params.timeoutMs ?? undefined, - signal - }) + // Binding can happen during recovery, before run-create/run-use can cancel this wait. + const waitResult = + activeDispatch && currentDispatchAssigneeRun(runtime, db, activeDispatch) + ? 'cancelled' + : await runtime.waitForMessage(address, { + typeFilter: typeFilter as string[] | undefined, + timeoutMs: params.timeoutMs ?? undefined, + signal + }) await revalidateWorkerMailbox() if (readCurrentGeneration() !== workerMailbox.generation) { throw dispatchFenced() } if (waitResult === 'timed_out' || waitResult === 'cancelled') { return { - ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}), - dispatchId: workerMailbox.dispatchId, + ...mailboxIdentity, messages: [], count: 0, acknowledged: acknowledged?.delivery.id ?? null, @@ -265,8 +281,7 @@ export async function checkWorkerMailbox(args: { if (params.peek) { const arrived = readPeek() return { - ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}), - dispatchId: workerMailbox.dispatchId, + ...mailboxIdentity, messages: exposeMessages(arrived), count: arrived.length, acknowledged: acknowledged?.delivery.id ?? null, @@ -277,8 +292,7 @@ export async function checkWorkerMailbox(args: { } const arrived = readDelivery(typeFilter) return { - ...(workerMailbox.runId ? { runId: workerMailbox.runId } : {}), - dispatchId: workerMailbox.dispatchId, + ...mailboxIdentity, deliveryId: arrived?.delivery.id ?? null, messages: exposeMessages(arrived?.messages ?? []), count: arrived?.messages.length ?? 0, diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/dispatch-recipient-identity.test.ts b/src/main/runtime/rpc/methods/orchestration/messaging/dispatch-recipient-identity.test.ts new file mode 100644 index 00000000000..4ab4b6555a0 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/messaging/dispatch-recipient-identity.test.ts @@ -0,0 +1,137 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { testOrcaSessionId } from '../../../../../../shared/orca-session-address-test-fixture' +import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture' +import { createOrchestrationRpcHarness } from '../rpc-test-harness' + +const PANE = 'tab_lead:22222222-2222-4222-9222-222222222222' +const OTHER = 'tab_other:33333333-3333-4333-8333-333333333333' +const SESSION = testOrcaSessionId('4bd46b4a-035b-41dd-a122-a9c29122ff11') + +describe.each([false, true])('Dispatch recipient identity (settled=%s)', (settled) => { + const h = createOrchestrationRpcHarness() + let state: ReturnType + let dispatch: ReturnType + + beforeEach(() => { + state = h.setup() + const task = state.db.createTask({ spec: 'lead' }) + dispatch = createRootDispatch(state.db, task.id, 'term_lead', PANE) + if (settled) { + state.db.completeDispatch(dispatch.id) + } + vi.spyOn(state.runtime, 'getLiveTerminalPaneKey').mockImplementation((handle) => + handle === 'term_lead' ? PANE : h.coordinatorPaneKey + ) + }) + afterEach(() => h.cleanup()) + + function send() { + return h.call( + 'orchestration.send', + { from: 'term_coord', to: `dispatch:${dispatch.id}`, subject: 'follow up' }, + state.ctx + ) + } + function sessionRun() { + state.db.db + .prepare('UPDATE dispatch_contexts SET assignee_orca_session_id = ? WHERE id = ?') + .run(SESSION, dispatch.id) + return state.db.createRun({ + objective: 'session lead', + coordinatorHandle: 'term_lead', + coordinatorPaneKey: null, + coordinatorOrcaSessionId: SESSION + }) + } + async function expectRun(runId: string) { + if (settled) { + await expect(send()).rejects.toMatchObject({ + code: 'dispatch_inactive', + message: expect.stringContaining(`Send to run:${runId} instead`) + }) + expect(state.db.getInbox()).toEqual([]) + } else { + expect(await send()).toMatchObject({ message: { to_handle: `run:${runId}`, run_id: runId } }) + } + } + async function expectNoRedirect(unrelatedRun: string) { + if (settled) { + await expect(send()).rejects.toMatchObject({ + code: 'dispatch_inactive', + message: expect.not.stringContaining(unrelatedRun) + }) + expect(state.db.getInbox()).toEqual([]) + } else { + expect(await send()).toMatchObject({ + message: { to_handle: `dispatch:${dispatch.id}`, run_id: dispatch.run_id } + }) + } + } + + it('uses a durable session binding without a live pane', async () => { + const run = sessionRun() + vi.mocked(state.runtime.getLiveTerminalPaneKey).mockReturnValue(null) + await expectRun(run.id) + }) + + it('uses the recorded session instead of an unrelated Run now occupying the saved pane', async () => { + const run = sessionRun() + state.db.createRun({ + objective: 'new occupant', + coordinatorHandle: 'term_other', + coordinatorPaneKey: PANE + }) + await expectRun(run.id) + }) + + it('ignores an old session column left behind by an older binary rebind', async () => { + const run = sessionRun() + state.db.db + .prepare( + 'UPDATE runs SET coordinator_handle = ?, coordinator_pane_key = ?, consumer_generation = consumer_generation + 1 WHERE id = ?' + ) + .run('term_other', OTHER, run.id) + await expectNoRedirect(run.id) + }) + + it('does not follow a closed handle to another occupant of its old pane', async () => { + const run = state.db.createRun({ + objective: 'new occupant', + coordinatorHandle: 'term_other', + coordinatorPaneKey: PANE + }) + vi.mocked(state.runtime.getLiveTerminalPaneKey).mockReturnValue(null) + await expectNoRedirect(run.id) + }) + + it.each(['replacement:pty:2', null])( + 'does not redirect with a replaced or unverifiable process: %s', + async (process) => { + const run = state.db.createRun({ + objective: 'pane run', + coordinatorHandle: 'term_lead', + coordinatorPaneKey: PANE + }) + state.db.db + .prepare('UPDATE dispatch_contexts SET process_incarnation = ? WHERE id = ?') + .run('original:pty:1', dispatch.id) + vi.mocked(state.runtime.getTerminalProcessIncarnation).mockReturnValue(process) + await expectNoRedirect(run.id) + expect(state.db.getDispatchContextById(dispatch.id)?.status).toBe( + settled ? 'completed' : dispatch.status + ) + } + ) + + it('accepts a reminted tab half with the same pane leaf and process', async () => { + const run = state.db.createRun({ + objective: 'same pane', + coordinatorHandle: 'term_lead', + coordinatorPaneKey: PANE + }) + vi.mocked(state.runtime.getLiveTerminalPaneKey).mockReturnValue( + PANE.replace('tab_lead', 'tab_restored') + ) + await expectRun(run.id) + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/message-methods.ts b/src/main/runtime/rpc/methods/orchestration/messaging/message-methods.ts index 3e59073349b..e079596f80d 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/message-methods.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/message-methods.ts @@ -12,6 +12,7 @@ import { import { exposeMessage } from './mailbox-message-receipt' import { resolveOrchestrationParty } from '../../../../orchestration/orchestration-party' import { recordReceiptBeforeNudge, replayMutationNudge } from './mutation-replay-nudge' +import { resolveReplyRecipient } from './recipient-routing' import { ReplyParams, InboxParams, @@ -113,15 +114,20 @@ export const ORCHESTRATION_MESSAGE_METHODS = [ ) } + const recipient = resolveReplyRecipient({ + runtime, + db, + originalFrom: original.from_handle, + originalRunId: original.run_id + }) db.markAsRead([original.id]) - const reply = db.insertMessage({ from: params.from ?? original.to_handle, - to: original.from_handle, + to: recipient.to, subject: `Re: ${original.subject}`, body: params.body, threadId: original.thread_id ?? original.id, - runId: original.run_id + runId: recipient.runId }) const receipt = { message: exposeMessage(reply) } diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/recipient-routing.ts b/src/main/runtime/rpc/methods/orchestration/messaging/recipient-routing.ts index 861cfadeac6..c499f743576 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/recipient-routing.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/recipient-routing.ts @@ -5,6 +5,8 @@ import type { OrcaRuntimeService } from '../../../../orca-runtime' import { readStructuredAgentSessionRecord } from '../../../../structured-worker-authority' import { structuredWorkerHostScope } from '../../../../structured-worker-identity' import { resolveOrchestrationParty } from '../../../../orchestration/orchestration-party' +import { isEquivalentPaneKey } from '../../../../orchestration/db/pane-key-match' +import { CURRENT_CONTRACT_VERSION } from '../../../../orchestration/db/contract-constants' const ACTIVE_DISPATCH_STATUSES: readonly DispatchStatus[] = ['pending', 'dispatched'] @@ -15,20 +17,123 @@ const ACTIVE_DISPATCH_STATUSES: readonly DispatchStatus[] = ['pending', 'dispatc * mailbox, so accepting the message reports success for a delivery that cannot * happen. Federated targets keep their own liveness check. */ -export function assertDispatchMailboxDeliverable(db: OrchestrationDb, dispatchId: string): void { +export function assertDispatchMailboxDeliverable( + runtime: OrcaRuntimeService, + db: OrchestrationDb, + dispatchId: string +): void { const dispatch = db.getDispatchContextById(dispatchId) if (!dispatch || ACTIVE_DISPATCH_STATUSES.includes(dispatch.status)) { return } + const recipientRun = currentDispatchAssigneeRun(runtime, db, dispatch)?.id ?? dispatch.run_id throw new OrchestrationError( 'dispatch_inactive', - `Dispatch ${dispatchId} is ${dispatch.status}; its worker will never read that mailbox. Send to run:${dispatch.run_id} instead, or start a new Dispatch for follow-up work.` + `Dispatch ${dispatchId} is ${dispatch.status}; its worker will never read that mailbox. Send to run:${recipientRun} instead, or start a new Dispatch for follow-up work.` ) } +// A saved pane alone cannot identify its occupant after reuse. +export function currentDispatchAssigneeRun( + runtime: OrcaRuntimeService, + db: OrchestrationDb, + dispatch: DispatchContextRow +) { + if ( + dispatch.contract_version !== CURRENT_CONTRACT_VERSION || + db.getFederatedDispatch(dispatch.id) + ) { + return undefined + } + if (dispatch.assignee_orca_session_id !== null) { + return db.getCurrentRunForCoordinator({ + terminalHandle: dispatch.assignee_handle, + paneKey: null, + orcaSessionId: dispatch.assignee_orca_session_id + }) + } + if (dispatch.assignee_handle === null) { + return undefined + } + const paneKey = runtime.getLiveTerminalPaneKey(dispatch.assignee_handle) + if ( + !paneKey || + (dispatch.assignee_pane_key && !isEquivalentPaneKey(dispatch.assignee_pane_key, paneKey)) || + (dispatch.process_incarnation !== null && + runtime.getTerminalProcessIncarnation(dispatch.assignee_handle) !== + dispatch.process_incarnation) + ) { + return undefined + } + return db.getCurrentRunForPane(paneKey) +} + +// Nested coordinators receive new mail where their current Run check waits. +export function resolveRunBoundDispatchRecipient( + runtime: OrcaRuntimeService, + db: OrchestrationDb, + dispatchId: string, + explicitRunId?: string +): { to: string; runId: string; warning: SendRecipientWarning } | undefined { + const dispatch = db.getDispatchContextById(dispatchId) + if (!dispatch || !ACTIVE_DISPATCH_STATUSES.includes(dispatch.status)) { + return undefined + } + const boundRun = currentDispatchAssigneeRun(runtime, db, dispatch) + if (!boundRun || boundRun.id === dispatch.run_id) { + return undefined + } + const recipient = `dispatch:${dispatchId}` + const mismatch = runMismatch(recipient, boundRun.id, explicitRunId) + if (mismatch && !mismatch.ok) { + throw new OrchestrationError(mismatch.code, mismatch.message) + } + return { + to: `run:${boundRun.id}`, + runId: boundRun.id, + warning: { + code: 'recipient_run_bound_redirect', + recipient, + message: `${recipient} is assigned to a terminal that now coordinates Run ${boundRun.id}; queued for run:${boundRun.id}, the mailbox that terminal reads.` + } + } +} + +// Replies share send routing; unresolved historical senders keep their original address. +export function resolveReplyRecipient(params: { + runtime: OrcaRuntimeService + db: OrchestrationDb + originalFrom: string + originalRunId: string | undefined +}): { to: string; runId: string | undefined } { + const { runtime, db, originalFrom, originalRunId } = params + const unchanged = { to: originalFrom, runId: originalRunId } + if (originalFrom.startsWith('run:')) { + return { to: originalFrom, runId: originalFrom.slice('run:'.length) } + } + if (originalFrom.startsWith('dispatch:')) { + const dispatchId = originalFrom.slice('dispatch:'.length) + // Federation owns its own recipient and liveness checks. + if (db.getFederatedDispatch(dispatchId)) { + return unchanged + } + assertDispatchMailboxDeliverable(runtime, db, dispatchId) + const runBound = resolveRunBoundDispatchRecipient(runtime, db, dispatchId) + return runBound ?? unchanged + } + const recipient = resolveBareOrchestrationRecipient({ + runtime, + db, + handle: originalFrom, + senderRunId: originalRunId + }) + return recipient.ok ? { to: recipient.to, runId: recipient.runId ?? originalRunId } : unchanged +} + export type SendRecipientWarning = { code: | 'legacy_terminal_recipient' + | 'recipient_run_bound_redirect' | 'recipient_unreachable' | 'recipient_ambiguous' | 'recipient_run_mismatch' diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-history.test.ts b/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-history.test.ts new file mode 100644 index 00000000000..fba01d0d455 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-history.test.ts @@ -0,0 +1,69 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture' +import { createOrchestrationRpcHarness } from '../rpc-test-harness' + +const LEAD = 'tab_lead:22222222-2222-4222-9222-222222222222' +function deliveryId(result: unknown): string { + if ( + typeof result === 'object' && + result && + 'deliveryId' in result && + typeof result.deliveryId === 'string' + ) { + return result.deliveryId + } + throw new Error('Expected a Delivery') +} + +describe('Run history after residual Dispatch acknowledgment', () => { + const h = createOrchestrationRpcHarness() + let state: ReturnType + beforeEach(() => { + state = h.setup() + vi.mocked(state.runtime.getTerminalPaneKey).mockImplementation((handle) => + handle === 'term_lead' ? LEAD : h.coordinatorPaneKey + ) + }) + afterEach(() => h.cleanup()) + it.each([ + { mode: { all: true }, acknowledgeTogether: false }, + { mode: { all: true }, acknowledgeTogether: true }, + { mode: { unread: false }, acknowledgeTogether: false }, + { mode: { unread: false }, acknowledgeTogether: true } + ])('shows Run history after old mail is read: %j', async ({ mode, acknowledgeTogether }) => { + const task = state.db.createTask({ spec: 'nested lead' }) + const dispatch = createRootDispatch(state.db, task.id, 'term_lead', LEAD) + const run = state.db.createRun({ + objective: 'child Run', + coordinatorHandle: 'term_lead', + coordinatorPaneKey: LEAD + }) + const old = state.db.insertMessage({ + from: 'term_coord', + to: `dispatch:${dispatch.id}`, + runId: dispatch.run_id, + subject: 'old mail' + }) + const check = (params = {}) => + h.call('orchestration.check', { terminal: 'term_lead', ...params }, state.ctx) + expect(await check(mode)).toMatchObject({ runId: dispatch.run_id, messages: [{ id: old.id }] }) + expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false) + const delivery = deliveryId(await check()) + if (!acknowledgeTogether) { + await check({ ack: delivery }) + } + const fresh = state.db.insertMessage({ + from: 'term_child', + to: `run:${run.id}`, + runId: run.id, + subject: 'new mail' + }) + expect( + await check({ ...mode, ...(acknowledgeTogether ? { ack: delivery } : {}) }) + ).toMatchObject({ runId: run.id, messages: [{ id: fresh.id }] }) + expect(state.db.getMessageById(old.id)?.read).toBe(1) + expect(state.db.getMessageById(fresh.id)?.read).toBe(0) + expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false) + expect(state.db.hasOutstandingMailboxDelivery(`run:${run.id}`)).toBe(false) + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-interleavings.test.ts b/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-interleavings.test.ts new file mode 100644 index 00000000000..4e92d7c35c1 --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-interleavings.test.ts @@ -0,0 +1,175 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture' +import { ORCHESTRATION_DELIVERY_BATCH_LIMIT } from '../../../../orchestration/db/messages/mailbox-routing-page' +import { createOrchestrationRpcHarness } from '../rpc-test-harness' + +const LEAD = 'tab_lead:22222222-2222-4222-9222-222222222222' +const OTHER = 'tab_other:33333333-3333-4333-8333-333333333333' + +function deliveryId(result: unknown): string { + if ( + typeof result === 'object' && + result !== null && + 'deliveryId' in result && + typeof result.deliveryId === 'string' + ) { + return result.deliveryId + } + throw new Error('Expected a Delivery') +} + +describe('Run binding during Dispatch checks', () => { + const h = createOrchestrationRpcHarness() + let state: ReturnType + let dispatch: ReturnType + + beforeEach(() => { + state = h.setup() + vi.mocked(state.runtime.getTerminalPaneKey).mockImplementation((handle) => + handle === 'term_lead' ? LEAD : handle === 'term_coord' ? h.coordinatorPaneKey : OTHER + ) + dispatch = createRootDispatch( + state.db, + state.db.createTask({ spec: 'lead' }).id, + 'term_lead', + LEAD + ) + }) + afterEach(() => { + state.runtime.cancelMessageWaiters(`dispatch:${dispatch.id}`) + h.cleanup() + vi.restoreAllMocks() + }) + + function check(params: Record = {}) { + return h.call('orchestration.check', { terminal: 'term_lead', ...params }, state.ctx) + } + function residual(subject: string) { + return state.db.insertMessage({ + from: 'term_coord', + to: `dispatch:${dispatch.id}`, + runId: dispatch.run_id, + subject + }) + } + async function bind(method: 'runCreate' | 'runUse') { + const params = + method === 'runCreate' + ? { objective: 'child' } + : { + id: state.db.createRun({ + objective: 'adopt', + coordinatorHandle: 'term_other', + coordinatorPaneKey: OTHER + }).id + } + await h.call(`orchestration.${method}`, { from: 'term_lead', ...params }, state.ctx) + const run = state.db.getCurrentRunForPane(LEAD) + if (!run) { + throw new Error('Expected bound Run') + } + return run + } + + describe.each(['runCreate', 'runUse'] as const)('%s', (method) => { + it.each(['parked', 'recovering'] as const)( + 'cancels a %s Dispatch wait and preserves its ack', + async (phase) => { + const old = residual('before bind') + const ack = deliveryId(await check()) + const record = vi.fn() + state.ctx.recordMutationReceipt = record + const wait = vi.spyOn(state.runtime, 'waitForMessage') + const waiting = check({ ack, wait: true, timeoutMs: 500 }) + if (phase === 'parked') { + await vi.waitFor(() => + expect(wait).toHaveBeenCalledWith(`dispatch:${dispatch.id}`, expect.anything()) + ) + } + const run = await bind(method) + await h.call( + 'orchestration.send', + { + from: 'term_coord', + to: `dispatch:${dispatch.id}`, + subject: 'after bind' + }, + state.ctx + ) + expect(await waiting).toMatchObject({ acknowledged: ack, cancelled: true, timedOut: false }) + expect(record).toHaveBeenCalledWith(expect.objectContaining({ acknowledged: ack })) + expect(state.db.getMessageById(old.id)).toMatchObject({ run_id: dispatch.run_id, read: 1 }) + expect(await check()).toMatchObject({ + runId: run.id, + messages: [{ subject: 'after bind' }] + }) + if (phase === 'recovering') { + expect(wait).not.toHaveBeenCalled() + } + } + ) + }) + + it.each(['before', 'during recovery'] as const)( + 'replays Run delivery created %s residual acknowledgment', + async (phase) => { + const ids = Array.from( + { length: ORCHESTRATION_DELIVERY_BATCH_LIMIT + 1 }, + (_, i) => residual(`old ${i}`).id + ) + const last = ids.at(-1) + if (!last) { + throw new Error('Expected residual tail') + } + const ack = deliveryId(await check()) + const run = await bind('runCreate') + state.db.insertMessage({ + from: 'term_child', + to: `run:${run.id}`, + runId: run.id, + subject: 'Run batch' + }) + let runDelivery: string | undefined + const readRun = () => { + const result = state.db.getOrCreateRunDelivery({ + runId: run.id, + consumerGeneration: run.consumer_generation + }) + if (!result) { + throw new Error('Expected Run delivery') + } + runDelivery = result.delivery.id + } + if (phase === 'before') { + runDelivery = deliveryId(await check({ run: run.id })) + } else { + state.db.db.prepare('UPDATE messages SET to_handle = ? WHERE id = ?').run('term_lead', last) + const route = state.db.routeUnreadDirectMessagesToDispatchMailbox.bind(state.db) + vi.spyOn(state.db, 'routeUnreadDirectMessagesToDispatchMailbox').mockImplementationOnce( + (...args) => { + const result = route(...args) + readRun() + return result + } + ) + } + const result = await check({ ack }) + expect(result).toMatchObject({ + runId: run.id, + deliveryId: runDelivery, + acknowledged: ack, + replayed: true + }) + expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false) + for (const id of ids.slice(0, -1)) { + expect(state.db.getMessageById(id)).toMatchObject({ read: 1, run_id: dispatch.run_id }) + } + expect(state.db.getMessageById(last)).toMatchObject({ read: 0, run_id: dispatch.run_id }) + expect(await check({ ack })).toMatchObject({ deliveryId: runDelivery, replayed: true }) + const tail = await check({ ack: runDelivery }) + expect(tail).toMatchObject({ runId: dispatch.run_id, messages: [{ id: last }], count: 1 }) + expect(await check({ ack: deliveryId(tail) })).toMatchObject({ count: 0 }) + expect(state.db.getMessageById(last)?.read).toBe(1) + } + ) +}) diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-safety.test.ts b/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-safety.test.ts new file mode 100644 index 00000000000..f5f2c57575f --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-mailbox-safety.test.ts @@ -0,0 +1,290 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture' +import { createOrchestrationRpcHarness } from '../rpc-test-harness' + +const LEAD = 'tab_lead:22222222-2222-4222-9222-222222222222' +const OTHER = 'tab_other:33333333-3333-4333-8333-333333333333' + +function deliveryId(result: unknown): string { + if ( + typeof result === 'object' && + result && + 'deliveryId' in result && + typeof result.deliveryId === 'string' + ) { + return result.deliveryId + } + throw new Error('Expected a Delivery') +} + +describe('Run-bound lead mailbox boundaries', () => { + const h = createOrchestrationRpcHarness() + let state: ReturnType + let dispatch: ReturnType + let leadRun: ReturnType + + beforeEach(() => { + state = h.setup() + vi.mocked(state.runtime.getTerminalPaneKey).mockImplementation((handle) => + handle === 'term_lead' ? LEAD : handle === 'term_coord' ? h.coordinatorPaneKey : OTHER + ) + const task = state.db.createTask({ spec: 'nested lead' }) + dispatch = createRootDispatch(state.db, task.id, 'term_lead', LEAD) + leadRun = state.db.createRun({ + objective: 'child Run', + coordinatorHandle: 'term_lead', + coordinatorPaneKey: LEAD + }) + }) + afterEach(() => h.cleanup()) + + function check(params: Record = {}) { + return h.call('orchestration.check', { terminal: 'term_lead', ...params }, state.ctx) + } + function residual(subject: string, type: 'status' | 'question' = 'status') { + return state.db.insertMessage({ + from: 'term_coord', + to: `dispatch:${dispatch.id}`, + runId: dispatch.run_id, + subject, + type + }) + } + function runMail(subject = 'Run report') { + return state.db.insertMessage({ + from: 'term_child', + to: `run:${leadRun.id}`, + runId: leadRun.id, + subject, + type: 'question' + }) + } + function handoff() { + state.db.bindRun({ + runId: leadRun.id, + coordinatorHandle: 'term_other', + coordinatorPaneKey: OTHER + }) + } + + it('recovers old raw-handle replies only from the active Dispatch Run', async () => { + const old = residual('old reply') + state.db.db.prepare('UPDATE messages SET to_handle = ? WHERE id = ?').run('term_lead', old.id) + const foreign = state.db.createRun({ + objective: 'unrelated', + coordinatorHandle: 'term_other', + coordinatorPaneKey: OTHER + }) + const privateMail = state.db.insertMessage({ + from: 'term_other', + to: 'term_lead', + runId: foreign.id, + subject: 'foreign' + }) + runMail() + const first = await check() + expect(first).toMatchObject({ runId: dispatch.run_id, messages: [{ id: old.id }] }) + expect(await check({ ack: deliveryId(first) })).toMatchObject({ + runId: leadRun.id, + messages: [{ subject: 'Run report' }] + }) + expect(state.db.getMessageById(privateMail.id)).toMatchObject({ read: 0, run_id: foreign.id }) + }) + + it('treats types as a wake condition and replays the entire residual FIFO batch', async () => { + residual('older status') + residual('decision needed', 'question') + runMail() + const first = await check({ wait: true, types: 'question' }) + expect(first).toMatchObject({ + messages: [{ subject: 'older status' }, { subject: 'decision needed' }] + }) + expect(await check({ wait: true, types: 'worker_done' })).toMatchObject({ + deliveryId: deliveryId(first), + replayed: true + }) + expect(await check({ ack: deliveryId(first) })).toMatchObject({ + runId: leadRun.id, + acknowledged: deliveryId(first), + messages: [{ subject: 'Run report' }] + }) + }) + + it('does not filter a non-waiting consuming residual check', async () => { + residual('status') + expect(await check({ types: 'question' })).toMatchObject({ messages: [{ subject: 'status' }] }) + }) + + it.each([{ peek: true }, { all: true }])( + 'filters residual inspection without consuming it: %j', + async (mode) => { + const status = residual('status') + residual('question', 'question') + expect(await check({ ...mode, types: 'question' })).toMatchObject({ + messages: [{ subject: 'question' }] + }) + expect(state.db.getMessageById(status.id)?.read).toBe(0) + expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false) + } + ) + + it('replays an outstanding Run batch before exposing older nonmatching residual mail', async () => { + residual('status') + runMail() + const first = await check({ wait: true, types: 'question' }) + expect(first).toMatchObject({ runId: leadRun.id }) + expect(await check()).toMatchObject({ deliveryId: deliveryId(first), replayed: true }) + expect(await check({ ack: deliveryId(first) })).toMatchObject({ + acknowledged: deliveryId(first), + messages: [{ subject: 'status' }] + }) + }) + + it('keeps explicit Run checks scoped and refuses the parent Run', async () => { + const pending = residual('parent mail') + runMail() + expect(await check({ run: leadRun.id })).toMatchObject({ + runId: leadRun.id, + messages: [{ subject: 'Run report' }] + }) + await expect(check({ run: dispatch.run_id })).rejects.toMatchObject({ code: 'consumer_fenced' }) + expect(state.db.getMessageById(pending.id)?.read).toBe(0) + }) + + it('fences a Run handoff during residual direct-mail recovery before delivery or ack', async () => { + const pending = residual('raw before bind') + state.db.db + .prepare('UPDATE messages SET to_handle = ? WHERE id = ?') + .run('term_lead', pending.id) + const route = state.db.routeUnreadDirectMessagesToDispatchMailbox.bind(state.db) + vi.spyOn(state.db, 'routeUnreadDirectMessagesToDispatchMailbox').mockImplementationOnce( + (...args) => { + const result = route(...args) + handoff() + return result + } + ) + await expect(check()).rejects.toMatchObject({ code: 'consumer_fenced' }) + expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false) + expect(state.db.getMessageById(pending.id)?.read).toBe(0) + }) + + it('records a residual acknowledgment before a Run handoff interrupts the following wait', async () => { + residual('ack me') + const first = await check() + const record = vi.fn() + state.ctx.recordMutationReceipt = record + vi.spyOn(state.runtime, 'waitForMessage').mockImplementationOnce(async () => { + expect(record).toHaveBeenCalledWith( + expect.objectContaining({ acknowledged: deliveryId(first) }) + ) + handoff() + return 'cancelled' + }) + expect(await check({ ack: deliveryId(first), wait: true })).toMatchObject({ + acknowledged: deliveryId(first), + waitInterrupted: 'consumer_fenced', + messages: [] + }) + expect(state.db.getUnreadMessages(`dispatch:${dispatch.id}`)).toEqual([]) + }) + + it('retains the acknowledged receipt if the wait transport throws', async () => { + residual('ack me') + const first = await check() + const record = vi.fn() + state.ctx.recordMutationReceipt = record + vi.spyOn(state.runtime, 'waitForMessage').mockRejectedValueOnce( + new Error('connection interrupted') + ) + await expect(check({ ack: deliveryId(first), wait: true })).rejects.toThrow( + 'connection interrupted' + ) + expect(record).toHaveBeenCalledWith( + expect.objectContaining({ acknowledged: deliveryId(first) }) + ) + expect(state.db.getUnreadMessages(`dispatch:${dispatch.id}`)).toEqual([]) + }) + + it('does not give a reused process the previous assignee mail', async () => { + state.db.mintDispatchCapability({ + dispatchId: dispatch.id, + paneKey: LEAD, + processIncarnation: 'old:pty:1' + }) + const pending = residual('old process mail') + runMail() + expect(await check()).toMatchObject({ + runId: leadRun.id, + messages: [{ subject: 'Run report' }] + }) + expect(state.db.getMessageById(pending.id)?.read).toBe(0) + expect(state.db.hasOutstandingMailboxDelivery(`dispatch:${dispatch.id}`)).toBe(false) + }) + + it('refuses an explicit sender Run that would silently redirect into another Run', async () => { + await expect( + h.call( + 'orchestration.send', + { + from: 'term_coord', + to: `dispatch:${dispatch.id}`, + run: dispatch.run_id, + subject: 'wrong scope' + }, + state.ctx + ) + ).rejects.toMatchObject({ code: 'recipient_run_mismatch' }) + expect(state.db.getInbox()).toEqual([]) + }) + + it('wakes a parked Run check for Dispatch mail sent after binding', async () => { + const waiter = vi.spyOn(state.runtime, 'waitForMessage') + const waiting = check({ wait: true, timeoutMs: 1_000 }) + await vi.waitFor(() => + expect(waiter).toHaveBeenCalledWith(`run:${leadRun.id}`, expect.anything()) + ) + await h.call( + 'orchestration.send', + { from: 'term_coord', to: `dispatch:${dispatch.id}`, subject: 'wake up' }, + state.ctx + ) + expect(await waiting).toMatchObject({ timedOut: false, messages: [{ subject: 'wake up' }] }) + }) + + it('keeps a canonical Run reply in the recipient Run even across an original thread Run', async () => { + const note = state.db.insertMessage({ + from: `run:${leadRun.id}`, + to: `run:${dispatch.run_id}`, + runId: dispatch.run_id, + subject: 'report' + }) + expect( + await h.call( + 'orchestration.reply', + { from: 'term_coord', id: note.id, body: 'decision' }, + state.ctx + ) + ).toMatchObject({ message: { to_handle: `run:${leadRun.id}`, run_id: leadRun.id } }) + expect(await check()).toMatchObject({ messages: [{ subject: 'Re: report' }] }) + }) + + it('refuses replies to an inactive canonical Dispatch before reading or inserting mail', async () => { + const note = state.db.insertMessage({ + from: `dispatch:${dispatch.id}`, + to: `run:${dispatch.run_id}`, + runId: dispatch.run_id, + subject: 'old report' + }) + state.db.completeDispatch(dispatch.id) + await expect( + h.call( + 'orchestration.reply', + { from: 'term_coord', id: note.id, body: 'too late' }, + state.ctx + ) + ).rejects.toMatchObject({ code: 'dispatch_inactive' }) + expect(state.db.getMessageById(note.id)?.read).toBe(0) + expect(state.db.getInbox()).toHaveLength(1) + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-recipient-mail.test.ts b/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-recipient-mail.test.ts new file mode 100644 index 00000000000..bb6c7041cee --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/messaging/run-bound-recipient-mail.test.ts @@ -0,0 +1,189 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { RpcContext } from '../../../core' +import type { OrchestrationDb, RunRow } from '../../../../orchestration/db' +import type { OrcaRuntimeService } from '../../../../orca-runtime' +import type { DispatchContextRow } from '../../../../orchestration/types' +import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture' +import { createOrchestrationRpcHarness } from '../rpc-test-harness' + +// A lead is dispatched by a root coordinator and then coordinates its own Run from the same pane. +// That pane's `check` reads its own Run mailbox, so mail meant for it must land there. +describe('mail for a lead whose pane coordinates its own Run', () => { + const h = createOrchestrationRpcHarness() + const coordPane = 'tab_coord:11111111-1111-4111-8111-111111111111' + const leadPane = 'tab_lead:22222222-2222-4222-9222-222222222222' + let db: OrchestrationDb + let runtime: OrcaRuntimeService + let ctx: RpcContext + let rootRun: RunRow + let dispatch: DispatchContextRow + + afterEach(() => { + h.cleanup() + }) + + function setup(): void { + ;({ db, runtime, ctx } = h.setup(false)) + vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) => + handle === 'term_coord' ? coordPane : handle === 'term_lead' ? leadPane : null + ) + rootRun = db.createRun({ + objective: 'root', + coordinatorHandle: 'term_coord', + coordinatorPaneKey: coordPane + }) + const task = db.createTask({ spec: 'lead the sub-project', runId: rootRun.id }) + dispatch = createRootDispatch(db, task.id, 'term_lead', leadPane) + } + + function bindLeadRun(): RunRow { + return db.createRun({ + objective: 'lead', + coordinatorHandle: 'term_lead', + coordinatorPaneKey: leadPane + }) + } + + async function call(name: string, params: Record) { + return h.call(name, params, ctx) + } + + async function leadInbox(params: Record = {}): Promise { + return call('orchestration.check', { terminal: 'term_lead', ...params }) + } + + function deliveryIdOf(result: unknown): string { + if ( + typeof result === 'object' && + result !== null && + 'deliveryId' in result && + typeof result.deliveryId === 'string' + ) { + return result.deliveryId + } + throw new Error('check returned no delivery') + } + + it('routes dispatch: mail to the Run the assignee pane now coordinates', async () => { + setup() + const leadRun = bindLeadRun() + + const result = await call('orchestration.send', { + from: 'term_coord', + to: `dispatch:${dispatch.id}`, + subject: 'Follow-up for the lead' + }) + + expect(result).toMatchObject({ + message: { to_handle: `run:${leadRun.id}`, run_id: leadRun.id }, + warnings: [{ code: 'recipient_run_bound_redirect' }] + }) + expect(await leadInbox()).toMatchObject({ messages: [{ subject: 'Follow-up for the lead' }] }) + }) + + it('still reads dispatch mail that arrived before the pane bound its own Run', async () => { + setup() + db.insertMessage({ + from: 'term_coord', + to: `dispatch:${dispatch.id}`, + subject: 'Sent before the lead bound a Run', + runId: rootRun.id + }) + const leadRun = bindLeadRun() + db.insertMessage({ + from: 'term_worker', + to: `run:${leadRun.id}`, + subject: 'Sub-worker report', + runId: leadRun.id + }) + + const first = await leadInbox() + expect(first).toMatchObject({ messages: [{ subject: 'Sent before the lead bound a Run' }] }) + + const second = await leadInbox({ ack: deliveryIdOf(first) }) + expect(second).toMatchObject({ messages: [{ subject: 'Sub-worker report' }] }) + }) + + it('keeps the --types wake condition when older Dispatch mail does not match it', async () => { + setup() + db.insertMessage({ + from: 'term_coord', + to: `dispatch:${dispatch.id}`, + subject: 'Older status note', + type: 'status', + runId: rootRun.id + }) + const leadRun = bindLeadRun() + db.insertMessage({ + from: 'term_worker', + to: `run:${leadRun.id}`, + subject: 'Sub-worker finished', + type: 'worker_done', + runId: leadRun.id + }) + + const woke = await leadInbox({ wait: true, types: 'worker_done', timeoutMs: 500 }) + + expect(woke).toMatchObject({ + runId: leadRun.id, + messages: [{ subject: 'Sub-worker finished' }] + }) + }) + + it('delivers a reply to a Run-bound sender and wakes its waiting check', async () => { + setup() + const leadRun = bindLeadRun() + const report = db.insertMessage({ + from: 'term_lead', + to: `run:${rootRun.id}`, + subject: 'Lead report', + runId: rootRun.id + }) + + const waiting = leadInbox({ wait: true, timeoutMs: 2_000 }) + const reply = await call('orchestration.reply', { + id: report.id, + from: 'term_coord', + body: 'Decision' + }) + + expect(reply).toMatchObject({ + message: { to_handle: `run:${leadRun.id}`, run_id: leadRun.id } + }) + expect(await waiting).toMatchObject({ + timedOut: false, + messages: [{ subject: 'Re: Lead report' }] + }) + }) + + it('keeps dispatch: mail on the Dispatch mailbox while the assignee has no Run', async () => { + setup() + + const result = await call('orchestration.send', { + from: 'term_coord', + to: `dispatch:${dispatch.id}`, + subject: 'Plain worker follow-up' + }) + + expect(result).toMatchObject({ message: { to_handle: `dispatch:${dispatch.id}` } }) + expect(result).not.toHaveProperty('warnings') + }) + + it('keeps a reply on the raw handle when the sender has no Run or live pane', async () => { + setup() + const note = db.insertMessage({ + from: 'term_offline', + to: `run:${rootRun.id}`, + subject: 'Offline note', + runId: rootRun.id + }) + + const reply = await call('orchestration.reply', { + id: note.id, + from: 'term_coord', + body: 'Ack' + }) + + expect(reply).toMatchObject({ message: { to_handle: 'term_offline', run_id: rootRun.id } }) + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/send-methods.ts b/src/main/runtime/rpc/methods/orchestration/messaging/send-methods.ts index 28f5a4d67d5..527c8c6a509 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/send-methods.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/send-methods.ts @@ -7,6 +7,7 @@ import { resolveMessageRun } from '../routing' import { assertDispatchMailboxDeliverable, resolveBareOrchestrationRecipient, + resolveRunBoundDispatchRecipient, type SendRecipientWarning } from './recipient-routing' import { @@ -162,7 +163,18 @@ export const ORCHESTRATION_SEND_METHODS = [ : undefined // Federated targets perform their own liveness check before relaying. if (addressedDispatchId && !federatedTarget) { - assertDispatchMailboxDeliverable(db, addressedDispatchId) + assertDispatchMailboxDeliverable(runtime, db, addressedDispatchId) + const runBound = resolveRunBoundDispatchRecipient( + runtime, + db, + addressedDispatchId, + params.run + ) + if (runBound) { + to = runBound.to + messageRunId = runBound.runId + sendWarnings.push(runBound.warning) + } } const federatedControl = sendFederatedControlMail({ params, diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/settled-dispatch-mail.test.ts b/src/main/runtime/rpc/methods/orchestration/messaging/settled-dispatch-mail.test.ts index e91614061ea..ddffed46890 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/settled-dispatch-mail.test.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/settled-dispatch-mail.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it } from 'vitest' +import { afterEach, describe, expect, it, vi } from 'vitest' import type { RpcContext } from '../../../core' import type { OrchestrationDb } from '../../../../orchestration/db' import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture' @@ -58,6 +58,38 @@ describe('orchestration.send to a settled Dispatch mailbox', () => { ).rejects.toThrow(new RegExp(`run:${dispatch.run_id}`)) }) + it('names the Run a settled assignee now coordinates, not the sender Run', async () => { + setup() + const leadPane = 'tab_lead:22222222-2222-4222-9222-222222222222' + vi.spyOn(ctx.runtime, 'getLiveTerminalPaneKey').mockReturnValue(leadPane) + const task = db.createTask({ spec: 'lead that settled and kept coordinating' }) + const dispatch = createRootDispatch(db, task.id, 'term_lead', leadPane) + db.completeDispatch(dispatch.id) + const leadRun = db.createRun({ + objective: 'lead', + coordinatorHandle: 'term_lead', + coordinatorPaneKey: leadPane + }) + + const rejection = call('orchestration.send', { + from: 'term_coord', + to: `dispatch:${dispatch.id}`, + subject: 'One more thing' + }) + + await expect(rejection).rejects.toMatchObject({ code: 'dispatch_inactive' }) + await expect(rejection).rejects.toThrow(new RegExp(`run:${leadRun.id}`)) + await expect(rejection).rejects.not.toThrow(new RegExp(`run:${dispatch.run_id}`)) + expect(db.getInbox()).toEqual([]) + await expect( + call('orchestration.check', { terminal: 'term_coord', run: leadRun.id }) + ).rejects.toMatchObject({ code: 'consumer_fenced' }) + // Run addresses are already visible to callers; a hint grants no consuming authority. + expect(await call('orchestration.runShow', { id: leadRun.id })).toMatchObject({ + run: { id: leadRun.id } + }) + }) + it('does not write an undeliverable message row', async () => { setup() const task = db.createTask({ spec: 'worker that already reported' }) diff --git a/src/main/runtime/rpc/methods/orchestration/runs/runs.ts b/src/main/runtime/rpc/methods/orchestration/runs/runs.ts index f074b5eba03..88653af8ecd 100644 --- a/src/main/runtime/rpc/methods/orchestration/runs/runs.ts +++ b/src/main/runtime/rpc/methods/orchestration/runs/runs.ts @@ -2,6 +2,9 @@ import { defineMethod } from '../../../core' import { OrchestrationError } from '../../../../orchestration/orchestration-error' import { assertCallerHandleMatchesEvidence, resolveOrchestrationCaller } from './run-scope' import { exposeRun } from './run-receipt' +import type { OrcaRuntimeService } from '../../../../orca-runtime' +import type { OrchestrationCallerIdentity } from '../../../../orchestration/orchestration-caller-identity' +import { currentDispatchAssigneeRun } from '../messaging/recipient-routing' import { RunCreateParams, RunCurrentParams, @@ -10,6 +13,18 @@ import { RunUseParams } from '../../../../../../shared/rpc-contract/orchestration-runs-params' +function cancelBoundDispatchWaiters( + runtime: OrcaRuntimeService, + caller: OrchestrationCallerIdentity, + runId: string +): void { + const db = runtime.getOrchestrationDb() + const dispatch = db.getActiveDispatchForIdentity(caller.address, caller.paneKey ?? undefined) + if (dispatch && currentDispatchAssigneeRun(runtime, db, dispatch)?.id === runId) { + runtime.cancelMessageWaiters(`dispatch:${dispatch.id}`) + } +} + export const ORCHESTRATION_RUN_METHODS = [ defineMethod({ name: 'orchestration.runCreate', @@ -30,6 +45,7 @@ export const ORCHESTRATION_RUN_METHODS = [ coordinatorOrcaSessionId: caller.orcaSessionId }) runtime.cancelMessageWaiters(params.from) + cancelBoundDispatchWaiters(runtime, caller, run.id) if (priorRun) { runtime.cancelMessageWaiters(`run:${priorRun.id}`) } @@ -86,6 +102,7 @@ export const ORCHESTRATION_RUN_METHODS = [ ) } runtime.cancelMessageWaiters(params.from) + cancelBoundDispatchWaiters(runtime, caller, run.id) runtime.cancelMessageWaiters(`run:${params.id}`) if (priorRun && priorRun.id !== params.id) { runtime.cancelMessageWaiters(`run:${priorRun.id}`) diff --git a/src/main/runtime/rpc/orchestration-session-coordinator.test.ts b/src/main/runtime/rpc/orchestration-session-coordinator.test.ts index 1a3312f2013..a60c3a16916 100644 --- a/src/main/runtime/rpc/orchestration-session-coordinator.test.ts +++ b/src/main/runtime/rpc/orchestration-session-coordinator.test.ts @@ -91,13 +91,30 @@ describe('a structured chat coordinates through the same verbs as a terminal', ( to: WORKER_HANDLE, subject: 'more' }) - expect(outbound).toMatchObject({ from_handle: ADDRESS_X, run_id: runId }) + expect(outbound).toMatchObject({ + from_handle: ADDRESS_X, + to_handle: `dispatch:${idOf(dispatch)}`, + run_id: runId + }) const replied = await as(SESSION_X, 'orchestration.reply', { id: idOf(inbound), body: 'ack' }) - expect(replied).toMatchObject({ message: { from_handle: ADDRESS_X, to_handle: WORKER_HANDLE } }) + expect(replied).toMatchObject({ + message: { + from_handle: ADDRESS_X, + to_handle: `dispatch:${idOf(dispatch)}`, + run_id: runId, + thread_id: idOf(inbound) + } + }) + expect(await as(undefined, 'orchestration.check', { terminal: WORKER_HANDLE })).toMatchObject({ + runId, + dispatchId: idOf(dispatch), + count: 2, + messages: [{ id: idOf(outbound) }, { id: idOf(replied.message) }] + }) const { gate } = await as(SESSION_X, 'orchestration.gateCreate', { task: taskId, @@ -255,16 +272,55 @@ describe('a structured chat coordinates through the same verbs as a terminal', ( ) ) await vi.waitFor(() => expect(waiter).toHaveBeenCalledWith(`run:${runId}`, expect.anything())) + const deliver = vi.spyOn(h.db, 'getOrCreateRunDelivery') + const acknowledge = vi.spyOn(h.db, 'acknowledgeRunDelivery') await as(SESSION_Y, 'orchestration.runUse', { id: runId }) - expect(await waiting).toMatchObject({ - ok: false, - error: { - code: 'consumer_fenced', - message: 'This mailbox consumer was replaced while waiting.' - } + const fenced = await waiting + expect(fenced).toMatchObject({ ok: false, error: { code: 'consumer_fenced' } }) + expect(fenced).not.toHaveProperty('result') + + const { message } = await as(undefined, 'orchestration.send', { + from: WORKER_HANDLE, + to: ADDRESS_Y, + subject: 'after takeover', + run: runId }) + expect( + await h.dispatch( + orchestrationRequest('orchestration.check', { run: runId }, { sessionId: SESSION_X }) + ) + ).toMatchObject({ ok: false, error: { code: 'consumer_fenced' } }) + expect(deliver).not.toHaveBeenCalled() + + const replacement = await as(SESSION_Y, 'orchestration.check', {}) + expect(replacement).toMatchObject({ + runId, + count: 1, + messages: [{ id: idOf(message), to_handle: `run:${runId}`, run_id: runId }] + }) + const ack = replacement.deliveryId + if (typeof ack !== 'string') { + throw new Error('Expected a Run Delivery') + } + expect( + await h.dispatch( + orchestrationRequest('orchestration.check', { run: runId, ack }, { sessionId: SESSION_X }) + ) + ).toMatchObject({ ok: false, error: { code: 'consumer_fenced' } }) + expect(acknowledge).not.toHaveBeenCalled() + expect(h.db.getDeliveryRaw(ack)?.acknowledged_at).toBeNull() + expect(await as(SESSION_Y, 'orchestration.check', {})).toMatchObject({ + deliveryId: ack, + replayed: true, + messages: [{ id: idOf(message) }] + }) + expect(await as(SESSION_Y, 'orchestration.check', { ack })).toMatchObject({ + acknowledged: ack, + count: 0 + }) + expect(h.db.getDeliveryRaw(ack)?.status).toBe('acknowledged') }) it('stops counting a coordinator Orca session id once an older binary rebinds the Run to a terminal', async () => { diff --git a/src/main/runtime/rpc/orchestration-session-run-binding.test.ts b/src/main/runtime/rpc/orchestration-session-run-binding.test.ts new file mode 100644 index 00000000000..0f6b65d6571 --- /dev/null +++ b/src/main/runtime/rpc/orchestration-session-run-binding.test.ts @@ -0,0 +1,138 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + mintStructuredWorkerHandle, + mintStructuredWorkerPaneKey, + structuredWorkerIdentities, + structuredWorkerProcessIncarnation +} from '../structured-worker-identity' +import { + createSessionCallerHarness, + orchestrationRequest, + resultOf, + idOf, + SESSION_X, + SESSION_Y, + type SessionCallerHarness +} from './orchestration-session-caller-test-fixture' + +const hostRef = vi.hoisted((): { current: unknown } => ({ current: null })) +vi.mock('../../native-chat/agent-session-wire/structured-agent-session-registry', () => ({ + getStructuredAgentSessionHost: () => hostRef.current +})) + +function deliveryId(result: Record): string { + if (typeof result.deliveryId === 'string') { + return result.deliveryId + } + throw new Error('Expected a Delivery') +} + +describe('structured lead Run binding through the RPC dispatcher', () => { + let h: SessionCallerHarness + let dispatchId: string + + function call(sessionId: string, method: string, params: Record = {}) { + return h + .dispatch(orchestrationRequest(`orchestration.${method}`, params, { sessionId })) + .then(resultOf) + } + + beforeEach(async () => { + h = createSessionCallerHarness(hostRef) + const handle = mintStructuredWorkerHandle() + const paneKey = mintStructuredWorkerPaneKey(SESSION_Y) + structuredWorkerIdentities.register({ + handle, + sessionId: SESSION_Y, + agent: 'claude', + paneKey, + processIncarnation: structuredWorkerProcessIncarnation(SESSION_Y), + worktreeId: 'folder-workspace', + hostScope: { kind: 'local', hostId: 'local' } + }) + const runId = idOf((await call(SESSION_X, 'runCreate', { objective: 'parent' })).run) + dispatchId = h.db.createDispatchContext({ + taskId: h.db.createTask({ runId, spec: 'structured lead' }).id, + assigneeHandle: handle, + assigneePaneKey: paneKey, + processIncarnation: structuredWorkerProcessIncarnation(SESSION_Y), + creator: { kind: 'session', orcaSessionId: SESSION_X }, + maxDepth: Number.MAX_SAFE_INTEGER + }).id + }) + afterEach(() => { + h.runtime.cancelMessageWaiters(`dispatch:${dispatchId}`) + h.close() + vi.restoreAllMocks() + }) + + it.each(['runCreate', 'runUse'] as const)( + 'cancels a parked session Dispatch check on %s', + async (method) => { + const params = + method === 'runCreate' + ? { objective: 'child' } + : { + id: h.db.createRun({ + objective: 'unbound', + coordinatorHandle: null, + coordinatorPaneKey: null + }).id + } + const wait = vi.spyOn(h.runtime, 'waitForMessage') + const waiting = call(SESSION_Y, 'check', { wait: true, timeoutMs: 500 }) + await vi.waitFor(() => + expect(wait).toHaveBeenCalledWith(`dispatch:${dispatchId}`, expect.anything()) + ) + const child = idOf((await call(SESSION_Y, method, params)).run) + expect(await waiting).toMatchObject({ cancelled: true, timedOut: false }) + const sent = await call(SESSION_X, 'send', { + to: `dispatch:${dispatchId}`, + subject: 'after bind' + }) + expect(sent.message).toMatchObject({ to_handle: `run:${child}`, run_id: child }) + expect(await call(SESSION_Y, 'check')).toMatchObject({ + messages: [{ subject: 'after bind' }] + }) + expect(h.db.getInbox()).toHaveLength(1) + } + ) + + it.each(['cancel', 'throw'] as const)( + 'replays a pre-bind ack receipt after the following wait ends by %s', + async (ending) => { + await call(SESSION_X, 'send', { to: `dispatch:${dispatchId}`, subject: 'ack me' }) + const ack = deliveryId(await call(SESSION_Y, 'check')) + const acknowledge = vi.spyOn(h.db, 'acknowledgeMailboxDelivery') + const wait = vi.spyOn(h.runtime, 'waitForMessage').mockImplementationOnce(async () => { + await call(SESSION_Y, 'runCreate', { objective: 'child' }) + if (ending === 'throw') { + throw new Error('wait transport interrupted') + } + return 'cancelled' + }) + const request = orchestrationRequest( + 'orchestration.check', + { ack, wait: true, timeoutMs: 500 }, + { + sessionId: SESSION_Y, + requestId: `bind-during-ack-${ending}` + } + ) + const first = await h.dispatch(request) + if (ending === 'throw') { + expect(first).toMatchObject({ ok: false, error: { code: 'runtime_error' } }) + } else { + expect(resultOf(first)).toMatchObject({ acknowledged: ack, count: 0, cancelled: true }) + } + const replay = resultOf(await h.dispatch(request)) + expect(replay).toMatchObject({ acknowledged: ack, count: 0, mutation: { replayed: true } }) + if (ending === 'throw') { + expect(replay.waitInterrupted).toBe('outcome_unknown') + } + expect(acknowledge).toHaveBeenCalledTimes(1) + expect(wait).toHaveBeenCalledTimes(1) + expect(h.db.getDeliveryRaw(ack)?.status).toBe('acknowledged') + } + ) +})