From 5813cb640ed342e872dcec03aee0ee7ef7536e7f Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 28 Sep 2026 02:24:04 -0700 Subject: [PATCH] fix(orchestration): reland process-incarnation reap for stale worker terminal handles (#23583) * fix(orchestration): reland process-incarnation reap for stale worker terminal handles Relands the orchestration part of #18790, which #22601 reverted in full because that squash commit also bundled an unannounced agent. Only the worker terminal-reap fix returns; no agent catalog changes. When a worker's saved terminal handle stops resolving while its process is still running, worker-show/read, worker-stop and worker-release now re-mint a live handle from the recorded process incarnation (exact pty id + incarnation id, same host scope) and act on it, instead of reporting the terminal missing and leaving the process running on the host. Fixes STA-8493 * test(orchestration): spy on the public incarnation methods instead of casting the runtime --- ...me-get-orchestration-dispatch-authority.ts | 35 ++ ...untime-terminal-handle-incarnation.test.ts | 139 +++++ .../worker-terminal-process-liveness.test.ts | 70 +++ .../worker-terminal-process-liveness.ts | 28 +- ...orker-release-incarnation-fallback.test.ts | 534 ++++++++++++++++++ ...ion-worker-release-reap-fixed.func.test.ts | 358 ++++++++++++ .../worker/structured-worker-release-stop.ts | 8 +- .../orchestration/worker/worker-control.ts | 5 +- .../worker/worker-observation.ts | 66 ++- .../worker/worker-release-completion.ts | 65 ++- ...worker-release-runtime-incarnation.test.ts | 79 +++ .../orchestration/worker/worker-stop.ts | 3 +- .../worker/worker-terminal-release-lease.ts | 14 +- 13 files changed, 1344 insertions(+), 60 deletions(-) create mode 100644 src/main/runtime/orchestration/worker-terminal-process-liveness.test.ts create mode 100644 src/main/runtime/rpc/methods/orchestration-worker-release-incarnation-fallback.test.ts create mode 100644 src/main/runtime/rpc/methods/orchestration-worker-release-reap-fixed.func.test.ts create mode 100644 src/main/runtime/rpc/methods/orchestration/worker/worker-release-runtime-incarnation.test.ts diff --git a/src/main/runtime/orca-runtime-get-orchestration-dispatch-authority.ts b/src/main/runtime/orca-runtime-get-orchestration-dispatch-authority.ts index d61374c3466..5eb0bffb8e8 100644 --- a/src/main/runtime/orca-runtime-get-orchestration-dispatch-authority.ts +++ b/src/main/runtime/orca-runtime-get-orchestration-dispatch-authority.ts @@ -22,6 +22,7 @@ import { getAppEnvironment } from '../../shared/app-environment' import type { FleetAgentStatusEvidence } from '../../shared/orchestration-fleet-agent-status-evidence' import { readOrchestrationFleetAgentStatusSnapshot } from './orchestration-fleet-agent-status-snapshot' import { resolveStructuredWorkerAuthority } from './structured-worker-authority' +import { matchesProcessIncarnation } from './orchestration/worker-terminal-process-liveness' export class OrcaRuntimeWithGetOrchestrationDispatchAuthority extends OrcaRuntimeWithVerifyOrchestrationCompatibilityCaller { /** Every pane key this PTY could be addressed by, including restored receipts. */ @@ -270,4 +271,38 @@ export class OrcaRuntimeWithGetOrchestrationDispatchAuthority extends OrcaRuntim : undefined }) } + /** + * Recover a live terminal handle for a worker whose durable handle stopped resolving + * (renderer graph epoch bump / handle invalidation) while its PTY is still tracked. Fences on + * the recorded process incarnation EXACTLY — never a bare ptyId, worktree, or pane — so a + * reused ptyId belonging to a different process can never be closed, and fails closed on an + * unknown host scope (this is also consumed by workerShow, which does no lease re-check). + * Returns a freshly minted live handle, or null when no live PTY carries that exact incarnation. + */ + resolveTerminalHandleByProcessIncarnation( + processIncarnation: string, + serializedHostScope: string | null + ): string | null { + if (!processIncarnation || !serializedHostScope) { + return null + } + // Scan by the incarnation itself (startsWith + exact equality, mirroring + // classifyWorkerTerminalProcessIncarnation) rather than splitting on a colon, so relay/SSH + // ptyIds and colon-bearing incarnationIds still match. A pty with no incarnationId can never + // match, so the legacy `${runtimeId}:${ptyId}:${ptyGeneration}` fence stays fail-closed. + for (const [ptyId, pty] of this.ptysById) { + if (!matchesProcessIncarnation(ptyId, pty.incarnationId, processIncarnation)) { + continue + } + const hostScope = this.getOrchestrationCompatibilityHostScope(pty) + if (!hostScope || JSON.stringify(hostScope) !== serializedHostScope) { + // Keep scanning: a colon-ambiguous decoy pty in a different host scope that this + // incarnation string happens to prefix-match must not suppress the genuine same-scope + // pty later in ptysById. The scope check still fences the real match below. + continue + } + return this.issuePtyHandle(pty) + } + return null + } } diff --git a/src/main/runtime/orca-runtime-terminal-handle-incarnation.test.ts b/src/main/runtime/orca-runtime-terminal-handle-incarnation.test.ts index dd98d82083f..101ad7d4e4e 100644 --- a/src/main/runtime/orca-runtime-terminal-handle-incarnation.test.ts +++ b/src/main/runtime/orca-runtime-terminal-handle-incarnation.test.ts @@ -1,11 +1,13 @@ import { describe, expect, it, vi } from 'vitest' import { OrcaRuntimeService } from './orca-runtime' +import { makePaneKey } from '../../shared/stable-pane-id' const PTY_ID = 'ssh:target@@relay-pty' const WORKTREE_ID = 'repo::/worktree' const TAB_ID = 'tab-terminal' const LEAF_ID = '11111111-1111-4111-8111-111111111111' +/** A runtime whose pty controller captures every write for assertion. */ function makeRuntime(): { runtime: OrcaRuntimeService; writes: string[] } { const writes: string[] = [] const runtime = new OrcaRuntimeService(null) @@ -20,6 +22,7 @@ function makeRuntime(): { runtime: OrcaRuntimeService; writes: string[] } { return { runtime, writes } } +/** Attach a window and publish a one-tab, one-leaf terminal graph. */ function syncGraph(runtime: OrcaRuntimeService): void { runtime.attachWindow(1) runtime.syncWindowGraph(1, { @@ -44,6 +47,7 @@ function syncGraph(runtime: OrcaRuntimeService): void { }) } +/** (Re)register the fixture pty leaf under the given incarnation id. */ function register(runtime: OrcaRuntimeService, incarnationId: string): void { runtime.registerPty(PTY_ID, WORKTREE_ID, 'target', { tabId: TAB_ID, @@ -66,6 +70,8 @@ describe('runtime terminal handle incarnation fencing', () => { inspectProcess }) expect(runtime.markRendererReloading(1)).not.toBeNull() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. expect((runtime as unknown as { handles: Map }).handles.has(handle)).toBe( false ) @@ -89,6 +95,8 @@ describe('runtime terminal handle incarnation fencing', () => { }) }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. it('treats a null-to-known incarnation as the same un-fenced PTY', async () => { const { runtime } = makeRuntime() const handle = runtime.preAllocateHandleForPty(PTY_ID) @@ -132,6 +140,8 @@ describe('runtime terminal handle incarnation fencing', () => { // Rotate the record directly so reconcile is the only fence exercised. // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: test reaches the runtime's protected pty record map to bypass the registerPty fence. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. const internals = runtime as unknown as { ptysById: Map } @@ -219,6 +229,8 @@ describe('runtime terminal handle incarnation fencing', () => { runtime.registerPreAllocatedHandleForPty(PTY_ID, replacementHandle) syncGraph(runtime) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. const internals = runtime as unknown as { handles: Map handleByLeafKey: Map @@ -241,3 +253,130 @@ describe('runtime terminal handle incarnation fencing', () => { }) }) }) + +describe('resolveTerminalHandleByProcessIncarnation direct fencing', () => { + const LOCAL_SCOPE = JSON.stringify({ kind: 'local', hostId: 'local' }) + // PTY_ID ('ssh:target@@relay-pty') derives connectionId 'target' via parseAppSshPtyId, so its + // host scope is ssh:target rather than local. + const SSH_TARGET_SCOPE = JSON.stringify({ kind: 'ssh', targetId: 'target' }) + + /** Register a pty leaf directly, optionally under a connection and incarnation. */ + function seedPty( + runtime: OrcaRuntimeService, + ptyId: string, + incarnationId: string | null, + connectionId: string | null = null + ): void { + runtime.registerPty(ptyId, WORKTREE_ID, connectionId, { + tabId: TAB_ID, + leafId: LEAF_ID, + ...(incarnationId ? { incarnationId } : {}) + }) + } + + function resolve( + runtime: OrcaRuntimeService, + processIncarnation: string, + serializedHostScope: string | null + ): string | null { + return runtime.resolveTerminalHandleByProcessIncarnation( + processIncarnation, + serializedHostScope + ) + } + + it('mints a live handle for a pty whose exact incarnation and host scope match', () => { + const { runtime } = makeRuntime() + seedPty(runtime, PTY_ID, 'incarnation-1') + const handle = resolve(runtime, `${PTY_ID}:incarnation-1`, SSH_TARGET_SCOPE) + expect(handle).toMatch(/^term_/) + // Re-minting the same live pty is idempotent. + expect(resolve(runtime, `${PTY_ID}:incarnation-1`, SSH_TARGET_SCOPE)).toBe(handle) + }) + + it('returns null when the recorded incarnationId no longer matches the live pty', () => { + const { runtime } = makeRuntime() + seedPty(runtime, PTY_ID, 'incarnation-1') + expect(resolve(runtime, `${PTY_ID}:incarnation-2`, LOCAL_SCOPE)).toBeNull() + }) + + it('returns null when the live pty has no incarnationId (legacy runtimeId:ptyId:gen never reaps)', () => { + const { runtime } = makeRuntime() + seedPty(runtime, PTY_ID, null) + // A modern-shaped probe cannot match a pty that carries no incarnationId... + expect(resolve(runtime, `${PTY_ID}:incarnation-1`, LOCAL_SCOPE)).toBeNull() + // ...and the legacy `${runtimeId}:${ptyId}:${ptyGeneration}` shape stays fail-closed rather + // than reap on a ptyGeneration guess. + expect(resolve(runtime, `runtime_test:${PTY_ID}:0`, LOCAL_SCOPE)).toBeNull() + }) + + it('returns null when the host scope does not match', () => { + const { runtime } = makeRuntime() + seedPty(runtime, PTY_ID, 'incarnation-1') + expect( + resolve(runtime, `${PTY_ID}:incarnation-1`, JSON.stringify({ kind: 'ssh', targetId: 'nope' })) + ).toBeNull() + }) + + it('returns null when no host scope is supplied (fails closed)', () => { + const { runtime } = makeRuntime() + seedPty(runtime, PTY_ID, 'incarnation-1') + expect(resolve(runtime, `${PTY_ID}:incarnation-1`, null)).toBeNull() + }) + + it('resolves a colon-bearing SSH/relay incarnationId that lastIndexOf would mis-split', () => { + const { runtime } = makeRuntime() + // PTY_ID already carries ':' and '@@'; the incarnationId itself also carries colons. + seedPty(runtime, PTY_ID, 'relay:conn-3:incarnation-9', 'target') + const handle = resolve( + runtime, + `${PTY_ID}:relay:conn-3:incarnation-9`, + JSON.stringify({ kind: 'ssh', targetId: 'target' }) + ) + expect(handle).toMatch(/^term_/) + }) + + it('resolves a Windows repo::C:\\path@@1 ptyId', () => { + const { runtime } = makeRuntime() + const windowsPtyId = 'repo::C:\\path@@1' + seedPty(runtime, windowsPtyId, 'incarnation-win') + const handle = resolve(runtime, `${windowsPtyId}:incarnation-win`, LOCAL_SCOPE) + expect(handle).toMatch(/^term_/) + }) + + it('keeps scanning past a colon-ambiguous decoy in another host scope to the genuine match', () => { + const { runtime } = makeRuntime() + // One process-incarnation string, two colon-ambiguous pty ids that both satisfy the exact + // matcher: 'relay' + 'conn:incarnation-1' AND 'relay:conn' + 'incarnation-1' both stringify to + // 'relay:conn:incarnation-1'. The decoy is registered FIRST and lives in a different host scope + // (ssh:decoy); the genuine pty is local and registered later. + const processIncarnation = 'relay:conn:incarnation-1' + const DECOY_TAB_ID = 'tab-decoy' + const DECOY_LEAF_ID = '22222222-2222-4222-8222-222222222222' + const GENUINE_TAB_ID = 'tab-genuine' + const GENUINE_LEAF_ID = '33333333-3333-4333-8333-333333333333' + runtime.registerPty('relay', WORKTREE_ID, 'decoy', { + tabId: DECOY_TAB_ID, + leafId: DECOY_LEAF_ID, + incarnationId: 'conn:incarnation-1' + }) + runtime.registerPty('relay:conn', WORKTREE_ID, null, { + tabId: GENUINE_TAB_ID, + leafId: GENUINE_LEAF_ID, + incarnationId: 'incarnation-1' + }) + + // Before the fix, the earlier decoy's host-scope mismatch returned null and suppressed the + // genuine same-scope pty entirely. `continue` lets the scan reach it. + const handle = resolve(runtime, processIncarnation, LOCAL_SCOPE) + expect(handle).toMatch(/^term_/) + // getTerminalProcessIncarnation cannot separate the two (both stringify to + // 'relay:conn:incarnation-1'), so pin the remint to the GENUINE pane. The paneKey is the only + // terminal-specific observable that distinguishes the genuine leaf from the same-incarnation + // decoy; this fails if the resolver ever mints the decoy's handle instead of the genuine pty's. + const genuinePaneKey = makePaneKey(GENUINE_TAB_ID, GENUINE_LEAF_ID) + const decoyPaneKey = makePaneKey(DECOY_TAB_ID, DECOY_LEAF_ID) + expect(runtime.getTerminalPaneKey(handle!)).toBe(genuinePaneKey) + expect(runtime.getTerminalPaneKey(handle!)).not.toBe(decoyPaneKey) + }) +}) diff --git a/src/main/runtime/orchestration/worker-terminal-process-liveness.test.ts b/src/main/runtime/orchestration/worker-terminal-process-liveness.test.ts new file mode 100644 index 00000000000..e2bc2cc7d63 --- /dev/null +++ b/src/main/runtime/orchestration/worker-terminal-process-liveness.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, it } from 'vitest' +import { matchesProcessIncarnation } from './worker-terminal-process-liveness' + +describe('matchesProcessIncarnation', () => { + it.each([ + { + name: 'an exact ptyId:incarnation match', + ptyId: 'pty-1', + incarnationId: 'incarnation-1', + processIncarnation: 'pty-1:incarnation-1', + expected: true + }, + { + name: 'a Windows repo::C:\\path@@1 ptyId whose incarnation matches', + ptyId: 'repo::C:\\path@@1', + incarnationId: 'incarnation-win', + processIncarnation: 'repo::C:\\path@@1:incarnation-win', + expected: true + }, + { + name: 'a colon-bearing relay incarnationId that a lastIndexOf split would mangle', + ptyId: 'relay-pty', + incarnationId: 'relay:conn-3:incarnation-9', + processIncarnation: 'relay-pty:relay:conn-3:incarnation-9', + expected: true + }, + { + name: 'a whitespace-dirty incarnationId (lost contact is never a match)', + ptyId: 'pty-1', + incarnationId: ' incarnation-1', + processIncarnation: 'pty-1: incarnation-1', + expected: false + }, + { + name: 'an empty-string incarnationId', + ptyId: 'pty-1', + incarnationId: '', + processIncarnation: 'pty-1:', + expected: false + }, + { + name: 'an absent (null) incarnationId', + ptyId: 'pty-1', + incarnationId: null, + processIncarnation: 'pty-1:incarnation-1', + expected: false + }, + { + name: 'an absent (undefined) incarnationId', + ptyId: 'pty-1', + incarnationId: undefined, + processIncarnation: 'pty-1:incarnation-1', + expected: false + }, + { + name: 'a prefix-decoy ptyId (@@1) against a longer live pty (@@10)', + ptyId: 'repo::C:\\path@@1', + incarnationId: 'inc-1', + processIncarnation: 'repo::C:\\path@@10:inc-1', + expected: false + } + ])('returns $expected for $name', ({ ptyId, incarnationId, processIncarnation, expected }) => { + expect(matchesProcessIncarnation(ptyId, incarnationId, processIncarnation)).toBe(expected) + }) + + it('rejects a partial prefix that is not colon-delimited', () => { + // 'pty-10' is not the pty 'pty-1'; the `${ptyId}:` fence keeps them distinct. + expect(matchesProcessIncarnation('pty-1', 'inc', 'pty-10:inc')).toBe(false) + }) +}) diff --git a/src/main/runtime/orchestration/worker-terminal-process-liveness.ts b/src/main/runtime/orchestration/worker-terminal-process-liveness.ts index 72c5ce07666..4e652d6286a 100644 --- a/src/main/runtime/orchestration/worker-terminal-process-liveness.ts +++ b/src/main/runtime/orchestration/worker-terminal-process-liveness.ts @@ -5,6 +5,24 @@ import type { PtyProcessInfo } from '../../providers/pty-process-info' export type { WorkerTerminalHostScope } from '../../../shared/worker-terminal-host-scope' export { parseWorkerTerminalHostScope } from '../../../shared/worker-terminal-host-scope' +/** + * Does `processIncarnation` name exactly this pty's live incarnation? Requires exact + * `${ptyId}:${incarnationId}` equality, so it is immune to colons on either side (relay/SSH + * ptyIds, colon-bearing relay incarnationIds). A pty with no (or a whitespace-dirty) + * incarnationId can never match — the exact-incarnation fence stays intact. + */ +export function matchesProcessIncarnation( + ptyId: string, + incarnationId: string | null | undefined, + processIncarnation: string +): boolean { + if (!incarnationId || incarnationId !== incarnationId.trim()) { + return false + } + return `${ptyId}:${incarnationId}` === processIncarnation +} + +/** Classify a recorded incarnation against live sessions: live on exact match, unverifiable when a candidate pty has a dirty or absent incarnationId (lost contact is never a death certificate), else exited. */ export function classifyWorkerTerminalProcessIncarnation( processIncarnation: string, sessions: readonly PtyProcessInfo[] @@ -13,13 +31,9 @@ export function classifyWorkerTerminalProcessIncarnation( processIncarnation.startsWith(`${session.id}:`) ) if ( - possibleMatches.some((session) => { - const incarnationId = session.incarnationId - if (!incarnationId || incarnationId !== incarnationId.trim()) { - return false - } - return `${session.id}:${incarnationId}` === processIncarnation - }) + possibleMatches.some((session) => + matchesProcessIncarnation(session.id, session.incarnationId, processIncarnation) + ) ) { return 'live' } diff --git a/src/main/runtime/rpc/methods/orchestration-worker-release-incarnation-fallback.test.ts b/src/main/runtime/rpc/methods/orchestration-worker-release-incarnation-fallback.test.ts new file mode 100644 index 00000000000..e839c2e017d --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration-worker-release-incarnation-fallback.test.ts @@ -0,0 +1,534 @@ +import { afterEach, describe, expect, it, vi, type MockInstance } from 'vitest' +import { ORCHESTRATION_METHODS } from './orchestration' +import { eraseRpcMethods, type RpcContext } from '../core' +import { OrchestrationDb } from '../../orchestration/db' +import { OrcaRuntimeService } from '../../orca-runtime' +import { completeWorkerTerminalRelease } from './orchestration/worker/worker-release-completion' + +describe('orchestration worker release incarnation fallback', () => { + let db: OrchestrationDb + let dbOpen = false + let runtime: OrcaRuntimeService + let ctx: RpcContext + let activeRunId: string + let inspectProcessLiveness: MockInstance< + OrcaRuntimeService['inspectTerminalProcessIncarnationLiveness'] + > + + const coordinatorPaneKey = 'tab_coord:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' + const workerPaneKey = 'tab_worker:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb' + + /** Fresh in-memory db and a fully-stubbed runtime for one worker-release scenario. */ + function setup(): void { + db = new OrchestrationDb(':memory:') + dbOpen = true + runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb(db) + inspectProcessLiveness = vi + .spyOn(runtime, 'inspectTerminalProcessIncarnationLiveness') + .mockResolvedValue('live') + vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) => + handle === 'term_coord' + ? coordinatorPaneKey + : handle === 'term_worker' || handle === 'term_reminted' + ? workerPaneKey + : null + ) + vi.spyOn(runtime, 'getTerminalProcessIncarnation').mockImplementation((handle) => + handle === 'term_worker' || handle === 'term_reminted' ? 'runtime_test:term_worker:1' : null + ) + vi.spyOn(runtime, 'getOrchestrationDispatchAuthority').mockImplementation((handle) => + handle === 'term_worker' || handle === 'term_reminted' + ? // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This test fixture is deliberately shaped to exercise the private/runtime boundary. + ({ + terminalHandle: handle, + paneKey: workerPaneKey, + processIncarnation: 'runtime_test:term_worker:1', + hostScope: { kind: 'local', hostId: 'local' } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + } as never) + : null + ) + vi.spyOn(runtime, 'validateOrchestrationAgentLauncher').mockImplementation(() => {}) + vi.spyOn(runtime, 'showTerminal').mockImplementation( + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + async (handle) => ({ handle, worktreeId: 'repo::worktree', status: 'running' }) as never + ) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + vi.spyOn(runtime, 'showManagedTerminalWorkspace').mockResolvedValue({ + id: 'repo::worktree' + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + } as never) + vi.spyOn(runtime, 'createTerminal').mockResolvedValue({ + handle: 'term_worker', + worktreeId: 'repo::worktree', + title: 'worker' + }) + vi.spyOn(runtime, 'waitForTerminal').mockResolvedValue({ + handle: 'term_worker', + condition: 'tui-idle', + satisfied: true, + status: 'running', + exitCode: null + }) + vi.spyOn(runtime, 'getTerminalOrchestrationCliCommand').mockReturnValue('orca') + vi.spyOn(runtime, 'sendTerminalAgentPrompt').mockResolvedValue({ + handle: 'term_worker', + accepted: true, + bytesWritten: 1 + }) + vi.spyOn(runtime, 'isTerminalRunningAgent').mockResolvedValue(true) + vi.spyOn(runtime, 'getExactWorkerProviderSession').mockReturnValue(null) + vi.spyOn(runtime, 'readTerminal').mockResolvedValue({ + handle: 'term_worker', + status: 'running', + tail: ['worker output line 1', 'worker output line 2'], + truncated: false, + nextCursor: '2' + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + vi.spyOn(runtime, 'closeTerminal').mockResolvedValue({ + handle: 'term_worker', + tabId: 'tab-worker', + ptyKilled: true + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + } as never) + vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {}) + activeRunId = db.createRun({ + objective: 'Release test Run', + coordinatorHandle: 'term_coord', + coordinatorPaneKey + }).id + ctx = { runtime } + } + + afterEach(() => { + if (dbOpen) { + dbOpen = false + db.close() + } + vi.restoreAllMocks() + }) + + /** Look up a registered orchestration RPC method by name. */ + function findMethod(name: string) { + const method = eraseRpcMethods(ORCHESTRATION_METHODS).find((m) => m.name === name) + if (!method) { + throw new Error(`Method not found: ${name}`) + } + return method + } + + /** Parse a method's params and invoke its handler against the shared ctx. */ + async function call(name: string, params: Record) { + const method = findMethod(name) + const parsed = method.params ? method.params.parse(params) : undefined + return method.handler(parsed, ctx) + } + + /** Start a ready worker on a fresh task off the coordinator terminal. */ + async function startWorker(options: { terminal?: string } = {}): Promise<{ + taskId: string + dispatchId: string + }> { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const task = db.createTask({ spec: 'release fixture task', runId: activeRunId }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const result = (await call('orchestration.workerStart', { + task: task.id, + from: 'term_coord', + ...(options.terminal ? { terminal: options.terminal } : { agent: 'codex' }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + })) as { dispatchId: string; state: string } + expect(result.state).toBe('ready') + return { taskId: task.id, dispatchId: result.dispatchId } + } + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + /** Record the worker's report as settled — the precondition for release. */ + function settle(taskId: string, dispatchId: string, outcome: 'succeeded' | 'failed'): void { + const settlement = db.settleWorkerReport({ + taskId, + dispatchId, + outcome, + result: `worker ${outcome}` + }) + expect(settlement.action).toBe('settled') + } + + /** Start a worker and settle its report, the state a release acts on. */ + async function startSettledWorker( + outcome: 'succeeded' | 'failed' = 'succeeded', + options: { terminal?: string } = {} + ): Promise<{ taskId: string; dispatchId: string }> { + const worker = await startWorker(options) + settle(worker.taskId, worker.dispatchId, outcome) + return worker + } + + it('closes a live worker terminal whose durable handle no longer resolves but whose process incarnation still matches', async () => { + setup() + const { dispatchId } = await startSettledWorker() + // The durable handle stops resolving (renderer graph epoch bump / handle invalidation)... + vi.mocked(runtime.showTerminal).mockImplementation(async (handle) => + handle === 'term_worker' + ? Promise.reject(new Error('terminal_handle_stale')) + : // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This test fixture is deliberately shaped to exercise the private/runtime boundary. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + ({ handle, worktreeId: 'repo::worktree', status: 'running' } as never) + ) + // ...but the recorded process incarnation still names a live PTY, re-minted to a fresh handle. + const resolveByIncarnation = vi.fn().mockReturnValue('term_reminted') + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const receipt = (await call('orchestration.workerRelease', { dispatch: dispatchId })) as { + state: string + processAction: string + } + + expect(receipt).toMatchObject({ state: 'released', processAction: 'closed_agent_terminal' }) + expect(resolveByIncarnation).toHaveBeenCalledWith( + 'runtime_test:term_worker:1', + JSON.stringify({ kind: 'local', hostId: 'local' }) + ) + // The close targeted exactly the re-minted live handle for that PTY, never the stale one. + expect(runtime.closeTerminal).toHaveBeenCalledTimes(1) + expect(runtime.closeTerminal).toHaveBeenCalledWith('term_reminted') + expect(db.getWorkerTerminalResourceByOwner(dispatchId)).toMatchObject({ + ownership_state: 'released', + release_state: 'released' + }) + }) + + it('stays release_unknown and closes nothing when the recorded incarnation no longer matches a live pty', async () => { + setup() + const { dispatchId } = await startSettledWorker() + vi.mocked(runtime.showTerminal).mockRejectedValue(new Error('terminal_handle_stale')) + // A reused ptyId now belongs to a different process: the incarnation mismatch refuses a close. + const resolveByIncarnation = vi.fn().mockReturnValue(null) + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const receipt = (await call('orchestration.workerRelease', { dispatch: dispatchId })) as { + state: string + } + + expect(receipt.state).toBe('release_unknown') + expect(resolveByIncarnation).toHaveBeenCalledWith( + 'runtime_test:term_worker:1', + JSON.stringify({ kind: 'local', hostId: 'local' }) + ) + expect(runtime.closeTerminal).not.toHaveBeenCalled() + expect(db.getWorkerTerminalResourceByOwner(dispatchId)?.release_state).toBe('unknown') + }) + + it('does not plain-settle an exited missing worker when the archive was never committed', async () => { + setup() + const { dispatchId } = await startSettledWorker() + vi.mocked(runtime.showTerminal).mockRejectedValue(new Error('terminal_handle_stale')) + const resolveByIncarnation = vi.fn().mockReturnValue(null) + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + inspectProcessLiveness.mockResolvedValue('exited') + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const receipt = (await call('orchestration.workerRelease', { dispatch: dispatchId })) as { + state: string + processAction: string + } + + expect(receipt).toMatchObject({ state: 'release_unknown', processAction: 'none' }) + expect(inspectProcessLiveness).toHaveBeenCalled() + expect(runtime.closeTerminal).not.toHaveBeenCalled() + expect(db.getWorkerTerminalResourceByOwner(dispatchId)).toMatchObject({ + ownership_state: 'owned', + release_state: 'unknown' + }) + }) + + it('settles released without a close when exited missing and an archive is already committed', async () => { + setup() + const { dispatchId } = await startSettledWorker() + const requested = db.requestWorkerTerminalRelease(dispatchId) + if (requested.disposition !== 'requested') { + throw new Error(`expected requested, got ${requested.disposition}`) + } + db.commitWorkerTerminalArchiveForRelease({ + dispatchId, + resourceId: requested.resource.id, + kind: 'terminal_tail', + content: JSON.stringify({ lines: [] }), + archiveSource: 'terminal', + archiveStatus: 'empty' + }) + vi.mocked(runtime.showTerminal).mockRejectedValue(new Error('terminal_handle_stale')) + const resolveByIncarnation = vi.fn().mockReturnValue(null) + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + inspectProcessLiveness.mockResolvedValue('exited') + + const receipt = await completeWorkerTerminalRelease({ + runtime, + db, + dispatchId, + resource: db.getWorkerTerminalResource(requested.resource.id)!, + mode: 'interactive' + }) + + expect(receipt).toMatchObject({ state: 'released', processAction: 'none' }) + expect(runtime.closeTerminal).not.toHaveBeenCalled() + expect(db.getWorkerTerminalResourceByOwner(dispatchId)).toMatchObject({ + ownership_state: 'released', + release_state: 'released' + }) + }) + + it('reaches settleDead before the lease check when a gone worker is exited with no live authority', async () => { + // Without a committed archive settleDead retains; lease must not run first and force + // retained/identity_unproven. Disposition is release_unknown (interactive) after the retain. + setup() + const { dispatchId } = await startSettledWorker() + vi.mocked(runtime.showTerminal).mockRejectedValue(new Error('terminal_handle_stale')) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const resolveByIncarnation = vi.fn().mockReturnValue(null) + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + vi.mocked(runtime.getOrchestrationDispatchAuthority).mockReturnValue(null) + inspectProcessLiveness.mockResolvedValue('exited') + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const receipt = (await call('orchestration.workerRelease', { dispatch: dispatchId })) as { + state: string + processAction: string + } + + expect(receipt).toMatchObject({ state: 'release_unknown', processAction: 'none' }) + expect(receipt.state).not.toBe('retained') + expect(inspectProcessLiveness).toHaveBeenCalled() + expect(runtime.closeTerminal).not.toHaveBeenCalled() + }) + + it('concedes release_unknown before the lease check when a gone worker has no live authority and liveness is unproven', async () => { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + setup() + const { dispatchId } = await startSettledWorker() + vi.mocked(runtime.showTerminal).mockRejectedValue(new Error('terminal_handle_stale')) + const resolveByIncarnation = vi.fn().mockReturnValue(null) + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + vi.mocked(runtime.getOrchestrationDispatchAuthority).mockReturnValue(null) + // Liveness is unresolvable/not-exited: the process may have been re-homed, so concede rather + // than retain or guess at a live process. + inspectProcessLiveness.mockResolvedValue('live') + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const receipt = (await call('orchestration.workerRelease', { dispatch: dispatchId })) as { + state: string + } + + expect(receipt.state).toBe('release_unknown') + expect(receipt.state).not.toBe('retained') + expect(runtime.closeTerminal).not.toHaveBeenCalled() + expect(db.getWorkerTerminalResourceByOwner(dispatchId)?.release_state).toBe('unknown') + }) + + it('workerStop closes a live worker via the reminted handle when the durable handle is stale', async () => { + setup() + const { dispatchId } = await startWorker() + // The durable handle stops resolving, but the recorded incarnation still names a live PTY. + vi.mocked(runtime.showTerminal).mockImplementation(async (handle) => + handle === 'term_worker' + ? Promise.reject(new Error('terminal_handle_stale')) + : // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This test fixture is deliberately shaped to exercise the private/runtime boundary. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + ({ handle, worktreeId: 'repo::worktree', status: 'running' } as never) + ) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const resolveByIncarnation = vi.fn().mockReturnValue('term_reminted') + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const receipt = (await call('orchestration.workerStop', { dispatch: dispatchId })) as { + processAction: string + } + + expect(receipt.processAction).toBe('closed_agent_terminal') + // The kill targeted exactly the reminted live handle, never the stale durable one — closing + // the stale handle would throw terminal_handle_stale and leak the PTY. + expect(runtime.closeTerminal).toHaveBeenCalledTimes(1) + expect(runtime.closeTerminal).toHaveBeenCalledWith('term_reminted') + }) + + it('workerRead reads a live worker via the reminted handle when the durable handle is stale', async () => { + setup() + const { dispatchId } = await startWorker() + vi.mocked(runtime.showTerminal).mockImplementation( + async (handle) => + handle === 'term_worker' + ? Promise.reject(new Error('terminal_handle_stale')) + : // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This test fixture is deliberately shaped to exercise the private/runtime boundary. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + ({ handle, worktreeId: 'repo::worktree', status: 'running' } as never) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + ) + const resolveByIncarnation = vi.fn().mockReturnValue('term_reminted') + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const output = (await call('orchestration.workerRead', { dispatch: dispatchId })) as { + terminal?: { tail: string[] } + } + + // Both the exact-session probe and the terminal read addressed the reminted handle. + expect(runtime.getExactWorkerProviderSession).toHaveBeenCalledWith( + 'term_reminted', + expect.any(Number) + ) + expect(runtime.readTerminal).toHaveBeenCalledWith('term_reminted', expect.anything()) + expect(output.terminal?.tail).toEqual(['worker output line 1', 'worker output line 2']) + }) + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + it('recovery-mode: defers when exited missing has no archive rather than plain-settling', async () => { + // Proof of death still runs settleDead first; when it retains (no archive), recovery must + // stay release_pending — not plain-settle, not unknown. + setup() + const { dispatchId } = await startSettledWorker() + vi.mocked(runtime.showTerminal).mockRejectedValue(new Error('terminal_handle_stale')) + const resolveByIncarnation = vi.fn().mockReturnValue(null) + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + inspectProcessLiveness.mockResolvedValue('exited') + const requested = db.requestWorkerTerminalRelease(dispatchId) + if (requested.disposition !== 'requested') { + throw new Error(`expected a requested release, got ${requested.disposition}`) + } + + const receipt = await completeWorkerTerminalRelease({ + runtime, + db, + dispatchId, + resource: requested.resource, + mode: 'recovery' + }) + + expect(receipt).toMatchObject({ state: 'release_pending', processAction: 'none' }) + expect(runtime.closeTerminal).not.toHaveBeenCalled() + expect(db.getWorkerTerminalResourceByOwner(dispatchId)).toMatchObject({ + ownership_state: 'owned', + release_state: 'requested' + }) + }) + + it('recovery-mode: settles released before the defer when exited missing already has an archive', async () => { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + setup() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const { dispatchId } = await startSettledWorker() + vi.mocked(runtime.showTerminal).mockRejectedValue(new Error('terminal_handle_stale')) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const resolveByIncarnation = vi.fn().mockReturnValue(null) + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + inspectProcessLiveness.mockResolvedValue('exited') + const requested = db.requestWorkerTerminalRelease(dispatchId) + if (requested.disposition !== 'requested') { + throw new Error(`expected a requested release, got ${requested.disposition}`) + } + db.commitWorkerTerminalArchiveForRelease({ + dispatchId, + resourceId: requested.resource.id, + kind: 'terminal_tail', + content: JSON.stringify({ lines: [] }), + archiveSource: 'terminal', + archiveStatus: 'empty' + }) + + const receipt = await completeWorkerTerminalRelease({ + runtime, + db, + dispatchId, + resource: db.getWorkerTerminalResource(requested.resource.id)!, + mode: 'recovery' + }) + + expect(receipt).toMatchObject({ state: 'released', processAction: 'none' }) + expect(runtime.closeTerminal).not.toHaveBeenCalled() + expect(db.getWorkerTerminalResourceByOwner(dispatchId)).toMatchObject({ + ownership_state: 'released', + release_state: 'released' + }) + }) + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + it('recovery-mode: defers release_pending when liveness is unverifiable rather than provably exited', async () => { + setup() + const { dispatchId } = await startSettledWorker() + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + vi.mocked(runtime.showTerminal).mockRejectedValue(new Error('terminal_handle_stale')) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const resolveByIncarnation = vi.fn().mockReturnValue(null) + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + resolveByIncarnation + ) + // Not a death certificate: inventory may still be incomplete, so recovery must defer. + inspectProcessLiveness.mockResolvedValue('unverifiable') + const requested = db.requestWorkerTerminalRelease(dispatchId) + if (requested.disposition !== 'requested') { + throw new Error(`expected a requested release, got ${requested.disposition}`) + } + + const receipt = await completeWorkerTerminalRelease({ + runtime, + db, + dispatchId, + resource: requested.resource, + mode: 'recovery' + }) + + expect(receipt.state).toBe('release_pending') + expect(runtime.closeTerminal).not.toHaveBeenCalled() + expect(db.getWorkerTerminalResourceByOwner(dispatchId)?.release_state).not.toBe('released') + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration-worker-release-reap-fixed.func.test.ts b/src/main/runtime/rpc/methods/orchestration-worker-release-reap-fixed.func.test.ts new file mode 100644 index 00000000000..579f37eaf9e --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration-worker-release-reap-fixed.func.test.ts @@ -0,0 +1,358 @@ +// PRB-0219 / upstream #18737 — FUNCTIONAL (integration) reproduction of the steady-state +// worker-release reap LEAK. +// +// This is the "functional/integration" tier of the PRB-0219 test strategy. It wires the REAL +// orchestration RPC surface (orchestration.workerStart / workerRelease / workerList), the REAL +// OrchestrationDb, and the REAL release completion + observation modules against a fake runtime +// that faithfully models the two identity planes involved in the bug: +// +// * VOLATILE, epoch-fenced handle table — `handleTable` keyed by terminal handle, each entry +// stamped with the `rendererGraphEpoch` at which it was issued. `showTerminal` resolves a +// handle ONLY while its stamped epoch matches the current epoch (mirrors getLiveLeafForHandle +// throwing 'terminal_handle_stale' on a rendererGraphEpoch bump). A relay reconnect / renderer +// remount on headless serve bumps the epoch. +// * DURABLE, incarnation-addressed process table — `ptysById` keyed by the ptyId embedded in the +// worker's persisted process_incarnation (`${ptyId}:${incarnationId}`). The pty stays LIVE +// across an epoch bump; nothing about the graph epoch kills the process. +// +// The bug (mechanism): when the epoch bumps, the durable db handle stops resolving through +// showTerminal WHILE THE PTY IS STILL ALIVE. inspectWorkerTerminal swallows the throw and reports +// `missing`; completeWorkerTerminalRelease then commits `release_unknown` and returns WITHOUT ever +// calling runtime.closeTerminal — so the process/PTY leaks. On mtl-02 those orphans accumulate in +// the orca-serve@factory cgroup until TasksMax=4096 is hit and Bun/omp abort() on EAGAIN. +// +// The observable leak signal asserted here: state 'release_unknown' + processAction 'none' + +// closeTerminal NEVER called + the pty STILL alive in ptysById + worker-list terminalState stuck +// on 'release_unknown' (never 'released'). + +import { afterEach, describe, expect, it, vi } from 'vitest' +import { ORCHESTRATION_METHODS } from './orchestration' +import { eraseRpcMethods, type RpcContext } from '../core' +import { OrchestrationDb } from '../../orchestration/db' +import { OrcaRuntimeService } from '../../orca-runtime' + +describe('PRB-0219 worker-release reap FIX (functional verification)', () => { + let db: OrchestrationDb + let dbOpen = false + let runtime: OrcaRuntimeService + let ctx: RpcContext + let activeRunId: string + + const coordinatorPaneKey = 'tab_coord:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' + const workerPaneKey = 'tab_worker:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb' + + // Dispatch/automation ptyId shape: `${repoId}::${worktreePath}@@${suffix}` (executionHostId:null). + const PTY_ID = 'repo-7f3a::/data/wt/factory-task-1@@a1b2c3d4' + const INCARNATION_ID = 1 + const PROCESS_INCARNATION = `${PTY_ID}:${INCARNATION_ID}` + + // ---- fake runtime process/handle planes (module-level so spies can mutate them) ---- + let ptysById: Map + let handleTable: Map + let rendererGraphEpoch: number + let closedPtyIds: string[] + + /** Resolve a handle to its live pty only while its stamped graph epoch is current. */ + function resolveHandleToLivePty( + handle: string + ): { ptyId: string; pty: { incarnationId: number; alive: boolean } } | null { + const entry = handleTable.get(handle) + if (!entry) { + return null + } + if (entry.epoch !== rendererGraphEpoch) { + // rendererGraphEpoch fence: the durable handle no longer resolves to a live leaf. + return null + } + const pty = ptysById.get(entry.ptyId) + if (!pty || !pty.alive) { + return null + } + return { ptyId: entry.ptyId, pty } + } + + /** Wire the real RPC surface and db against the two-plane fake runtime. */ + function setup(): void { + ptysById = new Map([ + ['coord-pty', { incarnationId: 1, alive: true }], + [PTY_ID, { incarnationId: INCARNATION_ID, alive: true }] + ]) + handleTable = new Map([ + ['term_coord', { ptyId: 'coord-pty', epoch: 0 }], + ['term_worker', { ptyId: PTY_ID, epoch: 0 }] + ]) + rendererGraphEpoch = 0 + closedPtyIds = [] + + db = new OrchestrationDb(':memory:') + dbOpen = true + runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb(db) + + // Incarnation-addressed liveness probe: reads the DURABLE plane, so it stays 'live' across the + // epoch bump (the process really is still running). Matches the real asymmetry. + vi.spyOn(runtime, 'inspectTerminalProcessIncarnationLiveness').mockImplementation( + async (incarnation: string) => { + const idx = incarnation.lastIndexOf(':') + const ptyId = incarnation.slice(0, idx) + const pty = ptysById.get(ptyId) + return pty?.alive ? 'live' : 'exited' + } + ) + + // The incarnation-addressed re-resolution primitive the fix adds. Present here as a spy so the + // same harness proves BOTH tiers: pre-fix completion never calls it (leak); post-fix completion + // calls it to remint a live handle (reap). Fence: EXACT incarnationId match only. + vi.spyOn(runtime, 'resolveTerminalHandleByProcessIncarnation').mockImplementation( + (incarnation: string): string | null => { + const idx = incarnation.lastIndexOf(':') + const ptyId = incarnation.slice(0, idx) + const inc = Number(incarnation.slice(idx + 1)) + const pty = ptysById.get(ptyId) + if (!pty || !pty.alive) { + return null + } + if (pty.incarnationId !== inc) { + // Fence: a reused ptyId with a different incarnation must NOT resolve. + return null + } + // Remint a live handle at the current graph epoch. + handleTable.set('term_reminted', { ptyId, epoch: rendererGraphEpoch }) + return 'term_reminted' + } + ) + + // Identity plane (durable) — answers for the original AND any reminted handle. Independent of + // the graph epoch, exactly like the real getTerminal* accessors that read dispatch authority. + const knownWorkerHandle = (handle: string): boolean => + handle === 'term_worker' || handle === 'term_reminted' + vi.spyOn(runtime, 'getTerminalPaneKey').mockImplementation((handle) => + handle === 'term_coord' + ? coordinatorPaneKey + : knownWorkerHandle(handle) + ? workerPaneKey + : null + ) + vi.spyOn(runtime, 'getTerminalProcessIncarnation').mockImplementation((handle) => + knownWorkerHandle(handle) ? PROCESS_INCARNATION : null + ) + vi.spyOn(runtime, 'getOrchestrationDispatchAuthority').mockImplementation((handle) => + knownWorkerHandle(handle) + ? // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: This test fixture is deliberately shaped to exercise the private/runtime boundary. + ({ + terminalHandle: handle, + paneKey: workerPaneKey, + processIncarnation: PROCESS_INCARNATION, + hostScope: { kind: 'local', hostId: 'local' } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + } as never) + : null + ) + vi.spyOn(runtime, 'validateOrchestrationAgentLauncher').mockImplementation(() => {}) + + // Volatile handle resolution — epoch-fenced. Throws 'terminal_handle_stale' once the epoch + // moves past the epoch at which the handle was issued. + vi.spyOn(runtime, 'showTerminal').mockImplementation(async (handle) => { + if (!resolveHandleToLivePty(handle)) { + throw new Error('terminal_handle_stale') + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + return { handle, worktreeId: 'repo::worktree', status: 'running' } as never + }) + + // The reap: closing a handle kills exactly the pty it resolves to. + vi.spyOn(runtime, 'closeTerminal').mockImplementation(async (handle) => { + const live = resolveHandleToLivePty(handle) + if (!live) { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + return { handle, tabId: null, ptyKilled: false } as never + } + live.pty.alive = false + ptysById.delete(live.ptyId) + closedPtyIds.push(live.ptyId) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + return { handle, tabId: `tab:${live.ptyId}`, ptyKilled: true } as never + }) + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // Remaining runtime surface required to start + settle a worker (mirrors the unit harness). + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + vi.spyOn(runtime, 'showManagedTerminalWorkspace').mockResolvedValue({ + id: 'repo::worktree' + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + } as never) + vi.spyOn(runtime, 'createTerminal').mockResolvedValue({ + handle: 'term_worker', + worktreeId: 'repo::worktree', + title: 'worker' + }) + vi.spyOn(runtime, 'waitForTerminal').mockResolvedValue({ + handle: 'term_worker', + condition: 'tui-idle', + satisfied: true, + status: 'running', + exitCode: null + }) + vi.spyOn(runtime, 'getTerminalOrchestrationCliCommand').mockReturnValue('orca') + vi.spyOn(runtime, 'sendTerminalAgentPrompt').mockResolvedValue({ + handle: 'term_worker', + accepted: true, + bytesWritten: 1 + }) + vi.spyOn(runtime, 'isTerminalRunningAgent').mockResolvedValue(true) + vi.spyOn(runtime, 'getExactWorkerProviderSession').mockReturnValue(null) + vi.spyOn(runtime, 'readTerminal').mockResolvedValue({ + handle: 'term_worker', + status: 'running', + tail: ['worker output line 1', 'worker output line 2'], + truncated: false, + nextCursor: '2' + }) + vi.spyOn(runtime, 'notifyMessageArrived').mockImplementation(() => {}) + + activeRunId = db.createRun({ + objective: 'PRB-0219 reap leak fixture', + coordinatorHandle: 'term_coord', + coordinatorPaneKey + }).id + ctx = { runtime } + } + + afterEach(() => { + if (dbOpen) { + dbOpen = false + db.close() + } + vi.restoreAllMocks() + }) + + /** Look up a registered orchestration RPC method by name. */ + function findMethod(name: string) { + const method = eraseRpcMethods(ORCHESTRATION_METHODS).find((m) => m.name === name) + if (!method) { + throw new Error(`Method not found: ${name}`) + } + return method + } + + /** Parse a method's params and invoke its handler against the shared ctx. */ + async function call(name: string, params: Record) { + const method = findMethod(name) + const parsed = method.params ? method.params.parse(params) : undefined + return method.handler(parsed, ctx) + } + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + /** Start a worker and settle its report, the state a release acts on. */ + async function startSettledWorker(): Promise<{ taskId: string; dispatchId: string }> { + const task = db.createTask({ spec: 'reap-leak fixture task', runId: activeRunId }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const result = (await call('orchestration.workerStart', { + task: task.id, + from: 'term_coord', + agent: 'codex' + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + })) as { dispatchId: string; state: string } + expect(result.state).toBe('ready') + const settlement = db.settleWorkerReport({ + taskId: task.id, + dispatchId: result.dispatchId, + outcome: 'succeeded', + result: 'worker succeeded' + }) + expect(settlement.action).toBe('settled') + return { taskId: task.id, dispatchId: result.dispatchId } + } + + /** The terminalState workerList projects for a dispatch, or null. */ + async function workerTerminalState(dispatchId: string): Promise { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const listed = (await call('orchestration.workerList', { run: activeRunId })) as { + workers: { dispatchId: string; terminalState: string | null }[] + } + return listed.workers.find((w) => w.dispatchId === dispatchId)?.terminalState ?? null + } + + it('REAP (fixed): a rendererGraphEpoch bump strands the durable handle, but the incarnation fallback remints a live handle and reaps the process', async () => { + setup() + const { dispatchId } = await startSettledWorker() + const resource = db.getWorkerTerminalResourceByOwner(dispatchId) + expect(resource?.process_incarnation).toBe(PROCESS_INCARNATION) + + // Relay reconnect / renderer remount bumps the graph epoch: the durable handle goes stale + // while the PTY stays alive. + rendererGraphEpoch = 1 + await expect(runtime.showTerminal('term_worker')).rejects.toThrow('terminal_handle_stale') + expect(ptysById.get(PTY_ID)?.alive).toBe(true) + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const receipt = (await call('orchestration.workerRelease', { dispatch: dispatchId })) as { + state: string + processAction: string + } + + // The fix: inspectWorkerTerminal re-resolved the live PTY by process incarnation, reminted a + // handle, and completion closed THAT handle — the process is actually reaped. + expect(receipt.state).toBe('released') + expect(receipt.processAction).toBe('closed_agent_terminal') + expect(runtime.closeTerminal).toHaveBeenCalledWith('term_reminted') + expect(closedPtyIds).toEqual([PTY_ID]) + expect(ptysById.has(PTY_ID)).toBe(false) + expect(await workerTerminalState(dispatchId)).toBe('released') + }) + + it('CONTROL: with the graph epoch intact the same release reaps exactly that PTY', async () => { + setup() + const { dispatchId } = await startSettledWorker() + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const receipt = (await call('orchestration.workerRelease', { dispatch: dispatchId })) as { + state: string + processAction: string + } + + expect(receipt.state).toBe('released') + expect(receipt.processAction).toBe('closed_agent_terminal') + expect(closedPtyIds).toEqual([PTY_ID]) + expect(ptysById.has(PTY_ID)).toBe(false) + expect(await workerTerminalState(dispatchId)).toBe('released') + }) + + it('FENCE (fixed): a reused ptyId carrying a different incarnation must NOT remint or close — stays release_unknown', async () => { + setup() + const { dispatchId } = await startSettledWorker() + + // The graph epoch bumps AND the ptyId has been reused by a different process (incarnation 2), + // while the worker's recorded incarnation is still 1. The exact-incarnation fence must refuse. + rendererGraphEpoch = 1 + const reused = ptysById.get(PTY_ID) + if (reused) { + reused.incarnationId = 2 + } + + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Test fixture crosses a private/runtime boundary with a verified shape. + const receipt = (await call('orchestration.workerRelease', { dispatch: dispatchId })) as { + state: string + processAction: string + } + + expect(receipt.state).toBe('release_unknown') + expect(receipt.processAction).toBe('none') + expect(runtime.closeTerminal).not.toHaveBeenCalled() + // The other lane's live process is left untouched (never reaped by an over-broad match). + expect(ptysById.get(PTY_ID)?.alive).toBe(true) + expect(await workerTerminalState(dispatchId)).toBe('release_unknown') + }) +}) diff --git a/src/main/runtime/rpc/methods/orchestration/worker/structured-worker-release-stop.ts b/src/main/runtime/rpc/methods/orchestration/worker/structured-worker-release-stop.ts index 4a32147d1a6..91e524e4bf1 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/structured-worker-release-stop.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/structured-worker-release-stop.ts @@ -4,15 +4,15 @@ import type { WorkerTerminalResourceRow } from '../../../../orchestration/worker import { stopStructuredWorker } from '../../orchestration-structured-worker-lifecycle' import type { StructuredWorkerIdentity } from '../../../../structured-worker-identity' import { archiveSummary } from './worker-terminal-resource-presentation' -import type { WorkerReleaseReceipt } from './worker-release-completion' +import { releaseUnknownRecovery, type WorkerReleaseReceipt } from './worker-release-completion' /** * The close half of a release for a worker that IS a structured session. * * Separate from the PTY close for the same reason the delivery lane is: there is no terminal to * close and no exit to observe, so the host's own settlement is the only proof available. Only a - * proven close may settle; an unproven one reports `release_unknown` and stays retryable under the - * same request id. + * proven close may settle; an unproven one reports `release_unknown` and stays retryable, but only + * under a fresh request id — replaying the prior one just returns this same receipt. */ export async function stopStructuredWorkerForRelease(args: { structured: StructuredWorkerIdentity @@ -36,7 +36,7 @@ export async function stopStructuredWorkerForRelease(args: { processAction: stop.closeAttempted ? 'closed_agent_terminal' : 'none', archive: { source: args.archiveSource, status: args.archiveStatus }, lastError: unknown.release_error ?? stop.reason, - recovery: `Inspect with: orca orchestration worker-show --dispatch ${dispatchId} --json — then repeat worker-release with the same --retry-request.` + recovery: releaseUnknownRecovery(dispatchId) } } const settled = db.settleWorkerTerminalRelease(resource.id) diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-control.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-control.ts index 3187d55cccf..a9211532626 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-control.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-control.ts @@ -143,6 +143,9 @@ export const ORCHESTRATION_WORKER_CONTROL_METHODS = [ `Worker Dispatch ${params.dispatch} no longer resolves to its exact process.` ) } + // Read via the handle inspectWorkerTerminal proved live: the durable one, or a handle + // re-minted from the recorded incarnation after the durable handle went stale. + const liveHandle = observation.terminalHandle ?? terminalHandle const structured = await readStructuredWorkerOutput({ db, dispatchId: params.dispatch, @@ -163,7 +166,7 @@ export const ORCHESTRATION_WORKER_CONTROL_METHODS = [ const output = await readExactWorkerOutput({ runtime, dispatchId: params.dispatch, - terminalHandle, + terminalHandle: liveHandle, workerState: worker?.state ?? 'unsupervised', terminalStatus: observation.status === 'exited' diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-observation.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-observation.ts index 81f3a2c2975..f1202c6e7cf 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-observation.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-observation.ts @@ -16,6 +16,7 @@ import type { WorkerDispatchRow } from '../../../../orchestration/types' +/** Observe a worker terminal, re-minting a live handle from the recorded process incarnation when the durable handle went stale, so a still-running worker is never reported missing and leaked. */ export async function inspectWorkerTerminal( runtime: OrcaRuntimeService, db: OrchestrationDb, @@ -31,12 +32,15 @@ export async function inspectWorkerTerminal( /** Structured workers only: whether mail still reaches it — at rest included, since the mail * starts it. Absent when ownership cannot be read. `status` stays the process verdict. */ addressable?: boolean + /** The handle that actually resolved: the durable one, or a live handle re-minted from the + * recorded process incarnation after the durable handle went stale. Null when none resolved. */ + terminalHandle: string | null }> { const worker = db.getWorkerDispatch(dispatchId) const terminalHandle = worker?.agent_terminal_handle ?? db.getDispatchContextById(dispatchId)?.assignee_handle if (!terminalHandle) { - return { terminal: null, exact: false, status: 'unattached' } + return { terminal: null, exact: false, status: 'unattached', terminalHandle: null } } const structured = resolveStructuredWorkerForDispatch(db, dispatchId) if (structured) { @@ -63,20 +67,44 @@ export async function inspectWorkerTerminal( exact, status: exact ? observation.status : 'identity_changed', ...(exact && observation.reason ? { reason: observation.reason } : {}), - ...(exact && addressable !== null ? { addressable } : {}) + ...(exact && addressable !== null ? { addressable } : {}), + terminalHandle: null } } - const terminal = await runtime.showTerminal(terminalHandle).catch(() => null) + let effectiveHandle = terminalHandle + let terminal = await runtime.showTerminal(effectiveHandle).catch(() => null) if (!terminal) { - return { terminal: null, exact: false, status: 'missing' } + // Why: the durable handle resolves nowhere after a renderer graph epoch bump or handle + // invalidation, yet the recorded process incarnation may still name a live PTY. Re-mint a + // live handle (incarnation-fenced) so worker-show and release act on the still-running + // process instead of reporting it missing — which would leak the agent process tree. + // (workerList is a pure DB projection and never calls inspectWorkerTerminal.) + const resource = db.getWorkerTerminalResourceByOwner(dispatchId) + const reminted = resource?.process_incarnation + ? runtime.resolveTerminalHandleByProcessIncarnation( + resource.process_incarnation, + resource.host_scope + ) + : null + if (reminted) { + const remintedTerminal = await runtime.showTerminal(reminted).catch(() => null) + if (remintedTerminal) { + effectiveHandle = reminted + terminal = remintedTerminal + } + } + } + if (!terminal) { + // The re-mint above failed, so no live handle resolved; report the durable handle unresolved. + return { terminal: null, exact: false, status: 'missing', terminalHandle: null } } const exact = db.isDispatchProcessCurrent({ dispatchId, - paneKey: runtime.getTerminalPaneKey(terminalHandle), - processIncarnation: runtime.getTerminalProcessIncarnation(terminalHandle) + paneKey: runtime.getTerminalPaneKey(effectiveHandle), + processIncarnation: runtime.getTerminalProcessIncarnation(effectiveHandle) }) if (!exact) { - return { terminal, exact, status: 'identity_changed' } + return { terminal, exact, status: 'identity_changed', terminalHandle: effectiveHandle } } // Why: the aggregate inventory only iterates registered providers, so a dropped // relay clears `connected` for every remote PTY at once. Lost contact is not a @@ -86,38 +114,48 @@ export async function inspectWorkerTerminal( // Exact-gated by the early return above: a replaced process's prompt would attribute another // lane's blocker to this worker. const agentWait = terminal.agentWait - const verdict = runtime.getTerminalLivenessVerdict?.(terminalHandle) ?? null + const verdict = runtime.getTerminalLivenessVerdict?.(effectiveHandle) ?? null if (verdict?.status === 'unverifiable') { - return { terminal, exact, status: 'unverifiable', reason: verdict.reason, agentWait } + return { + terminal, + exact, + status: 'unverifiable', + reason: verdict.reason, + agentWait, + terminalHandle: effectiveHandle + } } if (verdict?.status === 'live') { - return { terminal, exact, status: 'live', agentWait } + return { terminal, exact, status: 'live', agentWait, terminalHandle: effectiveHandle } } if (!verdict) { const dispatch = db.getDispatchContextById?.(dispatchId) const persistedHostScope = parseWorkerTerminalHostScope(dispatch?.host_scope ?? null) - const currentHostScope = runtime.getOrchestrationDispatchAuthority?.(terminalHandle)?.hostScope + const currentHostScope = runtime.getOrchestrationDispatchAuthority?.(effectiveHandle)?.hostScope if (persistedHostScope?.kind === 'ssh' || currentHostScope?.kind === 'ssh') { return { terminal, exact, status: 'unverifiable', reason: 'missing_liveness_verdict', - agentWait + agentWait, + terminalHandle: effectiveHandle } } return { terminal, exact, status: terminal.connected === false ? 'exited' : 'live', - agentWait + agentWait, + terminalHandle: effectiveHandle } } return { terminal, exact, status: 'exited', - agentWait + agentWait, + terminalHandle: effectiveHandle } } diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-release-completion.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-release-completion.ts index 0ae55f8a938..4aca4d0420a 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-release-completion.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-release-completion.ts @@ -120,6 +120,9 @@ async function completeWorkerTerminalReleaseOnce( } } const observation = await inspectWorkerTerminal(runtime, db, dispatchId) + // The live handle to act on: the durable one, or a handle re-minted from the recorded process + // incarnation when the durable handle went stale (inspectWorkerTerminal proved it live). + const terminalHandle = observation.terminalHandle ?? resource.terminal_handle if (observation.status === 'identity_changed') { const retained = db.revertWorkerTerminalReleaseToRetained(resource.id, 'identity_unproven') return { @@ -131,33 +134,39 @@ async function completeWorkerTerminalReleaseOnce( } } if (observation.status === 'missing' || observation.status === 'unattached') { - if (args.mode === 'recovery') { - // A close can succeed before the process crashes, leaving `releasing` durable state while - // terminal inventory no longer resolves the handle. Only a positive host liveness verdict - // may settle that exact incarnation; contact loss remains pending/unverifiable. - if (resource.process_incarnation) { - const processLiveness = await runtime.inspectTerminalProcessIncarnationLiveness( - resource.process_incarnation, - resource.host_scope - ) - if (processLiveness === 'exited') { - const reconciled = db.settleDeadWorkerTerminalRelease({ - requestingDispatchId: dispatchId, - resourceId: resource.id, - processIncarnation: resource.process_incarnation - }) - if (reconciled.disposition === 'released') { - runtime.notifyMessageArrived(`dispatch:${dispatchId}`, 'status') - return { - dispatchId, - state: 'released', - processAction: 'closed_exited_terminal', - archive: archiveSummary(reconciled.resource) - } + // Re-resolution by process incarnation (inspectWorkerTerminal) already failed, so no live PTY + // carries this worker's exact incarnation. If that incarnation is provably gone, settle + // released BEFORE the recovery defer: proof of death outranks deferral, so a provably-exited + // worker never languishes in release_pending across recovery passes. + if (resource.process_incarnation) { + const processLiveness = await runtime.inspectTerminalProcessIncarnationLiveness( + resource.process_incarnation, + resource.host_scope + ) + if (processLiveness === 'exited') { + // Prefer incarnation-fenced settle (dispatch relation + process_incarnation CAS). + const reconciled = db.settleDeadWorkerTerminalRelease({ + requestingDispatchId: dispatchId, + resourceId: resource.id, + processIncarnation: resource.process_incarnation + }) + if (reconciled.disposition === 'released') { + runtime.notifyMessageArrived(`dispatch:${dispatchId}`, 'status') + return { + dispatchId, + state: 'released', + processAction: 'none', + archive: archiveSummary(reconciled.resource) } } + // settleDead retains when the archive is still mandatory and missing (e.g. requested but + // never committed). Do NOT plain-settle: that would discard output and break recovery's + // "archive is mandatory" invariant. Fall through to recovery pending / unknown instead. } - // Inventory may still be incomplete during startup/reconnect discovery; defer. + } + if (args.mode === 'recovery') { + // No death certificate yet: inventory may still be incomplete during startup/reconnect + // discovery, so defer instead of guessing. return { dispatchId, state: 'release_pending', @@ -183,7 +192,7 @@ async function completeWorkerTerminalReleaseOnce( } } - if (!workerTerminalLeaseIsCurrent(runtime, db, dispatchId, resource)) { + if (!workerTerminalLeaseIsCurrent(runtime, db, dispatchId, resource, terminalHandle)) { const retained = db.revertWorkerTerminalReleaseToRetained(resource.id, 'identity_unproven') return { dispatchId, @@ -202,7 +211,7 @@ async function completeWorkerTerminalReleaseOnce( const captured = await captureWorkerOutputArchive({ runtime, dispatchId, - terminalHandle: resource.terminal_handle, + terminalHandle, attachedAtMs: orchestrationTimestampToMs(worker.created_at), structuredWorker: structured }) @@ -230,7 +239,7 @@ async function completeWorkerTerminalReleaseOnce( archive: archiveSummary(releasing) } } - if (!workerTerminalLeaseIsCurrent(runtime, db, dispatchId, releasing)) { + if (!workerTerminalLeaseIsCurrent(runtime, db, dispatchId, releasing, terminalHandle)) { const retained = db.revertWorkerTerminalReleaseToRetained(resource.id, 'identity_unproven') return { dispatchId, @@ -253,7 +262,7 @@ async function completeWorkerTerminalReleaseOnce( archiveStatus }) } - const close = await runtime.closeTerminal(resource.terminal_handle) + const close = await runtime.closeTerminal(terminalHandle) if (!close.ptyKilled) { const reason = describeUnconfirmedAgentStop(close) const unknown = db.markWorkerTerminalReleaseUnknown(resource.id, reason) diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-release-runtime-incarnation.test.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-release-runtime-incarnation.test.ts new file mode 100644 index 00000000000..4caac62f5da --- /dev/null +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-release-runtime-incarnation.test.ts @@ -0,0 +1,79 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { inspectWorkerTerminal } from './worker-observation' +import { createOrchestrationWorkerReleaseHarness } from './worker-release.test-support' +import { makePaneKey } from '../../../../../../shared/stable-pane-id' + +const PTY_ID = 'runtime_test:term_worker' +const LEAF_ID = 'bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb' + +describe('worker release through runtime incarnation recovery', () => { + const harness = createOrchestrationWorkerReleaseHarness() + afterEach(() => harness.cleanup()) + + it.each(['release', 'stop', 'read', 'replacement', 'disconnected'] as const)( + '%s uses runtime identity and close paths', + async (scenario) => { + harness.setup() + const { dispatchId } = + scenario === 'stop' ? await harness.startWorker() : await harness.startSettledWorker() + const runtime = harness.runtime + vi.mocked(runtime.showTerminal).mockRestore() + vi.mocked(runtime.getTerminalPaneKey).mockRestore() + vi.mocked(runtime.getTerminalProcessIncarnation).mockRestore() + vi.mocked(runtime.getOrchestrationDispatchAuthority).mockRestore() + vi.mocked(runtime.readTerminal).mockRestore() + vi.mocked(runtime.closeTerminal).mockRestore() + const kill = vi.fn(() => true) + runtime.setPtyController({ write: () => true, kill, getForegroundProcess: async () => null }) + runtime.registerPty(PTY_ID, 'repo::worktree', null, { + tabId: 'tab_worker', + leafId: LEAF_ID, + incarnationId: scenario === 'replacement' ? '2' : '1' + }) + await expect(runtime.showTerminal('term_worker')).rejects.toThrow() + if (scenario === 'disconnected') { + vi.spyOn(runtime, 'getTerminalLivenessVerdict').mockReturnValue({ + status: 'unverifiable', + reason: 'transport unavailable' + }) + const observed = await inspectWorkerTerminal(runtime, harness.db, dispatchId) + expect(observed).toMatchObject({ exact: true, status: 'unverifiable' }) + expect(kill).not.toHaveBeenCalled() + return + } + if (scenario === 'stop') { + const receipt = await harness.call('orchestration.workerStop', { dispatch: dispatchId }) + expect(receipt).toMatchObject({ state: 'stopped', processAction: 'closed_agent_terminal' }) + expect(kill).toHaveBeenCalledExactlyOnceWith(PTY_ID) + return + } + if (scenario === 'read') { + const readTerminal = vi.spyOn(runtime, 'readTerminal') + const output = await harness.call('orchestration.workerRead', { dispatch: dispatchId }) + // Pins the read to the reminted terminal: the handle it reached must resolve to the + // registered pane and incarnation, which the stale durable handle never does. + const [[readHandle]] = readTerminal.mock.calls + expect(runtime.getTerminalPaneKey(readHandle)).toBe(makePaneKey('tab_worker', LEAF_ID)) + expect(runtime.getTerminalProcessIncarnation(readHandle)).toBe(`${PTY_ID}:1`) + expect(output).toMatchObject({ + source: 'terminal', + fallbackReason: 'session_not_reported' + }) + expect(kill).not.toHaveBeenCalled() + return + } + const receipt = await harness.call('orchestration.workerRelease', { dispatch: dispatchId }) + if (scenario === 'replacement') { + expect(receipt).toMatchObject({ state: 'release_unknown', processAction: 'none' }) + expect(kill).not.toHaveBeenCalled() + return + } + expect(receipt).toMatchObject({ state: 'released', processAction: 'closed_agent_terminal' }) + expect(kill).toHaveBeenCalledExactlyOnceWith(PTY_ID) + expect(harness.db.getWorkerTerminalResourceByOwner(dispatchId)).toMatchObject({ + ownership_state: 'released', + release_state: 'released' + }) + } + ) +}) diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-stop.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-stop.ts index 98d3c376f61..ee0e73c904f 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-stop.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-stop.ts @@ -141,6 +141,7 @@ export const ORCHESTRATION_WORKER_STOP_METHODS = [ }) } const observation = await inspectWorkerTerminal(runtime, db, params.dispatch) + const liveHandle = observation.terminalHandle ?? handle // The host exit can settle this stop while terminal inspection is awaiting inventory. if (db.getWorkerDispatch(params.dispatch)?.state === 'stopped') { runtime.notifyMessageArrived(`dispatch:${params.dispatch}`, 'status') @@ -201,7 +202,7 @@ export const ORCHESTRATION_WORKER_STOP_METHODS = [ } } const closed = await runtime - .closeTerminal(handle) + .closeTerminal(liveHandle) .then((close) => ({ close }) as const) .catch( (error: unknown) => diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-terminal-release-lease.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-terminal-release-lease.ts index 50a97c9de4b..cad0ceb4128 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-terminal-release-lease.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-terminal-release-lease.ts @@ -9,23 +9,27 @@ export function workerTerminalLeaseIsCurrent( runtime: OrcaRuntimeService, db: OrchestrationDb, dispatchId: string, - resource: WorkerTerminalResourceRow + resource: WorkerTerminalResourceRow, + // Live (possibly reminted) handle for runtime probes. Durable identity still compares + // worker.agent_terminal_handle to resource.terminal_handle; after beginGraphReload the durable + // string is absent from the handle table, so authority/pane/incarnation must use the live one. + liveTerminalHandle: string ): boolean { const worker = db.getWorkerDispatch(dispatchId) if (isStructuredWorkerHandle(resource.terminal_handle)) { return structuredWorkerTerminalLeaseIsCurrent(db, dispatchId, worker, resource) } - const authority = runtime.getOrchestrationDispatchAuthority(resource.terminal_handle) + const authority = runtime.getOrchestrationDispatchAuthority(liveTerminalHandle) // Exited PTYs retain identity and host evidence but no longer mint launch authority. return Boolean( worker?.agent_terminal_handle === resource.terminal_handle && (authority ? resource.host_scope === JSON.stringify(authority.hostScope) - : runtime.getTerminalLivenessVerdict(resource.terminal_handle)?.status === 'exited') && + : runtime.getTerminalLivenessVerdict(liveTerminalHandle)?.status === 'exited') && db.isDispatchProcessCurrent({ dispatchId, - paneKey: runtime.getTerminalPaneKey(resource.terminal_handle), - processIncarnation: runtime.getTerminalProcessIncarnation(resource.terminal_handle) + paneKey: runtime.getTerminalPaneKey(liveTerminalHandle), + processIncarnation: runtime.getTerminalProcessIncarnation(liveTerminalHandle) }) && !db.workerTerminalResourceHasIdentityConflict(resource.id) )