fix(orchestration): admit worker reports after a restart reissues the worker's handle

When Orca restarts while a worker's PTY survives in the daemon, the renderer
graph can sync before the daemon inventory re-adopts the PTY's exported
handle. If the pane's incarnation was not yet persisted, the pane gets a fresh
synthetic handle and the worker's env handle (ORCA_TERMINAL_HANDLE) stops
resolving. Every worker_done, escalation and ask from that same process was
then rejected with worker_identity_changed, because the sender's pane was
looked up only through the stale handle.

Report admission now recovers such a reporter the way worker show, output and
stop already do: by the Dispatch's exact recorded process incarnation. It only
applies when the declared handle names no terminal at all, the caller's env
names that handle and the Dispatch's own pane, and the incarnation still lives
in that pane. A replacement process in the pane, a scrubbed or foreign env, or
a handle that still names another pane are all refused as before.

src/main/runtime/orca-runtime-tests/restart-reissued-worker-handle.test.ts

@@ -0,0 +1,175 @@
+// A restarted runtime can reissue a surviving worker PTY's handle; the worker still reports with its env handle.
+import { withDurableRuntimeStore } from './runtime-durable-store-fixture'
+import { afterEach, describe, expect, it } from 'vitest'
+import './runtime-test-module-registration.spec'
+import { OrcaRuntimeService } from '../orca-runtime'
+import { OrchestrationDb } from '../orchestration/db/orchestration-db'
+import '../orchestration/db/contract-constants'
+import '../orchestration/db/messages/mailbox-routing-page'
+import '../orchestration/db/dispatch-row-writer'
+import '../orchestration/db/attempt-outcome-projection'
+import { createRootDispatch } from '../orchestration/db/root-dispatch-test-fixture'
+import { getDefaultWorkspaceSession } from '../../../shared/constants'
+import { createOrchestrationRpcHarness } from '../rpc/methods/orchestration/rpc-test-harness'
+import {
+  HEADLESS_LEAF_ID,
+  TEST_REPO_ID,
+  TEST_WORKTREE_ID,
+  TEST_WORKTREE_PATH,
+  makeHeadlessTerminalLayout,
+  makeRuntimeStoreWithWorkspaceSession
+} from './orca-runtime-test-fixtures.spec'
+
+const TAB = '44444444-4444-4444-8444-444444444444'
+const PTY = `${TEST_WORKTREE_ID}@@restored1`
+const EXPORTED_HANDLE = 'term_exported-before-restart'
+const PANE = `${TAB}:${HEADLESS_LEAF_ID}`
+const INCARNATION_ID = 'inc-restored'
+const HOST_SCOPE = JSON.stringify({ kind: 'local', hostId: 'local' })
+
+const databases: OrchestrationDb[] = []
+afterEach(() => {
+  for (const db of databases.splice(0)) {
+    db.close()
+  }
+})
+
+/** Restart with the worker PTY alive in the daemon; the renderer graph lands before inventory adoption. */
+async function restartWithSurvivingWorker(processIncarnationId = INCARNATION_ID) {
+  const session = {
+    ...getDefaultWorkspaceSession(),
+    activeRepoId: TEST_REPO_ID,
+    activeWorktreeId: TEST_WORKTREE_ID,
+    tabsByWorktree: {
+      [TEST_WORKTREE_ID]: [
+        {
+          id: TAB,
+          ptyId: PTY,
+          worktreeId: TEST_WORKTREE_ID,
+          title: 'Terminal',
+          customTitle: null,
+          color: null,
+          sortOrder: 0,
+          createdAt: 1
+        }
+      ]
+    },
+    terminalLayoutsByTabId: { [TAB]: makeHeadlessTerminalLayout({ [HEADLESS_LEAF_ID]: PTY }) },
+    // The worker started shortly before the restart, before its pane incarnation was persisted.
+    terminalPtyIncarnationsByPaneKey: {}
+  }
+  const db = new OrchestrationDb(':memory:')
+  databases.push(db)
+  const run = db.createRun({
+    objective: 'restart',
+    coordinatorHandle: 'term_coord',
+    coordinatorPaneKey: 'tab_coord:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'
+  })
+  const task = db.createTask({ spec: 'long worker task', runId: run.id })
+  const dispatch = createRootDispatch(
+    db,
+    task.id,
+    EXPORTED_HANDLE,
+    PANE,
+    undefined,
+    `${PTY}:${INCARNATION_ID}`
+  )
+  db.db
+    .prepare('UPDATE dispatch_contexts SET host_scope = ? WHERE id = ?')
+    .run(HOST_SCOPE, dispatch.id)
+
+  const { runtimeStore } = makeRuntimeStoreWithWorkspaceSession(session)
+  const runtime = new OrcaRuntimeService(withDurableRuntimeStore(runtimeStore))
+  runtime.setOrchestrationDb(db)
+  runtime.setPtyController({
+    write: () => true,
+    kill: () => true,
+    getForegroundProcess: async () => null,
+    listProcesses: async () => [
+      {
+        id: PTY,
+        incarnationId: processIncarnationId,
+        terminalHandle: EXPORTED_HANDLE,
+        title: 'claude',
+        cwd: TEST_WORKTREE_PATH,
+        worktreeId: TEST_WORKTREE_ID,
+        wslDistro: null
+      }
+    ]
+  })
+  runtime.syncWindowGraph(1, {
+    tabs: [
+      {
+        tabId: TAB,
+        worktreeId: TEST_WORKTREE_ID,
+        title: 'Terminal',
+        activeLeafId: HEADLESS_LEAF_ID,
+        layout: null
+      }
+    ],
+    leaves: [
+      {
+        tabId: TAB,
+        worktreeId: TEST_WORKTREE_ID,
+        leafId: HEADLESS_LEAF_ID,
+        paneRuntimeId: 1,
+        ptyId: PTY
+      }
+    ]
+  })
+  await runtime.listTerminals(`id:${TEST_WORKTREE_ID}`)
+  return { runtime, db, taskId: task.id, dispatchId: dispatch.id }
+}
+
+async function sendWorkerDone(
+  runtime: OrcaRuntimeService,
+  taskId: string,
+  dispatchId: string
+): Promise<unknown> {
+  const method = createOrchestrationRpcHarness().findMethod('orchestration.send')
+  const params = method.params?.parse({
+    from: EXPORTED_HANDLE,
+    subject: 'Done',
+    type: 'worker_done',
+    payload: JSON.stringify({ taskId, dispatchId, outcome: 'succeeded' })
+  })
+  return await method.handler(params, {
+    runtime,
+    // What the worker's CLI reads from the env its PTY was launched with.
+    orchestrationCompatibilityEvidence: {
+      terminalHandle: EXPORTED_HANDLE,
+      paneKey: PANE,
+      launchToken: 'launch-token'
+    }
+  })
+}
+
+describe('worker_done after a restart reissues the surviving worker PTY handle', () => {
+  it('strands the exported handle when the graph lands before inventory adoption', async () => {
+    const { runtime } = await restartWithSurvivingWorker()
+
+    expect(runtime.getTerminalPaneKey(EXPORTED_HANDLE)).toBeNull()
+    expect(runtime.getTerminalHandleForPaneKey(PANE)).not.toBe(EXPORTED_HANDLE)
+  })
+
+  it('settles worker_done from the same surviving process', async () => {
+    const { runtime, db, taskId, dispatchId } = await restartWithSurvivingWorker()
+
+    const result = await sendWorkerDone(runtime, taskId, dispatchId)
+
+    expect(result).not.toMatchObject({ lifecycle: { action: 'rejected' } })
+    expect(db.getDispatchContextById(dispatchId)?.status).toBe('completed')
+    expect(db.getTask(taskId)?.status).toBe('completed')
+  })
+
+  it('refuses worker_done when the pane now runs a replacement process', async () => {
+    const { runtime, db, taskId, dispatchId } = await restartWithSurvivingWorker('inc-replacement')
+
+    const result = await sendWorkerDone(runtime, taskId, dispatchId)
+
+    expect(result).toMatchObject({
+      lifecycle: { action: 'rejected', code: 'worker_identity_changed' }
+    })
+    expect(db.getDispatchContextById(dispatchId)?.status).toBe('dispatched')
+  })
+})

