diff --git a/src/main/runtime/orca-runtime-tests/restart-reissued-worker-handle.test.ts b/src/main/runtime/orca-runtime-tests/restart-reissued-worker-handle.test.ts new file mode 100644 index 00000000000..cda76d2de15 --- /dev/null +++ b/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 { + 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') + }) +}) diff --git a/src/main/runtime/orchestration/workers/reports/reminted-worker-reporter.ts b/src/main/runtime/orchestration/workers/reports/reminted-worker-reporter.ts new file mode 100644 index 00000000000..53fcc35039b --- /dev/null +++ b/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 + | 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)) + } +} 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 ee868bd05cb..ff6ce3422ee 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/ask-methods.ts +++ b/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() }) }) } diff --git a/src/main/runtime/rpc/methods/orchestration/messaging/reminted-worker-report.test.ts b/src/main/runtime/rpc/methods/orchestration/messaging/reminted-worker-report.test.ts new file mode 100644 index 00000000000..e413b6411ad --- /dev/null +++ b/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 { + 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' }) + }) +}) 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 7899cea2921..97281fd9378 100644 --- a/src/main/runtime/rpc/methods/orchestration/messaging/send-methods.ts +++ b/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