diff --git a/src/main/runtime/orchestration/db/federation/remote-dispatch-attachment-create.ts b/src/main/runtime/orchestration/db/federation/remote-dispatch-attachment-create.ts index 2e58738848b..a319b55f0e0 100644 --- a/src/main/runtime/orchestration/db/federation/remote-dispatch-attachment-create.ts +++ b/src/main/runtime/orchestration/db/federation/remote-dispatch-attachment-create.ts @@ -101,6 +101,20 @@ export function getRemoteDispatchAttachment( .get(dispatchId) as RemoteDispatchAttachmentRow | undefined } +/** The newest remote Dispatch this host runs for a Run whose home is another computer. */ +export function findRemoteAttachmentForHomeRun( + this: OrchestrationDb, + runId: string +): RemoteDispatchAttachmentRow | undefined { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The schema-pinned attachment projection returns RemoteDispatchAttachmentRow rows; the adapter exposes unknown. + return this.db + .prepare( + `SELECT ${REMOTE_DISPATCH_ATTACHMENT_COLUMN_LIST} FROM remote_dispatch_attachments + WHERE home_run_id = ? ORDER BY rowid DESC LIMIT 1` + ) + .get(runId) as RemoteDispatchAttachmentRow | undefined +} + export function recordRemoteAttachmentStage( this: OrchestrationDb, params: { @@ -180,6 +194,7 @@ export function updateRemoteAttachmentSetupEvidence( export type RemoteDispatchAttachmentCreateMethods = { createRemoteDispatchAttachment: typeof createRemoteDispatchAttachment getRemoteDispatchAttachment: typeof getRemoteDispatchAttachment + findRemoteAttachmentForHomeRun: typeof findRemoteAttachmentForHomeRun recordRemoteAttachmentStage: typeof recordRemoteAttachmentStage updateRemoteAttachmentSetupEvidence: typeof updateRemoteAttachmentSetupEvidence } @@ -188,6 +203,7 @@ export function attachRemoteDispatchAttachmentCreate(ctor: { prototype: object } Object.assign(ctor.prototype, { createRemoteDispatchAttachment, getRemoteDispatchAttachment, + findRemoteAttachmentForHomeRun, recordRemoteAttachmentStage, updateRemoteAttachmentSetupEvidence }) diff --git a/src/main/runtime/rpc/errors.ts b/src/main/runtime/rpc/errors.ts index affc97b5cc4..ebc0bdb6347 100644 --- a/src/main/runtime/rpc/errors.ts +++ b/src/main/runtime/rpc/errors.ts @@ -114,6 +114,7 @@ const STRUCTURED_RUNTIME_PASSTHROUGH_CODES: ReadonlySet = new Set([ AGENT_LAUNCH_SURFACE_UNAVAILABLE_CODE, 'worktree_id_requires_full_path', 'run_not_found', + 'run_home_unreachable', 'run_required', 'stable_pane_required', 'consumer_fenced', diff --git a/src/main/runtime/rpc/methods/orchestration/federation/federated-message-targeting.test.ts b/src/main/runtime/rpc/methods/orchestration/federation/federated-message-targeting.test.ts index 6d92c69e4b9..b9bd02cd2cd 100644 --- a/src/main/runtime/rpc/methods/orchestration/federation/federated-message-targeting.test.ts +++ b/src/main/runtime/rpc/methods/orchestration/federation/federated-message-targeting.test.ts @@ -17,38 +17,8 @@ describe('orchestration federated message targeting', () => { }) it('rejects explicit send and ask targets without enqueueing a relay', async () => { - db = new OrchestrationDb(':memory:') - runtime = new OrcaRuntimeService() - runtime.setOrchestrationDb(db) - const paneKey = 'tab_worker:leaf_worker' - const processIncarnation = 'worker_epoch:pty:1' + ;({ db, runtime } = attachedRemoteWorker()) const dispatchId = 'ctx_remote_targeting' - vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue(paneKey) - vi.spyOn(runtime, 'getTerminalProcessIncarnation').mockReturnValue(processIncarnation) - db.createRemoteDispatchAttachment({ - runId: 'run-home', - dispatchId, - taskId: 'task_remote_targeting', - homePeerFingerprint: 'home_peer', - protocolVersion: 1, - runtimeEpoch: runtime.getRuntimeId(), - mutationReceipt: { - callerFingerprint: 'home_peer', - requestId: 'attach_request', - method: 'orchestration.federationAttachStart', - payloadHash: 'attach_payload' - } - }) - db.prepareRemoteAttachmentAuthority({ - dispatchId, - paneKey, - processIncarnation, - worktreeId: 'repo::remote-worktree', - terminalHandle: 'term_remote_worker', - setupState: 'not_applicable', - effects: [] - }) - db.markRemoteAttachmentReady(dispatchId) const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) const requests: RpcRequest[] = [ request('send_to', 'orchestration.send', { @@ -86,11 +56,251 @@ describe('orchestration federated message targeting', () => { db.listFederationRelay({ dispatchId, direction: 'to_home', afterSequence: 0 }) ).toHaveLength(0) }) + + it('delivers a send that names its own Run home to that home', async () => { + ;({ db, runtime } = attachedRemoteWorker()) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + + for (const [id, target] of [ + ['home_to', { to: 'run:run-home' }], + ['home_run', { run: 'run-home' }] + ] as const) { + await expect( + dispatcher.dispatch( + request(id, 'orchestration.send', { + from: 'term_remote_worker', + ...target, + subject: 'Progress' + }) + ) + ).resolves.toMatchObject({ ok: true }) + } + await expect( + dispatcher.dispatch( + request('home_ask', 'orchestration.ask', { + from: 'term_remote_worker', + to: 'run:run-home', + question: 'Which branch?', + timeoutMs: 0 + }) + ) + ).resolves.toMatchObject({ ok: true, result: { timedOut: true } }) + expect( + db + .listFederationRelay({ + dispatchId: 'ctx_remote_targeting', + direction: 'to_home', + afterSequence: 0 + }) + .map((item) => item.kind) + ).toEqual(['status', 'status', 'question']) + expect(db.getInbox()).toHaveLength(0) + }) + + it('refuses a Run a remote Dispatch here reports to once the link to it is lost', async () => { + ;({ db, runtime } = attachedRemoteWorker()) + // The agent now runs in a pane its remote Dispatch was never attached to. + vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue('tab_other:leaf_other') + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + + for (const [id, target] of [ + ['lost_to', { to: 'run:run-home' }], + ['lost_run', { run: 'run-home' }] + ] as const) { + await expect( + dispatcher.dispatch( + request(id, 'orchestration.send', { + from: 'term_remote_worker', + ...target, + type: 'worker_done', + subject: 'Done', + payload: JSON.stringify({ outcome: 'succeeded' }) + }) + ) + ).resolves.toMatchObject({ + ok: false, + error: { + code: 'run_home_unreachable', + message: + "Run run-home lives on the computer that coordinates it, which isn't reachable from here: this terminal is no longer attached to its Dispatch ctx_remote_targeting. Nothing was sent." + } + }) + } + expect(db.getInbox()).toHaveLength(0) + expect( + db.listFederationRelay({ + dispatchId: 'ctx_remote_targeting', + direction: 'to_home', + afterSequence: 0 + }) + ).toHaveLength(0) + }) + + it('still delivers to a local Run no remote Dispatch reports to', async () => { + ;({ db, runtime } = attachedRemoteWorker()) + vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue('tab_other:leaf_other') + const local = db.createRun({ + objective: 'local run', + coordinatorHandle: 'term_coordinator', + coordinatorPaneKey: 'tab_coordinator:leaf_coordinator' + }) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + + await expect( + dispatcher.dispatch( + request('local_to', 'orchestration.send', { + from: 'term_local_sender', + to: `run:${local.id}`, + subject: 'Hello' + }) + ) + ).resolves.toMatchObject({ ok: true }) + expect(db.getInbox().map((message) => message.to_handle)).toEqual([`run:${local.id}`]) + }) + + it('delivers to a local Run a loopback remote Dispatch also reports to', async () => { + const loopbackDb = new OrchestrationDb(':memory:') + db = loopbackDb + runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb(loopbackDb) + vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue('tab_coordinator:leaf_coordinator') + const local = loopbackDb.createRun({ + objective: 'loopback run', + coordinatorHandle: 'term_coordinator', + coordinatorPaneKey: 'tab_coordinator:leaf_coordinator' + }) + attachRemoteDispatch(loopbackDb, runtime, local.id) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + + await expect( + dispatcher.dispatch( + request('loopback_run', 'orchestration.send', { + from: 'term_coordinator', + run: local.id, + subject: 'Hello' + }) + ) + ).resolves.toMatchObject({ ok: true }) + expect(loopbackDb.getInbox().map((message) => message.run_id)).toEqual([local.id]) + }) + + it('answers coordinator mail at the Run home, or refuses once the link is lost', async () => { + ;({ db, runtime } = attachedRemoteWorker()) + const mirrorDb = db + const coordinatorMail = mirrorDb.insertMessage({ + id: 'msg_home_control', + runId: 'run-home', + from: 'term_coordinator', + to: 'dispatch:ctx_remote_targeting', + subject: 'Status?', + body: 'How far along?' + }) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + + await expect( + dispatcher.dispatch( + request('reply_linked', 'orchestration.reply', { + from: 'term_remote_worker', + id: coordinatorMail.id, + body: 'Halfway.' + }) + ) + ).resolves.toMatchObject({ ok: true, result: { message: { subject: 'Re: Status?' } } }) + const relayed = mirrorDb.listFederationRelay({ + dispatchId: 'ctx_remote_targeting', + direction: 'to_home', + afterSequence: 0 + }) + expect(relayed.map((item) => JSON.parse(item.payload).threadId)).toEqual([coordinatorMail.id]) + + vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue('tab_other:leaf_other') + await expect( + dispatcher.dispatch( + request('reply_lost', 'orchestration.reply', { + from: 'term_remote_worker', + id: coordinatorMail.id, + body: 'Done.' + }) + ) + ).resolves.toMatchObject({ ok: false, error: { code: 'run_home_unreachable' } }) + expect(mirrorDb.getInbox().map((message) => message.id)).toEqual([coordinatorMail.id]) + }) + + it("refuses a reply to mail sent to another worker's Dispatch of the same Run", async () => { + ;({ db, runtime } = attachedRemoteWorker()) + const mirrorDb = db + const otherWorkerMail = mirrorDb.insertMessage({ + id: 'msg_home_other_worker', + runId: 'run-home', + from: 'term_coordinator', + to: 'dispatch:ctx_other_worker', + subject: 'Status?', + body: 'How far along?' + }) + const dispatcher = new RpcDispatcher({ runtime, methods: ORCHESTRATION_METHODS }) + + await expect( + dispatcher.dispatch( + request('reply_other', 'orchestration.reply', { + from: 'term_remote_worker', + id: otherWorkerMail.id, + body: 'Not mine.' + }) + ) + ).resolves.toMatchObject({ ok: false, error: { code: 'run_home_unreachable' } }) + expect( + mirrorDb.listFederationRelay({ + dispatchId: 'ctx_remote_targeting', + direction: 'to_home', + afterSequence: 0 + }) + ).toHaveLength(0) + }) }) +/** A worker pane bound to remote Dispatch ctx_remote_targeting, whose Run home is `run-home`. */ +function attachedRemoteWorker(): { db: OrchestrationDb; runtime: OrcaRuntimeService } { + const db = new OrchestrationDb(':memory:') + const runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb(db) + vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue('tab_worker:leaf_worker') + attachRemoteDispatch(db, runtime, 'run-home') + return { db, runtime } +} + +function attachRemoteDispatch(db: OrchestrationDb, runtime: OrcaRuntimeService, runId: string) { + const processIncarnation = 'worker_epoch:pty:1' + const dispatchId = 'ctx_remote_targeting' + vi.spyOn(runtime, 'getTerminalProcessIncarnation').mockReturnValue(processIncarnation) + db.createRemoteDispatchAttachment({ + runId, + dispatchId, + taskId: 'task_remote_targeting', + homePeerFingerprint: 'home_peer', + protocolVersion: 1, + runtimeEpoch: runtime.getRuntimeId(), + mutationReceipt: { + callerFingerprint: 'home_peer', + requestId: 'attach_request', + method: 'orchestration.federationAttachStart', + payloadHash: 'attach_payload' + } + }) + db.prepareRemoteAttachmentAuthority({ + dispatchId, + paneKey: 'tab_worker:leaf_worker', + processIncarnation, + worktreeId: 'repo::remote-worktree', + terminalHandle: 'term_remote_worker', + setupState: 'not_applicable', + effects: [] + }) + db.markRemoteAttachmentReady(dispatchId) +} + function request( id: string, - method: 'orchestration.send' | 'orchestration.ask', + method: 'orchestration.send' | 'orchestration.ask' | 'orchestration.reply', params: Record ): RpcRequest { return { diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/ask-methods.ts b/src/main/runtime/rpc/methods/orchestration/messaging/ask-methods.ts index bb6ba8b1099..0c6ad865aad 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/ask-methods.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/ask-methods.ts @@ -51,7 +51,7 @@ export const ORCHESTRATION_ASK_METHODS = [ } const remoteAttachment = paneKey ? db.findActiveRemoteAttachmentForPane(paneKey) : undefined if (remoteAttachment && paneKey) { - rejectFederatedExplicitTarget(params) + rejectFederatedExplicitTarget(params, remoteAttachment.home_run_id) return askRemoteRunHome({ params: { ...params, timeoutMs }, runtime, 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 bedc1778272..24ca80267a1 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/message-methods.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/message-methods.ts @@ -14,6 +14,7 @@ import { resolveOrchestrationParty } from '../../../../orchestration/orchestrati import { recordReceiptBeforeNudge, replayMutationNudge } from './mutation-replay-nudge' import { resolveReplyRecipient } from './recipient-routing' import { helperQuestionAnswerer } from './helper-question-answerer' +import { replyToRemoteRunHome } from './reply-remote' import { ReplyParams, InboxParams, @@ -35,7 +36,8 @@ export const ORCHESTRATION_MESSAGE_METHODS = [ runtime, legacyCoordinatorRunId, recordMutationReceipt, - replayedMutationReceipt + replayedMutationReceipt, + signal } ) => { const replayNudge = readMutationReplayNudge(replayedMutationReceipt) @@ -138,6 +140,9 @@ export const ORCHESTRATION_MESSAGE_METHODS = [ ) } + if (db.getRun(original.run_id)?.home_database === 'remote') { + return await replyToRemoteRunHome({ runtime, db, original, params, signal }) + } const recipient = resolveReplyRecipient({ runtime, db, diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/reply-remote.ts b/src/main/runtime/rpc/methods/orchestration/messaging/reply-remote.ts new file mode 100644 index 00000000000..822c457bc8b --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/messaging/reply-remote.ts @@ -0,0 +1,51 @@ +import type { OrchestrationDb } from '../../../../orchestration/db' +import type { MessageRow } from '../../../../orchestration/types' +import type { OrcaRuntimeService } from '../../../../orca-runtime' +import { remoteRunHomeUnreachable } from '../routing' +import { sendRemoteMessage } from './send-remote' + +/** + * Coordinator mail imported under a remote Run's mirror is answered at the Run's home: through the + * replying pane's own Dispatch link, or not at all. + */ +export async function replyToRemoteRunHome(args: { + runtime: OrcaRuntimeService + db: OrchestrationDb + original: MessageRow + params: { body: string; from?: string; run?: string } + signal?: AbortSignal +}): Promise { + const { runtime, db, original, params } = args + const from = params.from ?? original.to_handle + const paneKey = runtime.getTerminalPaneKey(from) + const attachment = paneKey ? db.findActiveRemoteAttachmentForPane(paneKey) : undefined + if ( + !paneKey || + !attachment || + attachment.home_run_id !== original.run_id || + // Why: another worker of the same Run here must not answer mail sent to this one's Dispatch. + original.to_handle !== `dispatch:${attachment.dispatch_id}` + ) { + throw remoteRunHomeUnreachable(db, original.run_id) + } + const subject = `Re: ${original.subject}` + const sent = await sendRemoteMessage({ + params: { + from, + subject, + body: params.body, + threadId: original.thread_id ?? original.id, + ...(params.run ? { run: params.run } : {}) + }, + runtime, + db, + from, + senderPaneKey: paneKey, + remoteAttachment: attachment, + processIncarnation: runtime.getTerminalProcessIncarnation(from) ?? undefined, + signal: args.signal + }) + db.markAsRead([original.id]) + // Why `message`: every shipped CLI prints the reply's id from it. + return { message: { id: sent.relay.messageId, subject }, ...sent } +} diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/send-remote.ts b/src/main/runtime/rpc/methods/orchestration/messaging/send-remote.ts index 1f0b4369e94..3701ad1ce3f 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/send-remote.ts +++ b/src/main/runtime/rpc/methods/orchestration/messaging/send-remote.ts @@ -15,6 +15,7 @@ type SendParamsInput = z.infer type RemoteAttachment = { dispatch_id: string + home_run_id: string state: WorkerDispatchState protocol_version: number } @@ -28,9 +29,9 @@ export async function sendRemoteMessage(args: { remoteAttachment: RemoteAttachment processIncarnation?: string | null signal?: AbortSignal -}): Promise { +}) { const { params, runtime, db, from, senderPaneKey, remoteAttachment } = args - rejectFederatedExplicitTarget(params) + rejectFederatedExplicitTarget(params, remoteAttachment.home_run_id) assertWorkerCanReport({ dispatchId: remoteAttachment.dispatch_id, from, diff --git a/src/main/runtime/rpc/methods/orchestration/routing.ts b/src/main/runtime/rpc/methods/orchestration/routing.ts index 7bc565e6333..7de187de0c8 100644 --- a/src/main/runtime/rpc/methods/orchestration/routing.ts +++ b/src/main/runtime/rpc/methods/orchestration/routing.ts @@ -62,6 +62,11 @@ export function resolveMessageRun( ) } const targetRunId = params.to?.startsWith('run:') ? params.to.slice('run:'.length) : undefined + for (const namedRunId of new Set([params.runId, targetRunId])) { + if (namedRunId) { + rejectRemotelyHomedRun(db, namedRunId) + } + } const resolvedRunId = params.runId ?? targetRunId ?? dispatch?.run_id let run = resolvedRunId ? db.getRun(resolvedRunId) : undefined @@ -151,8 +156,41 @@ export function interruptedAcknowledgedCheck( } } -export function rejectFederatedExplicitTarget(params: { to?: string; run?: string }): void { - if (params.to || params.run) { +/** + * A `remote` Run row is only this host's mirror of a Run that lives on its coordinator's computer; + * mail written into it reaches no one (#32292). + */ +export function rejectRemotelyHomedRun( + db: ReturnType, + runId: string +): void { + if (db.getRun(runId)?.home_database === 'remote') { + throw remoteRunHomeUnreachable(db, runId) + } +} + +export function remoteRunHomeUnreachable( + db: ReturnType, + runId: string +): OrchestrationError { + const dispatchId = db.findRemoteAttachmentForHomeRun(runId)?.dispatch_id + const detached = dispatchId + ? `this terminal is no longer attached to its Dispatch ${dispatchId}` + : 'this terminal is not attached to one of its Dispatches' + return new OrchestrationError( + 'run_home_unreachable', + `Run ${runId} lives on the computer that coordinates it, which isn't reachable from here: ${detached}. Nothing was sent.` + ) +} + +/** A federated worker reaches only its Run home; naming that home is the same as omitting it. */ +export function rejectFederatedExplicitTarget( + params: { to?: string; run?: string }, + homeRunId: string +): void { + const namesOnlyHome = + (!params.to || params.to === `run:${homeRunId}`) && (!params.run || params.run === homeRunId) + if (!namesOnlyHome) { throw new OrchestrationError( 'invalid_argument', 'Federated Dispatch messages route to their Run home; omit --to and --run.'