src/main/runtime/orchestration/workers/reports/reminted-worker-reporter.ts

@@ -0,0 +1,101 @@
+import type { OrchestrationCompatibilityEvidence } from '../../../../../shared/orchestration/orchestration-compatibility-evidence'
+import type { OrcaRuntimeService } from '../../../orca-runtime'
+import type { OrchestrationCompatibilityCallerAuthority } from '../../terminal-authority/orchestration-compatibility-authority'
+import type { OrchestrationDb } from '../../db/orchestration-db'
+import { isEquivalentPaneKey } from '../../db/pane-key-match'
+
+type RemintRuntime = Pick<
+  OrcaRuntimeService,
+  | 'getTerminalPaneKey'
+  | 'getTerminalProcessIncarnation'
+  | 'resolveTerminalHandleByProcessIncarnation'
+>
+
+export type RemintedWorkerReporter = {
+  dispatchId: string
+  handle: string
+  paneKey: string
+  processIncarnation: string
+}
+
+/**
+ * A worker's env handle goes stale when a restarted runtime reissues the handle of its surviving PTY.
+ * Recover the reporter only when its declared handle no longer names any terminal, its env names the
+ * Dispatch's own pane, and the Dispatch's exact recorded process incarnation is still live in that pane.
+ * A replacement process in the same pane has a new incarnation and is never recovered.
+ */
+export function resolveRemintedWorkerReporter(
+  runtime: RemintRuntime,
+  db: OrchestrationDb,
+  args: { from: string; evidence: OrchestrationCompatibilityEvidence | undefined }
+): RemintedWorkerReporter | null {
+  const { from, evidence } = args
+  if (!from.startsWith('term_') || runtime.getTerminalPaneKey(from)) {
+    return null
+  }
+  // The caller's own env must name the stale handle and the Dispatch pane; scrubbed or foreign env proves nothing.
+  if (evidence?.terminalHandle !== from || !evidence.paneKey) {
+    return null
+  }
+  const dispatch = db.getActiveDispatchForIdentity(from)
+  if (
+    !dispatch ||
+    dispatch.assignee_handle !== from ||
+    !dispatch.assignee_pane_key ||
+    !dispatch.process_incarnation ||
+    !isEquivalentPaneKey(dispatch.assignee_pane_key, evidence.paneKey)
+  ) {
+    return null
+  }
+  const handle = runtime.resolveTerminalHandleByProcessIncarnation(
+    dispatch.process_incarnation,
+    dispatch.host_scope
+  )
+  if (!handle || handle === from) {
+    return null
+  }
+  const paneKey = runtime.getTerminalPaneKey(handle)
+  const processIncarnation = runtime.getTerminalProcessIncarnation(handle)
+  if (
+    !paneKey ||
+    !processIncarnation ||
+    !db.isDispatchProcessCurrent({ dispatchId: dispatch.id, paneKey, processIncarnation })
+  ) {
+    return null
+  }
+  return { dispatchId: dispatch.id, handle, paneKey, processIncarnation }
+}
+
+/**
+ * The pane and process a terminal reporter speaks for. Attested hook identity wins; otherwise the
+ * declared handle's own terminal; otherwise, for a restart-reissued handle, its exact-incarnation remint.
+ * The incarnation is re-read at admission so a process replaced after entry is still refused.
+ */
+export function resolveWorkerReporterIdentity(
+  runtime: RemintRuntime,
+  db: OrchestrationDb,
+  args: {
+    from: string
+    attested:
+      | Pick<OrchestrationCompatibilityCallerAuthority, 'paneKey' | 'processIncarnation'>
+      | undefined
+    hasSessionCaller: boolean
+    evidence: OrchestrationCompatibilityEvidence | undefined
+  }
+): { paneKey: string | null; processIncarnation: () => string | null } {
+  const { from, attested } = args
+  const declaredPaneKey = attested?.paneKey ?? runtime.getTerminalPaneKey(from)
+  const remint = () =>
+    args.hasSessionCaller || attested || declaredPaneKey
+      ? null
+      : resolveRemintedWorkerReporter(runtime, db, { from, evidence: args.evidence })
+  const reminted = remint()
+  return {
+    paneKey: declaredPaneKey ?? reminted?.paneKey ?? null,
+    processIncarnation: () =>
+      attested?.processIncarnation ??
+      (reminted
+        ? (remint()?.processIncarnation ?? null)
+        : runtime.getTerminalProcessIncarnation(from))
+  }
+}

