fix(orchestration): refuse sends to a Run whose home is another computer (#33005)

* fix(orchestration): refuse send to a Run whose home is another computer

A remote Dispatch's Run id also names a mirror row on the worker's host. Once the
worker lost its link to the home, send --to run:<id> wrote into that mirror and the
message reached no one. Now the host refuses with run_home_unreachable, and a linked
worker that names its own Run home is delivered there.

Reported by @cookapps-jhlee8 in #32292.

* fix(orchestration): refuse only mirror Runs, and answer coordinator mail at its Run home

A loopback runtime's attachment names a real local Run, so the refusal now keys on the
Run row's remote home, not on any attachment. A reply to coordinator mail imported under
a remote Run relays to the home through the replying pane's Dispatch link, and is refused
with run_home_unreachable when that link is gone.

* fix(orchestration): a remote reply answers only mail sent to the replier's own Dispatch

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
   })

src/main/runtime/rpc/errors.ts

@@ -114,6 +114,7 @@ const STRUCTURED_RUNTIME_PASSTHROUGH_CODES: ReadonlySet<string> = 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',

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<string, unknown>
 ): RpcRequest {
   return {

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,

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,

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<unknown> {
+  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 }
+}

src/main/runtime/rpc/methods/orchestration/messaging/send-remote.ts

@@ -15,6 +15,7 @@ type SendParamsInput = z.infer<typeof SendParams>
 
 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<unknown> {
+}) {
   const { params, runtime, db, from, senderPaneKey, remoteAttachment } = args
-  rejectFederatedExplicitTarget(params)
+  rejectFederatedExplicitTarget(params, remoteAttachment.home_run_id)
   assertWorkerCanReport({
     dispatchId: remoteAttachment.dispatch_id,
     from,

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<OrcaRuntimeService['getOrchestrationDb']>,
+  runId: string
+): void {
+  if (db.getRun(runId)?.home_database === 'remote') {
+    throw remoteRunHomeUnreachable(db, runId)
+  }
+}
+
+export function remoteRunHomeUnreachable(
+  db: ReturnType<OrcaRuntimeService['getOrchestrationDb']>,
+  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.'