diff --git a/src/main/daemon/daemon-exit-receipt-cleanup.test.ts b/src/main/daemon/daemon-exit-receipt-cleanup.test.ts new file mode 100644 index 000000000000..d8a730978a4c --- /dev/null +++ b/src/main/daemon/daemon-exit-receipt-cleanup.test.ts @@ -0,0 +1,183 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { rmSync } from 'node:fs' +import { createMockSubprocess, startDaemonAdapterHarness } from './daemon-pty-adapter-test-harness' +import type { DaemonRequestRouter } from './daemon-request-router' +import type { DaemonPtySpawnPreparations } from './daemon-pty-spawn-preparations' +import type { DaemonRequest } from './types' +import { MAX_EXIT_RECEIPTS, type TerminalHostSessionRecord } from './terminal-host-session-record' +import { localProvider, setLocalPtyProvider } from '../ipc/pty/provider/registry' +import { + inspectExitedIncarnationFromRuntimeController, + releaseExitedIncarnationFromRuntimeController +} from '../ipc/pty/runtime/operations' + +describe('exit receipt consumption is bookkeeping, not terminal shutdown', () => { + let harness: Awaited> + let subprocess: ReturnType + let previousProvider: typeof localProvider + + beforeEach(async () => { + previousProvider = localProvider + harness = await startDaemonAdapterHarness(() => { + subprocess = createMockSubprocess() + return subprocess + }) + setLocalPtyProvider(harness.adapter) + }) + + afterEach(async () => { + vi.restoreAllMocks() + setLocalPtyProvider(previousProvider) + harness?.adapter.dispose() + await harness?.server.shutdown() + if (harness) { + rmSync(harness.dir, { recursive: true, force: true }) + } + }) + + function router(): DaemonRequestRouter { + return (harness.server as unknown as { requestRouter: DaemonRequestRouter }).requestRouter + } + + async function exited(id: string) { + const spawned = await harness.adapter.spawn({ sessionId: id, cols: 80, rows: 24 }) + const incarnation = spawned.incarnationId! + subprocess._simulateExit(0) + expect(await inspectExitedIncarnationFromRuntimeController(id, incarnation)).toBe(true) + return incarnation + } + + it.each(['current', 'legacy without kill fence'] as const)( + 'preserves a replacement process on a %s daemon', + async (version) => { + const id = 'receipt-replaced' + const oldIncarnation = await exited(id) + const replacement = await harness.adapter.spawn({ sessionId: id, cols: 80, rows: 24 }) + const child = subprocess + expect(replacement.incarnationId).not.toBe(oldIncarnation) + if (version === 'legacy without kill fence') { + const originalRoute = router().route.bind(router()) + vi.spyOn(router(), 'route').mockImplementation((clientId, request) => { + if (request.type === 'consumeExitReceipt') { + throw new Error('Unknown request type: consumeExitReceipt') + } + // v1.4.199 reads sessionId/immediate but ignores the new kill condition. + const compatible: DaemonRequest = + request.type === 'kill' + ? { + ...request, + payload: { + sessionId: request.payload.sessionId, + immediate: request.payload.immediate + } + } + : request + return originalRoute(clientId, compatible) + }) + } + + await releaseExitedIncarnationFromRuntimeController(id, oldIncarnation) + + expect(child.forceKill).not.toHaveBeenCalled() + expect(child.kill).not.toHaveBeenCalled() + expect(await harness.adapter.probePtyLiveness(id)).toBe(true) + } + ) + + it('leaves another client’s replacement spawn preparation untouched', async () => { + const id = 'receipt-preparing' + const incarnation = await exited(id) + const preparations = ( + router() as unknown as { + options: { preparations: DaemonPtySpawnPreparations } + } + ).options.preparations + // Register at the real admission boundary before native spawn has completed. + const preparation = preparations.register(id, 'replacement-client', 'spawn-request', undefined) + try { + await releaseExitedIncarnationFromRuntimeController(id, incarnation) + expect(preparation.canceled).toBe(false) + expect(preparation.controller.signal.aborted).toBe(false) + await expect(preparations.prepareUnlessCanceled(id, preparation)).resolves.toBeUndefined() + } finally { + preparations.finish(id, preparation) + } + }) + + it('cannot stop even a live process with the matching incarnation', async () => { + const id = 'receipt-still-live' + const spawned = await harness.adapter.spawn({ sessionId: id, cols: 80, rows: 24 }) + const child = subprocess + await releaseExitedIncarnationFromRuntimeController(id, spawned.incarnationId!) + expect(child.kill).not.toHaveBeenCalled() + expect(child.forceKill).not.toHaveBeenCalled() + expect(await harness.adapter.probePtyLiveness(id)).toBe(true) + }) + + it('returns no exit proof after eviction and safely accepts late acknowledgement', async () => { + await harness.adapter.spawn({ sessionId: 'live-sentinel', cols: 80, rows: 24 }) + const host = ( + harness.server as unknown as { + host: { sessions: Map } + } + ).host + for (let i = 0; i < MAX_EXIT_RECEIPTS; i++) { + host.sessions.set(`old-${i}`, { incarnationId: `old-inc-${i}`, code: 0 }) + } + expect(await inspectExitedIncarnationFromRuntimeController('old-0', 'old-inc-0')).toBe(true) + const fresh = await exited('fresh-exit') + expect(await inspectExitedIncarnationFromRuntimeController('old-0', 'old-inc-0')).toBe(false) + expect(await inspectExitedIncarnationFromRuntimeController('old-1', 'old-inc-1')).toBe(true) + expect(await inspectExitedIncarnationFromRuntimeController('fresh-exit', fresh)).toBe(true) + expect(await harness.adapter.probePtyLiveness('live-sentinel')).toBe(true) + await expect(harness.adapter.consumeExitReceipt('old-0', 'old-inc-0')).resolves.toBeUndefined() + }) + + it('consumes only matching receipts and tolerates duplicate or missing acknowledgements', async () => { + const id = 'receipt-repeat' + const incarnation = await exited(id) + await harness.adapter.consumeExitReceipt(id, 'different-incarnation') + expect(await inspectExitedIncarnationFromRuntimeController(id, incarnation)).toBe(true) + await harness.adapter.consumeExitReceipt(id, incarnation) + expect(await inspectExitedIncarnationFromRuntimeController(id, incarnation)).toBe(false) + await expect(harness.adapter.consumeExitReceipt(id, incarnation)).resolves.toBeUndefined() + await expect(harness.adapter.consumeExitReceipt('absent', incarnation)).resolves.toBeUndefined() + }) + + it('leaves evidence intact when an old host rejects consumption', async () => { + const id = 'receipt-unsupported' + const incarnation = await exited(id) + const originalRoute = router().route.bind(router()) + const calls = vi.spyOn(router(), 'route').mockImplementation((clientId, request) => { + if (request.type === 'consumeExitReceipt') { + throw new Error('Unknown request type: consumeExitReceipt') + } + return originalRoute(clientId, request) + }) + await expect( + releaseExitedIncarnationFromRuntimeController(id, incarnation) + ).resolves.toBeUndefined() + expect(await inspectExitedIncarnationFromRuntimeController(id, incarnation)).toBe(true) + expect(calls.mock.calls.some(([, request]) => request.type === 'kill')).toBe(false) + }) + + it('preserves history and does not mark an observed exit as an explicit kill', async () => { + const id = 'receipt-history' + const incarnation = await exited(id) + const state = harness.adapter as unknown as { + historyManager: unknown + killedSessionTombstones: Map + } + const originalHistory = state.historyManager + const removeSession = vi.fn(async () => {}) + state.historyManager = { removeSession, closeSession: vi.fn(async () => {}) } + try { + await releaseExitedIncarnationFromRuntimeController(id, incarnation) + expect(removeSession).not.toHaveBeenCalled() + expect(state.killedSessionTombstones.has(id)).toBe(false) + expect(await inspectExitedIncarnationFromRuntimeController(id, incarnation)).toBe(false) + } finally { + state.historyManager = originalHistory + } + }) +}) diff --git a/src/main/daemon/daemon-inspection-incarnation-routing.test.ts b/src/main/daemon/daemon-inspection-incarnation-routing.test.ts new file mode 100644 index 000000000000..3aaeafad51fc --- /dev/null +++ b/src/main/daemon/daemon-inspection-incarnation-routing.test.ts @@ -0,0 +1,132 @@ +// A worker whose shell ended while Orca was closed leaves no route behind: nothing in this +// process claims its id any more. The incarnation-scoped question is exactly the one that has to +// survive that, so both daemon-fronting providers must carry it (and its `expectedIncarnationId`) +// to a daemon instead of answering `terminal_gone` from their own bookkeeping. +import { describe, expect, it, vi } from 'vitest' +import { DaemonPtyRouter } from './daemon-pty-router' +import { DegradedDaemonPtyProvider } from './degraded-daemon-pty-provider' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import type { IPtyProvider } from '../providers/types' +import type { PtyProcessInspection } from '../providers/pty-process-inspection' + +const INCARNATION = 'c0ffee00-0000-4000-8000-000000000001' + +const EXITED: PtyProcessInspection = { + foregroundProcess: null, + hasChildProcesses: false, + foregroundProcessEvidence: { + authorityGeneration: 'gen-1', + observationEpoch: 1, + capturedAgeMs: 0, + ptyId: 'pty-away', + ptyIncarnationId: INCARNATION, + verdict: 'exited', + reason: 'pty_exit_0' + } +} + +function createAdapter(sessions: string[] = []): DaemonPtyAdapter { + return { + hasPty: vi.fn((id: string) => sessions.includes(id)), + inspectProcess: vi.fn(async () => EXITED), + consumeExitReceipt: vi.fn(async () => {}), + listProcesses: vi.fn(async () => sessions.map((id) => ({ id, cwd: '', title: 'daemon' }))), + onData: vi.fn(() => () => {}), + onExit: vi.fn(() => () => {}), + onWriteUnavailable: vi.fn(() => () => {}), + onBackgroundStreamEvent: vi.fn(() => () => {}) + } as unknown as DaemonPtyAdapter +} + +function createFallbackProvider(): IPtyProvider { + return { + hasPty: vi.fn(() => false), + inspectProcess: vi.fn(async () => EXITED), + listProcesses: vi.fn(async () => []), + onData: vi.fn(() => () => {}), + onExit: vi.fn(() => () => {}), + onWriteUnavailable: vi.fn(() => () => {}), + onBackgroundStreamEvent: vi.fn(() => () => {}) + } as unknown as IPtyProvider +} + +describe('DaemonPtyRouter incarnation-scoped inspection', () => { + it('routes receipt consumption like inspection without retiring its route', async () => { + const current = createAdapter() + const legacy = createAdapter(['pty-live']) + const router = new DaemonPtyRouter({ current, legacy: [legacy] }) + await router.consumeExitReceipt('pty-away', INCARNATION) + expect(current.consumeExitReceipt).toHaveBeenCalledWith('pty-away', INCARNATION) + await router.consumeExitReceipt('pty-live', INCARNATION) + expect(legacy.consumeExitReceipt).toHaveBeenCalledWith('pty-live', INCARNATION) + await router.inspectProcess('pty-live', { expectedIncarnationId: INCARNATION }) + expect(legacy.inspectProcess).toHaveBeenCalled() + }) + it('asks the current daemon about an unclaimed id when the caller names an incarnation', async () => { + const current = createAdapter() + const router = new DaemonPtyRouter({ current, legacy: [createAdapter()] }) + + await expect( + router.inspectProcess('pty-away', { expectedIncarnationId: INCARNATION }) + ).resolves.toEqual(EXITED) + expect(current.inspectProcess).toHaveBeenCalledWith('pty-away', { + expectedIncarnationId: INCARNATION + }) + }) + + it('forwards the incarnation to the daemon that still claims the id', async () => { + const legacy = createAdapter(['pty-live']) + const router = new DaemonPtyRouter({ current: createAdapter(), legacy: [legacy] }) + + await router.inspectProcess('pty-live', { expectedIncarnationId: INCARNATION }) + + expect(legacy.inspectProcess).toHaveBeenCalledWith('pty-live', { + expectedIncarnationId: INCARNATION + }) + }) +}) + +describe('DegradedDaemonPtyProvider incarnation-scoped inspection', () => { + it('consumes unclaimed daemon evidence without using fallback shutdown', async () => { + const current = createAdapter() + const fallback = createFallbackProvider() + const provider = new DegradedDaemonPtyProvider({ current, legacy: [], fallback }) + await provider.consumeExitReceipt('pty-away', INCARNATION) + expect(current.consumeExitReceipt).toHaveBeenCalledWith('pty-away', INCARNATION) + }) + it('asks the current daemon about an unclaimed id when the caller names an incarnation', async () => { + const current = createAdapter() + const provider = new DegradedDaemonPtyProvider({ + current, + legacy: [], + fallback: createFallbackProvider() + }) + + await expect( + provider.inspectProcess('pty-away', { expectedIncarnationId: INCARNATION }) + ).resolves.toEqual(EXITED) + expect(current.inspectProcess).toHaveBeenCalledWith('pty-away', { + expectedIncarnationId: INCARNATION + }) + }) + + it('forwards inspection options to the provider that owns the session', async () => { + const current = createAdapter(['pty-live']) + const provider = new DegradedDaemonPtyProvider({ + current, + legacy: [], + fallback: createFallbackProvider() + }) + await provider.discoverDaemonSessions() + + await provider.inspectProcess('pty-live', { + expectedIncarnationId: INCARNATION, + steadyState: true + }) + + expect(current.inspectProcess).toHaveBeenCalledWith('pty-live', { + expectedIncarnationId: INCARNATION, + steadyState: true + }) + }) +}) diff --git a/src/main/daemon/daemon-pty-adapter-kill-incarnation-fence.test.ts b/src/main/daemon/daemon-pty-adapter-kill-incarnation-fence.test.ts new file mode 100644 index 000000000000..99accb3d851c --- /dev/null +++ b/src/main/daemon/daemon-pty-adapter-kill-incarnation-fence.test.ts @@ -0,0 +1,47 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { rmSync } from 'node:fs' +import type { DaemonPtyAdapter } from './daemon-pty-adapter' +import { createMockSubprocess, startDaemonAdapterHarness } from './daemon-pty-adapter-test-harness' + +describe('DaemonPtyAdapter shutdown incarnation fence', () => { + let adapter: DaemonPtyAdapter + let server: Awaited>['server'] + let tempDir: string + let lastSubprocess: ReturnType + + beforeEach(async () => { + const harness = await startDaemonAdapterHarness(() => { + lastSubprocess = createMockSubprocess() + return lastSubprocess + }) + adapter = harness.adapter + server = harness.server + tempDir = harness.dir + }) + + afterEach(async () => { + adapter?.dispose() + await server?.shutdown() + if (tempDir) { + rmSync(tempDir, { recursive: true, force: true }) + } + }) + + it('refuses over the wire to end a shell that is not the named incarnation', async () => { + const sessionId = 'kill-fence' + const spawned = await adapter.spawn({ cols: 80, rows: 24, sessionId }) + const live = lastSubprocess + + await expect( + adapter.shutdown(sessionId, { immediate: true, expectedIncarnationId: 'an-older-shell' }) + ).rejects.toThrow('PTY incarnation mismatch') + expect(live.kill).not.toHaveBeenCalled() + expect(live.forceKill).not.toHaveBeenCalled() + + await adapter.shutdown(sessionId, { + immediate: true, + expectedIncarnationId: spawned.incarnationId + }) + expect(live.forceKill).toHaveBeenCalled() + }) +}) diff --git a/src/main/daemon/daemon-pty-process-inspection.ts b/src/main/daemon/daemon-pty-process-inspection.ts index 335c641da627..97ed5110490d 100644 --- a/src/main/daemon/daemon-pty-process-inspection.ts +++ b/src/main/daemon/daemon-pty-process-inspection.ts @@ -11,6 +11,11 @@ import type { PtyProcessInspection } from '../providers/pty-process-inspection' import { clientOnlyUnverifiableInspection } from '../../shared/terminal-process-inspection' export abstract class DaemonPtyProcessInspection extends DaemonPtyBufferSnapshots { + async consumeExitReceipt(id: string, incarnationId: string): Promise { + // Predecessors reject this distinct request; never fall back to shutdown. + await this.client.request('consumeExitReceipt', { sessionId: id, incarnationId }) + } + // Why: daemon-backed PTYs can host long-lived agents while detached; cleanup prompts must not treat them as idle shells. protected hasChildProcessesFromForeground(foregroundProcess: string | null): boolean { return foregroundProcess !== null && !isShellProcess(foregroundProcess) diff --git a/src/main/daemon/daemon-pty-router.ts b/src/main/daemon/daemon-pty-router.ts index 2fbfd17a0391..53c17440146f 100644 --- a/src/main/daemon/daemon-pty-router.ts +++ b/src/main/daemon/daemon-pty-router.ts @@ -114,10 +114,7 @@ export class DaemonPtyRouter implements IPtyProvider { this.adapterFor(id).setPtyBackgrounded(id, background) } - async shutdown( - id: string, - opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number } - ): Promise { + async shutdown(id: string, opts: Parameters[1]): Promise { const adapter = this.adapterFor(id) const migrateHistory = shouldHandoffDaemonHistory(opts.keepHistory, adapter, this.current) await adapter.shutdown(id, opts) @@ -131,6 +128,10 @@ export class DaemonPtyRouter implements IPtyProvider { } } + async consumeExitReceipt(id: string, incarnationId: string): Promise { + await this.adapterForInspection(id, incarnationId).consumeExitReceipt(id, incarnationId) + } + async sendSignal(id: string, signal: string): Promise { await this.adapterFor(id).sendSignal(id, signal) } @@ -182,7 +183,7 @@ export class DaemonPtyRouter implements IPtyProvider { id: string, options?: { expectedIncarnationId?: string; steadyState?: boolean } ): Promise { - return this.adapterForInspection(id).inspectProcess(id, options) + return this.adapterForInspection(id, options?.expectedIncarnationId).inspectProcess(id, options) } async confirmForegroundProcess(id: string): Promise { @@ -325,15 +326,25 @@ export class DaemonPtyRouter implements IPtyProvider { return this.sessionAdapters.get(sessionId) ?? this.current } - private adapterForInspection(sessionId: string): DaemonPtyAdapter { + private adapterForInspection( + sessionId: string, + expectedIncarnationId?: string + ): DaemonPtyAdapter { const adapter = this.sessionAdapters.get(sessionId) ?? this.allAdapters().find((candidate) => candidate.hasPty(sessionId)) - if (!adapter) { + if (adapter) { + this.sessionAdapters.set(sessionId, adapter) + return adapter + } + // An unclaimed id is exactly what a caller naming a remembered incarnation asks about: the + // session that died while this client was away, so no route survived. A daemon that never held + // it answers not-found, so routing the question cannot manufacture an exit. A caller that names + // no incarnation is an ordinary poll and must not borrow a route it never owned. + if (expectedIncarnationId === undefined) { throw new Error('terminal_gone') } - this.sessionAdapters.set(sessionId, adapter) - return adapter + return this.current } private allAdapters(): DaemonPtyAdapter[] { diff --git a/src/main/daemon/daemon-pty-session-control.ts b/src/main/daemon/daemon-pty-session-control.ts index c10be5739263..4a6e4d6066ca 100644 --- a/src/main/daemon/daemon-pty-session-control.ts +++ b/src/main/daemon/daemon-pty-session-control.ts @@ -9,7 +9,7 @@ import type { ColdRestoreInfo } from './history-reader' import { normalizeWslColdRestoreCwd } from './wsl-cold-restore-cwd' import { SessionNotFoundError, type ListSessionsResult } from './types' import { resolveSafePtyDefaultCwd } from '../providers/pty-default-cwd' -import type { PtySpawnResult } from '../providers/types' +import type { IPtyProvider, PtySpawnResult } from '../providers/types' export const LIVENESS_PROBE_TIMEOUT_MS = 2_000 const MAX_TOMBSTONES = 1000 @@ -85,10 +85,7 @@ export abstract class DaemonPtySessionControl extends DaemonPtySessionInput { } } - async shutdown( - id: string, - opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number } - ): Promise { + async shutdown(id: string, opts: Parameters[1]): Promise { if (opts.keepHistory && this.disconnectOnlyPromise) { throw new Error('Cannot keep history after daemon disconnect has started') } @@ -107,7 +104,7 @@ export abstract class DaemonPtySessionControl extends DaemonPtySessionInput { protected async shutdownWithHistoryLock( id: string, - opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number } + opts: Parameters[1] ): Promise { // Why: shutdown can be the first lazy-client operation after restart; connect // before killing so a healthy daemon session is not orphaned (#7742). Connect, @@ -162,7 +159,13 @@ export abstract class DaemonPtySessionControl extends DaemonPtySessionInput { } await this.client.request( 'kill', - { sessionId: id, immediate: opts.immediate ?? false }, + { + sessionId: id, + immediate: opts.immediate ?? false, + ...(opts.expectedIncarnationId === undefined + ? {} + : { expectedIncarnationId: opts.expectedIncarnationId }) + }, remainingDaemonRequestTimeoutMs(opts.deadlineMs) ) this.activeSessionIds.delete(id) diff --git a/src/main/daemon/daemon-request-router.ts b/src/main/daemon/daemon-request-router.ts index ae007366dd02..4305fa4c1a00 100644 --- a/src/main/daemon/daemon-request-router.ts +++ b/src/main/daemon/daemon-request-router.ts @@ -90,8 +90,18 @@ export class DaemonRequestRouter { request.payload.sessionId, request.payload.background === true ) + case 'consumeExitReceipt': { + const { sessionId, incarnationId } = request.payload + this.options.host.consumeExitReceipt(sessionId, incarnationId) + return {} + } case 'kill': - return this.kill(clientId, request.payload.sessionId, request.payload.immediate) + return this.kill( + clientId, + request.payload.sessionId, + request.payload.immediate, + request.payload.expectedIncarnationId + ) case 'signal': this.options.host.signal(request.payload.sessionId, request.payload.signal) return {} @@ -194,13 +204,14 @@ export class DaemonRequestRouter { private async kill( clientId: string, sessionId: string, - immediate: boolean | undefined + immediate: boolean | undefined, + expectedIncarnationId: string | undefined ): Promise> { const canceledPendingSpawn = this.options.preparations.cancel(sessionId) this.options.attachments.clearInput(sessionId) const attribution = { sessionId, immediate: immediate === true, clientId } try { - await this.options.host.kill(sessionId, { immediate }) + await this.options.host.kill(sessionId, { immediate, expectedIncarnationId }) } catch (error) { if (!(canceledPendingSpawn && error instanceof SessionNotFoundError)) { this.options.log.log('session-kill-failed', { diff --git a/src/main/daemon/daemon-session-control-protocol.ts b/src/main/daemon/daemon-session-control-protocol.ts new file mode 100644 index 000000000000..4febd93742ae --- /dev/null +++ b/src/main/daemon/daemon-session-control-protocol.ts @@ -0,0 +1,16 @@ +export type KillRequest = { + id: string + type: 'kill' + payload: { + sessionId: string + immediate?: boolean + /** Refuse unless the session on this id is still this incarnation. */ + expectedIncarnationId?: string + } +} + +export type ConsumeExitReceiptRequest = { + id: string + type: 'consumeExitReceipt' + payload: { sessionId: string; incarnationId: string } +} diff --git a/src/main/daemon/degraded-daemon-pty-provider.ts b/src/main/daemon/degraded-daemon-pty-provider.ts index 2b50424ae830..50ba0a287e20 100644 --- a/src/main/daemon/degraded-daemon-pty-provider.ts +++ b/src/main/daemon/degraded-daemon-pty-provider.ts @@ -1,7 +1,7 @@ import type { DaemonPtyAdapter } from './daemon-pty-adapter' import { combineUnsubscribes } from './combine-unsubscribes' import { shutdownDegradedFallbackSessions } from './degraded-daemon-fallback-shutdown' -import { inspectPtyProviderProcess } from '../providers/pty-process-inspection' +import type { PtyProcessInspectionOptions } from '../providers/pty-process-inspection' import type { IPtyProvider, PtyBackgroundStreamEvent, @@ -15,6 +15,7 @@ import { adoptOwningProvider, attachDaemonOwnedSession, findDaemonAdapter, + inspectRoutedDaemonProcess, listProviderSessionIds } from './degraded-daemon-session-routing' import { DegradedDaemonFreshSpawnRouter } from './degraded-daemon-fresh-spawn-routing' @@ -133,10 +134,7 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { this.providerFor(id).setPtyBackgrounded?.(id, background) } - async shutdown( - id: string, - opts: { immediate?: boolean; keepHistory?: boolean; deadlineMs?: number } - ): Promise { + async shutdown(id: string, opts: Parameters[1]): Promise { await this.providerFor(id).shutdown(id, opts) if (!opts.keepHistory) { this.sessionProviders.delete(id) @@ -184,11 +182,15 @@ export class DegradedDaemonPtyProvider implements IPtyProvider { async getForegroundProcess(id: string): Promise { return this.providerFor(id).getForegroundProcess(id) } - inspectProcess(id: string) { - return this.hasPty(id) - ? inspectPtyProviderProcess(this.providerFor(id), id) - : Promise.reject(new Error('terminal_gone')) + inspectProcess(id: string, options?: PtyProcessInspectionOptions) { + const routed = this.hasPty(id) ? this.providerFor(id) : null + return inspectRoutedDaemonProcess(routed, this.current, id, options) + } + async consumeExitReceipt(id: string, incarnationId: string): Promise { + const owner = this.hasPty(id) ? this.providerFor(id) : this.current + await owner.consumeExitReceipt?.(id, incarnationId) } + async confirmForegroundProcess(id: string): Promise { return this.providerFor(id).confirmForegroundProcess?.(id) ?? null } diff --git a/src/main/daemon/degraded-daemon-session-routing.ts b/src/main/daemon/degraded-daemon-session-routing.ts index 7fb5a9a00c8c..f4ec0d0d1608 100644 --- a/src/main/daemon/degraded-daemon-session-routing.ts +++ b/src/main/daemon/degraded-daemon-session-routing.ts @@ -1,6 +1,31 @@ import type { IPtyProvider } from '../providers/types' import type { DaemonPtyAdapter } from './daemon-pty-adapter' import { SessionNotFoundError } from './daemon-errors' +import { + inspectPtyProviderProcess, + type PtyProcessInspection, + type PtyProcessInspectionOptions +} from '../providers/pty-process-inspection' + +/** + * An id no route claims may only be asked about by a caller naming a remembered incarnation: that + * is exactly the session that died while this client was away, so no route survived it. A daemon + * that never held it answers not-found, so routing the question cannot manufacture an exit. An + * ordinary poll must not borrow a route it never owned. Mirrors DaemonPtyRouter. + */ +export function inspectRoutedDaemonProcess( + owner: IPtyProvider | null, + currentDaemon: DaemonPtyAdapter, + sessionId: string, + options?: PtyProcessInspectionOptions +): Promise { + if (owner) { + return inspectPtyProviderProcess(owner, sessionId, options) + } + return options?.expectedIncarnationId === undefined + ? Promise.reject(new Error('terminal_gone')) + : currentDaemon.inspectProcess(sessionId, options) +} export function listProviderSessionIds( sessionProviders: ReadonlyMap, diff --git a/src/main/daemon/session-output-plane.ts b/src/main/daemon/session-output-plane.ts index 507ca01a927e..890a92c07bff 100644 --- a/src/main/daemon/session-output-plane.ts +++ b/src/main/daemon/session-output-plane.ts @@ -269,6 +269,9 @@ export class SessionOutputPlane { * which outlives fd teardown so an already-exited session can still be snapshotted. */ markDisposed(): void { this.disposed = true + // A disposed plane refuses takePendingOutput, so anything still buffered is unreachable. + this.pendingOutputRecords = [] + this.pendingOutputBytes = 0 } disposeEmulator(): void { diff --git a/src/main/daemon/session-termination-controller.ts b/src/main/daemon/session-termination-controller.ts index 814be8228c3c..f0d239115bb6 100644 --- a/src/main/daemon/session-termination-controller.ts +++ b/src/main/daemon/session-termination-controller.ts @@ -20,6 +20,7 @@ export type SessionTerminationControllerDeps = { * force-kill fallback, and the one-shot release of the native PTY handle. */ export class SessionTerminationController { private _isTerminating = false + private _killRequested = false private killTimer: ReturnType | null = null private forceKillSent = false private subprocessDisposed = false @@ -31,6 +32,12 @@ export class SessionTerminationController { return this._isTerminating } + /** The owner asked this session to end. Unlike `isTerminating` it survives the exit, so the + * reaper can tell an owner-ended session from one whose exit the owner may still need. */ + get killRequested(): boolean { + return this._killRequested + } + markPhysicalExit(): void { this.physicalExit.markExited() } @@ -46,6 +53,7 @@ export class SessionTerminationController { return false } this._isTerminating = true + this._killRequested = true // Why: a paused child can be blocked inside write(); resume before any async snapshot so it handles termination promptly. this.deps.releaseProducerPause({ resume: true }) return true @@ -118,6 +126,7 @@ export class SessionTerminationController { } if (!this._isTerminating) { this._isTerminating = true + this._killRequested = true this.deps.releaseProducerPause({ resume: true }) } // Why: escalate a graceful termination now; waiting for the 5s timer would spend most of the physical-exit budget. @@ -147,6 +156,7 @@ export class SessionTerminationController { private resetTerminationAfterSignalFailure(): void { this._isTerminating = false + this._killRequested = false this.cancelForceKillFallback() } diff --git a/src/main/daemon/session.ts b/src/main/daemon/session.ts index 9265b2acfefd..0e604889b036 100644 --- a/src/main/daemon/session.ts +++ b/src/main/daemon/session.ts @@ -4,10 +4,7 @@ import { createSessionOutputPipeline } from './session-output-pipeline' import { SessionProducerPause } from './session-producer-pause' import { SessionShellReadyBarrier } from './session-shell-ready-barrier' import type { TerminalShellRecoveryBarrier } from './terminal-shell-recovery-barrier' -import { - SessionTerminationController, - IMMEDIATE_KILL_PHYSICAL_EXIT_TIMEOUT_MS -} from './session-termination-controller' +import { SessionTerminationController } from './session-termination-controller' import type { SubprocessHandle } from './session-subprocess-handle' import type { JobTerminationOutcome } from '../windows/windows-pty-job' import type { SessionOptions } from './session-options' @@ -124,6 +121,10 @@ export class Session { return this.termination.isTerminating } + get killRequested(): boolean { + return this.termination.killRequested + } + /** Claims termination synchronously so attach/re-entry cannot race async * teardown preparation. Returns false when another owner already claimed it. */ beginTermination(): boolean { @@ -194,9 +195,7 @@ export class Session { this.termination.scheduleForceDisposeFallback() } - async forceKillAndWaitForExit( - timeoutMs = IMMEDIATE_KILL_PHYSICAL_EXIT_TIMEOUT_MS - ): Promise { + async forceKillAndWaitForExit(timeoutMs?: number): Promise { await this.termination.forceKillAndWaitForExit(timeoutMs) } @@ -385,7 +384,7 @@ export class Session { this.output.broadcastExit(code, this.incarnationId, cause) - // Why: hand off to the owner's reaper (disposes emulator, drops session from host map); else dead sessions accumulate. + // Why: hand off to the owner's reaper (disposes the emulator and the subprocess handle); else dead sessions pin their scrollback. this.onSessionExit?.(code) } diff --git a/src/main/daemon/terminal-host-exit-receipt-budget.test.ts b/src/main/daemon/terminal-host-exit-receipt-budget.test.ts new file mode 100644 index 000000000000..527796d0a0ed --- /dev/null +++ b/src/main/daemon/terminal-host-exit-receipt-budget.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, it, vi } from 'vitest' +import type { Session } from './session' +import { + consumeExitReceipt, + MAX_EXIT_RECEIPTS, + reapSessionRecord, + type TerminalHostSessionRecord +} from './terminal-host-session-record' + +function session(id: string, isAlive = false): Session { + return { + incarnationId: id, + isAlive, + killRequested: false, + exitCode: 0, + dispose: vi.fn() + } as unknown as Session +} + +function fillReceipts(records: Map): void { + for (let i = 0; i < MAX_EXIT_RECEIPTS; i++) { + records.set(`exit-${i}`, { incarnationId: `inc-${i}`, code: 0 }) + } +} + +describe('host exit receipt budget', () => { + it('bounds sustained unacknowledged exit churn without evicting live sessions', () => { + const live = session('live', true) + const records = new Map([['live', live]]) + for (let i = 0; i < MAX_EXIT_RECEIPTS * 3; i++) { + records.set(`exit-${i}`, session(`inc-${i}`)) + expect(reapSessionRecord(records, `exit-${i}`)).toBe(true) + expect(records.size).toBeLessThanOrEqual(MAX_EXIT_RECEIPTS + 1) + } + expect(records.get('live')).toBe(live) + expect(live.dispose).not.toHaveBeenCalled() + expect(records.has(`exit-${MAX_EXIT_RECEIPTS * 2 - 1}`)).toBe(false) + expect(records.has(`exit-${MAX_EXIT_RECEIPTS * 2}`)).toBe(true) + expect(records.get(`exit-${MAX_EXIT_RECEIPTS * 3 - 1}`)).toEqual({ + incarnationId: `inc-${MAX_EXIT_RECEIPTS * 3 - 1}`, + code: 0 + }) + }) + + it('orders receipts by exit rather than process creation', () => { + const records = new Map([ + ['long-running', session('old-launch')] + ]) + fillReceipts(records) + reapSessionRecord(records, 'long-running') + expect(records.size).toBe(MAX_EXIT_RECEIPTS) + expect(records.has('exit-0')).toBe(false) + expect(records.get('long-running')).toEqual({ incarnationId: 'old-launch', code: 0 }) + }) + + it('does not count live sessions or acknowledged receipts against the budget', () => { + const records = new Map() + for (let i = 0; i < MAX_EXIT_RECEIPTS + 1; i++) { + records.set(`live-${i}`, session(`live-${i}`, true)) + } + fillReceipts(records) + consumeExitReceipt(records, 'exit-0', 'inc-0') + records.set('new-exit', session('new-incarnation')) + reapSessionRecord(records, 'new-exit') + expect(records.has('exit-1')).toBe(true) + expect(records.size).toBe(MAX_EXIT_RECEIPTS * 2 + 1) + for (let i = 0; i < MAX_EXIT_RECEIPTS + 1; i++) { + expect(records.has(`live-${i}`)).toBe(true) + } + }) + + it('does not evict evidence when reaping an explicitly killed process', () => { + const records = new Map() + fillReceipts(records) + const killed = session('killed') + Object.assign(killed, { killRequested: true }) + records.set('killed', killed) + reapSessionRecord(records, 'killed') + expect(records.size).toBe(MAX_EXIT_RECEIPTS) + expect(records.has('exit-0')).toBe(true) + expect(records.has('killed')).toBe(false) + }) +}) diff --git a/src/main/daemon/terminal-host-options.ts b/src/main/daemon/terminal-host-options.ts index 9bb5ff48c0f7..5112a09bf107 100644 --- a/src/main/daemon/terminal-host-options.ts +++ b/src/main/daemon/terminal-host-options.ts @@ -33,6 +33,4 @@ export type TerminalHostOptions = { snapshot: TerminalSnapshot, records: TakePendingOutputResult['records'] ) => void - // Why: tests need deterministic tombstone eviction without thousands of sessions. - maxTombstones?: number } diff --git a/src/main/daemon/terminal-host-process-inspection.test.ts b/src/main/daemon/terminal-host-process-inspection.test.ts index 50858a89ae16..8d96d4ed25fb 100644 --- a/src/main/daemon/terminal-host-process-inspection.test.ts +++ b/src/main/daemon/terminal-host-process-inspection.test.ts @@ -1,8 +1,11 @@ import { describe, expect, it, vi } from 'vitest' import type { SubprocessHandle } from './session-subprocess-handle' import { TerminalHost } from './terminal-host' +import { SessionNotFoundError } from './types' -function createSubprocess(): SubprocessHandle { +type MockSubprocess = SubprocessHandle & { exit(code: number): void } + +function createSubprocess(): MockSubprocess { let onExit: ((code: number) => void) | null = null return { pid: 99_999, @@ -17,8 +20,20 @@ function createSubprocess(): SubprocessHandle { onExit: (callback) => { onExit = callback }, - dispose: vi.fn() - } + dispose: vi.fn(), + exit: (code: number) => onExit?.(code) + } as MockSubprocess +} + +function createHost(): { host: TerminalHost; lastSubprocess: () => MockSubprocess } { + let last: MockSubprocess | undefined + const host = new TerminalHost({ + spawnSubprocess: () => { + last = createSubprocess() + return last + } + }) + return { host, lastSubprocess: () => last as MockSubprocess } } describe('TerminalHost process inspection', () => { @@ -47,3 +62,252 @@ describe('TerminalHost process inspection', () => { } }) }) + +describe('TerminalHost undelivered exits', () => { + /** Orca quits (the client's attachments drop), the shell ends, Orca reopens and asks. */ + async function exitWhileClientIsAway( + host: TerminalHost, + subprocess: () => MockSubprocess, + sessionId: string, + code = 0 + ): Promise { + const created = await host.createOrAttach({ + sessionId, + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit: vi.fn() } + }) + host.detach(sessionId, created.attachToken as symbol) + subprocess().exit(code) + return created.incarnationId + } + + it('answers during exit broadcast and retains the same proof after reaping', async () => { + const { host, lastSubprocess } = createHost() + let proofDuringBroadcast: ReturnType | undefined + let incarnationId = '' + try { + const created = await host.createOrAttach({ + sessionId: 'broadcast-exit', + cols: 80, + rows: 24, + streamClient: { + onData: vi.fn(), + onExit: () => { + proofDuringBroadcast = host.inspectProcess('broadcast-exit', { + expectedIncarnationId: incarnationId + }) + } + } + }) + incarnationId = created.incarnationId + lastSubprocess().exit(17) + const expected = { + foregroundProcessEvidence: { + verdict: 'exited', + ptyIncarnationId: incarnationId, + reason: 'pty_exit_17' + } + } + await expect(proofDuringBroadcast).resolves.toMatchObject(expected) + await expect( + host.inspectProcess('broadcast-exit', { expectedIncarnationId: incarnationId }) + ).resolves.toMatchObject(expected) + } finally { + await host.dispose() + } + }) + + it('keeps handing a close-and-reopen caller the exit its client never received', async () => { + const { host, lastSubprocess } = createHost() + try { + const incarnationId = await exitWhileClientIsAway(host, lastSubprocess, 'session-away', 3) + + // The exit is gone from every liveness surface, but the evidence is still owed to its owner. + expect(host.listSessions()).toHaveLength(0) + await expect( + host.inspectProcess('session-away', { expectedIncarnationId: incarnationId }) + ).resolves.toMatchObject({ + foregroundProcess: null, + hasChildProcesses: false, + foregroundProcessEvidence: { + ptyId: 'session-away', + ptyIncarnationId: incarnationId, + verdict: 'exited', + reason: 'pty_exit_3' + } + }) + + // Reading is not delivery: a sweep that reads the proof and then fails to persist the + // settlement must find it again next time. + await expect( + host.inspectProcess('session-away', { expectedIncarnationId: incarnationId }) + ).resolves.toMatchObject({ foregroundProcessEvidence: { verdict: 'exited' } }) + } finally { + await host.dispose() + } + }) + + it('answers nothing to a caller with the wrong incarnation or none at all', async () => { + const { host, lastSubprocess } = createHost() + try { + const incarnationId = await exitWhileClientIsAway(host, lastSubprocess, 'session-away') + + expect(() => + host.inspectProcess('session-away', { expectedIncarnationId: 'someone-else' }) + ).toThrow(SessionNotFoundError) + expect(() => host.inspectProcess('session-away')).toThrow(SessionNotFoundError) + + // Neither refusal consumed the evidence the real owner is still owed. + await expect( + host.inspectProcess('session-away', { expectedIncarnationId: incarnationId }) + ).resolves.toMatchObject({ foregroundProcessEvidence: { verdict: 'exited' } }) + } finally { + await host.dispose() + } + }) + + it('still answers an exit that was broadcast to an attached client', async () => { + const { host, lastSubprocess } = createHost() + try { + const onExit = vi.fn() + const created = await host.createOrAttach({ + sessionId: 'session-attached', + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit } + }) + lastSubprocess().exit(0) + + // Broadcast is not the record's lifecycle: whether anyone was listening, the exited session + // stays answerable to a caller naming its exact incarnation. + expect(onExit).toHaveBeenCalledWith(0, created.incarnationId, expect.anything()) + expect(host.listSessions()).toHaveLength(0) + await expect( + host.inspectProcess('session-attached', { expectedIncarnationId: created.incarnationId }) + ).resolves.toMatchObject({ + foregroundProcessEvidence: { + ptyId: 'session-attached', + ptyIncarnationId: created.incarnationId, + verdict: 'exited', + reason: 'pty_exit_0' + } + }) + expect(() => host.inspectProcess('session-attached')).toThrow(SessionNotFoundError) + } finally { + await host.dispose() + } + }) + + it('does not let a session recreated under the same id inherit the old exit', async () => { + const { host, lastSubprocess } = createHost() + try { + const incarnationId = await exitWhileClientIsAway(host, lastSubprocess, 'session-reused') + const replacement = await host.createOrAttach({ + sessionId: 'session-reused', + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit: vi.fn() } + }) + + expect(replacement.incarnationId).not.toBe(incarnationId) + await expect( + host.inspectProcess('session-reused', { expectedIncarnationId: incarnationId }) + ).resolves.toMatchObject({ + foregroundProcessEvidence: { verdict: 'unverifiable', reason: 'incarnation_mismatch' } + }) + } finally { + await host.dispose() + } + }) + + it('drops the record of a session its owner killed, even with no client attached', async () => { + const { host, lastSubprocess } = createHost() + try { + const created = await host.createOrAttach({ + sessionId: 'session-killed', + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit: vi.fn() } + }) + host.detach('session-killed', created.attachToken as symbol) + await host.kill('session-killed') + lastSubprocess().exit(0) + + // The owner asked for this exit; there is nothing left to tell it, so nothing is retained. + expect(() => + host.inspectProcess('session-killed', { expectedIncarnationId: created.incarnationId }) + ).toThrow(SessionNotFoundError) + } finally { + await host.dispose() + } + }) + + it('drops a held exit once its owner kills the session it belonged to', async () => { + const { host, lastSubprocess } = createHost() + try { + const incarnationId = await exitWhileClientIsAway(host, lastSubprocess, 'session-away') + await expect( + host.inspectProcess('session-away', { expectedIncarnationId: incarnationId }) + ).resolves.toMatchObject({ foregroundProcessEvidence: { verdict: 'exited' } }) + + // Nothing killed the process; the owner is telling the host it has acted on the exit. + await expect( + host.kill('session-away', { expectedIncarnationId: incarnationId }) + ).resolves.toBeUndefined() + expect(() => + host.inspectProcess('session-away', { expectedIncarnationId: incarnationId }) + ).toThrow(SessionNotFoundError) + expect(() => host.kill('session-never')).toThrow(SessionNotFoundError) + } finally { + await host.dispose() + } + }) + + it('refuses to end a newer shell when the owner releases an older incarnation', async () => { + const { host, lastSubprocess } = createHost() + try { + const older = await exitWhileClientIsAway(host, lastSubprocess, 'session-reused') + // The pane respawned onto its stable id before the owner settled the old exit. + const replacement = await host.createOrAttach({ + sessionId: 'session-reused', + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit: vi.fn() } + }) + const live = lastSubprocess() + + expect(() => host.kill('session-reused', { expectedIncarnationId: older })).toThrow( + 'PTY incarnation mismatch' + ) + expect(live.kill).not.toHaveBeenCalled() + expect(live.forceKill).not.toHaveBeenCalled() + expect(host.listSessions()).toMatchObject([ + { sessionId: 'session-reused', incarnationId: replacement.incarnationId, isAlive: true } + ]) + } finally { + await host.dispose() + } + }) + + it('keeps a session the owner kills mid-exit out of the retained set', async () => { + const { host } = createHost() + try { + const created = await host.createOrAttach({ + sessionId: 'session-immediate', + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit: vi.fn() } + }) + host.detach('session-immediate', created.attachToken as symbol) + await host.kill('session-immediate', { immediate: true }) + + expect(() => + host.inspectProcess('session-immediate', { expectedIncarnationId: created.incarnationId }) + ).toThrow(SessionNotFoundError) + expect(host.listSessions()).toHaveLength(0) + } finally { + await host.dispose() + } + }) +}) diff --git a/src/main/daemon/terminal-host-process-inspection.ts b/src/main/daemon/terminal-host-process-inspection.ts index 2f9fb1491d90..cc12cee27497 100644 --- a/src/main/daemon/terminal-host-process-inspection.ts +++ b/src/main/daemon/terminal-host-process-inspection.ts @@ -6,6 +6,7 @@ import { getStrictProcessTableSnapshotWithAge } from '../../shared/process-table import { resolveRemoteForegroundEvidence } from '../providers/agent-foreground-process' import { buildPaneProcessFingerprint } from '../providers/posix-pane-foreground-fingerprint' import type { Session } from './session' +import type { ExitedSession } from './terminal-host-session-record' import { clearSteadyStateAnchor, getSteadyStateAnchor, @@ -19,8 +20,6 @@ export type TerminalHostProcessInspection = { foregroundProcessEvidence?: RemoteForegroundEvidence } -type RetiredIncarnation = { incarnationId: string; code: number; expiresAt: number } - /** * Tick tiers for a POSIX pane. `cheap` forks `ps` without `tty=`/`command=` (11-38x cheaper) * and answers from the anchored identity when the pane fingerprint is unchanged; anything it @@ -34,18 +33,14 @@ export async function inspectTerminalHostProcess(args: { expectedIncarnationId?: string /** The caller is a self-correcting poll that only reads the process name, never evidence. */ steadyState?: boolean - retiredIncarnation?: RetiredIncarnation + exitedSession?: ExitedSession authorityGeneration: string nextObservationEpoch: () => number onTier?: (tier: TerminalHostInspectionTier) => void }): Promise { - const { sessionId, session, expectedIncarnationId, retiredIncarnation } = args + const { sessionId, session, expectedIncarnationId, exitedSession } = args if (!session || !session.isAlive) { - if ( - retiredIncarnation && - retiredIncarnation.expiresAt > Date.now() && - expectedIncarnationId === retiredIncarnation.incarnationId - ) { + if (exitedSession) { return { foregroundProcess: null, hasChildProcesses: false, @@ -54,9 +49,9 @@ export async function inspectTerminalHostProcess(args: { observationEpoch: args.nextObservationEpoch(), capturedAgeMs: 0, ptyId: sessionId, - ptyIncarnationId: retiredIncarnation.incarnationId, + ptyIncarnationId: exitedSession.incarnationId, verdict: 'exited', - reason: `pty_exit_${retiredIncarnation.code}` + reason: `pty_exit_${exitedSession.code}` } } } diff --git a/src/main/daemon/terminal-host-session-create.ts b/src/main/daemon/terminal-host-session-create.ts index 8f6833c3d9fe..e4c090221c47 100644 --- a/src/main/daemon/terminal-host-session-create.ts +++ b/src/main/daemon/terminal-host-session-create.ts @@ -4,11 +4,11 @@ import { resolvePtyOwnerBackend } from '../../shared/pty-owner-backend' import { getDaemonSessionResultMetadata } from './daemon-create-or-attach-result' import { normalizePtySize } from './daemon-pty-size' import { Session } from './session' +import { sessionFromRecord, type TerminalHostSessionRecord } from './terminal-host-session-record' import { shellPathSupportsPtyStartupBarrier } from './shell-ready' import type { InternalCreateOrAttachOptions } from './terminal-host-agent-session-claim' import type { CreateOrAttachResult } from './terminal-host-create-contract' import type { TerminalHostOptions } from './terminal-host-options' -import type { TerminalHostTombstones } from './terminal-host-tombstones' import type { TerminalSessionTeardown } from './terminal-session-teardown' import { resolveDaemonSessionScrollbackRows } from './daemon-session-scrollback-window' import { TerminalAttachCanceledError } from './daemon-errors' @@ -17,11 +17,10 @@ import { SessionNotFoundError } from './types' import { resolveWslSessionContext } from './wsl-session-context' type TerminalHostSessionCreateDependencies = { - sessions: Map + sessions: Map /** Re-checks the host's shutdown fence and this request's cancellation after any await. */ assertCreateAllowed: () => void sessionTeardown: TerminalSessionTeardown - killedTombstones: TerminalHostTombstones spawnSubprocess: TerminalHostOptions['spawnSubprocess'] onDeadSessionRemoved: (sessionId: string) => void onSessionCreated: (sessionId: string, generation: string | undefined, isAlive: boolean) => void @@ -34,7 +33,7 @@ export async function createOrAttachTerminalSession( deps: TerminalHostSessionCreateDependencies ): Promise { opts.onSessionResolved?.(opts.sessionId) - let existing = deps.sessions.get(opts.sessionId) + let existing = sessionFromRecord(deps.sessions.get(opts.sessionId)) // Why: descendant capture must finish before attach or recreation, or the // caller could receive a doomed session while teardown owns its process. @@ -52,7 +51,7 @@ export async function createOrAttachTerminalSession( rejectOnAbort(opts.cancelSignal, opts.sessionId) ]) deps.assertCreateAllowed() - existing = deps.sessions.get(opts.sessionId) + existing = sessionFromRecord(deps.sessions.get(opts.sessionId)) // Unkillable child, or a fresh teardown claimed it while we waited: still nobody's to recreate. if (existing?.isAlive && existing.isTerminating) { throw new SessionNotFoundError(opts.sessionId) @@ -91,13 +90,12 @@ export async function createOrAttachTerminalSession( throw new SessionNotFoundError(opts.sessionId) } - if (existing) { - existing.dispose() + if (deps.sessions.has(opts.sessionId)) { + existing?.dispose() deps.sessions.delete(opts.sessionId) deps.onDeadSessionRemoved(opts.sessionId) } - deps.killedTombstones.clearForCreate(opts.sessionId) const size = normalizePtySize(opts.cols, opts.rows) const wslDistro = resolveWslSessionContext(opts)?.distro return await spawnAndPublishSession(opts, deps, { size, wslDistro }) diff --git a/src/main/daemon/terminal-host-session-listing.ts b/src/main/daemon/terminal-host-session-listing.ts index 3ae2d5e845ee..1465a93b4df8 100644 --- a/src/main/daemon/terminal-host-session-listing.ts +++ b/src/main/daemon/terminal-host-session-listing.ts @@ -1,14 +1,15 @@ import type { ClaimedAgentPtyOwnerRegistry } from '../../shared/claimed-agent-pty-owner' -import type { Session } from './session' +import { sessionFromRecord, type TerminalHostSessionRecord } from './terminal-host-session-record' import type { SessionInfo } from './types' export function listLiveTerminalHostSessions( - sessions: ReadonlyMap, + sessions: ReadonlyMap, agentSessionOwners: ClaimedAgentPtyOwnerRegistry ): SessionInfo[] { const result: SessionInfo[] = [] - for (const session of sessions.values()) { - if (!session.isAlive) { + for (const record of sessions.values()) { + const session = sessionFromRecord(record) + if (!session?.isAlive) { continue } const size = session.getAppliedSize() diff --git a/src/main/daemon/terminal-host-session-reaping-leak.test.ts b/src/main/daemon/terminal-host-session-reaping-leak.test.ts index c937607a33a7..0d22f479e215 100644 --- a/src/main/daemon/terminal-host-session-reaping-leak.test.ts +++ b/src/main/daemon/terminal-host-session-reaping-leak.test.ts @@ -1,15 +1,4 @@ -/** - * Memory-leak regression: TerminalHost must reap dead sessions. - * - * SessionIds are minted fresh per pane and never reused, so a `TerminalHost` - * that never removes exited sessions from its `sessions` map leaks one dead - * `Session` and its `@xterm/headless` scrollback grid per terminal for the - * lifetime of the long-lived daemon process. - * - * The fix wires a Session `onExit` hook to `TerminalHost.reapSession`, which - * disposes the emulator and drops the entry from the map. These tests assert the - * emulator is disposed when a subprocess exits (before the fix it never was). - */ +// Disposal alone does not release xterm buffers while the host still owns the Session. import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { TerminalHost } from './terminal-host' import type { SubprocessHandle } from './session-subprocess-handle' @@ -98,25 +87,31 @@ describe('TerminalHost dead-session reaping (leak regression)', () => { // Natural exit. lastSubprocess._onExitCb?.(0) - // The dead session's emulator (its scrollback buffer) is freed and the - // session is gone from the map — not merely skipped by listSessions. + // Live surfaces disappear even though the host still owes the exit evidence. expect(emulatorDispose).toHaveBeenCalledTimes(1) expect(host.listSessions()).toHaveLength(0) }) it('does not retain dead-session emulators across many create/exit cycles', async () => { const CYCLES = 5 + const expectedRecords: [string, { incarnationId: string; code: number }][] = [] for (let i = 0; i < CYCLES; i++) { - await host.createOrAttach({ + const created = await host.createOrAttach({ sessionId: `session-${i}`, cols: 80, rows: 24, streamClient: streamClient() }) lastSubprocess._onExitCb?.(0) + expectedRecords.push([`session-${i}`, { incarnationId: created.incarnationId, code: 0 }]) } - // Every dead session was reaped: one emulator disposed per cycle, none retained. + const records = (host as unknown as { sessions: Map }).sessions + expect([...records]).toEqual(expectedRecords) + expect( + [...records.values()].every((record) => Object.getPrototypeOf(record) === Object.prototype) + ).toBe(true) + // Owning only scalar evidence prevents any emulator/subprocess graph from being retained. expect(emulatorDispose).toHaveBeenCalledTimes(CYCLES) expect(host.listSessions()).toHaveLength(0) }) @@ -142,10 +137,9 @@ describe('TerminalHost dead-session reaping (leak regression)', () => { lastSubprocess._onExitCb?.(137) await killed - // Emulator freed and session dropped from the map (no lingering dead entry). + // Emulator freed and the session gone from every live surface. expect(emulatorDispose).toHaveBeenCalledTimes(1) expect(host.listSessions()).toHaveLength(0) - expect(host.isKilled('session-1')).toBe(true) }) it('retains a graceful-timeout session until the forced child physically exits', async () => { diff --git a/src/main/daemon/terminal-host-session-record.ts b/src/main/daemon/terminal-host-session-record.ts new file mode 100644 index 000000000000..b87c3def1a88 --- /dev/null +++ b/src/main/daemon/terminal-host-session-record.ts @@ -0,0 +1,83 @@ +import type { Session } from './session' + +/** Only the host-observed result survives the operational session. */ +export type ExitedSession = { incarnationId: string; code: number } + +export type TerminalHostSessionRecord = Session | ExitedSession + +// A host retains at most this many unconsumed exits, regardless of client availability. +export const MAX_EXIT_RECEIPTS = 1024 + +function pruneExitReceipts(records: Map): void { + if (records.size <= MAX_EXIT_RECEIPTS) { + return + } + let excess = -MAX_EXIT_RECEIPTS + for (const record of records.values()) { + if ('code' in record) { + excess++ + } + } + for (const [id, record] of records) { + if (excess <= 0) { + break + } + if ('code' in record) { + records.delete(id) + excess-- + } + } +} + +export function consumeExitReceipt( + records: Map, + sessionId: string, + incarnationId: string +): void { + const record = records.get(sessionId) + if (record && 'code' in record && record.incarnationId === incarnationId) { + records.delete(sessionId) + } +} + +export function reapSessionRecord( + records: Map, + sessionId: string +): boolean { + const session = sessionFromRecord(records.get(sessionId)) + if (!session || session.isAlive) { + return false + } + const exit = session.killRequested ? undefined : exitFromRecord(session) + session.dispose() + if (records.get(sessionId) !== session) { + return false + } + if (exit) { + // Reinsert so receipt order follows exit time, not process creation time. + records.delete(sessionId) + records.set(sessionId, exit) + pruneExitReceipts(records) + } else { + records.delete(sessionId) + } + return true +} + +export function exitFromRecord( + record: TerminalHostSessionRecord | undefined +): ExitedSession | undefined { + // Exit broadcast precedes reaping, so a reentrant reader can still see the exited Session. + if (record && 'code' in record) { + return record + } + return record && !record.isAlive && record.exitCode !== null + ? { incarnationId: record.incarnationId, code: record.exitCode } + : undefined +} + +export function sessionFromRecord( + record: TerminalHostSessionRecord | undefined +): Session | undefined { + return record && 'isAlive' in record ? record : undefined +} diff --git a/src/main/daemon/terminal-host-session-shutdown.ts b/src/main/daemon/terminal-host-session-shutdown.ts index 41080ff12494..98f57b7a34a3 100644 --- a/src/main/daemon/terminal-host-session-shutdown.ts +++ b/src/main/daemon/terminal-host-session-shutdown.ts @@ -1,8 +1,8 @@ -import type { Session } from './session' +import { sessionFromRecord, type TerminalHostSessionRecord } from './terminal-host-session-record' import type { TakePendingOutputResult, TerminalSnapshot } from './types' function checkpointTerminalHostSessions( - sessions: ReadonlyMap, + sessions: ReadonlyMap, onFinalCheckpoint?: ( sessionId: string, snapshot: TerminalSnapshot, @@ -12,8 +12,9 @@ function checkpointTerminalHostSessions( if (!onFinalCheckpoint) { return } - for (const [sessionId, session] of sessions) { - if (!session.isAlive) { + for (const [sessionId, record] of sessions) { + const session = sessionFromRecord(record) + if (!session?.isAlive) { continue } const take = session.takePendingOutput(true, { teardownSnapshot: true }) @@ -28,9 +29,15 @@ function checkpointTerminalHostSessions( } } -async function disposeTerminalHostSessions(sessions: Iterable): Promise { +async function disposeTerminalHostSessions( + sessions: Iterable +): Promise { const results = await Promise.allSettled( - [...sessions].map(async (session) => { + [...sessions].map(async (record) => { + const session = sessionFromRecord(record) + if (!session) { + return + } session.detachAllClients() // Why: live children retain native ownership until physical exit, while // exited children must release handles without signalling a recycled pid. @@ -50,7 +57,7 @@ async function disposeTerminalHostSessions(sessions: Iterable): Promise } export async function shutdownTerminalHostSessions( - sessions: Map, + sessions: Map, onFinalCheckpoint?: ( sessionId: string, snapshot: TerminalSnapshot, diff --git a/src/main/daemon/terminal-host-tombstones.ts b/src/main/daemon/terminal-host-tombstones.ts deleted file mode 100644 index 145962db90b0..000000000000 --- a/src/main/daemon/terminal-host-tombstones.ts +++ /dev/null @@ -1,29 +0,0 @@ -export class TerminalHostTombstones { - private readonly killed = new Map() - - constructor(private readonly capacity: number) {} - - clearForCreate(sessionId: string): void { - this.killed.delete(sessionId) - } - - record(sessionId: string): void { - this.killed.delete(sessionId) - this.killed.set(sessionId, Date.now()) - if (this.killed.size <= this.capacity) { - return - } - const oldest = this.killed.keys().next().value - if (oldest) { - this.killed.delete(oldest) - } - } - - has(sessionId: string): boolean { - return this.killed.has(sessionId) - } - - clear(): void { - this.killed.clear() - } -} diff --git a/src/main/daemon/terminal-host.test.ts b/src/main/daemon/terminal-host.test.ts index 142b9b2c5a2a..5e8ebd3cfe29 100644 --- a/src/main/daemon/terminal-host.test.ts +++ b/src/main/daemon/terminal-host.test.ts @@ -388,7 +388,7 @@ describe('TerminalHost', () => { }) describe('kill', () => { - it('kills the session and tombstones it', async () => { + it('kills the session', async () => { await host.createOrAttach({ sessionId: 'session-1', cols: 80, @@ -398,10 +398,9 @@ describe('TerminalHost', () => { host.kill('session-1') expect(lastSubprocess.kill).toHaveBeenCalled() - expect(host.isKilled('session-1')).toBe(true) }) - it('does not tombstone a session when graceful kill admission fails', async () => { + it('keeps the session attachable when graceful kill admission fails', async () => { await host.createOrAttach({ sessionId: 'session-1', cols: 80, @@ -413,7 +412,6 @@ describe('TerminalHost', () => { }) expect(() => host.kill('session-1')).toThrow('signal rejected') - expect(host.isKilled('session-1')).toBe(false) await expect( host.createOrAttach({ sessionId: 'session-1', @@ -516,7 +514,6 @@ describe('TerminalHost', () => { expect.objectContaining({ ownsRoot: expect.any(Function) }) ) expect(lastSubprocess.forceKill).not.toHaveBeenCalled() - expect(host.isKilled('agent-1')).toBe(true) const finish = killWithDescendantSweepMock.mock.calls[0][1] as () => void finish() @@ -721,7 +718,6 @@ describe('TerminalHost', () => { host.signal('session-1', 'SIGINT') expect(lastSubprocess.signal).toHaveBeenCalledWith('SIGINT') - expect(host.isKilled('session-1')).toBe(false) }) }) @@ -787,27 +783,6 @@ describe('TerminalHost', () => { }) }) - describe('tombstones', () => { - it('caps tombstones at limit', async () => { - await host.dispose() - host = new TerminalHost({ spawnSubprocess: spawnFn as MockSpawnFn, maxTombstones: 3 }) - - for (let i = 0; i < 5; i++) { - await host.createOrAttach({ - sessionId: `session-${i}`, - cols: 80, - rows: 24, - streamClient: { onData: vi.fn(), onExit: vi.fn() } - }) - host.kill(`session-${i}`) - } - - // Oldest tombstones should be evicted - expect(host.isKilled('session-0')).toBe(false) - expect(host.isKilled('session-4')).toBe(true) - }) - }) - describe('dispose', () => { it('force-kills live subprocesses and releases PTY fds on dispose', async () => { await host.createOrAttach({ diff --git a/src/main/daemon/terminal-host.ts b/src/main/daemon/terminal-host.ts index 95bedd1a7fde..550a6f0027dc 100644 --- a/src/main/daemon/terminal-host.ts +++ b/src/main/daemon/terminal-host.ts @@ -1,4 +1,11 @@ import type { Session } from './session' +import { + consumeExitReceipt, + exitFromRecord, + reapSessionRecord, + sessionFromRecord, + type TerminalHostSessionRecord +} from './terminal-host-session-record' import { SessionNotFoundError, type SessionInfo, @@ -16,13 +23,11 @@ import { } from './terminal-host-agent-session-claim' import { TerminalHostAgentSessionGenerations } from './terminal-host-agent-session-generations' import { resolveTerminalHostSessionCwd } from './terminal-host-session-cwd' -import { TerminalHostTombstones } from './terminal-host-tombstones' import { listLiveTerminalHostSessions } from './terminal-host-session-listing' import { createOrAttachTerminalSession } from './terminal-host-session-create' import { TerminalAttachCanceledError } from './daemon-errors' import { rejectOnAbort } from './terminal-attach-cancellation' import { randomUUID } from 'node:crypto' -import { pruneRetiredPtyIncarnations } from '../../shared/retired-pty-incarnations' import { inspectTerminalHostProcess, type TerminalHostProcessInspection @@ -41,38 +46,27 @@ export type { CreateOrAttachOptions, CreateOrAttachResult } from './terminal-hos export type { TerminalHostOptions } from './terminal-host-options' -const DEFAULT_MAX_TOMBSTONES = 1000 -const REMOTE_FOREGROUND_TOMBSTONE_RETENTION_MS = 2_000 - export class TerminalHost { - private sessions = new Map() + private sessions = new Map() // Serializes creates for one id across async spawn validation. private pendingCreations = new Map>() private sessionTeardown = new TerminalSessionTeardown(this.sessions) - private killedTombstones: TerminalHostTombstones private spawnSubprocess: TerminalHostOptions['spawnSubprocess'] private onSessionReaped: TerminalHostOptions['onSessionReaped'] private reportReadinessEvent: TerminalHostOptions['reportReadinessEvent'] private onFinalCheckpoint: TerminalHostOptions['onFinalCheckpoint'] - private maxTombstones: number private creationFenced = false private disposePromise: Promise | null = null private readonly agentSessionOwners = new ClaimedAgentPtyOwnerRegistry() private readonly agentSessionGenerations = new TerminalHostAgentSessionGenerations() private readonly authorityGeneration = randomUUID() private observationEpoch = 0 - private readonly retiredIncarnations = new Map< - string, - { incarnationId: string; code: number; expiresAt: number } - >() constructor(opts: TerminalHostOptions) { this.spawnSubprocess = opts.spawnSubprocess this.onSessionReaped = opts.onSessionReaped this.reportReadinessEvent = opts.reportReadinessEvent this.onFinalCheckpoint = opts.onFinalCheckpoint - this.maxTombstones = opts.maxTombstones ?? DEFAULT_MAX_TOMBSTONES - this.killedTombstones = new TerminalHostTombstones(this.maxTombstones) } async createOrAttach(opts: InternalCreateOrAttachOptions): Promise { @@ -104,18 +98,17 @@ export class TerminalHost { isLive: (owner) => this.agentSessionGenerations.isCurrent( owner, - Boolean(this.sessions.get(owner.ptyId)?.isAlive) + Boolean(this.getSession(owner.ptyId)?.isAlive) ), createOrAttach: async (options) => { this.assertCreateOrAttachAllowed(options) - if (options.agentSessionGeneration && this.sessions.get(options.sessionId)?.isAlive) { + if (options.agentSessionGeneration && this.getSession(options.sessionId)?.isAlive) { throw new Error('agent_session_claim_unavailable') } return await createOrAttachTerminalSession(options, { sessions: this.sessions, assertCreateAllowed: () => this.assertCreateOrAttachAllowed(options), sessionTeardown: this.sessionTeardown, - killedTombstones: this.killedTombstones, spawnSubprocess: this.spawnSubprocess, onDeadSessionRemoved: (sessionId) => this.agentSessionGenerations.forget(sessionId), onSessionCreated: (sessionId, generation, isAlive) => @@ -124,15 +117,6 @@ export class TerminalHost { ? { reportReadinessEvent: this.reportReadinessEvent } : {}), onSessionExit: (sessionId, generation) => { - const session = this.sessions.get(sessionId) - if (session) { - pruneRetiredPtyIncarnations(this.retiredIncarnations) - this.retiredIncarnations.set(sessionId, { - incarnationId: session.incarnationId, - code: session.exitCode ?? 0, - expiresAt: Date.now() + REMOTE_FOREGROUND_TOMBSTONE_RETENTION_MS - }) - } this.agentSessionOwners.release(sessionId, generation) this.agentSessionGenerations.forget(sessionId, generation) this.reapSession(sessionId) @@ -169,7 +153,7 @@ export class TerminalHost { // Why null-not-throw (unlike write/resize): pause/resume are best-effort hints against a session that may have exited. pauseProducer(sessionId: string): void { - const session = this.sessions.get(sessionId) + const session = this.getSession(sessionId) if (!session || !session.isAlive) { return } @@ -177,31 +161,48 @@ export class TerminalHost { } resumeProducer(sessionId: string): void { - this.sessions.get(sessionId)?.resumeProducer() + this.getSession(sessionId)?.resumeProducer() } - kill(sessionId: string, opts: { immediate?: boolean } = {}): Promise { + kill( + sessionId: string, + opts: { immediate?: boolean; expectedIncarnationId?: string } = {} + ): Promise { + const record = this.sessions.get(sessionId) + const session = sessionFromRecord(record) + // A stale close must not stop a replacement shell. + if ( + record && + opts.expectedIncarnationId !== undefined && + record.incarnationId !== opts.expectedIncarnationId + ) { + throw new Error(`PTY incarnation mismatch for ${sessionId}`) + } const pending = this.sessionTeardown.get(sessionId) if (pending) { return Promise.resolve( opts.immediate ? this.sessionTeardown.requestImmediate(sessionId) : pending ) } - const session = this.getAliveSession(sessionId) - const killed = this.sessionTeardown.killSession(sessionId, session, opts.immediate === true) - this.killedTombstones.record(sessionId) - return Promise.resolve(killed) + if (record && !session?.isAlive) { + this.sessions.delete(sessionId) + return Promise.resolve() + } + const alive = this.getAliveSession(sessionId) + return Promise.resolve( + this.sessionTeardown.killSession(sessionId, alive, opts.immediate === true) + ) + } + + consumeExitReceipt(sessionId: string, incarnationId: string): void { + consumeExitReceipt(this.sessions, sessionId, incarnationId) } - // Why: dispose a dead session's emulator so exited terminals don't pin their scrollback window for the daemon's life. + // Natural exits retain evidence, never the session's operational object graph. private reapSession(sessionId: string): void { - const session = this.sessions.get(sessionId) - if (!session || session.isAlive) { - return + if (reapSessionRecord(this.sessions, sessionId)) { + this.onSessionReaped?.(sessionId) } - session.dispose() - this.sessions.delete(sessionId) - this.onSessionReaped?.(sessionId) } signal(sessionId: string, sig: string): void { @@ -214,7 +215,7 @@ export class TerminalHost { detachClients(attachments: readonly { sessionId: string; token: symbol }[]): void { for (const { sessionId, token } of attachments) { - this.sessions.get(sessionId)?.detachClient(token) + this.getSession(sessionId)?.detachClient(token) } } @@ -224,7 +225,7 @@ export class TerminalHost { // Why: null-not-throw — fetched for the tab-bar icon, so a vanished pane should quietly yield "no agent". getForegroundProcess(sessionId: string): string | null { - const session = this.sessions.get(sessionId) + const session = this.getSession(sessionId) if (!session || !session.isAlive) { return null } @@ -235,15 +236,12 @@ export class TerminalHost { sessionId: string, options?: { expectedIncarnationId?: string; steadyState?: boolean } ): Promise { - pruneRetiredPtyIncarnations(this.retiredIncarnations) - const session = this.sessions.get(sessionId) - if ( - (!session || !session.isAlive) && - !( - (this.retiredIncarnations.get(sessionId)?.expiresAt ?? 0) > Date.now() && - options?.expectedIncarnationId === this.retiredIncarnations.get(sessionId)?.incarnationId - ) - ) { + const record = this.sessions.get(sessionId) + const session = sessionFromRecord(record) + const observedExit = exitFromRecord(record) + const exitedSession = + observedExit?.incarnationId === options?.expectedIncarnationId ? observedExit : undefined + if (!session?.isAlive && !exitedSession) { // Preserve the historical synchronous missing-session failure. throw new SessionNotFoundError(sessionId) } @@ -254,19 +252,19 @@ export class TerminalHost { ? { expectedIncarnationId: options.expectedIncarnationId } : {}), ...(options?.steadyState === true ? { steadyState: true } : {}), - retiredIncarnation: this.retiredIncarnations.get(sessionId), + ...(exitedSession ? { exitedSession } : {}), authorityGeneration: this.authorityGeneration, nextObservationEpoch: () => ++this.observationEpoch }) } async confirmForegroundProcess(sessionId: string): Promise { - return confirmTerminalHostForegroundProcess(this.sessions.get(sessionId)) + return confirmTerminalHostForegroundProcess(this.getSession(sessionId)) } async confirmShellForeground(sessionId: string): Promise { - return confirmTerminalHostShellForeground(this.sessions.get(sessionId), () => - this.sessions.get(sessionId) + return confirmTerminalHostShellForeground(this.getSession(sessionId), () => + this.getSession(sessionId) ) } @@ -276,24 +274,24 @@ export class TerminalHost { // Why: null-not-throw (unlike getAliveSession) — checkpoint is best-effort against a session that may have just exited. getSnapshot(sessionId: string, opts: { scrollbackRows?: number } = {}): TerminalSnapshot | null { - return getTerminalHostSnapshot(this.sessions.get(sessionId), opts) + return getTerminalHostSnapshot(this.getSession(sessionId), opts) } async getSettledSnapshot( sessionId: string, opts: { scrollbackRows?: number } = {} ): Promise { - return getSettledTerminalHostSnapshot(this.sessions.get(sessionId), opts) + return getSettledTerminalHostSnapshot(this.getSession(sessionId), opts) } // Why: scan-authority handoff seed (null-not-throw like getSnapshot) — emulator's dangling incomplete escape at the stream position. getPartialEscapeTailAnsi(sessionId: string): string { - return getTerminalHostPartialEscapeTail(this.sessions.get(sessionId)) + return getTerminalHostPartialEscapeTail(this.getSession(sessionId)) } // Why: renderer diffs this against xterm to detect a dropped/coerced daemon-side resize; null-not-throw like getSnapshot. getAppliedSize(sessionId: string): { cols: number; rows: number } | null { - return getTerminalHostAppliedSize(this.sessions.get(sessionId)) + return getTerminalHostAppliedSize(this.getSession(sessionId)) } // Why: null-not-throw like getSnapshot — incremental checkpoints are best-effort against a just-exited session. @@ -302,11 +300,7 @@ export class TerminalHost { includeSnapshot: boolean, opts: { teardownSnapshot?: boolean } = {} ): TakePendingOutputResult | null { - return takeTerminalHostPendingOutput(this.sessions.get(sessionId), includeSnapshot, opts) - } - - isKilled(sessionId: string): boolean { - return this.killedTombstones.has(sessionId) + return takeTerminalHostPendingOutput(this.getSession(sessionId), includeSnapshot, opts) } listSessions(): SessionInfo[] { @@ -335,11 +329,14 @@ export class TerminalHost { await Promise.all(this.pendingCreations.values()) } await shutdownTerminalHostSessions(this.sessions, this.onFinalCheckpoint) - this.killedTombstones.clear() + } + + private getSession(sessionId: string): Session | undefined { + return sessionFromRecord(this.sessions.get(sessionId)) } private getAliveSession(sessionId: string): Session { - const session = this.sessions.get(sessionId) + const session = this.getSession(sessionId) if (!session || !session.isAlive) { throw new SessionNotFoundError(sessionId) } diff --git a/src/main/daemon/terminal-session-teardown.ts b/src/main/daemon/terminal-session-teardown.ts index 017f841a5e0c..afe48ebc1f13 100644 --- a/src/main/daemon/terminal-session-teardown.ts +++ b/src/main/daemon/terminal-session-teardown.ts @@ -1,5 +1,6 @@ import { killWithDescendantSweep } from '../pty-descendant-termination' import type { Session } from './session' +import type { TerminalHostSessionRecord } from './terminal-host-session-record' type TeardownOperation = { promise: Promise @@ -14,7 +15,7 @@ type TeardownOperation = { export class TerminalSessionTeardown { private operations = new Map() - constructor(private sessions: ReadonlyMap) {} + constructor(private sessions: ReadonlyMap) {} get(sessionId: string): Promise | undefined { return this.operations.get(sessionId)?.promise diff --git a/src/main/daemon/types.ts b/src/main/daemon/types.ts index bffdc1d2ed59..53c19892156b 100644 --- a/src/main/daemon/types.ts +++ b/src/main/daemon/types.ts @@ -1,3 +1,5 @@ +import type { ConsumeExitReceiptRequest, KillRequest } from './daemon-session-control-protocol' +export type { KillRequest } from './daemon-session-control-protocol' import type { ConfirmForegroundProcessRequest, ConfirmShellForegroundRequest, @@ -161,15 +163,6 @@ export type SetSessionBackgroundRequest = { } } -export type KillRequest = { - id: string - type: 'kill' - payload: { - sessionId: string - immediate?: boolean - } -} - export type SignalRequest = { id: string type: 'signal' @@ -311,6 +304,7 @@ export type DaemonRequest = | ResumePtyRequest | SetSessionBackgroundRequest | KillRequest + | ConsumeExitReceiptRequest | SignalRequest | ListSessionsRequest | ShutdownIfIdleRequest diff --git a/src/main/ipc/pty/runtime/controller.ts b/src/main/ipc/pty/runtime/controller.ts index 633f4c91b4c4..701bf8fc4a85 100644 --- a/src/main/ipc/pty/runtime/controller.ts +++ b/src/main/ipc/pty/runtime/controller.ts @@ -20,6 +20,8 @@ import { hasChildProcessesFromRuntimeController, hasPtyFromRuntimeController, hasRendererSerializerFromRuntimeController, + inspectExitedIncarnationFromRuntimeController, + releaseExitedIncarnationFromRuntimeController, inspectProcessFromRuntimeController, probePtyLivenessFromRuntimeController, resizePtyFromRuntimeController, @@ -63,6 +65,10 @@ export function installPtyRuntimeController(deps: PtyRuntimeControllerDeps): voi stopAndWait: (ptyId, opts) => stopAndWaitPtyFromRuntimeController(deps, ptyId, opts), getForegroundProcess: (ptyId) => getForegroundProcessFromRuntimeController(ptyId), inspectProcess: (ptyId, options) => inspectProcessFromRuntimeController(ptyId, options), + inspectExitedIncarnation: (ptyId, incarnationId) => + inspectExitedIncarnationFromRuntimeController(ptyId, incarnationId), + releaseExitedIncarnation: (ptyId, incarnationId) => + releaseExitedIncarnationFromRuntimeController(ptyId, incarnationId), confirmForegroundProcess: (ptyId) => confirmForegroundProcessFromRuntimeController(ptyId), confirmShellForeground: (ptyId) => confirmShellForegroundFromRuntimeController(ptyId), getCwd: (ptyId) => getCwdFromRuntimeController(ptyId), diff --git a/src/main/ipc/pty/runtime/operations.ts b/src/main/ipc/pty/runtime/operations.ts index 4bcc4f4e6432..d30dc4f7f656 100644 --- a/src/main/ipc/pty/runtime/operations.ts +++ b/src/main/ipc/pty/runtime/operations.ts @@ -5,7 +5,11 @@ import { ptyOwnership } from '../provider/ownership-state' import { ptySizes } from '../delivery/visibility-state' import { rendererSerializerReadiness } from '../pane/serializer-state' import { getProviderForPty, localProvider } from '../provider/registry' -import { inspectPtyProviderProcess } from '../../../providers/pty-process-inspection' +import { + inspectPtyProviderProcess, + providerObservedIncarnationExit +} from '../../../providers/pty-process-inspection' +import type { PtyIncarnationId } from '../../../../shared/pty-incarnation' import type { PtyRuntimeControllerDeps } from './controller-deps' import { agentSessionPtyWriteGate } from '../../../runtime/agent-session-pty-write-gate' import { reportAgentSessionWriteRefusal } from '../agent-session-write-refusal-report' @@ -112,6 +116,44 @@ export async function probePtyLivenessFromRuntimeController( } } +/** + * True only when the execution host reported an observed exit for this exact incarnation. A thrown + * transport, a `terminal_gone`, an unverifiable verdict and a `remote:` id are all doubt, and doubt + * never settles a worker (docs/reference/ssh-execution-boundary.md). + */ +export async function inspectExitedIncarnationFromRuntimeController( + ptyId: string, + incarnationId: PtyIncarnationId +): Promise { + // Why: no locally routed provider can authoritatively answer for a remote host's PTY. + if (ptyId.startsWith('remote:')) { + return false + } + try { + return await providerObservedIncarnationExit(getProviderForPty(ptyId), ptyId, incarnationId) + } catch { + return false + } +} + +/** + * Best-effort acknowledgement after durable recovery. Unsupported providers/hosts retain their + * existing policy; failures leave evidence subject to host retention, not a retry queue. + */ +export async function releaseExitedIncarnationFromRuntimeController( + ptyId: string, + incarnationId: PtyIncarnationId +): Promise { + if (ptyId.startsWith('remote:')) { + return + } + try { + await getProviderForPty(ptyId).consumeExitReceipt?.(ptyId, incarnationId) + } catch { + // Never substitute process shutdown for unsupported or failed evidence consumption. + } +} + export async function attachPtyFromRuntimeController( deps: PtyRuntimeControllerDeps, ptyId: string diff --git a/src/main/ipc/pty/runtime/remembered-incarnation-exit.test.ts b/src/main/ipc/pty/runtime/remembered-incarnation-exit.test.ts new file mode 100644 index 000000000000..35486c5c6da8 --- /dev/null +++ b/src/main/ipc/pty/runtime/remembered-incarnation-exit.test.ts @@ -0,0 +1,139 @@ +// Recovery settles a worker on this answer, so every shape of doubt has to read as `false`: +// a host that says unverifiable, a host that answers about a different incarnation, a thrown +// transport, and an id no locally routed provider owns. +import { afterEach, describe, expect, it, vi } from 'vitest' +import { inspectExitedIncarnationFromRuntimeController } from './operations' +import { sshProviders } from '../provider/registry' +import { ptyOwnership } from '../provider/ownership-state' +import type { PtyIncarnationId } from '../../../../shared/pty-incarnation' + +const CONNECTION = 'exit-evidence-host' +const PTY_ID = `ssh:${CONNECTION}@@pty-1` +const INCARNATION = '45454545-4545-4545-8545-454545454545' as PtyIncarnationId +const OTHER_INCARNATION = '56565656-5656-4656-8656-565656565656' as PtyIncarnationId + +function hostAnswering(inspectProcess: unknown): ReturnType { + const spy = vi.fn(inspectProcess as never) + sshProviders.set(CONNECTION, { inspectProcess: spy } as never) + return spy +} + +const REMOTE_PTY_ID = 'remote:host-1/pty-1' + +afterEach(() => { + sshProviders.delete(CONNECTION) + ptyOwnership.delete(REMOTE_PTY_ID) +}) + +describe('inspectExitedIncarnationFromRuntimeController', () => { + it('accepts an exited verdict the host stamped with the requested incarnation', async () => { + const inspect = hostAnswering(async () => ({ + foregroundProcess: null, + hasChildProcesses: false, + foregroundProcessEvidence: { + authorityGeneration: 'gen-1', + observationEpoch: 1, + capturedAgeMs: 0, + ptyId: 'pty-1', + ptyIncarnationId: INCARNATION, + verdict: 'exited', + reason: 'pty_exit_0' + } + })) + + await expect(inspectExitedIncarnationFromRuntimeController(PTY_ID, INCARNATION)).resolves.toBe( + true + ) + expect(inspect).toHaveBeenCalledWith(PTY_ID, { + expectedIncarnationId: INCARNATION + }) + }) + + it('refuses an exited verdict stamped with a different incarnation', async () => { + hostAnswering(async () => ({ + foregroundProcess: null, + hasChildProcesses: false, + foregroundProcessEvidence: { + authorityGeneration: 'gen-1', + observationEpoch: 1, + capturedAgeMs: 0, + ptyId: 'pty-1', + ptyIncarnationId: OTHER_INCARNATION, + verdict: 'exited', + reason: 'pty_exit_0' + } + })) + + await expect(inspectExitedIncarnationFromRuntimeController(PTY_ID, INCARNATION)).resolves.toBe( + false + ) + }) + + it.each([ + [ + 'unverifiable', + async () => ({ + foregroundProcess: null, + hasChildProcesses: false, + foregroundProcessEvidence: { + authorityGeneration: 'gen-1', + observationEpoch: 1, + capturedAgeMs: 0, + ptyId: 'pty-1', + ptyIncarnationId: INCARNATION, + verdict: 'unverifiable', + reason: 'incarnation_mismatch' + } + }) + ], + ['no evidence at all', async () => ({ foregroundProcess: null, hasChildProcesses: false })], + [ + 'terminal_gone', + async () => { + throw new Error('terminal_gone') + } + ], + [ + 'a lost transport', + async () => { + throw new Error('connection lost') + } + ] + ])('defers when the host answers %s', async (_label, inspectProcess) => { + hostAnswering(inspectProcess) + + await expect(inspectExitedIncarnationFromRuntimeController(PTY_ID, INCARNATION)).resolves.toBe( + false + ) + }) + + it('never asks a locally routed provider about a remote-scoped id', async () => { + // Routed on purpose: without the remote-scope guard this provider WOULD answer, and its + // verdict describes its own host, not the remote runtime that owns this session. + const inspect = hostAnswering(async () => ({ + foregroundProcess: null, + hasChildProcesses: false, + foregroundProcessEvidence: { + authorityGeneration: 'gen-1', + observationEpoch: 1, + capturedAgeMs: 0, + ptyId: REMOTE_PTY_ID, + ptyIncarnationId: INCARNATION, + verdict: 'exited', + reason: 'pty_exit_0' + } + })) + ptyOwnership.set(REMOTE_PTY_ID, CONNECTION) + + await expect( + inspectExitedIncarnationFromRuntimeController(REMOTE_PTY_ID, INCARNATION) + ).resolves.toBe(false) + expect(inspect).not.toHaveBeenCalled() + }) + + it('defers when no provider is attached for the id', async () => { + await expect(inspectExitedIncarnationFromRuntimeController(PTY_ID, INCARNATION)).resolves.toBe( + false + ) + }) +}) diff --git a/src/main/providers/pty-process-inspection.ts b/src/main/providers/pty-process-inspection.ts index 2c316ae05f94..d12a6655874a 100644 --- a/src/main/providers/pty-process-inspection.ts +++ b/src/main/providers/pty-process-inspection.ts @@ -47,6 +47,30 @@ export async function inspectPtyProviderProcess( return { foregroundProcess, hasChildProcesses } } +/** + * Whether the execution host reports that THIS remembered incarnation exited. + * + * Deliberately skips the `hasPty` precheck above: that is this process's cache of what it last + * saw, so it answers `terminal_gone` for exactly the ids a caller asks about here — the ones whose + * death it was disconnected for. Only the host that watched the process end may say `exited`, and + * only for the incarnation the caller stored (docs/reference/ssh-execution-boundary.md). + */ +export async function providerObservedIncarnationExit( + provider: IPtyProvider, + ptyId: string, + incarnationId: PtyIncarnationId +): Promise { + const inspectProcess = (provider as CompletionSensitivePtyProvider).inspectProcess + if (!inspectProcess) { + return false + } + const inspection = await inspectProcess.call(provider, ptyId, { + expectedIncarnationId: incarnationId + }) + const evidence = inspection.foregroundProcessEvidence + return evidence?.verdict === 'exited' && evidence.ptyIncarnationId === incarnationId +} + export async function inspectPtyProviderProcessForRenderer( provider: IPtyProvider, ptyId: string, diff --git a/src/main/providers/pty-provider-contract.ts b/src/main/providers/pty-provider-contract.ts index 9adaa580e6af..5fbf96c26b97 100644 --- a/src/main/providers/pty-provider-contract.ts +++ b/src/main/providers/pty-provider-contract.ts @@ -200,6 +200,9 @@ export type IPtyProvider = { /** Optional host capability used to suppress expensive legacy remote inventory polls. */ supportsForegroundProcessEvidence?(options?: { signal?: AbortSignal }): Promise + /** Best-effort evidence acknowledgement; must never stop a process or retire its resources. */ + consumeExitReceipt?(id: string, incarnationId: PtyIncarnationId): Promise + // Why: deadlineMs (absolute epoch ms) bounds the underlying RPCs so destructive // teardown fails fast inside its sweep budget instead of tripping the outer sweep // deadline; each RPC leaf converts to a relative timeout when it actually issues. diff --git a/src/main/providers/ssh-pty-reattach-absence-discrimination.test.ts b/src/main/providers/ssh-pty-reattach-absence-discrimination.test.ts index bfcc174a4b66..400e10b2bdc1 100644 --- a/src/main/providers/ssh-pty-reattach-absence-discrimination.test.ts +++ b/src/main/providers/ssh-pty-reattach-absence-discrimination.test.ts @@ -25,12 +25,15 @@ import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer' const CONNECTION = 'conn-1' const SESSION = 'pty-1' -function reattachAgainst(attach: () => Promise): Promise { +function reattachAgainst( + attach: (method: string, params: Record) => Promise, + options: Record = {} +): Promise { return reattachSshPtySessionForSpawn({ mux: { request: vi.fn(attach) } as unknown as SshChannelMultiplexer, connectionId: CONNECTION, sessionId: SESSION, - options: { cols: 80, rows: 24 }, + options: { cols: 80, rows: 24, ...options }, exitRaceTracker: { begin: () => 1, didMatchingExitArrive: () => false, @@ -50,6 +53,23 @@ async function refusalFrom(attach: () => Promise): Promise { } describe('an SSH reattach refusal says whether the host observed the PTY', () => { + it('names the incarnation the caller remembers so the relay can mark its refusal', async () => { + // Without it the relay cannot tell this reattach apart from one for an id it never minted, so + // its "not found" stays the ambiguous union and no caller may act on it. + const request = vi.fn(async () => { + throw new Error(`PTY "${SESSION}" not found (${PTY_ATTACH_PROVEN_EXITED_MARKER})`) + }) + const incarnationId = '45454545-4545-4545-8545-454545454545' + + await reattachAgainst(request, { expectedIncarnationId: incarnationId }).catch(() => {}) + + expect(request).toHaveBeenCalledWith( + 'pty.attach', + expect.objectContaining({ id: SESSION, expectedIncarnationId: incarnationId }), + expect.anything() + ) + }) + it('marks a relay that answered "not found" as positive evidence of absence', async () => { const error = await refusalFrom(async () => { throw new Error(`PTY "${SESSION}" not found`) diff --git a/src/main/providers/ssh-pty-session-reattach.ts b/src/main/providers/ssh-pty-session-reattach.ts index f05373a03dc3..53e3bddf0eda 100644 --- a/src/main/providers/ssh-pty-session-reattach.ts +++ b/src/main/providers/ssh-pty-session-reattach.ts @@ -194,6 +194,9 @@ export async function reattachSshPtySession(args: { // Why: expected pane identity prevents a reused relay id from attaching the wrong shell. const expectedPaneKey = args.options.paneKey ?? args.options.env?.ORCA_PANE_KEY const expectedTabId = args.options.tabId ?? args.options.env?.ORCA_TAB_ID + // Additive: lets the relay mark its refusal as a death it observed for THIS incarnation rather + // than the ambiguous "no such id". Omitted when the caller remembers no incarnation. + const expectedIncarnationId = args.options.expectedIncarnationId const attachResult = await requestSshPtyAttach({ mux: args.mux, relayPtyId: relaySessionId, @@ -208,7 +211,8 @@ export async function reattachSshPtySession(args: { // answers "you already have this", and the pane stays blank until new output arrives. requireReplay: true, ...(expectedPaneKey ? { expectedPaneKey } : {}), - ...(expectedTabId ? { expectedTabId } : {}) + ...(expectedTabId ? { expectedTabId } : {}), + ...(expectedIncarnationId ? { expectedIncarnationId } : {}) }, installSourceActivation: args.installSourceActivation, rememberPtyIncarnation: args.rememberPtyIncarnation diff --git a/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts b/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts index 2e1357e34507..58acc755fcbf 100644 --- a/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts +++ b/src/main/runtime/orca-runtime-preserved-branch-cleanup.ts @@ -156,6 +156,14 @@ export class OrcaRuntimeWithPreservedBranchCleanup extends OrcaRuntimeWithTermin connectionId ), runMutation: (worktreeId, operation) => this.runWorktreeTerminalMutation(worktreeId, operation), + proveTerminalExited: async (candidate) => + (await this.ptyController?.inspectExitedIncarnation?.( + candidate.ptyId, + candidate.incarnationId + )) === true, + releaseProvenExit: async (candidate) => { + await this.ptyController?.releaseExitedIncarnation?.(candidate.ptyId, candidate.incarnationId) + }, getActivation: (worktreeId) => this.getLegacyWorkerRecoveryActivation(worktreeId), hasExactPersistedSurface: (candidate) => this.hasExactPersistedTerminalSurfaceIdentity(candidate), diff --git a/src/main/runtime/orca-runtime-test-scenario-builders.spec.ts b/src/main/runtime/orca-runtime-test-scenario-builders.spec.ts index 1f87d901c817..e983422e5e99 100644 --- a/src/main/runtime/orca-runtime-test-scenario-builders.spec.ts +++ b/src/main/runtime/orca-runtime-test-scenario-builders.spec.ts @@ -175,7 +175,9 @@ function makePostRevealWorkerRecoveryHarness( worktreeId: string wslDistro: null }[] - > + >, + /** The execution host's exit answer. Omitted means the host said nothing, which never settles. */ + inspectExitedIncarnation?: (ptyId: string, incarnationId: string) => Promise ): { runtime: RuntimeService getSession: () => WorkspaceSessionState @@ -238,6 +240,7 @@ function makePostRevealWorkerRecoveryHarness( kill, getForegroundProcess: async () => null, hasPty, + ...(inspectExitedIncarnation ? { inspectExitedIncarnation } : {}), listProcesses: listProcesses ?? (async () => [ diff --git a/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-03.spec.ts b/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-03.spec.ts index 00a083108775..f9f1bb731429 100644 --- a/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-03.spec.ts +++ b/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-03.spec.ts @@ -199,7 +199,13 @@ describe('OrcaRuntimeService', () => { .mockResolvedValueOnce([liveProcess]) .mockResolvedValueOnce([liveProcess]) .mockResolvedValueOnce([]) - const harness = makePostRevealWorkerRecoveryHarness(() => false, listProcesses) + // The listing going empty is not the evidence; the host answering for this exact incarnation is. + const harness = makePostRevealWorkerRecoveryHarness( + () => false, + listProcesses, + async (ptyId, incarnationId) => + ptyId === 'pty-post-reveal' && incarnationId === '45454545-4545-4545-8545-454545454545' + ) harness.revealTerminalSession.mockImplementation(() => publishLegacyWorkerReveal(harness.runtime, { worktreeId: TEST_WORKTREE_ID, @@ -326,24 +332,32 @@ describe('OrcaRuntimeService', () => { .fn() .mockResolvedValueOnce([exactProcess]) .mockResolvedValueOnce([replacement]) - const harness = makePostRevealWorkerRecoveryHarness(() => false, listProcesses) + // The host answers for the id, not for this incarnation: a replacement holds it now, and + // nothing here watched the worker's own shell end. + const harness = makePostRevealWorkerRecoveryHarness( + () => false, + listProcesses, + async () => false + ) await expect(harness.runtime.reconcileLegacyWorkerTerminals()).resolves.toMatchObject({ adoptedDispatchIds: [], - exitedDispatchIds: ['dispatch-post-reveal'], - deferredDispatchIds: [] + exitedDispatchIds: [], + deferredDispatchIds: ['dispatch-post-reveal'] }) expect(harness.getSession().tabsByWorktree[TEST_WORKTREE_ID]).toEqual([]) expect( harness.getSession().sleepingAgentSessionsByPaneKey?.[harness.workerPaneKey] - ).toBeUndefined() + ).toBeDefined() + // The surface unbinds from an id a replacement now holds — that is our own tab bookkeeping — + // while the dispatch defers and no exit is ever certified. expect(harness.resolveLegacyWorkerTerminalRecovery).toHaveBeenCalledWith( harness.workerPaneKey, 'rolled_back', harness.ptyId ) - expect(harness.resolveLegacyWorkerTerminalRecovery).toHaveBeenCalledWith( + expect(harness.resolveLegacyWorkerTerminalRecovery).not.toHaveBeenCalledWith( harness.workerPaneKey, 'exited' ) @@ -369,7 +383,11 @@ describe('OrcaRuntimeService', () => { .mockResolvedValueOnce([exactProcess]) .mockResolvedValueOnce([exactProcess]) .mockResolvedValueOnce([replacement]) - const harness = makePostRevealWorkerRecoveryHarness(() => false, listProcesses) + const harness = makePostRevealWorkerRecoveryHarness( + () => false, + listProcesses, + async () => false + ) harness.revealTerminalSession.mockImplementation(() => publishLegacyWorkerReveal(harness.runtime, { worktreeId: TEST_WORKTREE_ID, @@ -379,19 +397,20 @@ describe('OrcaRuntimeService', () => { }) ) + // The pane surface is still rolled back — the id no longer routes to this worker — but the + // dispatch defers: a replacement holding the id is not an observation of this shell's exit. await expect( harness.runtime.reconcileLegacyWorkerTerminals({ materializeRenderer: true }) ).resolves.toMatchObject({ adoptedDispatchIds: [], - exitedDispatchIds: ['dispatch-post-reveal'], - deferredDispatchIds: [] + exitedDispatchIds: [], + deferredDispatchIds: ['dispatch-post-reveal'] }) expect( harness.getSession().sleepingAgentSessionsByPaneKey?.[harness.workerPaneKey] - ).toBeUndefined() - expect(harness.getSession().tabsByWorktree[TEST_WORKTREE_ID]).toEqual([]) - expect(harness.getSession().terminalLayoutsByTabId['legacy-post-reveal']).toBeUndefined() + ).toBeDefined() + // The adopted surface IS retired: leaving it bound points the pane at the replacement shell. const runtimeState = harness.runtime as unknown as { tabs: Map leaves: Map @@ -411,7 +430,7 @@ describe('OrcaRuntimeService', () => { 'rolled_back', harness.ptyId ) - expect(harness.resolveLegacyWorkerTerminalRecovery).toHaveBeenCalledWith( + expect(harness.resolveLegacyWorkerTerminalRecovery).not.toHaveBeenCalledWith( harness.workerPaneKey, 'exited' ) diff --git a/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-04.spec.ts b/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-04.spec.ts index ea70b730388f..48231fbeea10 100644 --- a/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-04.spec.ts +++ b/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-04.spec.ts @@ -17,6 +17,7 @@ import { } from '../orca-runtime-test-fixtures.spec' import { publishLegacyWorkerReveal } from '../orca-runtime-test-scenario-builders.spec' +/** A daemon that outlived the app close, fronted by the router main actually asks. */ describe('OrcaRuntimeService', () => { it('requeues an active Task before clearing recovery for an authoritatively missing worker', async () => { const workerPaneKey = `legacy-missing:${HEADLESS_LEAF_ID}` @@ -82,6 +83,9 @@ describe('OrcaRuntimeService', () => { kill: vi.fn(() => true), getForegroundProcess: async () => null, hasPty: () => false, + // The empty listing settles nothing; this host answer for the stored incarnation does. + inspectExitedIncarnation: async (ptyId, expected) => + ptyId === 'pty-missing-worker' && expected === incarnationId, listProcesses: async () => [] }) const resolveLegacyWorkerTerminalRecovery = vi.fn() @@ -123,6 +127,79 @@ describe('OrcaRuntimeService', () => { } }) + it('leaves an absent worker dispatched when no host answered for its incarnation', async () => { + const workerPaneKey = `legacy-unanswered:${HEADLESS_LEAF_ID}` + const incarnationId = '36363636-3636-4636-8636-363636363636' + const { runtimeStore, getSession } = makeRuntimeStoreWithWorkspaceSession({ + ...getDefaultWorkspaceSession(), + tabsByWorktree: { [TEST_WORKTREE_ID]: [] }, + sleepingAgentSessionsByPaneKey: { + [workerPaneKey]: { + paneKey: workerPaneKey, + tabId: 'legacy-unanswered', + worktreeId: TEST_WORKTREE_ID, + agent: 'codex', + providerSession: { key: 'session_id', id: 'legacy-unanswered-session' }, + prompt: 'continue', + state: 'working', + capturedAt: 1, + updatedAt: 1, + origin: 'live' + } + } + }) + const runtime = new OrcaRuntimeService( + { ...runtimeStore, flushOrThrow: vi.fn() } as never, + undefined, + { canRecoverPersistentLocalPtys: () => true } + ) + const db = new OrchestrationDb(':memory:') + try { + const task = db.createTask({ runId: 'run_legacy_local', spec: 'unanswered worker' }) + const started = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: task.id, + startOptions: { topology: 'current', agent: 'codex' } + }) + db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: 'term_unanswered', + paneKey: workerPaneKey, + processIncarnation: `pty-unanswered:${incarnationId}`, + worktreeId: TEST_WORKTREE_ID, + setupState: 'not_applicable', + effects: [] + }) + db.markWorkerDispatchReady(started.dispatch.id) + runtime.setOrchestrationDb(db) + // The pty is absent from the listing and the host says nothing about this incarnation, which + // is every disconnected host. Absence alone must not retire the worker. + runtime.setPtyController({ + write: vi.fn(() => true), + kill: vi.fn(() => true), + getForegroundProcess: async () => null, + hasPty: () => false, + inspectExitedIncarnation: async () => false, + listProcesses: async () => [] + }) + const resolveLegacyWorkerTerminalRecovery = vi.fn() + runtime.setNotifier({ resolveLegacyWorkerTerminalRecovery } as never) + + await expect(runtime.reconcileLegacyWorkerTerminals()).resolves.toMatchObject({ + adoptedDispatchIds: [], + exitedDispatchIds: [], + deferredDispatchIds: [started.dispatch.id] + }) + expect(db.getWorkerDispatch(started.dispatch.id)?.state).toBe('ready') + expect(db.getDispatchContextById(started.dispatch.id)?.status).toBe('dispatched') + expect(getSession().sleepingAgentSessionsByPaneKey?.[workerPaneKey]).toBeDefined() + expect(resolveLegacyWorkerTerminalRecovery).not.toHaveBeenCalled() + } finally { + db.close() + } + }) + it('waits for durability before retry settles a resolution already present in memory', async () => { const workerPaneKey = `legacy-missing-retry:${HEADLESS_LEAF_ID}` const incarnationId = '34343434-3434-4434-8434-343434343434' @@ -192,6 +269,8 @@ describe('OrcaRuntimeService', () => { kill: vi.fn(() => true), getForegroundProcess: async () => null, hasPty: () => false, + inspectExitedIncarnation: async (ptyId, expected) => + ptyId === 'pty-missing-retry' && expected === incarnationId, listProcesses: async () => [] }) const resolveLegacyWorkerTerminalRecovery = vi.fn() @@ -325,6 +404,8 @@ describe('OrcaRuntimeService', () => { kill: vi.fn(() => true), getForegroundProcess: async () => null, hasPty: () => false, + inspectExitedIncarnation: async (ptyId, expected) => + ptyId === 'pty-exited' && expected === incarnationId, listProcesses }) const revealTerminalSession = vi.fn() diff --git a/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-05.spec.ts b/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-05.spec.ts index 389c70d4f42f..5aaceaf2eec2 100644 --- a/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-05.spec.ts +++ b/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-05.spec.ts @@ -141,23 +141,26 @@ describe('OrcaRuntimeService', () => { listProcesses }) + // Wrong handle and wrong incarnation are LIVE ids now held by a different terminal identity. + // That is a recycled id, not an observation of this worker's death, and this controller offers + // no host answer for the stored incarnation, so all four defer rather than settle. await expect(runtime.reconcileLegacyWorkerTerminals()).resolves.toMatchObject({ adoptedDispatchIds: [], - exitedDispatchIds: ['dispatch-wrong-handle', 'dispatch-wrong-incarnation'], - deferredDispatchIds: ['dispatch-missing', 'dispatch-ambiguous'] + exitedDispatchIds: [], + deferredDispatchIds: [ + 'dispatch-missing', + 'dispatch-ambiguous', + 'dispatch-wrong-handle', + 'dispatch-wrong-incarnation' + ] }) expect(listProcesses).toHaveBeenCalledOnce() expect(listProcesses).toHaveBeenCalledWith(null, LIST_PROVIDER_DEADLINE) - for (const { name, leafId } of cases.slice(0, 2)) { + for (const { name, leafId } of cases) { expect( getSession().sleepingAgentSessionsByPaneKey?.[`legacy-${name}:${leafId}`] ).toBeDefined() } - for (const { name, leafId } of cases.slice(2)) { - expect( - getSession().sleepingAgentSessionsByPaneKey?.[`legacy-${name}:${leafId}`] - ).toBeUndefined() - } }) it('adopts an exact live legacy worker in a folder workspace', async () => { diff --git a/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-08.spec.ts b/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-08.spec.ts new file mode 100644 index 000000000000..da97e290d2c7 --- /dev/null +++ b/src/main/runtime/orca-runtime-tests/terminal-output-and-worker-recovery-part-08.spec.ts @@ -0,0 +1,438 @@ +import { describe, expect, it, vi } from 'vitest' +import { + OrcaRuntimeService, + OrchestrationDb, + getDefaultWorkspaceSession +} from '../orca-runtime-test-mocks.spec' +import { + HEADLESS_LEAF_ID, + TEST_WORKTREE_ID, + makeRuntimeStoreWithWorkspaceSession +} from '../orca-runtime-test-fixtures.spec' +import { TerminalHost } from '../../daemon/terminal-host' +import { DaemonPtyRouter } from '../../daemon/daemon-pty-router' +import type { DaemonPtyAdapter } from '../../daemon/daemon-pty-adapter' +import type { SubprocessHandle } from '../../daemon/session-subprocess-handle' +import { getLocalPtyProvider, setLocalPtyProvider } from '../../ipc/pty/provider/registry' +import { + inspectExitedIncarnationFromRuntimeController, + releaseExitedIncarnationFromRuntimeController +} from '../../ipc/pty/runtime/operations' + +function daemonRouterOver( + host: TerminalHost, + beforeConsumption: () => Promise = async () => {} +): DaemonPtyRouter { + const adapter = { + // Nothing in this process routes the id any more: the app restarted after the shell ended. + hasPty: () => false, + inspectProcess: (id: string, options?: { expectedIncarnationId?: string }) => + host.inspectProcess(id, options), + consumeExitReceipt: async (id: string, incarnationId: string) => { + await beforeConsumption() + host.consumeExitReceipt(id, incarnationId) + }, + listProcesses: async () => [], + onData: () => () => {}, + onExit: () => () => {}, + onWriteUnavailable: () => () => {}, + onBackgroundStreamEvent: () => () => {} + } as unknown as DaemonPtyAdapter + return new DaemonPtyRouter({ current: adapter, legacy: [] }) +} + +function daemonSubprocess(): SubprocessHandle & { exit(code: number): void } { + let onExit: ((code: number) => void) | null = null + return { + pid: 4242, + getForegroundProcess: () => null, + write: vi.fn(), + resize: vi.fn(), + kill: vi.fn(), + terminateOwnedTree: () => 'unavailable', + forceKill: vi.fn(() => onExit?.(137)), + signal: vi.fn(), + onData: vi.fn(), + onExit: (callback: (code: number) => void) => { + onExit = callback + }, + dispose: vi.fn(), + exit: (code: number) => onExit?.(code) + } as unknown as SubprocessHandle & { exit(code: number): void } +} + +describe('OrcaRuntimeService', () => { + it('settles a local worker whose daemon session exited while the app was closed', async () => { + const workerPaneKey = `legacy-daemon-exit:${HEADLESS_LEAF_ID}` + const ptyId = 'pty-daemon-worker' + const subprocess = daemonSubprocess() + const host = new TerminalHost({ spawnSubprocess: () => subprocess }) + // Orca quits (the client's attachment drops), then the worker's shell ends. + const created = await host.createOrAttach({ + sessionId: ptyId, + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit: vi.fn() } + }) + host.detach(ptyId, created.attachToken as symbol) + subprocess.exit(0) + + const { runtimeStore, getSession } = makeRuntimeStoreWithWorkspaceSession({ + ...getDefaultWorkspaceSession(), + tabsByWorktree: { [TEST_WORKTREE_ID]: [] }, + sleepingAgentSessionsByPaneKey: { + [workerPaneKey]: { + paneKey: workerPaneKey, + tabId: 'legacy-daemon-exit', + worktreeId: TEST_WORKTREE_ID, + agent: 'codex', + providerSession: { key: 'session_id', id: 'legacy-daemon-exit-session' }, + prompt: 'continue', + state: 'working', + capturedAt: 1, + updatedAt: 1, + origin: 'live' + } + } + }) + const runtime = new OrcaRuntimeService( + { ...runtimeStore, flushOrThrow: vi.fn() } as never, + undefined, + { canRecoverPersistentLocalPtys: () => true } + ) + const db = new OrchestrationDb(':memory:') + const previousProvider = getLocalPtyProvider() + setLocalPtyProvider(daemonRouterOver(host)) + try { + const task = db.createTask({ runId: 'run_legacy_local', spec: 'daemon worker' }) + const started = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: task.id, + startOptions: { topology: 'current', agent: 'codex' } + }) + db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: 'term_daemon_exit', + paneKey: workerPaneKey, + processIncarnation: `${ptyId}:${created.incarnationId}`, + worktreeId: TEST_WORKTREE_ID, + setupState: 'not_applicable', + effects: [] + }) + db.markWorkerDispatchReady(started.dispatch.id) + runtime.setOrchestrationDb(db) + // No stub: this is the shipped proof chain, from the recovery port through + // providerObservedIncarnationExit and the daemon router to the host that watched it die. + runtime.setPtyController({ + write: vi.fn(() => true), + kill: vi.fn(() => true), + getForegroundProcess: async () => null, + hasPty: () => false, + inspectExitedIncarnation: (candidatePtyId, incarnationId) => + inspectExitedIncarnationFromRuntimeController(candidatePtyId, incarnationId), + releaseExitedIncarnation: releaseExitedIncarnationFromRuntimeController, + listProcesses: async () => [] + }) + runtime.setNotifier({ resolveLegacyWorkerTerminalRecovery: vi.fn() } as never) + + await expect(runtime.reconcileLegacyWorkerTerminals()).resolves.toMatchObject({ + adoptedDispatchIds: [], + exitedDispatchIds: [started.dispatch.id], + deferredDispatchIds: [] + }) + expect(db.getDispatchContextById(started.dispatch.id)?.status).not.toBe('dispatched') + expect(getSession().sleepingAgentSessionsByPaneKey?.[workerPaneKey]).toBeUndefined() + // Settled durably, so the owner told the host and the host holds nothing for this process. + expect(() => + host.inspectProcess(ptyId, { expectedIncarnationId: created.incarnationId }) + ).toThrow() + } finally { + setLocalPtyProvider(previousProvider) + await host.dispose() + db.close() + } + }) + + it('keeps the daemon exit when the orchestration write failed after the surface persisted', async () => { + const workerPaneKey = `legacy-daemon-dbfail:${HEADLESS_LEAF_ID}` + const ptyId = 'pty-daemon-dbfail' + const subprocess = daemonSubprocess() + const host = new TerminalHost({ spawnSubprocess: () => subprocess }) + const created = await host.createOrAttach({ + sessionId: ptyId, + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit: vi.fn() } + }) + host.detach(ptyId, created.attachToken as symbol) + subprocess.exit(0) + + const { runtimeStore } = makeRuntimeStoreWithWorkspaceSession({ + ...getDefaultWorkspaceSession(), + tabsByWorktree: { [TEST_WORKTREE_ID]: [] }, + sleepingAgentSessionsByPaneKey: { + [workerPaneKey]: { + paneKey: workerPaneKey, + tabId: 'legacy-daemon-dbfail', + worktreeId: TEST_WORKTREE_ID, + agent: 'codex', + providerSession: { key: 'session_id', id: 'legacy-daemon-dbfail-session' }, + prompt: 'continue', + state: 'working', + capturedAt: 1, + updatedAt: 1, + origin: 'live' + } + } + }) + const runtime = new OrcaRuntimeService( + { ...runtimeStore, flushOrThrow: vi.fn() } as never, + undefined, + { canRecoverPersistentLocalPtys: () => true } + ) + const db = new OrchestrationDb(':memory:') + const previousProvider = getLocalPtyProvider() + setLocalPtyProvider(daemonRouterOver(host)) + try { + const task = db.createTask({ runId: 'run_legacy_local', spec: 'daemon worker' }) + const started = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: task.id, + startOptions: { topology: 'current', agent: 'codex' } + }) + db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: 'term_daemon_dbfail', + paneKey: workerPaneKey, + processIncarnation: `${ptyId}:${created.incarnationId}`, + worktreeId: TEST_WORKTREE_ID, + setupState: 'not_applicable', + effects: [] + }) + db.markWorkerDispatchReady(started.dispatch.id) + // The surface persists, then the orchestration write is refused (a locked database). + vi.spyOn(db, 'reconcileMissingWorkerTerminal').mockImplementationOnce(() => { + throw new Error('database is locked') + }) + runtime.setOrchestrationDb(db) + runtime.setPtyController({ + write: vi.fn(() => true), + kill: vi.fn(() => true), + getForegroundProcess: async () => null, + hasPty: () => false, + inspectExitedIncarnation: (candidatePtyId, incarnationId) => + inspectExitedIncarnationFromRuntimeController(candidatePtyId, incarnationId), + releaseExitedIncarnation: releaseExitedIncarnationFromRuntimeController, + listProcesses: async () => [] + }) + runtime.setNotifier({ resolveLegacyWorkerTerminalRecovery: vi.fn() } as never) + + await expect(runtime.reconcileLegacyWorkerTerminals()).resolves.toMatchObject({ + adoptedDispatchIds: [], + exitedDispatchIds: [], + deferredDispatchIds: [started.dispatch.id] + }) + expect(db.getDispatchContextById(started.dispatch.id)?.status).toBe('dispatched') + // Release must follow the durable write, or this retry has nothing to read. + await expect( + host.inspectProcess(ptyId, { expectedIncarnationId: created.incarnationId }) + ).resolves.toMatchObject({ foregroundProcessEvidence: { verdict: 'exited' } }) + } finally { + setLocalPtyProvider(previousProvider) + await host.dispose() + db.close() + } + }) + + it('keeps the daemon exit when the settlement did not persist, so the next sweep can retry', async () => { + const workerPaneKey = `legacy-daemon-unsettled:${HEADLESS_LEAF_ID}` + const ptyId = 'pty-daemon-unsettled' + const subprocess = daemonSubprocess() + const host = new TerminalHost({ spawnSubprocess: () => subprocess }) + const created = await host.createOrAttach({ + sessionId: ptyId, + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit: vi.fn() } + }) + host.detach(ptyId, created.attachToken as symbol) + subprocess.exit(0) + + const { runtimeStore } = makeRuntimeStoreWithWorkspaceSession({ + ...getDefaultWorkspaceSession(), + tabsByWorktree: { [TEST_WORKTREE_ID]: [] }, + sleepingAgentSessionsByPaneKey: { + [workerPaneKey]: { + paneKey: workerPaneKey, + tabId: 'legacy-daemon-unsettled', + worktreeId: TEST_WORKTREE_ID, + agent: 'codex', + providerSession: { key: 'session_id', id: 'legacy-daemon-unsettled-session' }, + prompt: 'continue', + state: 'working', + capturedAt: 1, + updatedAt: 1, + origin: 'live' + } + } + }) + // The workspace-session flush fails, so persist reports nothing settled. + const runtime = new OrcaRuntimeService( + { + ...runtimeStore, + flushPendingOrThrowAsync: vi.fn(() => Promise.reject(new Error('disk full'))) + } as never, + undefined, + { canRecoverPersistentLocalPtys: () => true } + ) + const db = new OrchestrationDb(':memory:') + const previousProvider = getLocalPtyProvider() + setLocalPtyProvider(daemonRouterOver(host)) + try { + const task = db.createTask({ runId: 'run_legacy_local', spec: 'daemon worker' }) + const started = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: task.id, + startOptions: { topology: 'current', agent: 'codex' } + }) + db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: 'term_daemon_unsettled', + paneKey: workerPaneKey, + processIncarnation: `${ptyId}:${created.incarnationId}`, + worktreeId: TEST_WORKTREE_ID, + setupState: 'not_applicable', + effects: [] + }) + db.markWorkerDispatchReady(started.dispatch.id) + runtime.setOrchestrationDb(db) + runtime.setPtyController({ + write: vi.fn(() => true), + kill: vi.fn(() => true), + getForegroundProcess: async () => null, + hasPty: () => false, + inspectExitedIncarnation: (candidatePtyId, incarnationId) => + inspectExitedIncarnationFromRuntimeController(candidatePtyId, incarnationId), + releaseExitedIncarnation: releaseExitedIncarnationFromRuntimeController, + listProcesses: async () => [] + }) + runtime.setNotifier({ resolveLegacyWorkerTerminalRecovery: vi.fn() } as never) + + await expect(runtime.reconcileLegacyWorkerTerminals()).resolves.toMatchObject({ + adoptedDispatchIds: [], + exitedDispatchIds: [], + deferredDispatchIds: [started.dispatch.id] + }) + expect(db.getDispatchContextById(started.dispatch.id)?.status).toBe('dispatched') + // Reading the proof was not acting on it: the host still holds it for the next sweep. + await expect( + host.inspectProcess(ptyId, { expectedIncarnationId: created.incarnationId }) + ).resolves.toMatchObject({ foregroundProcessEvidence: { verdict: 'exited' } }) + } finally { + setLocalPtyProvider(previousProvider) + await host.dispose() + db.close() + } + }) + + it('never ends a shell that took the pane id between the proof and the release', async () => { + const workerPaneKey = `legacy-daemon-respawn:${HEADLESS_LEAF_ID}` + const ptyId = 'pty-daemon-respawn' + let subprocess = daemonSubprocess() + const host = new TerminalHost({ spawnSubprocess: () => subprocess }) + const created = await host.createOrAttach({ + sessionId: ptyId, + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit: vi.fn() } + }) + host.detach(ptyId, created.attachToken as symbol) + subprocess.exit(0) + + const { runtimeStore } = makeRuntimeStoreWithWorkspaceSession({ + ...getDefaultWorkspaceSession(), + tabsByWorktree: { [TEST_WORKTREE_ID]: [] }, + sleepingAgentSessionsByPaneKey: { + [workerPaneKey]: { + paneKey: workerPaneKey, + tabId: 'legacy-daemon-respawn', + worktreeId: TEST_WORKTREE_ID, + agent: 'codex', + providerSession: { key: 'session_id', id: 'legacy-daemon-respawn-session' }, + prompt: 'continue', + state: 'working', + capturedAt: 1, + updatedAt: 1, + origin: 'live' + } + } + }) + const runtime = new OrcaRuntimeService( + { ...runtimeStore, flushOrThrow: vi.fn() } as never, + undefined, + { canRecoverPersistentLocalPtys: () => true } + ) + const db = new OrchestrationDb(':memory:') + const previousProvider = getLocalPtyProvider() + // Pane restore raced the sweep: its respawn onto the same stable id reached the host first. + let replacement: typeof subprocess | undefined + setLocalPtyProvider( + daemonRouterOver(host, async () => { + subprocess = daemonSubprocess() + replacement = subprocess + await host.createOrAttach({ + sessionId: ptyId, + cols: 80, + rows: 24, + streamClient: { onData: vi.fn(), onExit: vi.fn() } + }) + }) + ) + try { + const task = db.createTask({ runId: 'run_legacy_local', spec: 'daemon worker' }) + const started = db.createStartingWorkerDispatch({ + creator: { kind: 'system' }, + maxDepth: Number.MAX_SAFE_INTEGER, + taskId: task.id, + startOptions: { topology: 'current', agent: 'codex' } + }) + db.prepareStartingWorkerAuthority({ + dispatchId: started.dispatch.id, + handle: 'term_daemon_respawn', + paneKey: workerPaneKey, + processIncarnation: `${ptyId}:${created.incarnationId}`, + worktreeId: TEST_WORKTREE_ID, + setupState: 'not_applicable', + effects: [] + }) + db.markWorkerDispatchReady(started.dispatch.id) + runtime.setOrchestrationDb(db) + runtime.setPtyController({ + write: vi.fn(() => true), + kill: vi.fn(() => true), + getForegroundProcess: async () => null, + hasPty: () => false, + inspectExitedIncarnation: (candidatePtyId, incarnationId) => + inspectExitedIncarnationFromRuntimeController(candidatePtyId, incarnationId), + releaseExitedIncarnation: releaseExitedIncarnationFromRuntimeController, + listProcesses: async () => [] + }) + runtime.setNotifier({ resolveLegacyWorkerTerminalRecovery: vi.fn() } as never) + + await expect(runtime.reconcileLegacyWorkerTerminals()).resolves.toMatchObject({ + exitedDispatchIds: [started.dispatch.id] + }) + // The old exit settled; the shell the user just got back is untouched. + expect(replacement?.kill).not.toHaveBeenCalled() + expect(replacement?.forceKill).not.toHaveBeenCalled() + expect(host.listSessions()).toMatchObject([{ sessionId: ptyId, isAlive: true }]) + } finally { + setLocalPtyProvider(previousProvider) + await host.dispose() + db.close() + } + }) +}) diff --git a/src/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts index 829c2ca321e1..18c485a400ab 100644 --- a/src/main/runtime/orca-runtime.test.ts +++ b/src/main/runtime/orca-runtime.test.ts @@ -57,6 +57,7 @@ await import('./orca-runtime-tests/terminal-output-and-worker-recovery-part-04.s await import('./orca-runtime-tests/terminal-output-and-worker-recovery-part-05.spec') await import('./orca-runtime-tests/terminal-output-and-worker-recovery-part-06.spec') await import('./orca-runtime-tests/terminal-output-and-worker-recovery-part-07.spec') +await import('./orca-runtime-tests/terminal-output-and-worker-recovery-part-08.spec') await import('./orca-runtime-tests/terminal-handles-and-agent-status.spec') await import('./orca-runtime-tests/terminal-handles-and-agent-status-part-02.spec') await import('./orca-runtime-tests/terminal-settled-prompt-foreground-confirmation.spec') diff --git a/src/main/runtime/runtime-legacy-worker-terminal-recovery-candidate.ts b/src/main/runtime/runtime-legacy-worker-terminal-recovery-candidate.ts index c25ca8deab69..989e7088c8e8 100644 --- a/src/main/runtime/runtime-legacy-worker-terminal-recovery-candidate.ts +++ b/src/main/runtime/runtime-legacy-worker-terminal-recovery-candidate.ts @@ -8,6 +8,20 @@ import type { LegacyWorkerRecoveryWorkspace } from './runtime-legacy-worker-terminal-recovery-types' +/** + * A listing that does not name the PTY is not an observation of its exit, and an id now held by a + * different terminal identity is a RECYCLED id, not a death. Both are this client's bookkeeping; + * only the execution host that watched the process end may certify it, and only for the incarnation + * this dispatch stored. Everything else defers to the next sweep + * (docs/reference/ssh-execution-boundary.md). + */ +async function settleAbsentCandidate( + ports: LegacyWorkerRecoveryPorts, + candidate: LegacyWorkerRecoveryCandidate +): Promise<'exited' | 'unverifiable'> { + return (await ports.proveTerminalExited(candidate)) ? 'exited' : 'unverifiable' +} + export async function reconcileLegacyWorkerCandidate(args: { controller: RuntimeLegacyWorkerTerminalRecoveryController ports: LegacyWorkerRecoveryPorts @@ -21,19 +35,20 @@ export async function reconcileLegacyWorkerCandidate(args: { }): Promise { const { controller, ports, options, candidate, workspace, resolvedWorktrees } = args if (!args.inventory.livePtyIds.has(candidate.ptyId)) { - args.pendingResolutions.push({ candidate, resolution: 'exited' }) + if ((await settleAbsentCandidate(ports, candidate)) === 'exited') { + args.pendingResolutions.push({ candidate, resolution: 'exited' }) + } else { + args.deferredDispatchIds.add(candidate.dispatchId) + } return } const controllerIdentity = args.inventory.terminalIdentityByPtyId.get(candidate.ptyId) - if (!controllerIdentity) { - args.deferredDispatchIds.add(candidate.dispatchId) - return - } if ( + !controllerIdentity || controllerIdentity.handle !== candidate.terminalHandle || controllerIdentity.incarnationId !== candidate.incarnationId ) { - args.pendingResolutions.push({ candidate, resolution: 'exited' }) + args.deferredDispatchIds.add(candidate.dispatchId) return } let adoptionStatus: 'ready' | 'unverifiable' | 'exited' @@ -47,7 +62,7 @@ export async function reconcileLegacyWorkerCandidate(args: { return 'unverifiable' } if (!preAdoptionInventory.livePtyIds.has(candidate.ptyId)) { - return 'exited' + return await settleAbsentCandidate(ports, candidate) } const preAdoptionIdentity = preAdoptionInventory.terminalIdentityByPtyId.get(candidate.ptyId) if (!preAdoptionIdentity) { @@ -57,7 +72,11 @@ export async function reconcileLegacyWorkerCandidate(args: { preAdoptionIdentity.handle !== candidate.terminalHandle || preAdoptionIdentity.incarnationId !== candidate.incarnationId ) { - return 'exited' + // The pane's binding to this id is our own tab bookkeeping, and the id now routes to a + // different shell, so retire the surface. The dispatch still defers: a recycled id is not + // a death. + ports.rollback(candidate) + return 'unverifiable' } const exactSurfaceAlreadyPublished = ports.hasExactPersistedSurface(candidate) && ports.hasExactSurface(candidate) @@ -133,8 +152,12 @@ export async function reconcileLegacyWorkerCandidate(args: { } if (!finalInventory.livePtyIds.has(candidate.ptyId)) { controller.deleteReceipt(candidate.paneKey) - ports.onPtyExit(candidate) - args.pendingResolutions.push({ candidate, resolution: 'exited' }) + if ((await settleAbsentCandidate(ports, candidate)) === 'exited') { + ports.onPtyExit(candidate) + args.pendingResolutions.push({ candidate, resolution: 'exited' }) + } else { + args.deferredDispatchIds.add(candidate.dispatchId) + } return } const finalIdentity = finalInventory.terminalIdentityByPtyId.get(candidate.ptyId) @@ -148,7 +171,9 @@ export async function reconcileLegacyWorkerCandidate(args: { finalIdentity.incarnationId !== candidate.incarnationId ) { controller.deleteReceipt(candidate.paneKey) - args.pendingResolutions.push({ candidate, resolution: 'exited' }) + // Same as the pre-adoption mismatch: unbind the surface, but certify nothing about the process. + ports.rollback(candidate) + args.deferredDispatchIds.add(candidate.dispatchId) return } args.pendingResolutions.push({ candidate, resolution: 'adopted' }) diff --git a/src/main/runtime/runtime-legacy-worker-terminal-recovery-runner.ts b/src/main/runtime/runtime-legacy-worker-terminal-recovery-runner.ts index 6bbe3b2ed2f4..89b565301dce 100644 --- a/src/main/runtime/runtime-legacy-worker-terminal-recovery-runner.ts +++ b/src/main/runtime/runtime-legacy-worker-terminal-recovery-runner.ts @@ -102,6 +102,9 @@ export async function runLegacyWorkerTerminalRecovery( } ports.notifyResolution(candidate, 'exited') exitedDispatchIds.push(candidate.dispatchId) + // Why after reconcileMissing: the host keeps the exit until its owner has acted on it, and a + // settlement that failed to persist must find the same evidence on the next sweep. + await ports.releaseProvenExit(candidate) } const result = { adoptedDispatchIds, diff --git a/src/main/runtime/runtime-legacy-worker-terminal-recovery-types.ts b/src/main/runtime/runtime-legacy-worker-terminal-recovery-types.ts index 16ca330647c0..629bd782debd 100644 --- a/src/main/runtime/runtime-legacy-worker-terminal-recovery-types.ts +++ b/src/main/runtime/runtime-legacy-worker-terminal-recovery-types.ts @@ -48,6 +48,11 @@ export type LegacyWorkerRecoveryPorts = { ) => Promise /** Serializes the pre-adoption liveness probe and the adoption itself against other terminal mutations. */ runMutation: (worktreeId: string, operation: () => Promise) => Promise + /** True only when the execution host observed THIS candidate's stored incarnation exit. Every + * other answer — absence, mismatch, transport failure — is doubt and must defer. */ + proveTerminalExited: (candidate: LegacyWorkerRecoveryCandidate) => Promise + /** Tells the execution host the proven exit has been settled durably, so it can drop the record. */ + releaseProvenExit: (candidate: LegacyWorkerRecoveryCandidate) => Promise getActivation: (worktreeId: string) => { activeTabId?: string; activeGroupId?: string } hasExactPersistedSurface: (candidate: LegacyWorkerRecoveryCandidate) => boolean hasExactSurface: (candidate: LegacyWorkerRecoveryCandidate) => boolean diff --git a/src/main/runtime/runtime-pty-controller-contract.ts b/src/main/runtime/runtime-pty-controller-contract.ts index 73a75af017ed..ea370b776174 100644 --- a/src/main/runtime/runtime-pty-controller-contract.ts +++ b/src/main/runtime/runtime-pty-controller-contract.ts @@ -113,6 +113,12 @@ export type RuntimePtyController = { ptyId: string, options?: { expectedIncarnationId?: PtyIncarnationId; scanChildProcesses?: boolean } ): Promise + /** True only when the execution host observed THIS incarnation exit. Unlike `inspectProcess` it + * is not gated on `hasPty`, so it can answer for an id this process no longer has cached. */ + inspectExitedIncarnation?(ptyId: string, incarnationId: PtyIncarnationId): Promise + /** The owner has durably settled a proven exit of THIS incarnation; the host may drop its record + * and must refuse if the id now carries a newer one. Never throws. */ + releaseExitedIncarnation?(ptyId: string, incarnationId: PtyIncarnationId): Promise confirmForegroundProcess?(ptyId: string): Promise confirmShellForeground?(ptyId: string): Promise hasChildProcesses?(ptyId: string): Promise diff --git a/src/relay/pty-handler-observed-exit-evidence.test.ts b/src/relay/pty-handler-observed-exit-evidence.test.ts new file mode 100644 index 000000000000..388bfbbd404f --- /dev/null +++ b/src/relay/pty-handler-observed-exit-evidence.test.ts @@ -0,0 +1,296 @@ +// The relay answers `exited` from one place: the exit it observed and its owner has not received +// yet. There is no second copy and no expiry, so a laptop that reconnects an hour later still gets +// the verdict, and nothing that merely tore a record down can produce one. +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest' +import * as ptyShellUtils from './pty-shell-utils' +import { + PTY_ATTACH_PROVEN_EXITED_MARKER, + isProvenExitedPtyAttachRefusal +} from '../shared/pty-attach-absence-evidence' + +const { mockPtySpawn, mockPtyInstance, mockCreateShellPromptReadinessProbe } = vi.hoisted(() => ({ + mockPtySpawn: vi.fn(), + mockCreateShellPromptReadinessProbe: vi.fn(), + mockPtyInstance: { + pid: process.pid, + process: 'zsh', + onData: vi.fn(), + onExit: vi.fn(), + write: vi.fn(), + resize: vi.fn(), + kill: vi.fn(), + clear: vi.fn(), + pause: vi.fn(), + resume: vi.fn() + } +})) + +vi.mock('node-pty', () => ({ spawn: mockPtySpawn })) + +vi.mock('../main/pty/posix-pty-process-groups', () => ({ + forceKillPosixPtyProcessGroups: vi.fn((_pid: number, fallback: () => void) => fallback()) +})) + +vi.mock('../main/shell-prompt-readiness-probe', () => ({ + createShellPromptReadinessProbe: mockCreateShellPromptReadinessProbe +})) + +import type { PtyHandler } from './pty-handler' +import { + beginPtyHandlerTest, + createPtyRequestHelpers, + createTestPtyHandler, + endPtyHandlerTest, + testPtyId, + type MockDispatcher +} from './pty-handler-test-harness' + +const PTY_1 = testPtyId(1) + +type ExitEvidence = { + foregroundProcessEvidence?: { + verdict: string + reason?: string + ptyIncarnationId?: string + } +} + +describe('relay exit evidence', () => { + let dispatcher: MockDispatcher + let handler: PtyHandler + let originalPlatform: PropertyDescriptor | undefined + let exitCallback: ((info: { exitCode: number }) => void) | undefined + + const { spawnPty, attachPty } = createPtyRequestHelpers(() => dispatcher) + + beforeEach(() => { + ;({ dispatcher, handler, originalPlatform } = beginPtyHandlerTest({ + mockPtySpawn, + mockPtyInstance, + mockCreateShellPromptReadinessProbe + })) + exitCallback = undefined + mockPtySpawn.mockImplementation(() => ({ + ...mockPtyInstance, + onData: vi.fn(), + onExit: vi.fn((callback: (info: { exitCode: number }) => void) => { + exitCallback = callback + }) + })) + }) + + afterEach(async () => { + await endPtyHandlerTest(handler, originalPlatform) + }) + + /** Rebuild the handler with a legacy writer that refuses, i.e. the owning client is away. */ + function withAbsentOwner(): { capacity: () => void } { + let capacityListener: (() => void) | undefined + let accepting = false + Object.assign(dispatcher, { + onLegacyPtyCapacity: vi.fn((listener: () => void) => { + capacityListener = listener + return vi.fn() + }), + tryNotifyPtyData: vi.fn(() => true), + tryNotifyPtyExit: vi.fn(() => accepting), + legacyRetentionBelowLowWater: true + }) + handler = createTestPtyHandler(dispatcher) + return { + capacity: () => { + accepting = true + capacityListener?.() + } + } + } + + function inspect(params: Record): Promise { + return dispatcher.callRequest('pty.inspectProcess', params) as Promise + } + + it('answers exited for the stored incarnation long after the owner disconnected', async () => { + const owner = withAbsentOwner() + const { incarnationId } = await spawnPty() + + exitCallback?.({ exitCode: 3 }) + // Well past the 5s window the retired-incarnation tombstone used to impose. + await vi.advanceTimersByTimeAsync(600_000) + + expect(await inspect({ id: PTY_1, expectedIncarnationId: incarnationId })).toMatchObject({ + foregroundProcessEvidence: { + verdict: 'exited', + reason: 'pty_exit_3', + ptyIncarnationId: incarnationId + } + }) + + // The reconnect delivers it; after that the client holds the exit and the relay says nothing. + owner.capacity() + await expect(inspect({ id: PTY_1, expectedIncarnationId: incarnationId })).rejects.toThrow( + 'terminal_gone' + ) + }) + + it('spells an ESRCH death as an unknown status rather than a fabricated code', async () => { + withAbsentOwner() + const { incarnationId } = await spawnPty() + vi.spyOn(ptyShellUtils, 'isProcessAlive').mockReturnValue(false) + + // The listing probe is the reachable ESRCH path: a shell that exited without node-pty's onExit. + await dispatcher.callRequest('pty.listProcesses', { + includeForegroundProcessEvidence: false + }) + + expect(await inspect({ id: PTY_1, expectedIncarnationId: incarnationId })).toMatchObject({ + foregroundProcessEvidence: { + verdict: 'exited', + reason: 'pty_exit_status_unknown' + } + }) + }) + + it('says nothing once the exit reached the owner', async () => { + const { incarnationId } = await spawnPty() + + exitCallback?.({ exitCode: 0 }) + expect(dispatcher.notify).toHaveBeenCalledWith('pty.exit', { + id: PTY_1, + code: 0, + incarnationId + }) + + await expect(inspect({ id: PTY_1, expectedIncarnationId: incarnationId })).rejects.toThrow( + 'terminal_gone' + ) + }) + + it('refuses to answer a caller that named no incarnation or a different one', async () => { + withAbsentOwner() + await spawnPty() + + exitCallback?.({ exitCode: 0 }) + + await expect(inspect({ id: PTY_1 })).rejects.toThrow('terminal_gone') + await expect( + inspect({ + id: PTY_1, + expectedIncarnationId: '11111111-1111-4111-8111-111111111111' + }) + ).rejects.toThrow('terminal_gone') + }) + + it('never answers for an id the relay never minted', async () => { + withAbsentOwner() + const { incarnationId } = await spawnPty() + exitCallback?.({ exitCode: 0 }) + + // A restarted relay renumbers, and legacy hosts minted bare `pty-N`. Neither is this exit. + for (const id of ['pty-1', 'pty2:other-mint-epoch:1', testPtyId(9)]) { + await expect(inspect({ id, expectedIncarnationId: incarnationId })).rejects.toThrow( + 'terminal_gone' + ) + } + }) + + it('marks an attach refusal only when it holds that incarnation exit', async () => { + withAbsentOwner() + const { incarnationId } = await spawnPty() + + exitCallback?.({ exitCode: 0 }) + + const marked = await attachPty({ + id: PTY_1, + expectedIncarnationId: incarnationId + }).catch((error: Error) => error) + expect((marked as Error).message).toContain(PTY_ATTACH_PROVEN_EXITED_MARKER) + expect(isProvenExitedPtyAttachRefusal(marked)).toBe(true) + + // Without the incarnation the answer is the ambiguous "no such id" union, unchanged. + const unmarked = await attachPty({ id: PTY_1 }).catch((error: Error) => error) + expect(isProvenExitedPtyAttachRefusal(unmarked)).toBe(false) + expect((unmarked as Error).message).toBe(`PTY "${PTY_1}" not found`) + }) + + it('does not attribute a held exit to the incarnation that replaced it', async () => { + withAbsentOwner() + const { incarnationId: retired } = await spawnPty() + exitCallback?.({ exitCode: 0 }) + + const state = JSON.stringify([ + { id: PTY_1, pid: process.pid, cols: 80, rows: 24, cwd: process.cwd() } + ]) + await dispatcher.callRequest('pty.revive', { state }) + const revived = ( + (await dispatcher.callRequest('pty.listProcesses', { + includeForegroundProcessEvidence: false + })) as { id: string; incarnationId: string }[] + ).find((entry) => entry.id === PTY_1) + expect(revived?.incarnationId).not.toBe(retired) + + // The live record answers for itself, and the held exit answers for nobody but its own + // incarnation — which no longer owns this id. + expect(await inspect({ id: PTY_1, expectedIncarnationId: retired })).toMatchObject({ + foregroundProcessEvidence: { + verdict: 'unverifiable', + reason: 'incarnation_mismatch' + } + }) + expect( + ( + await inspect({ + id: PTY_1, + expectedIncarnationId: revived?.incarnationId + }) + ).foregroundProcessEvidence?.verdict + ).not.toBe('exited') + }) + + it('writes no death when the sweep drops a record we tore down ourselves', async () => { + withAbsentOwner() + const { incarnationId } = await spawnPty() + // State-space, not a user path: nothing today leaves a disposed record in the pool. The sweep + // branch exists for it, and this pins that it retires bookkeeping without certifying a death. + const managed = ( + handler as unknown as { + ptys: Map void } }> + } + ).ptys.get(PTY_1)! + managed.disposed = true + const markExited = vi.spyOn(managed.physicalExit, 'markExited') + + await dispatcher.callRequest('pty.listProcesses', { + includeForegroundProcessEvidence: false + }) + + expect(handler.activePtyCount).toBe(0) + // Not even the internal "this process ended" flag: a shutdown waiting on it would read our own + // bookkeeping as the host's observation. + expect(markExited).not.toHaveBeenCalled() + expect(dispatcher.notify).not.toHaveBeenCalledWith('pty.exit', expect.anything()) + await expect(inspect({ id: PTY_1, expectedIncarnationId: incarnationId })).rejects.toThrow( + 'terminal_gone' + ) + }) + + it('leaves a shutdown that timed out unverifiable, with a sibling still serving', async () => { + withAbsentOwner() + const { incarnationId } = await spawnPty() + const failedKill = vi.fn<() => void>(() => { + throw new Error('host refused kill') + }) + mockPtySpawn.mockReturnValueOnce({ ...mockPtyInstance, kill: failedKill }) + await spawnPty() + + const disposal = handler.dispose().catch((error: Error) => error) + await vi.advanceTimersByTimeAsync(8_001) + expect(await disposal).toMatchObject({ message: 'host refused kill' }) + failedKill.mockImplementation(() => {}) + + expect(() => process.kill(process.pid, 0)).not.toThrow() + expect(dispatcher.notify).not.toHaveBeenCalledWith('pty.exit', expect.anything()) + await expect(inspect({ id: PTY_1, expectedIncarnationId: incarnationId })).rejects.toThrow( + 'terminal_gone' + ) + }) +}) diff --git a/src/relay/pty-handler-ssh-recovery-convergence.test.ts b/src/relay/pty-handler-ssh-recovery-convergence.test.ts new file mode 100644 index 000000000000..451cbb5e1988 --- /dev/null +++ b/src/relay/pty-handler-ssh-recovery-convergence.test.ts @@ -0,0 +1,221 @@ +// End to end over the boundary the fix is about: an SSH worker dies while the laptop is away, the +// laptop comes back much later, and the orchestration sweep settles it. Relay, SSH provider and +// recovery candidate all run for real here, so a disagreement between them fails the test. +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockPtySpawn, mockPtyInstance, mockCreateShellPromptReadinessProbe } = vi.hoisted(() => ({ + mockPtySpawn: vi.fn(), + mockCreateShellPromptReadinessProbe: vi.fn(), + mockPtyInstance: { + pid: process.pid, + process: 'zsh', + onData: vi.fn(), + onExit: vi.fn(), + write: vi.fn(), + resize: vi.fn(), + kill: vi.fn(), + clear: vi.fn(), + pause: vi.fn(), + resume: vi.fn() + } +})) + +vi.mock('node-pty', () => ({ spawn: mockPtySpawn })) + +vi.mock('../main/pty/posix-pty-process-groups', () => ({ + forceKillPosixPtyProcessGroups: vi.fn((_pid: number, fallback: () => void) => fallback()) +})) + +vi.mock('../main/shell-prompt-readiness-probe', () => ({ + createShellPromptReadinessProbe: mockCreateShellPromptReadinessProbe +})) + +import type { PtyHandler } from './pty-handler' +import { SshPtyProvider } from '../main/providers/ssh-pty-provider' +import { inspectExitedIncarnationFromRuntimeController } from '../main/ipc/pty/runtime/operations' +import { sshProviders } from '../main/ipc/pty/provider/registry' +import { toAppSshPtyId } from '../shared/ssh-pty-id' +import type { PtyIncarnationId } from '../shared/pty-incarnation' +import { reconcileLegacyWorkerCandidate } from '../main/runtime/runtime-legacy-worker-terminal-recovery-candidate' +import type { LegacyWorkerRecoveryResolution } from '../main/runtime/runtime-legacy-worker-terminal-recovery-types' +import { OrchestrationDb } from '../main/runtime/orchestration/db' +import { + beginPtyHandlerTest, + createTestPtyHandler, + endPtyHandlerTest, + type MockDispatcher +} from './pty-handler-test-harness' + +const CONNECTION = 'recovery-convergence' + +describe('SSH worker recovery converges on the relay exit', () => { + let dispatcher: MockDispatcher + let handler: PtyHandler + let originalPlatform: PropertyDescriptor | undefined + let provider: SshPtyProvider + let exitCallback: ((info: { exitCode: number }) => void) | undefined + let ownerAccepts: boolean + + beforeEach(() => { + ;({ dispatcher, handler, originalPlatform } = beginPtyHandlerTest({ + mockPtySpawn, + mockPtyInstance, + mockCreateShellPromptReadinessProbe + })) + exitCallback = undefined + ownerAccepts = false + mockPtySpawn.mockImplementation(() => ({ + ...mockPtyInstance, + onData: vi.fn(), + onExit: vi.fn((callback: (info: { exitCode: number }) => void) => { + exitCallback = callback + }) + })) + Object.assign(dispatcher, { + onLegacyPtyCapacity: vi.fn(() => vi.fn()), + tryNotifyPtyData: vi.fn(() => true), + // The owning client is offline: the relay holds the exit instead of dropping it. + tryNotifyPtyExit: vi.fn(() => ownerAccepts), + legacyRetentionBelowLowWater: true + }) + handler = createTestPtyHandler(dispatcher) + provider = new SshPtyProvider(CONNECTION, { + request: (method: string, params: Record) => + dispatcher.callRequest(method, params), + onNotification: () => () => {}, + onRequest: () => () => {} + } as never) + sshProviders.set(CONNECTION, provider) + }) + + afterEach(async () => { + sshProviders.delete(CONNECTION) + provider.dispose() + await endPtyHandlerTest(handler, originalPlatform) + }) + + async function reconcile( + appPtyId: string, + incarnationId: PtyIncarnationId, + dispatchId: string + ): Promise<{ + pendingResolutions: LegacyWorkerRecoveryResolution[] + deferredDispatchIds: Set + }> { + const pendingResolutions: LegacyWorkerRecoveryResolution[] = [] + const deferredDispatchIds = new Set() + await reconcileLegacyWorkerCandidate({ + controller: {} as never, + ports: { + // The production port, so the registry, the provider and the relay all answer for real. + proveTerminalExited: (candidate) => + inspectExitedIncarnationFromRuntimeController(candidate.ptyId, candidate.incarnationId) + } as never, + options: { connectionId: CONNECTION }, + candidate: { dispatchId, ptyId: appPtyId, incarnationId } as never, + workspace: {} as never, + resolvedWorktrees: [], + // The client's listing is empty: it was disconnected when the shell ended. + inventory: { livePtyIds: new Set() } as never, + pendingResolutions, + deferredDispatchIds + }) + return { pendingResolutions, deferredDispatchIds } + } + + it('settles a worker whose exit the relay held through a long disconnect', async () => { + const { id, incarnationId } = (await dispatcher.callRequest('pty.spawn', {})) as { + id: string + incarnationId: PtyIncarnationId + } + exitCallback?.({ exitCode: 0 }) + expect(handler.activePtyCount).toBe(0) + expect( + await dispatcher.callRequest('pty.listProcesses', { + includeForegroundProcessEvidence: false + }) + ).toEqual([]) + await vi.advanceTimersByTimeAsync(600_000) + + const appPtyId = toAppSshPtyId(CONNECTION, id) + const db = new OrchestrationDb(':memory:') + try { + const run = db.createRun({ + objective: 'recover a terminated SSH worker', + coordinatorHandle: 'term_coordinator', + coordinatorPaneKey: 'coordinator:11111111-1111-4111-8111-111111111111' + }) + const task = db.createTask({ + spec: 'recover a terminated SSH worker', + runId: run.id + }) + const start = { + taskId: task.id, + startOptions: {}, + creator: { kind: 'system' as const }, + maxDepth: 9 + } + const { dispatch } = db.createStartingWorkerDispatch(start) + db.markWorkerDispatchReady(dispatch.id) + + // Before the sweep the dispatch owns the task; a second worker cannot be started for it. + expect(db.getWorkerDispatch(dispatch.id)?.state).toBe('ready') + expect(() => db.createStartingWorkerDispatch(start)).toThrow() + + const result = await reconcile(appPtyId, incarnationId, dispatch.id) + + expect(result.pendingResolutions).toEqual([ + { + candidate: { + dispatchId: dispatch.id, + ptyId: appPtyId, + incarnationId + }, + resolution: 'exited' + } + ]) + expect([...result.deferredDispatchIds]).toEqual([]) + } finally { + db.close() + } + }) + + it('defers when the relay already handed that exit to its owner', async () => { + ownerAccepts = true + const { id, incarnationId } = (await dispatcher.callRequest('pty.spawn', {})) as { + id: string + incarnationId: PtyIncarnationId + } + exitCallback?.({ exitCode: 0 }) + + // Main learned about this death through the stream; the sweep has no second source to consult. + const result = await reconcile(toAppSshPtyId(CONNECTION, id), incarnationId, 'worker-delivered') + + expect(result.pendingResolutions).toEqual([]) + expect([...result.deferredDispatchIds]).toEqual(['worker-delivered']) + }) + + it('defers a shutdown that timed out while a sibling kill failure kept the relay serving', async () => { + const { id, incarnationId } = (await dispatcher.callRequest('pty.spawn', {})) as { + id: string + incarnationId: PtyIncarnationId + } + const failedKill = vi.fn<() => void>(() => { + throw new Error('host refused kill') + }) + mockPtySpawn.mockReturnValueOnce({ ...mockPtyInstance, kill: failedKill }) + await dispatcher.callRequest('pty.spawn', {}) + + const disposal = handler.dispose().catch((error: Error) => error) + await vi.advanceTimersByTimeAsync(8_001) + expect(await disposal).toMatchObject({ message: 'host refused kill' }) + failedKill.mockImplementation(() => {}) + + // The record is gone from the relay's map and nothing watched the shell end. + expect(() => process.kill(process.pid, 0)).not.toThrow() + const result = await reconcile(toAppSshPtyId(CONNECTION, id), incarnationId, 'worker-timeout') + + expect(result.pendingResolutions).toEqual([]) + expect([...result.deferredDispatchIds]).toEqual(['worker-timeout']) + }) +}) diff --git a/src/relay/pty-handler.ts b/src/relay/pty-handler.ts index cdf436bca2ab..ca2698f96ad1 100644 --- a/src/relay/pty-handler.ts +++ b/src/relay/pty-handler.ts @@ -83,7 +83,6 @@ import type { RemoteForegroundEvidence } from '../shared/foreground-process-evidence' import { expandWindowsPathEnvironmentVariables } from '../shared/windows-environment-expansion' -import { pruneRetiredPtyIncarnations } from '../shared/retired-pty-incarnations' import { agentSessionOwnerBindingsEqual, ClaimedAgentPtyOwnerRegistry @@ -504,6 +503,9 @@ export type PtyEnvAugmenter = (ctx: { launchAgent?: TuiAgent }) => Record +/** An observed exit held until the PTY's owner has received it. */ +type PendingPtyExit = { id: string; code: number; incarnationId: string } + export type RelayPtyWorktreeRemovalCoordinator = { beginWorktreePtySpawn(operationPath: string): () => void } @@ -519,11 +521,7 @@ export class PtyHandler { private outputFlushTimer: ReturnType | null = null private pendingOutputByPty = new Map() private pendingProducerBytesByPty = new Map() - private pendingExitByPty = new Map() - private retiredIncarnations = new Map< - string, - { id: string; code: number; incarnationId: string; expiresAt: number } - >() + private pendingExitByPty = new Map() private pausedOutputPtys = new Set() private consumerPausedOutputPtys = new Set() private removeLegacyCapacityListener: (() => void) | null = null @@ -1029,13 +1027,6 @@ export class PtyHandler { code: exitCode, incarnationId: managed.incarnationId }) - this.retiredIncarnations.set(managed.id, { - id: managed.id, - code: exitCode, - incarnationId: managed.incarnationId, - expiresAt: Date.now() + 5_000 - }) - pruneRetiredPtyIncarnations(this.retiredIncarnations) this.publishPendingExit(managed.id) this.notifyExitListener(managed) this.agentSessionOwners.release(managed.id) @@ -1390,9 +1381,10 @@ export class PtyHandler { this.outputFlushTimer = null } + // Output flow only: an observed exit is evidence its owner has not received yet, and retiring + // the record it belonged to is not delivery. `publishPendingExit` owns that delete. private clearPtyFlowState(id: string): void { this.deletePendingOutput(id) - this.pendingExitByPty.delete(id) this.pausedOutputPtys.delete(id) this.consumerPausedOutputPtys.delete(id) this.clearPtyInputState(id) @@ -2025,7 +2017,6 @@ export class PtyHandler { } : {}) } - this.retiredIncarnations.delete(id) this.sourcePublication?.activate(id, managed.incarnationId, context) const sourceActivation = context && this.sourcePublication?.receivingActivation?.(id, context.clientId) @@ -2067,6 +2058,9 @@ export class PtyHandler { const managed = this.ptys.get(id) // Why: after dispose, pty.kill is a POSIX no-op; treat disposed as not-found so failures aren't silent. if (!managed || managed.disposed) { + if (this.undeliveredExitFor(id, params.expectedIncarnationId)) { + throw new Error(`PTY "${id}" not found (${PTY_ATTACH_PROVEN_EXITED_MARKER})`) + } throw new Error(`PTY "${id}" not found`) } @@ -2392,30 +2386,32 @@ export class PtyHandler { } /** - * Retire every record for a PTY whose process is proven gone. Shared by the attach probe, the - * listing probe and the post-shutdown sweep so the three cannot drift on what "gone" retires. - * - * `evidence` is not decoration: `exited` publishes a verdict to the client, and only ESRCH from - * the host that owns the pid earns it. The disposed-record sweep retires off our own - * bookkeeping, which says we tore the record down — not that the shell died — so it stays - * silent (docs/reference/ssh-execution-boundary.md). + * The exit this id's owner has not received yet, for the incarnation the caller named. Deleted + * the moment publication settles, so an empty answer after delivery is correct: that client + * already has the exit. The entry carries the id it was written for, so that is checked too. */ - private reapExitedPty(managed: ManagedPty, evidence: 'exited' | 'record-torn-down'): void { - managed.physicalExit?.markExited() + private undeliveredExitFor(id: string, expected: unknown): PendingPtyExit | undefined { + const exit = this.pendingExitByPty.get(id) + return typeof expected === 'string' && exit?.id === id && exit.incarnationId === expected + ? exit + : undefined + } + + /** + * Drop a record we tore down ourselves. `managed.disposed` says our bookkeeping is gone, not that + * the shell died, so this path publishes nothing, marks no physical exit, and never writes + * `pendingExitByPty` — only something that watched a process end may + * (docs/reference/ssh-execution-boundary.md). + */ + private dropTornDownPtyRecord(managed: ManagedPty): void { this.releaseRelayIngress(managed) this.flushPtyOutput(managed.id) - if (evidence === 'exited') { - this.publishReapedExit(managed) - } + this.retirePtyRecord(managed) + } + + private retirePtyRecord(managed: ManagedPty): void { this.notifyExitListener(managed) this.agentSessionOwners.release(managed.id) - this.retiredIncarnations.set(managed.id, { - id: managed.id, - code: 0, - incarnationId: managed.incarnationId, - expiresAt: Date.now() + 5_000 - }) - pruneRetiredPtyIncarnations(this.retiredIncarnations) disposeManagedPty(managed) this.removePty(managed.id) this.clearPtyFlowState(managed.id) @@ -2464,7 +2460,12 @@ export class PtyHandler { if (!managed.pty.pid || isProcessAlive(managed.pty.pid)) { return false } - this.reapExitedPty(managed, 'exited') + // The death write lives here and nowhere else: this is the only retirement backed by ESRCH. + managed.physicalExit?.markExited() + this.releaseRelayIngress(managed) + this.flushPtyOutput(managed.id) + this.publishReapedExit(managed) + this.retirePtyRecord(managed) return true } @@ -2629,17 +2630,11 @@ export class PtyHandler { childProcessEvidence?: PtyChildProcessVerdict foregroundProcessEvidence?: RemoteForegroundEvidence }> { - pruneRetiredPtyIncarnations(this.retiredIncarnations) const id = params.id as string const managed = this.ptys.get(id) if (!managed || managed.disposed) { - const tombstone = this.retiredIncarnations.get(id) - if ( - tombstone && - tombstone.expiresAt > Date.now() && - typeof params.expectedIncarnationId === 'string' && - params.expectedIncarnationId === tombstone.incarnationId - ) { + const exit = this.undeliveredExitFor(id, params.expectedIncarnationId) + if (exit) { return { foregroundProcess: null, hasChildProcesses: false, @@ -2648,9 +2643,11 @@ export class PtyHandler { observationEpoch: ++this.foregroundEvidenceEpoch, capturedAgeMs: 0, ptyId: id, - ptyIncarnationId: tombstone.incarnationId, + ptyIncarnationId: exit.incarnationId, verdict: 'exited', - reason: `pty_exit_${tombstone.code}` + // -1 is this wire's "gone, status unrecoverable": the pid was proven absent but nothing + // waited on the shell, so no status exists. Never spelled as a real code. + reason: exit.code === -1 ? 'pty_exit_status_unknown' : `pty_exit_${exit.code}` } } } @@ -2817,7 +2814,7 @@ export class PtyHandler { } for (const [entryIndex, [id, managed]] of managedEntries.entries()) { if (managed.disposed) { - this.reapExitedPty(managed, 'record-torn-down') + this.dropTornDownPtyRecord(managed) continue } if (this.reapPtyProvenExited(managed)) { diff --git a/src/shared/retired-pty-incarnations.test.ts b/src/shared/retired-pty-incarnations.test.ts deleted file mode 100644 index 69506fecc692..000000000000 --- a/src/shared/retired-pty-incarnations.test.ts +++ /dev/null @@ -1,30 +0,0 @@ -import { describe, expect, it } from 'vitest' -import { pruneRetiredPtyIncarnations } from './retired-pty-incarnations' - -describe('retired PTY incarnation retention', () => { - it('removes expired records before they can accumulate', () => { - const records = new Map([ - ['expired', { incarnationId: 'a', code: 0, expiresAt: 10 }], - ['live', { incarnationId: 'b', code: 0, expiresAt: 30 }] - ]) - - pruneRetiredPtyIncarnations(records, 20) - - expect([...records.keys()]).toEqual(['live']) - }) - - it('caps records when many distinct PTYs retire together', () => { - const records = new Map( - Array.from({ length: 1001 }, (_, index) => [ - `pty-${index}`, - { incarnationId: `inc-${index}`, code: 0, expiresAt: 100 } - ]) - ) - - pruneRetiredPtyIncarnations(records, 0) - - expect(records.size).toBe(1000) - expect(records.has('pty-0')).toBe(false) - expect(records.has('pty-1000')).toBe(true) - }) -}) diff --git a/src/shared/retired-pty-incarnations.ts b/src/shared/retired-pty-incarnations.ts deleted file mode 100644 index fd8241867f81..000000000000 --- a/src/shared/retired-pty-incarnations.ts +++ /dev/null @@ -1,26 +0,0 @@ -export type RetiredPtyIncarnation = { - incarnationId: string - code: number - expiresAt: number -} - -const MAX_RETIRED_PTY_INCARNATIONS = 1000 - -/** Drop expired exit evidence and cap retained records during long-lived hosts. */ -export function pruneRetiredPtyIncarnations( - records: Map, - now = Date.now() -): void { - for (const [id, record] of records) { - if (record.expiresAt <= now) { - records.delete(id) - } - } - while (records.size > MAX_RETIRED_PTY_INCARNATIONS) { - const oldest = records.keys().next().value - if (oldest === undefined) { - break - } - records.delete(oldest) - } -}