src/main/runtime/rpc/methods/orchestration/messaging/ask-methods.ts

@@ -13,6 +13,7 @@ import {
 import { resolveOrchestrationParty } from '../../../../orchestration/addressing/orchestration-party'
 import { assertLifecycleCallerIsNotAnotherParty } from './lifecycle-caller-fence'
 import { assertWorkerCanReport } from '../../../../orchestration/workers/reports/worker-report-admission'
+import { resolveWorkerReporterIdentity } from '../../../../orchestration/workers/reports/reminted-worker-reporter'
 import { finishedWorkerDispatch, refuseFinishedRemoteWorker } from './finished-worker-mail'
 import { isDispatchAssignee } from '../../../../orchestration/db/dispatch-context/dispatch-lookup'
 import {
@@ -52,7 +53,14 @@ export const ORCHESTRATION_ASK_METHODS = [
       const from = params.from ?? 'unknown'
       // Echo the clamped budget on every return so callers report time actually waited.
       const timeoutMs = clampOrchestrationAskTimeoutMs(params.timeoutMs)
-      const paneKey = runtime.getTerminalPaneKey(from) ?? undefined
+      // A runtime restart can reissue a surviving worker PTY's handle; recover it by exact incarnation.
+      const reporter = resolveWorkerReporterIdentity(runtime, db, {
+        from,
+        attested: undefined,
+        hasSessionCaller: Boolean(orchestrationCaller),
+        evidence: orchestrationCompatibilityEvidence
+      })
+      const paneKey = reporter.paneKey ?? undefined
       if (!orchestrationCaller) {
         assertLifecycleCallerIsNotAnotherParty(runtime, {
           from,
@@ -107,7 +115,7 @@ export const ORCHESTRATION_ASK_METHODS = [
             db.isDispatchProcessCurrent({
               dispatchId: workingDispatch.id,
               paneKey: paneKey ?? null,
-              processIncarnation: runtime.getTerminalProcessIncarnation(from)
+              processIncarnation: reporter.processIncarnation()
             })
         })
       }

src/main/runtime/rpc/methods/orchestration/messaging/reminted-worker-report.test.ts

@@ -0,0 +1,189 @@
+// A restarted runtime can reissue a surviving worker PTY's handle while the worker keeps reporting with its env handle.
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import type { RpcContext } from '../../../core'
+import type { OrchestrationDb } from '../../../../orchestration/db/orchestration-db'
+import type { OrcaRuntimeService } from '../../../../orca-runtime'
+import type { DispatchContextRow } from '../../../../orchestration/db/orchestration-records'
+import type { OrchestrationCompatibilityEvidence } from '../../../../../../shared/orchestration/orchestration-compatibility-evidence'
+import { createOrchestrationRpcHarness } from '../rpc-test-harness'
+import { createRootDispatch } from '../../../../orchestration/db/root-dispatch-test-fixture'
+
+const STALE_HANDLE = 'term_exported-before-restart'
+const REMINTED_HANDLE = 'term_reissued-after-restart'
+const WORKER_PANE = 'tab_worker:11111111-1111-4111-8111-111111111111'
+const OTHER_PANE = 'tab_other:22222222-2222-4222-8222-222222222222'
+const PTY_ID = 'wt::/repo/worker@@restored1'
+const INCARNATION = `${PTY_ID}:inc-original`
+const HOST_SCOPE = JSON.stringify({ kind: 'local', hostId: 'local' })
+
+describe('worker reports after a restart reissues the worker terminal handle', () => {
+  const harness = createOrchestrationRpcHarness()
+  const { coordinatorPaneKey } = harness
+  let db: OrchestrationDb
+  let runtime: OrcaRuntimeService
+  let ctx: RpcContext
+
+  afterEach(() => {
+    harness.cleanup()
+  })
+
+  /** The PTY survived the restart; only its handle changed. `paneIncarnation` is what now runs in the pane. */
+  function setup(paneIncarnation = INCARNATION): { taskId: string; dispatch: DispatchContextRow } {
+    ;({ db, runtime, ctx } = harness.setup())
+    const task = db.createTask({ spec: 'long worker task spanning a restart' })
+    const dispatch = createRootDispatch(
+      db,
+      task.id,
+      STALE_HANDLE,
+      WORKER_PANE,
+      undefined,
+      INCARNATION
+    )
+    db.db
+      .prepare('UPDATE dispatch_contexts SET host_scope = ? WHERE id = ?')
+      .run(HOST_SCOPE, dispatch.id)
+    vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) =>
+      handle === 'term_coord' ? coordinatorPaneKey : handle === REMINTED_HANDLE ? WORKER_PANE : null
+    )
+    vi.mocked(runtime.getTerminalProcessIncarnation).mockImplementation((handle) =>
+      handle === REMINTED_HANDLE ? paneIncarnation : null
+    )
+    vi.spyOn(runtime, 'getTerminalHandleForPaneKey').mockImplementation((paneKey) =>
+      paneKey === WORKER_PANE ? REMINTED_HANDLE : null
+    )
+    vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation(
+      (incarnation, hostScope) =>
+        incarnation === paneIncarnation && hostScope === HOST_SCOPE ? REMINTED_HANDLE : null
+    )
+    vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {})
+    return { taskId: task.id, dispatch }
+  }
+
+  function workerEnv(
+    evidence: OrchestrationCompatibilityEvidence = {
+      terminalHandle: STALE_HANDLE,
+      paneKey: WORKER_PANE,
+      launchToken: 'launch-token'
+    }
+  ): RpcContext {
+    return { ...ctx, orchestrationCompatibilityEvidence: evidence }
+  }
+
+  async function sendWorkerDone(
+    taskId: string,
+    dispatchId: string,
+    callerCtx: RpcContext
+  ): Promise<unknown> {
+    return await harness.call(
+      'orchestration.send',
+      {
+        from: STALE_HANDLE,
+        subject: 'Done',
+        type: 'worker_done',
+        payload: JSON.stringify({ taskId, dispatchId, outcome: 'succeeded' })
+      },
+      callerCtx
+    )
+  }
+
+  it('settles worker_done from the same process after its handle was reissued', async () => {
+    const { taskId, dispatch } = setup()
+
+    const result = await sendWorkerDone(taskId, dispatch.id, workerEnv())
+
+    expect(result).not.toMatchObject({ lifecycle: { action: 'rejected' } })
+    expect(result).toMatchObject({ message: { type: 'worker_done' } })
+    expect(db.getDispatchContextById(dispatch.id)?.status).toBe('completed')
+    expect(db.getTask(taskId)?.status).toBe('completed')
+  })
+
+  it('admits an escalation from the same process after its handle was reissued', async () => {
+    const { taskId, dispatch } = setup()
+
+    const escalation = await harness.call(
+      'orchestration.send',
+      {
+        from: STALE_HANDLE,
+        subject: 'Blocked',
+        type: 'escalation',
+        payload: JSON.stringify({ taskId, dispatchId: dispatch.id })
+      },
+      workerEnv()
+    )
+
+    expect(escalation).not.toMatchObject({ lifecycle: { action: 'rejected' } })
+    expect(escalation).toMatchObject({ message: { type: 'escalation' } })
+  })
+
+  it('still refuses a replacement process running in the worker pane', async () => {
+    const { taskId, dispatch } = setup(`${PTY_ID}:inc-replacement`)
+
+    const result = await sendWorkerDone(taskId, dispatch.id, workerEnv())
+
+    expect(result).toMatchObject({
+      lifecycle: { action: 'rejected', code: 'worker_identity_changed' }
+    })
+    expect(db.getDispatchContextById(dispatch.id)?.status).toBe('dispatched')
+  })
+
+  it.each([
+    ['no env evidence', undefined],
+    ['env naming another pane', { terminalHandle: STALE_HANDLE, paneKey: OTHER_PANE }],
+    ['env naming another handle', { terminalHandle: 'term_other', paneKey: WORKER_PANE }]
+  ] as const)('refuses a stale-handle report with %s', async (_label, evidence) => {
+    const { taskId, dispatch } = setup()
+    const callerCtx = evidence ? workerEnv(evidence) : ctx
+
+    const result = await sendWorkerDone(taskId, dispatch.id, callerCtx)
+
+    expect(result).toMatchObject({
+      lifecycle: { action: 'rejected', code: 'worker_identity_changed' }
+    })
+    expect(db.getDispatchContextById(dispatch.id)?.status).toBe('dispatched')
+  })
+
+  it('does not reinterpret a handle that still names another live pane', async () => {
+    const { taskId, dispatch } = setup()
+    vi.mocked(runtime.getTerminalPaneKey).mockImplementation((handle) =>
+      handle === 'term_coord'
+        ? coordinatorPaneKey
+        : handle === STALE_HANDLE
+          ? OTHER_PANE
+          : handle === REMINTED_HANDLE
+            ? WORKER_PANE
+            : null
+    )
+
+    const result = await sendWorkerDone(taskId, dispatch.id, workerEnv())
+
+    expect(result).toMatchObject({
+      lifecycle: { action: 'rejected', code: 'worker_identity_changed' }
+    })
+  })
+
+  it('creates a question from the same process after its handle was reissued', async () => {
+    setup()
+    vi.spyOn(runtime, 'waitForMessage').mockResolvedValue('timed_out')
+
+    const asked = await harness.call(
+      'orchestration.ask',
+      { from: STALE_HANDLE, question: 'Proceed?', timeoutMs: 1 },
+      workerEnv()
+    )
+
+    expect(asked).toMatchObject({ messageId: expect.stringMatching(/^msg_/), timedOut: true })
+  })
+
+  it('refuses a question from a replacement process in the worker pane', async () => {
+    setup(`${PTY_ID}:inc-replacement`)
+    vi.spyOn(runtime, 'waitForMessage').mockResolvedValue('timed_out')
+
+    await expect(
+      harness.call(
+        'orchestration.ask',
+        { from: STALE_HANDLE, question: 'Proceed?', timeoutMs: 1 },
+        workerEnv()
+      )
+    ).rejects.toMatchObject({ code: 'worker_identity_changed' })
+  })
+})

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

@@ -25,6 +25,7 @@ import { sendGroupMessage } from './send-group'
 import { sendFederatedControlMail } from '../../../../orchestration/legacy-remote-home/send-control-mail'
 import { orchestrationCallerIdentity } from '../runs/run-scope'
 import { assertLifecycleCallerIsNotAnotherParty } from './lifecycle-caller-fence'
+import { resolveWorkerReporterIdentity } from '../../../../orchestration/workers/reports/reminted-worker-reporter'
 import { planWorkerFollowUp, startWorkerFollowUp } from './worker-follow-up'
 import { finishedWorkerSend, refuseFinishedRemoteWorker } from './finished-worker-mail'
 import { relayedDeliveryUnknownWarning, withDeliveryHoldWarning } from './send-delivery-hold'
@@ -71,10 +72,16 @@ export const ORCHESTRATION_SEND_METHODS = [
             ? orchestrationCompatibilityCallerAuthority
             : undefined
         // Why: attested hook identity survives graph remount; caller params never supply lifecycle authority.
+        const reporter = resolveWorkerReporterIdentity(runtime, db, {
+          from,
+          attested: attestedCaller,
+          hasSessionCaller: Boolean(orchestrationCaller),
+          evidence: orchestrationCompatibilityEvidence
+        })
         const sender = orchestrationCallerIdentity(runtime, {
           handle: from,
           session: orchestrationCaller,
-          paneKey: attestedCaller?.paneKey ?? runtime.getTerminalPaneKey(from)
+          paneKey: reporter.paneKey
         })
         const senderPaneKey = sender.paneKey ?? undefined
         // Why: a session caller was already bound to its own identity at the dispatch entry.
@@ -85,10 +92,7 @@ export const ORCHESTRATION_SEND_METHODS = [
             evidence: orchestrationCompatibilityEvidence
           })
         }
-        const processIncarnation = () =>
-          attestedCaller?.processIncarnation ??
-          runtime.getTerminalProcessIncarnation(from) ??
-          undefined
+        const processIncarnation = () => reporter.processIncarnation() ?? undefined
         const remoteAttachment = senderPaneKey
           ? db.findActiveRemoteAttachmentForPane(senderPaneKey)
           : undefined