diff --git a/apps/cli/src/agent/AGENTS.md b/apps/cli/src/agent/AGENTS.md index bd930c053..3351aa9ef 100644 --- a/apps/cli/src/agent/AGENTS.md +++ b/apps/cli/src/agent/AGENTS.md @@ -81,8 +81,11 @@ arrive: context/message-flow.md "Upstream". answer before giving up on the upstream turn's response: the Codex adapter drains session notifications before refusing, so the turn's response routinely wins that race and would otherwise mask the refusal. -- `acp-runner.ts` — process spawn/restart around the client. Spawn + initialize + - `newSession`/`loadSession` share `acp-session-start-gate.ts` (default 2, +- `acp-runner.ts` — process spawn/restart around the client. + Auxiliary ACP shutdown shares one termination attempt per owned child, uses the + Windows process-tree cleanup helper, and reports termination failure; protocol + session-close failure must still proceed to process cleanup. + Spawn + initialize + `newSession`/`loadSession` share `acp-session-start-gate.ts` (default 2, `LODY_MAX_CONCURRENT_ACP_SESSION_STARTS`). Unbounded concurrent Codex starts each spawn a lody.exe adapter, a Codex app-server, and a lody.exe MCP child; they contend on `~/.codex` and freeze every in-flight session until Lody diff --git a/apps/cli/src/agent/acp-authentication.test.ts b/apps/cli/src/agent/acp-authentication.test.ts index dbd1ccfa4..ea6bd72b8 100644 --- a/apps/cli/src/agent/acp-authentication.test.ts +++ b/apps/cli/src/agent/acp-authentication.test.ts @@ -10,6 +10,12 @@ import type { Logger } from '@/utils/logger'; import { createStdinWritableStream, createStdoutReadableStream } from '@/utils/stream'; import { AcpAuthenticationManager, probeBuiltinAuthentication } from './acp-authentication'; +vi.mock('@/utils/windows-process-tree', () => ({ + terminateWindowsProcessTree: async (child: ChildProcess) => { + if (child.exitCode == null && child.signalCode == null) child.kill('SIGKILL'); + }, +})); + const createSilentLogger = (): Logger => ({ info: () => {}, warn: () => {}, @@ -316,8 +322,12 @@ describe('AcpAuthenticationManager', () => { disposition: 'error', error: 'Kimi Code authentication timed out. Please try again.', }); - expect(stuckChild.kill).toHaveBeenNthCalledWith(1, 'SIGTERM'); - expect(stuckChild.kill).toHaveBeenNthCalledWith(2, 'SIGKILL'); + if (process.platform === 'win32') { + expect(stuckChild.kill).toHaveBeenCalledWith('SIGKILL'); + } else { + expect(stuckChild.kill).toHaveBeenNthCalledWith(1, 'SIGTERM'); + expect(stuckChild.kill).toHaveBeenNthCalledWith(2, 'SIGKILL'); + } await expect(manager.authenticate({ requestId: 'auth-2', ...input })).resolves.toEqual({ success: true, @@ -531,7 +541,7 @@ describe('AcpAuthenticationManager', () => { success: true, disposition: 'cancelled', }); - expect(child.kill).toHaveBeenCalledWith('SIGTERM'); + expect(child.kill).toHaveBeenCalledWith(process.platform === 'win32' ? 'SIGKILL' : 'SIGTERM'); }); it('bridges ACP URL consent without retaining authentication process output', async () => { diff --git a/apps/cli/src/agent/acp-runner.test.ts b/apps/cli/src/agent/acp-runner.test.ts index d4c49a33e..41ed065ee 100644 --- a/apps/cli/src/agent/acp-runner.test.ts +++ b/apps/cli/src/agent/acp-runner.test.ts @@ -3,7 +3,11 @@ import type { ChildProcess } from 'child_process'; import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; -import { afterEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +const treeCleanup = vi.hoisted(() => vi.fn<() => Promise>()); +vi.mock('@/utils/windows-process-tree', () => ({ terminateWindowsProcessTree: treeCleanup })); +const nativePlatform = process.platform; import { __test__, shutdownLocalAcpAgent, spawnAcpProcess } from './acp-runner'; import type { Logger } from '@/utils/logger'; @@ -136,7 +140,12 @@ base_url = "https://gateway.example/v1" }); describe('shutdownLocalAcpAgent', () => { + beforeEach(() => { + Object.defineProperty(process, 'platform', { value: 'linux' }); + treeCleanup.mockReset(); + }); afterEach(() => { + Object.defineProperty(process, 'platform', { value: nativePlatform }); vi.useRealTimers(); vi.restoreAllMocks(); }); @@ -187,7 +196,7 @@ describe('shutdownLocalAcpAgent', () => { expect(child.kill).toHaveBeenNthCalledWith(2, 'SIGKILL'); }); - if (process.platform !== 'win32') { + { it('terminates the ACP process group on POSIX when the child has a PID', async () => { const child = createFakeChildProcess({ pid: 1234 }); const processKill = vi.spyOn(process, 'kill').mockImplementation((pid, signal) => { @@ -235,4 +244,109 @@ describe('shutdownLocalAcpAgent', () => { expect(child.kill).not.toHaveBeenCalled(); }); } + + it('recognizes a child that already exited from a signal', async () => { + const child = createFakeChildProcess(); + child.signalCode = 'SIGTERM'; + await shutdownLocalAcpAgent({ + agentProcess: child, + logger: createSilentLogger(), + sessionLabel: 'signal', + }); + expect(child.kill).not.toHaveBeenCalled(); + }); + + it('rejects when force termination never produces an exit', async () => { + vi.useFakeTimers(); + const child = createFakeChildProcess({ exitOnSigterm: false, exitOnSigkill: false }); + const result = shutdownLocalAcpAgent({ + agentProcess: child, + logger: createSilentLogger(), + sessionLabel: 'stuck', + exitTimeoutMs: 10, + }); + const assertion = expect(result).rejects.toThrow('did not exit after SIGKILL'); + await vi.advanceTimersByTimeAsync(20); + await assertion; + expect(child.listenerCount('exit')).toBe(0); + }); + + it('reports a signaling failure and allows a later cleanup attempt', async () => { + const child = createFakeChildProcess(); + vi.mocked(child.kill).mockImplementationOnce(() => { + throw new Error('permission denied'); + }); + const options = { agentProcess: child, logger: createSilentLogger(), sessionLabel: 'retry' }; + await expect(shutdownLocalAcpAgent(options)).rejects.toThrow('permission denied'); + await shutdownLocalAcpAgent(options); + expect(child.exitCode).toBe(0); + }); + + it('waits for shared Windows tree cleanup despite protocol close failure', async () => { + Object.defineProperty(process, 'platform', { value: 'win32' }); + const child = createFakeChildProcess({ pid: 1234 }); + let finish: (() => void) | undefined; + treeCleanup.mockImplementation( + () => + new Promise((resolve) => { + finish = resolve; + }) + ); + const options = { + agentProcess: child, + logger: createSilentLogger(), + sessionLabel: 'tree', + client: { + closeSession: async () => { + throw new Error('closed'); + }, + } as never, + acpSessionId: 'acp' as never, + }; + let settled = false; + const first = shutdownLocalAcpAgent(options).then(() => { + settled = true; + }); + const second = shutdownLocalAcpAgent(options); + await Promise.resolve(); + await Promise.resolve(); + expect(settled).toBe(false); + expect(treeCleanup).toHaveBeenCalledTimes(1); + child.signalCode = 'SIGKILL'; + finish?.(); + await Promise.all([first, second]); + expect(settled).toBe(true); + expect(child.kill).not.toHaveBeenCalled(); + }); + + it('reports Windows tree failure and permits retry', async () => { + Object.defineProperty(process, 'platform', { value: 'win32' }); + const child = createFakeChildProcess({ pid: 1234 }); + treeCleanup.mockRejectedValueOnce(new Error('tree failed')).mockImplementationOnce(async () => { + child.signalCode = 'SIGKILL'; + }); + const options = { + agentProcess: child, + logger: createSilentLogger(), + sessionLabel: 'tree-retry', + }; + await expect(shutdownLocalAcpAgent(options)).rejects.toThrow('tree failed'); + await shutdownLocalAcpAgent(options); + expect(child.signalCode).toBe('SIGKILL'); + }); + + it('does not equate Windows helper success with wrapper exit', async () => { + Object.defineProperty(process, 'platform', { value: 'win32' }); + vi.useFakeTimers(); + treeCleanup.mockResolvedValue(undefined); + const result = shutdownLocalAcpAgent({ + agentProcess: createFakeChildProcess({ pid: 1234 }), + logger: createSilentLogger(), + sessionLabel: 'tree-timeout', + exitTimeoutMs: 10, + }); + const assertion = expect(result).rejects.toThrow('did not exit after Windows tree termination'); + await vi.advanceTimersByTimeAsync(10); + await assertion; + }); }); diff --git a/apps/cli/src/agent/acp-runner.ts b/apps/cli/src/agent/acp-runner.ts index c6a934afb..3f026bc5a 100644 --- a/apps/cli/src/agent/acp-runner.ts +++ b/apps/cli/src/agent/acp-runner.ts @@ -13,6 +13,7 @@ import { v4 as uuidV4 } from 'uuid'; import { z } from 'zod'; import type { Logger } from '@/utils/logger'; +import { terminateWindowsProcessTree } from '@/utils/windows-process-tree'; import type { TerminalManager } from '@/session/terminal-manager'; import { AgentClient, @@ -162,8 +163,12 @@ export const createAcpClient = async (options: CreateAcpClientOptions) => { return { client, acpSessionId: sessionResponse.sessionId as ACPSessionId, sessionResponse }; }; +function hasChildExited(child: ChildProcess): boolean { + return child.exitCode !== null || child.signalCode != null; +} + function waitForChildProcessExit(child: ChildProcess, timeoutMs: number): Promise { - if (child.exitCode !== null) { + if (hasChildExited(child)) { return Promise.resolve(true); } @@ -174,7 +179,7 @@ function waitForChildProcessExit(child: ChildProcess, timeoutMs: number): Promis }; const onTimeout = () => { cleanup(); - resolve(child.exitCode !== null); + resolve(hasChildExited(child)); }; const cleanup = () => { clearTimeout(timeoutHandle); @@ -195,20 +200,48 @@ function signalChildProcess(child: ChildProcess, signal: NodeJS.Signals): void { child.kill(signal); } -async function terminateChildProcess( +const childTerminations = new WeakMap>(); + +function terminateChildProcess( + child: ChildProcess, + logger: Logger, + sessionLabel: string, + exitTimeoutMs: number +): Promise { + const pending = childTerminations.get(child); + if (pending) return pending; + const termination = terminateChildProcessOnce(child, logger, sessionLabel, exitTimeoutMs); + childTerminations.set(child, termination); + void termination.catch(() => { + if (childTerminations.get(child) === termination) childTerminations.delete(child); + }); + return termination; +} + +async function terminateChildProcessOnce( child: ChildProcess, logger: Logger, sessionLabel: string, exitTimeoutMs: number ): Promise { - if (child.exitCode !== null) { + if (process.platform === 'win32') { + await terminateWindowsProcessTree(child, true, { timeoutMs: exitTimeoutMs }); + if (!(await waitForChildProcessExit(child, exitTimeoutMs))) { + throw new Error( + `[${sessionLabel}] ACP agent process did not exit after Windows tree termination` + ); + } + return; + } + if (hasChildExited(child)) { return; } try { signalChildProcess(child, 'SIGTERM'); - } catch { - return; + } catch (error) { + if (hasChildExited(child)) return; + throw error; } if (await waitForChildProcessExit(child, exitTimeoutMs)) { @@ -220,10 +253,13 @@ async function terminateChildProcess( ); try { signalChildProcess(child, 'SIGKILL'); - } catch { - return; + } catch (error) { + if (hasChildExited(child)) return; + throw error; + } + if (!(await waitForChildProcessExit(child, exitTimeoutMs))) { + throw new Error(`[${sessionLabel}] ACP agent process did not exit after SIGKILL`); } - await waitForChildProcessExit(child, exitTimeoutMs); } export type SpawnAcpProcessOptions = { diff --git a/apps/cli/src/commands/start-session-shutdown.test.ts b/apps/cli/src/commands/start-session-shutdown.test.ts new file mode 100644 index 000000000..23c6162ca --- /dev/null +++ b/apps/cli/src/commands/start-session-shutdown.test.ts @@ -0,0 +1,81 @@ +import { afterEach, expect, it, vi } from 'vitest'; +import type { ChildProcess } from 'child_process'; +import { EventEmitter } from 'events'; +import type { SessionId, WorkspaceId } from '@lody/shared'; +import { Session } from '../session/session'; +import { createStartShutdownController } from './start-shutdown'; +import type { Logger } from '../utils/logger'; +import type { SessionProcessHandle } from '../session/session-sandbox'; + +afterEach(() => vi.useRealTimers()); + +it('runs the Session process phase before outer exit when terminal disposal hangs', async () => { + vi.useFakeTimers(); + const logger: Logger = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + success: vi.fn(), + setLevel: vi.fn(), + child: () => logger, + close: async () => {}, + }; + const session = new Session( + { + workspaceId: 'workspace' as WorkspaceId, + sessionId: 'session' as SessionId, + userId: 'user', + machineId: 'machine', + agentCliType: 'builtin', + agentType: 'codex', + userName: 'test', + userEmail: 'test@example.com', + }, + logger, + process.cwd() + ); + session.acpSessionId = 'acp' as never; + const dispose = vi + .spyOn(session.terminalManager, 'disposeAll') + .mockImplementation(() => new Promise(() => {})); + const child = new EventEmitter() as ChildProcess; + child.exitCode = null; + child.signalCode = null; + const terminate = vi.fn(async () => { + child.signalCode = 'SIGKILL'; + }); + const handle: SessionProcessHandle = { + child, + terminate, + inspectExit: async () => null, + onExit: () => () => {}, + onClose: () => () => {}, + onError: () => () => {}, + }; + // @ts-expect-error - synthetic owned process fixture + session.agentProcess = handle; + const exit = vi.fn(); + const controller = createStartShutdownController({ + signals: [], + logger, + shutdown: () => session.terminate(false), + forceShutdown: () => session.terminate(true), + flushTelemetry: async () => {}, + exit, + }); + const result = controller.shutdown('SIGTERM'); + await vi.advanceTimersByTimeAsync(14_999); + expect(terminate).not.toHaveBeenCalled(); + expect(exit).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(terminate).toHaveBeenCalledWith(true); + expect(exit).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(5_000); + await result; + expect(dispose).toHaveBeenCalledTimes(1); + expect(exit).toHaveBeenCalledWith(143); + // Failed terminal cleanup remains truthfully retryable despite root termination. + // @ts-expect-error - retained cleanup state + expect(session.status).toBe('stopping'); +}); diff --git a/apps/cli/src/commands/start-shutdown.ts b/apps/cli/src/commands/start-shutdown.ts index 1c19fdd0d..54e076660 100644 --- a/apps/cli/src/commands/start-shutdown.ts +++ b/apps/cli/src/commands/start-shutdown.ts @@ -1,6 +1,8 @@ import type { Logger } from '@/utils/logger'; export const START_SHUTDOWN_TIMEOUT_MS = 15_000; +export const START_FORCE_SHUTDOWN_TIMEOUT_MS = 15_000; +export const START_TELEMETRY_SHUTDOWN_TIMEOUT_MS = 2_000; type ShutdownExit = (code: number) => void; export type StartShutdownRequest = @@ -21,9 +23,13 @@ export interface StartShutdownControllerOptions { signals: NodeJS.Signals[]; logger: Logger; shutdown: () => Promise; + /** Bypass graceful drains and start owned process cleanup concurrently. */ + forceShutdown?: () => Promise; flushTelemetry: () => Promise; exit: ShutdownExit; timeoutMs?: number; + forceTimeoutMs?: number; + telemetryTimeoutMs?: number; } const SIGNAL_EXIT_CODES: Partial> = { @@ -60,6 +66,28 @@ export function createStartShutdownController( let isShuttingDown = false; let exitRequested = false; let shutdownTimeout: NodeJS.Timeout | null = null; + let forcedShutdown: Promise | null = null; + let resolveFinished: () => void = () => {}; + const finished = new Promise((resolve) => { + resolveFinished = resolve; + }); + + const withinDeadline = async (action: () => Promise, milliseconds: number) => { + let timer: ReturnType | undefined; + try { + await Promise.race([ + Promise.resolve().then(action), + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error('Shutdown phase deadline exceeded')), + milliseconds + ); + }), + ]); + } finally { + clearTimeout(timer); + } + }; const clearShutdownTimeout = () => { if (!shutdownTimeout) { @@ -86,13 +114,17 @@ export function createStartShutdownController( unregister(); try { - await options.flushTelemetry(); + await withinDeadline( + options.flushTelemetry, + options.telemetryTimeoutMs ?? START_TELEMETRY_SHUTDOWN_TIMEOUT_MS + ); } catch (error) { options.logger.debug( `Telemetry shutdown failed: ${error instanceof Error ? error.message : 'Unknown error'}` ); } finally { options.exit(code); + resolveFinished(); } }; @@ -100,11 +132,27 @@ export function createStartShutdownController( request: { signal?: NodeJS.Signals; exitCode: number }, reason: string ) => { + if (forcedShutdown) return forcedShutdown; options.logger.warn(reason); - await exitAfterTelemetry(request.exitCode || getExitCodeForSignal(request.signal)); + clearShutdownTimeout(); + forcedShutdown = Promise.resolve().then(async () => { + if (options.forceShutdown) { + try { + await withinDeadline( + options.forceShutdown, + options.forceTimeoutMs ?? START_FORCE_SHUTDOWN_TIMEOUT_MS + ); + } catch { + options.logger.warn('Forced process cleanup did not complete successfully before exit'); + } + } + await exitAfterTelemetry(request.exitCode || getExitCodeForSignal(request.signal) || 1); + }); + return forcedShutdown; }; const shutdown = async (request?: StartShutdownRequest) => { + if (exitRequested) return finished; const normalized = normalizeShutdownRequest(request); if (isShuttingDown) { await forceExit( @@ -131,15 +179,23 @@ export function createStartShutdownController( ); }, timeoutMs); - try { - await options.shutdown(); - } catch (error) { - options.logger.error( - `Shutdown error: ${error instanceof Error ? error.message : 'Unknown error'}` - ); - } finally { - await exitAfterTelemetry(normalized.exitCode); - } + const graceful = Promise.resolve().then(async () => { + try { + await options.shutdown(); + } catch (error) { + options.logger.error( + `Shutdown error: ${error instanceof Error ? error.message : 'Unknown error'}` + ); + await forceExit( + normalized, + 'Graceful shutdown failed; attempting forced process cleanup...' + ); + } finally { + if (forcedShutdown) await forcedShutdown; + else await exitAfterTelemetry(normalized.exitCode); + } + }); + await Promise.race([graceful, finished]); }; return { diff --git a/apps/cli/src/commands/start.test.ts b/apps/cli/src/commands/start.test.ts index 349cd35eb..37faad13b 100644 --- a/apps/cli/src/commands/start.test.ts +++ b/apps/cli/src/commands/start.test.ts @@ -297,3 +297,79 @@ describe('start shutdown controller', () => { expect(exits).toEqual([130]); }); }); + +it('reserves forced cleanup before exit even when graceful shutdown remains stuck', async () => { + vi.useFakeTimers(); + const force = createDeferred(); + const exit = vi.fn(); + const forceShutdown = vi.fn(() => force.promise); + const controller = createStartShutdownController({ + signals: [], + logger: createTestLogger(), + shutdown: () => new Promise(() => {}), + forceShutdown, + flushTelemetry: async () => {}, + exit, + timeoutMs: 15, + forceTimeoutMs: 15, + }); + const result = controller.shutdown('SIGTERM'); + await vi.advanceTimersByTimeAsync(15); + expect(forceShutdown).toHaveBeenCalledTimes(1); + expect(exit).not.toHaveBeenCalled(); + force.resolve(); + await result; + expect(exit).toHaveBeenCalledWith(143); + expect(vi.getTimerCount()).toBe(0); +}); + +it('bounds force cleanup and telemetry when both remain stuck', async () => { + vi.useFakeTimers(); + const exit = vi.fn(); + const controller = createStartShutdownController({ + signals: [], + logger: createTestLogger(), + shutdown: () => new Promise(() => {}), + forceShutdown: () => new Promise(() => {}), + flushTelemetry: () => new Promise(() => {}), + exit, + timeoutMs: 15, + forceTimeoutMs: 15, + telemetryTimeoutMs: 2, + }); + const result = controller.shutdown('SIGTERM'); + await vi.advanceTimersByTimeAsync(31); + expect(exit).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + await result; + expect(exit).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); +}); + +it('does not let late graceful completion bypass the forced phase', async () => { + vi.useFakeTimers(); + const graceful = createDeferred(); + const force = createDeferred(); + const exit = vi.fn(); + const forceShutdown = vi.fn(() => force.promise); + const controller = createStartShutdownController({ + signals: [], + logger: createTestLogger(), + shutdown: () => graceful.promise, + forceShutdown, + flushTelemetry: async () => {}, + exit, + timeoutMs: 15, + }); + const first = controller.shutdown('SIGINT'); + await vi.advanceTimersByTimeAsync(15); + const second = controller.shutdown('SIGINT'); + graceful.resolve(); + await vi.advanceTimersByTimeAsync(0); + expect(exit).not.toHaveBeenCalled(); + expect(forceShutdown).toHaveBeenCalledTimes(1); + force.resolve(); + await Promise.all([first, second]); + expect(exit).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); +}); diff --git a/apps/cli/src/commands/start.ts b/apps/cli/src/commands/start.ts index 1afe0f672..9b659f301 100644 --- a/apps/cli/src/commands/start.ts +++ b/apps/cli/src/commands/start.ts @@ -618,6 +618,7 @@ async function startAgentService( const shutdownController = createStartShutdownController({ signals: shutdownSignals, logger, + forceShutdown: async () => await fleet.forceTerminateSessions(), shutdown: async () => { unregisterSupervisorControl(); unregisterProcessCleanup(); diff --git a/apps/cli/src/lib/lody-fleet.ts b/apps/cli/src/lib/lody-fleet.ts index 7fc0c957c..f74264ad9 100644 --- a/apps/cli/src/lib/lody-fleet.ts +++ b/apps/cli/src/lib/lody-fleet.ts @@ -156,6 +156,7 @@ export class LodyFleet { private readonly workspaceWatchCoordinator: WorkspaceWatchCoordinator; private readonly runtimes = new Map(); + private shutdownPromise: Promise | null = null; private readonly reviewCredentialResolvers = new Map(); private readonly startInFlight = new Map>(); private readonly retryTimers = new Map(); @@ -529,8 +530,29 @@ export class LodyFleet { }); } - async shutdown(): Promise { - if (this.stopped) return; + async forceTerminateSessions(): Promise { + this.stopped = true; + const results = await Promise.allSettled( + Array.from(this.runtimes.values(), (runtime) => runtime.lody.forceTerminateSessions()) + ); + const failures = results.flatMap((result) => + result.status === 'rejected' ? [result.reason] : [] + ); + if (failures.length > 0) + throw new AggregateError(failures, 'Forced workspace process cleanup failed'); + } + + shutdown(): Promise { + if (this.shutdownPromise) return this.shutdownPromise; + const pending = this.runShutdown(); + this.shutdownPromise = pending; + void pending.catch(() => { + if (this.shutdownPromise === pending) this.shutdownPromise = null; + }); + return pending; + } + + private async runShutdown(): Promise { this.stopped = true; this.stopRuntimeStateLoop(); this.memoryPressure.stop(); @@ -560,23 +582,32 @@ export class LodyFleet { this.unsubscribeWorkspaces = null; const runtimes = Array.from(this.runtimes.values()); - this.runtimes.clear(); - for (const runtime of runtimes) { - try { - await runtime.lody.cleanup(); - runtime.unsubscribeTerminalCleanup(); - await runtime.prPollerWorkspace?.dispose(); - await runtime.taskAutomation?.dispose(); - await runtime.reviewAutomation?.dispose(); - } catch (error) { - runtime.unsubscribeTerminalCleanup(); - this.logger.debug( - `[fleet] Failed to cleanup workspace runtime ${runtime.workspace.id}: ${formatErrorMessage( - error - )}` - ); - } - } + const cleanupResults = await Promise.allSettled( + runtimes.map(async (runtime) => { + try { + await runtime.lody.cleanup(); + runtime.unsubscribeTerminalCleanup(); + await runtime.prPollerWorkspace?.dispose(); + await runtime.taskAutomation?.dispose(); + await runtime.reviewAutomation?.dispose(); + if (this.runtimes.get(runtime.workspace.id) === runtime) + this.runtimes.delete(runtime.workspace.id); + } catch (error) { + runtime.unsubscribeTerminalCleanup(); + this.logger.debug( + `[fleet] Failed to cleanup workspace runtime ${runtime.workspace.id}: ${formatErrorMessage( + error + )}` + ); + throw error; + } + }) + ); + const cleanupFailures = cleanupResults.flatMap((result) => + result.status === 'rejected' ? [result.reason] : [] + ); + if (cleanupFailures.length > 0) + throw new AggregateError(cleanupFailures, 'Workspace cleanup failed'); await this.workspaceWatchCoordinator.dispose(); await this.cloudPort.dispose(); diff --git a/apps/cli/src/lib/lody.ts b/apps/cli/src/lib/lody.ts index aa93e43d4..2425c60c4 100644 --- a/apps/cli/src/lib/lody.ts +++ b/apps/cli/src/lib/lody.ts @@ -294,6 +294,10 @@ export class Lody { ); } + async forceTerminateSessions(): Promise { + await this.runtime.forceTerminateSessions(); + } + cleanup = async () => { this.cleanedUp = true; if (this.builtinAgentConfigRetryTimer) { diff --git a/apps/cli/src/lib/machine-runtime.ts b/apps/cli/src/lib/machine-runtime.ts index 626ccec08..1770d83da 100644 --- a/apps/cli/src/lib/machine-runtime.ts +++ b/apps/cli/src/lib/machine-runtime.ts @@ -231,6 +231,13 @@ export class MachineRuntime { }); } + async forceTerminateSessions(): Promise { + this.gcManager?.stop(); + this.messageProcessor.stop(); + this.handler?.cancelPendingPermissionRequests(); + await this.sessionManager?.forceTerminateSessions(); + } + async cleanup(): Promise { this.options.workspaceDocument.clearMachineMonitorProvider(); this.resourceMonitor = null; diff --git a/apps/cli/src/session/AGENTS.md b/apps/cli/src/session/AGENTS.md index 86f5b7496..0cf175627 100644 --- a/apps/cli/src/session/AGENTS.md +++ b/apps/cli/src/session/AGENTS.md @@ -246,6 +246,14 @@ the frozen identity. Never fall back to the Session owner when the driving Turn session/preparation producers but deliberately leaves the document manager and credentials alive so MessageHandler can flush final ACP/Code Collab evidence; the later plain `cleanUp()` closes shared resources. Never restore document teardown ahead of session termination. + Failed process/terminal cleanup retains the Session for retry and blocks shared-resource + teardown; a root exit is not successful cleanup. Remove only the exact successfully + terminated instance. Terminal release reports observed exit only, retains failed releases, + and Session bounds terminal disposal before continuing process termination. + Concurrent termination shares one attempt; force requests upgrade it and bypass graceful + drains. Shutdown closes session admission before awaiting work. The process boundary + reserves a separate forced-cleanup deadline and sweeps retained workspace/preparation + owners concurrently before exiting. - `session-preparation-service.ts` — process-local speculative ACP lease/state owner. Peek/claim are synchronous published-resource snapshots and must never delay cold fallback; peek never transfers ownership. A prepared resource may reuse its open diff --git a/apps/cli/src/session/session-manager.test.ts b/apps/cli/src/session/session-manager.test.ts index e656e1b3d..485ac56f8 100644 --- a/apps/cli/src/session/session-manager.test.ts +++ b/apps/cli/src/session/session-manager.test.ts @@ -1,4 +1,5 @@ import os from 'node:os'; +import { EventEmitter } from 'node:events'; import path from 'node:path'; import { execFileSync } from 'node:child_process'; import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; @@ -186,6 +187,168 @@ const createSessionInner = async ( ).createSessionInner(config, undefined, preparedWorktree); describe('SessionManager cleanup phases', () => { + it('closes admission before waiting for preparation cleanup', async () => { + const workspaceDocument = createWorkspaceDocument(new Map()); + const manager = new SessionManager( + createLogger(), + 'token', + 'machine-1' as MachineId, + 'workspace-1' as WorkspaceId, + workspaceDocument, + { + sessionSandboxFactory: async () => createNoopSessionSandbox(), + cloudPort: createTestCloudPort(), + } + ); + const internals = manager as unknown as { preparationService: { disposeAll(): Promise } }; + const held = deferred(); + vi.spyOn(internals.preparationService, 'disposeAll').mockReturnValue(held.promise); + const cleanup = manager.cleanUp(); + await expect( + manager.createSession({ + sessionId: 'late-session' as SessionId, + assumeDocExisting: true, + } as SessionConfig) + ).rejects.toThrow('shutting down'); + held.resolve(); + await cleanup; + }); + const cleanupFixture = () => { + const workspaceDocument = createWorkspaceDocument(new Map()); + const manager = new SessionManager( + createLogger(), + 'token', + 'machine-1' as MachineId, + 'workspace-1' as WorkspaceId, + workspaceDocument, + { + sessionSandboxFactory: async () => createNoopSessionSandbox(), + cloudPort: createTestCloudPort(), + } + ); + const internals = manager as unknown as { sessions: Map }; + return { manager, workspaceDocument, sessions: internals.sessions }; + }; + + it('retains failed cleanup ownership, waits for all attempts, and retries before closing documents', async () => { + const { manager, workspaceDocument, sessions } = cleanupFixture(); + const successId = 'cleanup-success' as SessionId; + const failureId = 'cleanup-failure' as SessionId; + const completed = deferred(); + const failure = new Error('tree failure'); + const failureEvents = new EventEmitter(); + const retry = vi + .fn<() => Promise>() + .mockImplementationOnce(async () => { + failureEvents.emit('exit', { sessionId: failureId, exitCode: 0 }); + throw failure; + }) + .mockResolvedValue(undefined); + sessions.set(successId, { + sessionId: successId, + terminate: () => completed.promise, + } as unknown as ISession); + const failedSession = Object.assign(failureEvents, { + sessionId: failureId, + terminate: retry, + }) as unknown as ISession; + ( + manager as unknown as { registerSessionEvents(session: ISession): void } + ).registerSessionEvents(failedSession); + sessions.set(failureId, failedSession); + let settled = false; + const cleanup = manager.cleanUp().finally(() => { + settled = true; + }); + const assertion = expect(cleanup).rejects.toMatchObject({ errors: [failure] }); + await Promise.resolve(); + expect(settled).toBe(false); + completed.resolve(); + await assertion; + expect(sessions.has(successId)).toBe(false); + expect(sessions.get(failureId)).toBe(failedSession); + expect(manager.getSession(failureId)).toBeNull(); + await expect( + manager.createSession({ sessionId: failureId, assumeDocExisting: true } as SessionConfig) + ).rejects.toThrow('shutting down'); + failureEvents.emit('exit', { sessionId: failureId, exitCode: 0 }); + expect(sessions.get(failureId)).toBe(failedSession); + expect(workspaceDocument.cleanUp).not.toHaveBeenCalled(); + await manager.cleanUp(); + expect(sessions.size).toBe(0); + expect(workspaceDocument.cleanUp).toHaveBeenCalledOnce(); + }); + + it('force sweeps registered and preparing sessions without waiting for preparation disposal', async () => { + const { manager, sessions } = cleanupFixture(); + const internals = manager as unknown as { + preparationSessions: Map; + preparationService: { disposeAll(): Promise }; + }; + const preparationDisposal = deferred(); + vi.spyOn(internals.preparationService, 'disposeAll').mockReturnValue( + preparationDisposal.promise + ); + const terminated: string[] = []; + const residentId = 'force-resident' as SessionId; + const preparingId = 'force-preparing' as SessionId; + sessions.set(residentId, { + sessionId: residentId, + terminate: async () => { + terminated.push('resident'); + }, + } as unknown as ISession); + internals.preparationSessions.set(preparingId, { + sessionId: preparingId, + terminate: async () => { + terminated.push('preparing'); + }, + } as unknown as ISession); + await manager.forceTerminateSessions(); + expect(terminated.sort()).toEqual(['preparing', 'resident']); + expect(sessions.size).toBe(0); + expect(internals.preparationSessions.size).toBe(0); + await expect( + manager.createSession({ + sessionId: 'new' as SessionId, + assumeDocExisting: true, + } as SessionConfig) + ).rejects.toThrow('shutting down'); + preparationDisposal.resolve(); + }); + + it('preserves replacement and newly registered sessions during an in-flight cleanup', async () => { + const { manager, sessions } = cleanupFixture(); + const sessionId = 'cleanup-replaced' as SessionId; + const newId = 'cleanup-new' as SessionId; + const started = deferred(); + const completed = deferred(); + const oldEvents = new EventEmitter(); + const oldSession = Object.assign(oldEvents, { + sessionId, + terminate: () => { + started.resolve(); + return completed.promise; + }, + }) as unknown as ISession; + sessions.set(sessionId, oldSession); + ( + manager as unknown as { registerSessionEvents(session: ISession): void } + ).registerSessionEvents(oldSession); + const cleanup = manager.cleanUp({ keepWorkspaceDocumentOpen: true }); + await started.promise; + const replacement = { sessionId } as unknown as ISession; + const newSession = { sessionId: newId } as unknown as ISession; + sessions.set(sessionId, replacement); + sessions.set(newId, newSession); + oldEvents.emit('exit', { sessionId, exitCode: 0 }); + oldEvents.emit('terminated', { sessionId, exitCode: 0 }); + completed.resolve(); + await cleanup; + expect(sessions.get(sessionId)).toBe(replacement); + expect(sessions.get(newId)).toBe(newSession); + }); + it('stops session producers before closing the workspace document', async () => { const workspaceDocument = createWorkspaceDocument(new Map()); const manager = new SessionManager( diff --git a/apps/cli/src/session/session-manager.ts b/apps/cli/src/session/session-manager.ts index 89f47335c..51d8379fb 100644 --- a/apps/cli/src/session/session-manager.ts +++ b/apps/cli/src/session/session-manager.ts @@ -451,6 +451,16 @@ export class SessionManager extends EventEmitter { private githubTokenManager: CloudGithubTokenManager | null = null; private gitCredentialBroker: GitCredentialBroker | null = null; private readonly sessions = new Map(); + private readonly cleanupOwnedSessions = new WeakSet(); + private shuttingDown = false; + + private assertSessionAdmission(sessionId?: SessionId): void { + if (this.shuttingDown) throw new Error('Session manager is shutting down'); + const resident = sessionId ? this.sessions.get(sessionId) : undefined; + if (resident && this.cleanupOwnedSessions.has(resident)) { + throw new Error('Session cleanup must complete before replacement'); + } + } private readonly pendingSessionCreates = new Map>(); private readonly pendingTerminationPromises = new Map>(); private readonly preparationSessions = new Map(); @@ -612,6 +622,7 @@ export class SessionManager extends EventEmitter { } async createSession(config: SessionConfig, agentStart?: AgentStartConfig): Promise { + this.assertSessionAdmission(config.sessionId); if (!config.assumeDocExisting) { const sessionId = await this.workspaceDocument.createSession( config.machineId, @@ -623,6 +634,7 @@ export class SessionManager extends EventEmitter { } const sessionId = config.sessionId; + this.assertSessionAdmission(sessionId); if (!sessionId) { throw new Error('SessionId is required to create a session'); } @@ -1039,6 +1051,7 @@ export class SessionManager extends EventEmitter { session = new Session(config, this.logger, provisionalWorkdir, sandbox); sandbox = null; session.ghTokenInjected = ghTokenInjected; + this.assertSessionAdmission(sessionId); this.preparationSessions.set(sessionId, session); await this.rebalanceSessionSandboxes(); signal.throwIfAborted(); @@ -2043,6 +2056,7 @@ export class SessionManager extends EventEmitter { let session: Session; if (preparedSession) { + this.assertSessionAdmission(config.sessionId); session = preparedSession; if (workdir) { session.setWorkdir(workdir); @@ -2050,10 +2064,17 @@ export class SessionManager extends EventEmitter { this.preparationSessions.delete(config.sessionId!); } else { const sandbox = await this.sessionSandboxFactory(config.sessionId!); + try { + this.assertSessionAdmission(config.sessionId); + } catch (error) { + await sandbox.cleanup(); + throw error; + } this.logger.debug(`[${config.sessionId}] Session sandbox: ${sandbox.description}`); session = new Session(config, this.logger, workdir, sandbox); } this.registerSessionEvents(session); + this.assertSessionAdmission(config.sessionId); this.sessions.set(config.sessionId!, session); await this.rebalanceSessionSandboxes(); return session; @@ -2066,11 +2087,41 @@ export class SessionManager extends EventEmitter { return; } + this.cleanupOwnedSessions.add(session); await session.terminate(force); + this.cleanupOwnedSessions.delete(session); this.logger.debug(`[${sessionId}] Session terminated`); } + async forceTerminateSessions(): Promise { + this.shuttingDown = true; + this.preparationRecoveryGeneration += 1; + this.detachPreparationRecovery?.(); + this.detachPreparationRecovery = null; + // Expire preparation leases synchronously, without waiting ahead of process kills. + void this.preparationService.disposeAll().catch((error: unknown) => { + this.logger.error(`Preparation cleanup failed: ${formatErrorMessage(error)}`); + }); + const targets = new Set([...this.sessions.values(), ...this.preparationSessions.values()]); + const results = await Promise.allSettled( + [...targets].map(async (session) => { + this.cleanupOwnedSessions.add(session); + await session.terminate(true); + if (this.sessions.get(session.sessionId) === session) + this.sessions.delete(session.sessionId); + if (this.preparationSessions.get(session.sessionId) === session) + this.preparationSessions.delete(session.sessionId); + this.cleanupOwnedSessions.delete(session); + }) + ); + const failures = results.flatMap((result) => + result.status === 'rejected' ? [result.reason] : [] + ); + if (failures.length) throw new AggregateError(failures, 'Forced session cleanup failed'); + } + async cleanUp(options: { keepWorkspaceDocumentOpen?: boolean } = {}) { + this.shuttingDown = true; this.preparationRecoveryGeneration += 1; this.detachPreparationRecovery?.(); this.detachPreparationRecovery = null; @@ -2096,18 +2147,20 @@ export class SessionManager extends EventEmitter { private async cleanupSessions(): Promise { this.logger.debug('Cleaning up all sessions...'); - const terminations = Array.from(this.sessions.values()).map((session) => - session - .terminate(true) - .catch((error: unknown) => - this.logger.error( - `[${session.sessionId}] Failed to terminate session: ${error instanceof Error ? error.message : 'Unknown error'}` - ) - ) + const results = await Promise.allSettled( + Array.from(this.sessions.entries()).map(async ([sessionId, session]) => { + this.cleanupOwnedSessions.add(session); + await session.terminate(true); + if (this.sessions.get(sessionId) === session) this.sessions.delete(sessionId); + this.cleanupOwnedSessions.delete(session); + }) ); - - await Promise.allSettled(terminations); - this.sessions.clear(); + const failures = results.flatMap((result) => + result.status === 'rejected' ? [result.reason] : [] + ); + if (failures.length > 0) { + throw new AggregateError(failures, 'Failed to terminate all sessions'); + } } hasSession(sessionId: SessionId): boolean { @@ -2115,7 +2168,8 @@ export class SessionManager extends EventEmitter { } getSession(sessionId: SessionId): ISession | null { - return this.sessions.get(sessionId) ?? null; + const session = this.sessions.get(sessionId); + return session && !this.cleanupOwnedSessions.has(session) ? session : null; } async resolveSessionWorkdir(sessionId: SessionId): Promise { @@ -2250,12 +2304,16 @@ export class SessionManager extends EventEmitter { }); session.on('exit', (event: SessionExitEvent) => { + if (this.sessions.get(event.sessionId) !== session) return; + if (this.cleanupOwnedSessions.has(session)) return; this.sessions.delete(event.sessionId); void this.rebalanceSessionSandboxes(); this.emit('exit', event); }); session.on('terminated', (event: SessionExitEvent) => { + if (this.sessions.get(event.sessionId) !== session) return; + this.cleanupOwnedSessions.delete(session); this.sessions.delete(event.sessionId); void this.rebalanceSessionSandboxes(); const terminatedEvent: SessionTerminatedEvent = { diff --git a/apps/cli/src/session/session-sandbox.ts b/apps/cli/src/session/session-sandbox.ts index c58cfa4b6..888a0b13c 100644 --- a/apps/cli/src/session/session-sandbox.ts +++ b/apps/cli/src/session/session-sandbox.ts @@ -8,6 +8,7 @@ import { type SessionId } from '@lody/shared'; import type { Logger } from '@/utils/logger'; import { formatErrorMessage } from '@/utils/format-error'; import { applyExecutionProcessResourceProfile } from '@/utils/process-resource-profile'; +import { terminateWindowsProcessTree } from '@/utils/windows-process-tree'; const DEFAULT_CGROUP_MOUNT = '/sys/fs/cgroup'; const DEFAULT_SESSION_PARENT = 'lody-sessions'; @@ -285,7 +286,7 @@ export function calculateAutomaticSessionSandboxLimits( class NoopSessionSandbox implements SessionSandbox { readonly enabled = false; - private readonly trackedProcesses = new Map(); + private readonly trackedProcesses = new Map(); constructor( private readonly deps: Pick< @@ -330,7 +331,7 @@ class NoopSessionSandbox implements SessionSandbox { async () => null, async (force) => { if (typeof child.pid === 'number' && child.pid > 0) { - await this.terminateProcessTree(child.pid, force, detached); + await this.terminateProcessTree(child, force, detached); return; } await terminateChildProcessDirectly(child, force); @@ -338,7 +339,7 @@ class NoopSessionSandbox implements SessionSandbox { { captureOutput, logger: this.logger } ); if (typeof child.pid === 'number' && child.pid > 0) { - this.trackedProcesses.set(child.pid, { detached }); + this.trackedProcesses.set(child.pid, { child, detached }); const trackedPid = child.pid; const cleanupTrackedProcess = () => { this.trackedProcesses.delete(trackedPid); @@ -352,9 +353,16 @@ class NoopSessionSandbox implements SessionSandbox { } async terminate(force: boolean = false): Promise { - for (const [pid, processInfo] of this.trackedProcesses.entries()) { - await this.terminateProcessTree(pid, force, processInfo.detached); + const failures: unknown[] = []; + for (const { child, detached } of this.trackedProcesses.values()) { + try { + await this.terminateProcessTree(child, force, detached); + } catch (error) { + failures.push(error); + } } + if (failures.length > 0) + throw new AggregateError(failures, 'Session process tree termination failed'); } async cleanup(): Promise { @@ -362,15 +370,17 @@ class NoopSessionSandbox implements SessionSandbox { } private async terminateProcessTree( - pid: number, + child: ChildProcess, force: boolean, detached: boolean ): Promise { if (this.deps.platform === 'win32') { - await this.runWindowsTaskkill(pid, force); + await terminateWindowsProcessTree(child, force, { spawnProcess: this.deps.spawnProcess }); return; } + const pid = child.pid; + if (pid === undefined) return; const signal = force ? 'SIGKILL' : 'SIGTERM'; const targetPid = detached ? -pid : pid; try { @@ -382,21 +392,6 @@ class NoopSessionSandbox implements SessionSandbox { throw error; } } - - private async runWindowsTaskkill(pid: number, force: boolean): Promise { - await new Promise((resolve, reject) => { - const child = this.deps.spawnProcess( - 'taskkill', - ['/PID', String(pid), '/T', ...(force ? ['/F'] : [])], - { - stdio: 'ignore', - windowsHide: true, - } - ); - child.once('error', reject); - child.once('close', () => resolve()); - }); - } } class LinuxCgroupSessionSandbox implements SessionSandbox { diff --git a/apps/cli/src/session/session.ts b/apps/cli/src/session/session.ts index 22cf09fe6..6b412edc2 100644 --- a/apps/cli/src/session/session.ts +++ b/apps/cli/src/session/session.ts @@ -110,6 +110,11 @@ export class Session extends EventEmitter implements ISession { private readonly startedAtMs = getServerNow(); private activeProcess: SessionProcessHandle | null = null; private agentProcess: SessionProcessHandle | null = null; + private terminalDisposal: { + manager: TerminalManager; + sessionId: string; + promise: Promise; + } | null = null; private readonly sandbox: SessionSandbox; private gitIdentity: { id: string; name: string; email: string }; public agentClient: AgentClient | null = null; @@ -237,14 +242,53 @@ export class Session extends EventEmitter implements ISession { return execPromise; } - async terminate(force: boolean = false): Promise { + private terminationPromise: Promise | null = null; + private terminationForceRequested = false; + private forceTerminationSignal: Promise = Promise.resolve(); + private requestForceTermination: (() => void) | null = null; + + terminate(force: boolean = false): Promise { + if (this.terminationPromise) { + if (force) { + this.terminationForceRequested = true; + this.requestForceTermination?.(); + } + return this.terminationPromise; + } + if (this.status === 'terminated') return Promise.resolve(); + this.terminationForceRequested = force; + this.forceTerminationSignal = new Promise((resolve) => { + this.requestForceTermination = resolve; + }); + if (force) this.requestForceTermination?.(); + this.status = 'stopping'; + const termination = Promise.resolve().then(() => this.terminateOnce()); + this.terminationPromise = termination; + void termination.then( + () => {}, + () => { + if (this.terminationPromise === termination) this.terminationPromise = null; + } + ); + return termination; + } + + private async terminateOnce(): Promise { + const force = this.terminationForceRequested; this.logger.debug(`[${this.sessionId}] Terminating session${force ? ' (force)' : ''}`); this.status = 'stopping'; + const failures: unknown[] = []; + // Exit callbacks may release these references during terminal/ACP disposal. + const activeProcess = this.activeProcess; + const agentProcess = this.agentProcess; + let terminalDisposal: Promise | undefined; if (this.acpSessionId && this.terminalManager.disposeAll) { try { - await this.terminalManager.disposeAll(this.acpSessionId); + terminalDisposal = this.disposeTerminalsWithDeadline(this.acpSessionId); + await Promise.race([terminalDisposal, this.forceTerminationSignal]); } catch (error) { + failures.push(error); this.logger.debug( `[${ this.sessionId @@ -253,9 +297,12 @@ export class Session extends EventEmitter implements ISession { } } - if (!force && this.acpSessionId && this.agentClient?.isCreated()) { + if (!this.terminationForceRequested && this.acpSessionId && this.agentClient?.isCreated()) { try { - await this.agentClient.closeSession(this.acpSessionId); + await Promise.race([ + this.agentClient.closeSession(this.acpSessionId), + this.forceTerminationSignal, + ]); } catch (error) { this.logger.debug( `[${this.sessionId}] Failed to close ACP session during terminate: ${formatErrorMessage( @@ -265,32 +312,57 @@ export class Session extends EventEmitter implements ISession { } } - // Capture references before any async work, since onExit handlers may null them out - const activeProcess = this.activeProcess; - const agentProcess = this.agentProcess; - // Kill both processes and wait for them to actually exit before proceeding. // This prevents OS-level process leaks where SIGTERM is sent but the process // outlives this function (and all tracking of it). - await Promise.all([ - this.killAndWait(activeProcess, force), - this.killAndWait(agentProcess, force), + const processResults = await Promise.allSettled([ + this.killAndWait(activeProcess, this.terminationForceRequested), + this.killAndWait(agentProcess, this.terminationForceRequested), ]); + for (const result of processResults) { + if (result.status === 'rejected') failures.push(result.reason); + } + if (terminalDisposal && this.terminationForceRequested) { + let timer: ReturnType | undefined; + try { + await Promise.race([ + terminalDisposal, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error('Terminal disposal incomplete after forced cleanup')), + 5_000 + ); + }), + ]); + } catch (error) { + failures.push(error); + } finally { + clearTimeout(timer); + } + } try { - await this.sandbox.terminate(force); + await this.sandbox.terminate(this.terminationForceRequested); } catch (error) { + failures.push(error); this.logger.debug( `[${this.sessionId}] Failed to terminate sandbox process tree: ${formatErrorMessage(error)}` ); } + // Preserve ownership and stopping status when cleanup cannot be confirmed. + // A later terminate call can retry; do not discard the sandbox's tracked roots. + if (failures.length > 0) { + throw new AggregateError(failures, 'Session process termination failed'); + } + try { await this.sandbox.cleanup(); } catch (error) { this.logger.debug( `[${this.sessionId}] Failed to clean up sandbox state: ${formatErrorMessage(error)}` ); + throw error; } this.activeProcess = null; @@ -308,63 +380,106 @@ export class Session extends EventEmitter implements ISession { this.emit('terminated', event); } + private async disposeTerminalsWithDeadline(sessionId: string): Promise { + const manager = this.terminalManager; + let disposal = this.terminalDisposal; + if (!disposal || disposal.manager !== manager || disposal.sessionId !== sessionId) { + const promise = Promise.resolve().then(() => manager.disposeAll?.(sessionId)); + disposal = { manager, sessionId, promise }; + this.terminalDisposal = disposal; + const clear = () => { + if (this.terminalDisposal === disposal) this.terminalDisposal = null; + }; + void promise.then(clear, clear); + } + // Shell terminals have bounded graceful/forced phases. This outer bound also + // protects shutdown from other TerminalManager implementations that hang. + let timer: ReturnType | undefined; + try { + await Promise.race([ + disposal.promise, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error('Terminal disposal timed out after 30000ms')), + 30_000 + ); + void this.forceTerminationSignal.then(() => clearTimeout(timer)); + }), + ]); + } finally { + clearTimeout(timer); + } + } + /** * Kill a process and wait for it to actually exit. * - * With force=false: sends SIGTERM, waits up to SIGTERM_GRACE_MS, then + * With force=false: requests graceful termination, waits up to five seconds, then * escalates to SIGKILL if the process hasn't exited. * With force=true: sends SIGKILL directly. * - * Always awaits the actual OS process exit before returning, so callers can - * be certain no orphaned processes remain. + * Rejects if the direct process has not exited five seconds after forced + * termination. Direct-process exit does not prove descendant cleanup. */ private async killAndWait(proc: SessionProcessHandle | null, force: boolean): Promise { if (!proc?.child) return; - const child = proc.child; - // Already exited — nothing to do. - // Note: child.killed only means a signal was *sent*, not that the process - // exited. Only exitCode !== null proves the process has actually terminated. - if (child.exitCode !== null) return; - - const waitForExit = (): Promise => - new Promise((resolve) => { - const unsubscribe = proc.onExit(() => { + const hasExited = () => child.exitCode != null || child.signalCode != null; + if (hasExited()) return; + + const EXIT_TIMEOUT_MS = 5_000; + const waitForExit = (interruptible = false): Promise => { + if (hasExited()) return Promise.resolve(true); + return new Promise((resolve) => { + let settled = false; + let unsubscribe = () => {}; + const finish = (exited: boolean) => { + if (settled) return; + settled = true; + clearTimeout(timer); unsubscribe(); - resolve(); - }); - // Guard: if the process exited between the check above and - // registering the listener, resolve immediately. - if (child.exitCode !== null) { - unsubscribe(); - resolve(); - } + resolve(exited); + }; + const timer = setTimeout(() => finish(hasExited()), EXIT_TIMEOUT_MS); + if (interruptible) void this.forceTerminationSignal.then(() => finish(hasExited())); + unsubscribe = proc.onExit(() => finish(true)); + // onExit may replay an already observed exit synchronously. + if (settled) unsubscribe(); + else if (hasExited()) finish(true); }); + }; - if (force) { + try { + await proc.terminate(force); + } catch (error) { + // A refused graceful request can be retried forcibly only while the + // original root is still live. Its exit cannot erase a failed tree kill. + if (force || hasExited()) throw error; await proc.terminate(true); - await waitForExit(); - return; + if (await waitForExit()) return; + throw new Error( + `Session process did not exit within ${EXIT_TIMEOUT_MS}ms after forced termination`, + { cause: error } + ); } - - // Graceful path: SIGTERM → wait → SIGKILL fallback - const SIGTERM_GRACE_MS = 5_000; - await proc.terminate(false); - - const outcome = await Promise.race([ - waitForExit().then(() => 'exited' as const), - new Promise<'timeout'>((resolve) => setTimeout(() => resolve('timeout'), SIGTERM_GRACE_MS)), - ]); - - if (outcome === 'timeout' && child.exitCode === null) { + if ( + await Promise.race([ + waitForExit(!force), + ...(force ? [] : [this.forceTerminationSignal.then(() => false)]), + ]) + ) + return; + if (!force) { this.logger.debug( - `[${this.sessionId}] Process did not exit within ${SIGTERM_GRACE_MS}ms of SIGTERM; escalating to SIGKILL` + `[${this.sessionId}] Process did not exit within ${EXIT_TIMEOUT_MS}ms of SIGTERM; escalating to SIGKILL` ); await proc.terminate(true); - await waitForExit(); + if (await waitForExit()) return; } + throw new Error( + `Session process did not exit within ${EXIT_TIMEOUT_MS}ms after forced termination` + ); } - /** * Update git identity for commits made in this session. * This should be called when a new user sends a chat request to an existing session. @@ -478,8 +593,15 @@ export class Session extends EventEmitter implements ISession { } async createAgent(callbacks: CreateAgentConfig): Promise { + const assertRunning = () => { + if (this.status === 'stopping' || this.status === 'terminated') { + throw new Error('Cannot launch agent while session is stopping'); + } + }; + assertRunning(); this.acpCapabilitySourceVersion = callbacks.capabilitySourceVersion ?? null; const loginShellEnv = await getLoginShellEnv(); + assertRunning(); callbacks.abortSignal?.throwIfAborted(); const env = withLodyNpmCacheForNpx( callbacks.command, @@ -532,6 +654,7 @@ export class Session extends EventEmitter implements ISession { let agentProcessHandle: SessionProcessHandle; try { callbacks.abortSignal?.throwIfAborted(); + assertRunning(); agentProcessHandle = await this.sandbox.spawn(callbacks.command, callbacks.args ?? [], { cwd: this.getWorkdir(), env, @@ -542,6 +665,10 @@ export class Session extends EventEmitter implements ISession { throw error; } const agentProcess = agentProcessHandle.child; + if (this.status === 'stopping' || this.status === 'terminated') { + await this.killAndWait(agentProcessHandle, true); + throw new Error('Agent launch cancelled by session shutdown'); + } this.agentProcess = agentProcessHandle; lastAgentProcessHandle = agentProcessHandle; diff --git a/apps/cli/src/session/terminal-manager.ts b/apps/cli/src/session/terminal-manager.ts index 8ccf1c0a1..1dab10851 100644 --- a/apps/cli/src/session/terminal-manager.ts +++ b/apps/cli/src/session/terminal-manager.ts @@ -41,6 +41,8 @@ interface TerminalState { truncated: boolean; exitStatus: TerminalExitStatus | null; waiters: Array<(status: TerminalExitStatus) => void>; + releasing?: Promise; + disposed: boolean; } interface TerminalHooks { @@ -91,6 +93,7 @@ abstract class BaseTerminalManager implements TerminalManager { truncated: false, exitStatus: null, waiters: [], + disposed: false, }; const hooks: TerminalHooks = { @@ -130,20 +133,57 @@ abstract class BaseTerminalManager implements TerminalManager { async releaseTerminal(acpSessionId: string, terminalId: string): Promise { const state = this.getTerminal(acpSessionId, terminalId); + if (state.releasing) return state.releasing; + const releasing = this.releaseState(state); + state.releasing = releasing; try { - await this.killHandle(state); - } catch (error) { - this.logger.debug( - `[${this.sessionLabel}] Failed to kill terminal ${terminalId} on release: ${error}` - ); + await releasing; + } finally { + state.releasing = undefined; } + } + + private async releaseState(state: TerminalState): Promise { if (!state.exitStatus) { - state.exitStatus = { exitCode: null, signal: 'SIGTERM' }; - this.resolveWaiters(state); + let forced = false; + try { + await this.killHandle(state); + } catch (error) { + if (!this.isHandleLive(state)) throw error; + await this.killHandle(state, true); + forced = true; + } + if (!(await this.waitForObservedExit(state))) { + if (!forced && this.isHandleLive(state)) { + await this.killHandle(state, true); + if (!(await this.waitForObservedExit(state))) { + throw new Error('Terminal process did not report exit after forced termination'); + } + } else { + throw new Error('Terminal process did not report exit after termination'); + } + } } await this.disposeHandle(state); - this.terminals.delete(terminalId); - this.logger.debug(`[${this.sessionLabel}] Terminal ${terminalId} released`); + state.disposed = true; + this.terminals.delete(state.id); + this.logger.debug(`[${this.sessionLabel}] Terminal ${state.id} released`); + } + + private waitForObservedExit(state: TerminalState): Promise { + if (state.exitStatus) return Promise.resolve(true); + return new Promise((resolve) => { + const waiter = () => { + clearTimeout(timer); + resolve(true); + }; + const timer = setTimeout(() => { + const index = state.waiters.indexOf(waiter); + if (index >= 0) state.waiters.splice(index, 1); + resolve(false); + }, 5_000); + state.waiters.push(waiter); + }); } async waitForTerminalExit(acpSessionId: string, terminalId: string): Promise { @@ -168,11 +208,15 @@ abstract class BaseTerminalManager implements TerminalManager { if (terminalIds.length === 0) { return; } - await Promise.allSettled( + const results = await Promise.allSettled( terminalIds.map(async (terminalId) => { await this.releaseTerminal(acpSessionId, terminalId); }) ); + const failures = results.flatMap((result) => + result.status === 'rejected' ? [result.reason] : [] + ); + if (failures.length > 0) throw new AggregateError(failures, 'Terminal disposal failed'); } protected abstract startProcess( @@ -186,12 +230,14 @@ abstract class BaseTerminalManager implements TerminalManager { hooks: TerminalHooks ): Promise; - protected abstract killHandle(state: TerminalState): Promise; + protected abstract killHandle(state: TerminalState, force?: boolean): Promise; + protected abstract isHandleLive(state: TerminalState): boolean; protected abstract disposeHandle(state: TerminalState): Promise; private resolveWaiters(state: TerminalState) { - const exitStatus = state.exitStatus ?? { exitCode: null, signal: null }; + const exitStatus = state.exitStatus; + if (!exitStatus) return; while (state.waiters.length) { const waiter = state.waiters.shift(); if (waiter) { @@ -224,7 +270,7 @@ abstract class BaseTerminalManager implements TerminalManager { exitCode: number | null, signal: NodeJS.Signals | null ) { - if (!this.terminals.has(state.id)) { + if (state.disposed || state.exitStatus) { return; } state.exitStatus = { @@ -303,6 +349,8 @@ export class ShellTerminalManager const stdoutListener = (chunk: Buffer) => hooks.onData(chunk); const stderrListener = (chunk: Buffer) => hooks.onData(chunk); const closeListener = (code: number | null, signal: NodeJS.Signals | null) => { + // OS close is authoritative even if resource accounting is slow or stuck. + hooks.onExit(code, signal); void processHandle .inspectExit(code, signal) .then((violation) => { @@ -316,9 +364,6 @@ export class ShellTerminalManager }) .catch((error: unknown) => { hooks.onError?.(error instanceof Error ? error : new Error(String(error))); - }) - .finally(() => { - hooks.onExit(code, signal); }); }; const errorListener = (error: Error) => hooks.onError?.(error); @@ -339,9 +384,17 @@ export class ShellTerminalManager }; } - protected async killHandle(state: TerminalState): Promise { + protected isHandleLive(state: TerminalState): boolean { + const child = state.handle.processHandle.child; + return child.exitCode == null && child.signalCode == null; + } + + protected async killHandle( + state: TerminalState, + force = false + ): Promise { if (!state.exitStatus) { - await state.handle.processHandle.terminate(false); + await state.handle.processHandle.terminate(force); } } diff --git a/apps/cli/src/utils/windows-process-tree.real-process.test.ts b/apps/cli/src/utils/windows-process-tree.real-process.test.ts new file mode 100644 index 000000000..4161c6d49 --- /dev/null +++ b/apps/cli/src/utils/windows-process-tree.real-process.test.ts @@ -0,0 +1,123 @@ +import { spawn, type ChildProcess } from 'node:child_process'; +import { once } from 'node:events'; +import { createInterface } from 'node:readline'; +import { describe, expect, it } from 'vitest'; +import { z } from 'zod'; +import { terminateWindowsProcessTree } from './windows-process-tree'; + +// IPC acknowledges the entire tree. Disconnecting IPC deliberately does NOT +// terminate descendants. Detached children prevent Windows parent-lifetime +// cleanup from masking a wrapper-only kill; taskkill /T must reach both levels. +const fixture = String.raw` + function run(depth) { + const { spawn } = require('node:child_process'); + setTimeout(() => process.exit(90), 60000); // Failure-only orphan watchdog. + if (depth === 0) { + process.send([process.pid]); + return; + } + const child = spawn(process.execPath, ['-e', '(' + run.toString() + ')(' + (depth - 1) + ')'], { + stdio: ['ignore', 'ignore', 'ignore', 'ipc'], windowsHide: true, detached: true, + }); + child.once('message', (pids) => process.send([process.pid, ...pids])); + child.once('error', () => process.exit(91)); + } + run(2); +`; + +function exitResult(child: ChildProcess): Promise { + return new Promise((resolve, reject) => { + child.once('error', reject); + child.once('close', resolve); + }); +} + +describe.skipIf(process.platform !== 'win32')('Windows process tree integration', () => { + it('terminates an owned wrapper, child, and grandchild, verified by OS handles', async () => { + const root = spawn(process.execPath, ['-e', fixture], { + stdio: ['ignore', 'ignore', 'ignore', 'ipc'], + windowsHide: true, + }); + const rootExit = exitResult(root); + // Attach rejection handlers immediately, including on early readiness failure. + void rootExit.catch(() => {}); + let observer: ReturnType | undefined; + let observerExit: Promise | undefined; + let lines: ReturnType | undefined; + try { + const [message] = await once(root, 'message', { signal: AbortSignal.timeout(10_000) }); + const pids = z.array(z.number().int().positive()).length(3).parse(message); + expect(pids[0]).toBe(root.pid); + expect(new Set(pids).size).toBe(3); + + // Capture handles BEFORE termination. WaitForExit observes the original + // objects even if Windows reuses a PID. Finally cleans up those same handles + // if an assertion fails; it never searches for or kills arbitrary processes. + const script = String.raw` + $ErrorActionPreference = 'Stop' + $owned = @() + try { + foreach ($processId in @(${pids.join(',')})) { + $item = [System.Diagnostics.Process]::GetProcessById($processId) + $null = $item.Handle + $owned += $item + if ($item.HasExited) { throw 'Fixture exited before verification' } + } + [Console]::WriteLine('handles-ready') + $command = [Console]::In.ReadLineAsync() + if (-not $command.Wait(10000) -or $command.Result -ne 'verify') { + throw 'Verification handshake failed' + } + foreach ($item in $owned) { + if (-not $item.WaitForExit(5000)) { throw 'Owned descendant remained alive' } + if ($item.ExitCode -eq 90) { throw 'Fixture watchdog fired' } + } + [Console]::WriteLine('all-three-exited') + } finally { + [Array]::Reverse($owned) + foreach ($item in $owned) { + try { + if (-not $item.HasExited) { $item.Kill() } + if (-not $item.WaitForExit(5000)) { throw 'Fixture cleanup failed' } + } finally { $item.Dispose() } + } + } + `; + observer = spawn('powershell.exe', ['-NoProfile', '-NonInteractive', '-Command', script], { + stdio: ['pipe', 'pipe', 'pipe'], + windowsHide: true, + }); + observerExit = exitResult(observer); + void observerExit.catch(() => {}); + if (!observer.stdout || !observer.stdin) throw new Error('Observer pipes unavailable'); + let output = ''; + observer.stdout.on('data', (chunk: Buffer) => { + output += chunk.toString(); + }); + let errors = ''; + observer.stderr?.on('data', (chunk: Buffer) => { + errors += chunk.toString(); + }); + lines = createInterface({ input: observer.stdout }); + const [ready] = await once(lines, 'line', { signal: AbortSignal.timeout(10_000) }); + expect(ready).toBe('handles-ready'); + await terminateWindowsProcessTree(root, true); + observer.stdin.end('verify\n'); + expect(await observerExit, errors).toBe(0); + expect(output).toContain('all-three-exited'); + await rootExit; + } finally { + // Closing stdin asks the observer to clean captured handles on every failure. + observer?.stdin?.end(); + try { + if (observerExit) await observerExit; + } finally { + lines?.close(); + if (root.exitCode === null && root.signalCode === null) { + await terminateWindowsProcessTree(root, true); + } + await rootExit; + } + } + }, 40_000); +}); diff --git a/apps/cli/src/utils/windows-process-tree.test.ts b/apps/cli/src/utils/windows-process-tree.test.ts new file mode 100644 index 000000000..0486beab0 --- /dev/null +++ b/apps/cli/src/utils/windows-process-tree.test.ts @@ -0,0 +1,119 @@ +import { EventEmitter } from 'events'; +import type { ChildProcess } from 'child_process'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { terminateWindowsProcessTree } from './windows-process-tree'; + +function processFixture(pid = 42): ChildProcess { + const child = new EventEmitter() as ChildProcess; + child.pid = pid; + child.exitCode = null; + child.signalCode = null; + child.kill = vi.fn(() => true); + return child; +} + +afterEach(() => vi.useRealTimers()); + +describe('terminateWindowsProcessTree', () => { + it.each([false, true])('requests recursive hidden termination (force=%s)', async (force) => { + const root = processFixture(); + const helper = processFixture(43); + const spawnProcess = vi.fn(() => helper); + const result = terminateWindowsProcessTree(root, force, { spawnProcess }); + expect(spawnProcess).toHaveBeenCalledWith( + 'taskkill', + ['/PID', '42', '/T', ...(force ? ['/F'] : [])], + { stdio: 'ignore', windowsHide: true } + ); + helper.emit('close', 0, null); + await expect(result).resolves.toBeUndefined(); + expect(helper.eventNames()).toEqual([]); + }); + + it.each(['exitCode', 'signalCode'] as const)( + 'does not target an exited root (%s)', + async (field) => { + const root = processFixture(); + if (field === 'exitCode') root.exitCode = 0; + else root.signalCode = 'SIGTERM'; + const spawnProcess = vi.fn(); + await terminateWindowsProcessTree(root, true, { spawnProcess }); + expect(spawnProcess).not.toHaveBeenCalled(); + } + ); + + it.each([ + [1, null], + [null, 'SIGTERM'], + [0, 'SIGTERM'], + ] as const)( + 'rejects unsuccessful helper exit %s/%s even if the root exits', + async (code, signal) => { + const root = processFixture(); + const helper = processFixture(43); + const result = terminateWindowsProcessTree(root, true, { spawnProcess: () => helper }); + root.exitCode = 0; + helper.emit('close', code, signal); + await expect(result).rejects.toThrow('did not succeed'); + expect(helper.eventNames()).toEqual([]); + } + ); + + it('redacts helper errors and synchronous spawn errors', async () => { + const helper = processFixture(); + const result = terminateWindowsProcessTree(processFixture(), true, { + spawnProcess: () => helper, + }); + helper.emit('error', new Error('secret command data')); + await expect(result).rejects.toThrow('helper failed'); + expect(helper.eventNames()).toEqual([]); + await expect( + terminateWindowsProcessTree(processFixture(), true, { + spawnProcess: () => { + throw new Error('secret command data'); + }, + }) + ).rejects.toThrow('Could not start Windows process tree termination'); + }); + + it('bounds a hung helper and kills only the helper, preserving timeout after synchronous close', async () => { + vi.useFakeTimers(); + const root = processFixture(); + const helper = processFixture(43); + helper.kill = vi.fn(() => { + helper.emit('close', 0, null); + return true; + }); + const result = terminateWindowsProcessTree(root, true, { + spawnProcess: () => helper, + timeoutMs: 25, + }); + const rejected = expect(result).rejects.toThrow('timed out'); + await vi.advanceTimersByTimeAsync(25); + await rejected; + expect(root.kill).not.toHaveBeenCalled(); + expect(helper.kill).toHaveBeenCalledWith('SIGKILL'); + expect(helper.eventNames()).toEqual([]); + expect(vi.getTimerCount()).toBe(0); + }); + + it('clears the deadline on successful completion', async () => { + vi.useFakeTimers(); + const helper = processFixture(); + const result = terminateWindowsProcessTree(processFixture(), false, { + spawnProcess: () => helper, + }); + helper.emit('close', 0, null); + await result; + expect(vi.getTimerCount()).toBe(0); + expect(helper.kill).not.toHaveBeenCalled(); + }); + + it('rejects invalid ownership before spawning', async () => { + const spawnProcess = vi.fn(); + await expect( + terminateWindowsProcessTree(processFixture(0), true, { spawnProcess }) + ).rejects.toThrow('owned process ID'); + expect(spawnProcess).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/cli/src/utils/windows-process-tree.ts b/apps/cli/src/utils/windows-process-tree.ts new file mode 100644 index 000000000..686879329 --- /dev/null +++ b/apps/cli/src/utils/windows-process-tree.ts @@ -0,0 +1,70 @@ +import type { ChildProcess, SpawnOptions } from 'child_process'; +import spawn from 'cross-spawn'; + +export interface WindowsProcessTreeOptions { + timeoutMs?: number; + spawnProcess?: (command: string, args: string[], options: SpawnOptions) => ChildProcess; +} + +/** + * Request recursive termination of a still-owned Windows child. A successful + * taskkill result is not independent confirmation that the process tree is empty. + * Never use an exited child's PID: Windows may have reassigned it. + */ +export async function terminateWindowsProcessTree( + child: ChildProcess, + force: boolean, + options: WindowsProcessTreeOptions = {} +): Promise { + if (child.exitCode != null || child.signalCode != null) return; + const pid = child.pid; + if (typeof pid !== 'number' || !Number.isSafeInteger(pid) || pid <= 0) { + throw new Error('Cannot terminate Windows process tree without an owned process ID'); + } + const timeoutMs = options.timeoutMs ?? 5_000; + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0 || timeoutMs > 2_147_483_647) { + throw new Error('Windows process tree termination timeout must be a positive bounded number'); + } + const spawnProcess = options.spawnProcess ?? spawn; + await new Promise((resolve, reject) => { + let helper: ChildProcess; + try { + helper = spawnProcess('taskkill', ['/PID', String(pid), '/T', ...(force ? ['/F'] : [])], { + stdio: 'ignore', + windowsHide: true, + }); + } catch { + reject(new Error('Could not start Windows process tree termination')); + return; + } + let settled = false; + let timedOut = false; + const finish = (error?: Error) => { + if (settled) return; + settled = true; + clearTimeout(timer); + helper.removeListener('error', onError); + helper.removeListener('close', onClose); + if (timedOut) reject(new Error('Windows process tree termination timed out')); + else if (error) reject(error); + else resolve(); + }; + const onError = () => finish(new Error('Windows process tree termination helper failed')); + const onClose = (code: number | null, signal: NodeJS.Signals | null) => { + if (code === 0 && signal === null) finish(); + else finish(new Error('Windows process tree termination did not succeed')); + }; + const timer = setTimeout(() => { + timedOut = true; + // Only terminate our taskkill helper, never another process by a cached PID. + try { + helper.kill('SIGKILL'); + } catch { + // Preserve the timeout result without exposing spawn arguments or output. + } + finish(new Error('Windows process tree termination timed out')); + }, timeoutMs); + helper.once('error', onError); + helper.once('close', onClose); + }); +} diff --git a/apps/cli/tests/lody-fleet-shutdown.test.ts b/apps/cli/tests/lody-fleet-shutdown.test.ts new file mode 100644 index 000000000..2ff6beb41 --- /dev/null +++ b/apps/cli/tests/lody-fleet-shutdown.test.ts @@ -0,0 +1,121 @@ +import { describe, expect, it, vi } from 'vitest'; +import { LodyFleet } from '../src/lib/lody-fleet'; +import { Lody } from '../src/lib/lody'; +import { MachineRuntime } from '../src/lib/machine-runtime'; + +vi.mock('@/lib/local-ipc-socket-server', async (original) => ({ + ...(await original()), + stopLocalIpcSocketServers: vi.fn(async () => {}), +})); +vi.mock('@/lib/local-terminal-server', async (original) => ({ + ...(await original()), + stopLocalTerminalServer: vi.fn(async () => {}), +})); +vi.mock('@/lib/local-loro-data-plane-server', async (original) => ({ + ...(await original()), + stopLocalLoroDataPlaneServer: vi.fn(async () => {}), +})); +vi.mock('@/mcp/lody-mcp-http-server', async (original) => ({ + ...(await original()), + stopLodyMcpHttpServer: vi.fn(async () => {}), +})); + +function runtime(id: string) { + return { + workspace: { id }, + unsubscribeTerminalCleanup: vi.fn(), + lody: { + cleanup: vi.fn(async () => {}), + forceTerminateSessions: vi.fn(async () => {}), + }, + }; +} +function fixture(entries: ReturnType[]) { + const runtimes = new Map(entries.map((entry) => [entry.workspace.id, entry])); + const fleet: LodyFleet = Object.assign(Object.create(LodyFleet.prototype), { + runtimes, + stopped: false, + shutdownPromise: null, + retryTimers: new Map(), + logger: { debug: vi.fn() }, + stopRuntimeStateLoop: vi.fn(), + memoryPressure: { stop: vi.fn() }, + cancelScheduledRemoteBridgeOffline: vi.fn(), + clearReconcileRetry: vi.fn(), + workspaceWatchCoordinator: { dispose: vi.fn(async () => {}) }, + cloudPort: { dispose: vi.fn(async () => {}) }, + terminalPtyService: { closeAll: vi.fn() }, + }); + return { fleet, runtimes }; +} + +describe('fleet process shutdown ownership', () => { + it('starts force cleanup across workspaces while graceful cleanup is hung', async () => { + const first = runtime('first'); + const second = runtime('second'); + let complete: () => void = () => {}; + first.lody.cleanup.mockImplementation( + () => + new Promise((resolve) => { + complete = resolve; + }) + ); + second.lody.cleanup.mockRejectedValue(new Error('retryable')); + const { fleet, runtimes } = fixture([first, second]); + const shutdown = fleet.shutdown(); + const rejected = expect(shutdown).rejects.toThrow('Workspace cleanup failed'); + await Promise.resolve(); + await fleet.forceTerminateSessions(); + expect(first.lody.forceTerminateSessions).toHaveBeenCalledTimes(1); + expect(second.lody.forceTerminateSessions).toHaveBeenCalledTimes(1); + expect(runtimes.has('first')).toBe(true); + complete(); + await rejected; + expect(runtimes.has('first')).toBe(false); + expect(runtimes.get('second')).toBe(second); + }); + + it('aggregates forced failures after attempting every retained runtime', async () => { + const first = runtime('first'); + const second = runtime('second'); + first.lody.forceTerminateSessions.mockRejectedValue(new Error('refused')); + const { fleet, runtimes } = fixture([first, second]); + await expect(fleet.forceTerminateSessions()).rejects.toThrow( + 'Forced workspace process cleanup failed' + ); + expect(second.lody.forceTerminateSessions).toHaveBeenCalledTimes(1); + expect(runtimes.size).toBe(2); + first.lody.forceTerminateSessions.mockResolvedValue(undefined); + await fleet.forceTerminateSessions(); + expect(first.lody.forceTerminateSessions).toHaveBeenCalledTimes(2); + }); + + it('forwards forced cleanup through Lody and active machine runtime', async () => { + const forceTerminateSessions = vi.fn(async () => {}); + const stop = vi.fn(); + const cancel = vi.fn(); + const machine: MachineRuntime = Object.assign(Object.create(MachineRuntime.prototype), { + gcManager: { stop }, + messageProcessor: { stop }, + handler: { cancelPendingPermissionRequests: cancel }, + sessionManager: { forceTerminateSessions }, + }); + const lody: Lody = Object.assign(Object.create(Lody.prototype), { runtime: machine }); + await lody.forceTerminateSessions(); + expect(forceTerminateSessions).toHaveBeenCalledTimes(1); + expect(stop).toHaveBeenCalledTimes(2); + expect(cancel).toHaveBeenCalledTimes(1); + }); +}); + +it('retries graceful cleanup of retained failures after a forced sweep', async () => { + const entry = runtime('retry'); + entry.lody.cleanup.mockRejectedValueOnce(new Error('refused')).mockResolvedValue(undefined); + const { fleet, runtimes } = fixture([entry]); + await expect(fleet.shutdown()).rejects.toThrow('Workspace cleanup failed'); + expect(runtimes.get('retry')).toBe(entry); + await fleet.forceTerminateSessions(); + await fleet.shutdown(); + expect(entry.lody.cleanup).toHaveBeenCalledTimes(2); + expect(runtimes.size).toBe(0); +}); diff --git a/apps/cli/tests/session-sandbox.test.ts b/apps/cli/tests/session-sandbox.test.ts index 93120ee14..957172a50 100644 --- a/apps/cli/tests/session-sandbox.test.ts +++ b/apps/cli/tests/session-sandbox.test.ts @@ -35,6 +35,7 @@ class FakeChildProcess extends EventEmitter { pid: number; killed = false; exitCode: number | null = null; + signalCode: NodeJS.Signals | null = null; readonly stdout = new EventEmitter(); readonly stderr = new EventEmitter(); readonly kill = vi.fn((_signal?: NodeJS.Signals) => { @@ -178,6 +179,83 @@ class FakeCgroupFs { } describe('session sandbox', () => { + it('awaits recursive Windows termination and reports taskkill failure', async () => { + const child = new FakeChildProcess(1234); + const helper = new FakeChildProcess(5678); + const spawnProcess = vi.fn( + (command: string) => (command === 'taskkill' ? helper : child) as unknown as ChildProcess + ) as typeof realSpawn; + const factory = createSessionSandboxFactory({ + logger: createSilentLogger(), + deps: { platform: 'win32', spawnProcess, configureExecutionProcess: vi.fn(async () => {}) }, + }); + const sandbox = await factory('windows-tree' as SessionId); + const handle = await sandbox.spawn('node', [], { + cwd: process.cwd(), + env: {}, + stdio: 'ignore', + }); + const settled = vi.fn(); + const termination = handle.terminate(true); + const result = termination.catch(settled); + await Promise.resolve(); + expect(settled).not.toHaveBeenCalled(); + expect(spawnProcess).toHaveBeenLastCalledWith('taskkill', ['/PID', '1234', '/T', '/F'], { + stdio: 'ignore', + windowsHide: true, + }); + helper.emit('close', 1, null); + await result; + expect(settled).toHaveBeenCalledWith( + expect.objectContaining({ message: 'Windows process tree termination did not succeed' }) + ); + }); + + it('never reuses an exited Windows handle PID for tree termination', async () => { + const child = new FakeChildProcess(1234); + const spawnProcess = vi.fn(() => child as unknown as ChildProcess) as typeof realSpawn; + const factory = createSessionSandboxFactory({ + logger: createSilentLogger(), + deps: { platform: 'win32', spawnProcess, configureExecutionProcess: vi.fn(async () => {}) }, + }); + const sandbox = await factory('windows-exited' as SessionId); + const handle = await sandbox.spawn('node', [], { + cwd: process.cwd(), + env: {}, + stdio: 'ignore', + }); + child.signalCode = 'SIGTERM'; + child.emit('exit', null, 'SIGTERM'); + await handle.terminate(true); + await sandbox.terminate(true); + expect(spawnProcess).toHaveBeenCalledTimes(1); + }); + + it('attempts every Windows root when one taskkill fails and retains tracking for retry', async () => { + const first = new FakeChildProcess(1234); + const second = new FakeChildProcess(2345); + const spawnProcess = vi.fn((command: string, args: string[]) => { + if (command !== 'taskkill') + return (command === 'first' ? first : second) as unknown as ChildProcess; + const helper = new FakeChildProcess(5678); + queueMicrotask(() => helper.emit('close', args.includes('1234') ? 1 : 0, null)); + return helper as unknown as ChildProcess; + }) as typeof realSpawn; + const factory = createSessionSandboxFactory({ + logger: createSilentLogger(), + deps: { platform: 'win32', spawnProcess, configureExecutionProcess: vi.fn(async () => {}) }, + }); + const sandbox = await factory('windows-multiple' as SessionId); + for (const command of ['first', 'second']) { + await sandbox.spawn(command, [], { cwd: process.cwd(), env: {}, stdio: 'ignore' }); + } + await expect(sandbox.terminate(true)).rejects.toThrow( + 'Session process tree termination failed' + ); + expect(spawnProcess).toHaveBeenCalledTimes(4); + expect(await sandbox.readResourceAccounting()).toMatchObject({ rootPids: [1234, 2345] }); + }); + it('applies process resource profiles on Linux', async () => { const setPriority = vi.fn(); const writeFile = vi.fn(async () => {}); @@ -639,7 +717,11 @@ describe('session sandbox', () => { }, }); const sandbox = await factory('session-capture-cap' as SessionId); - const handle = await sandbox.spawn('noisy', [], { cwd: process.cwd(), env: {}, captureOutput: true }); + const handle = await sandbox.spawn('noisy', [], { + cwd: process.cwd(), + env: {}, + captureOutput: true, + }); let bytes = 0; let tail = ''; diff --git a/apps/cli/tests/session-terminate-cleanup.test.ts b/apps/cli/tests/session-terminate-cleanup.test.ts index c35155902..df318372c 100644 --- a/apps/cli/tests/session-terminate-cleanup.test.ts +++ b/apps/cli/tests/session-terminate-cleanup.test.ts @@ -76,6 +76,41 @@ function createProcessHandle(terminate: SessionProcessHandle['terminate']): Sess } describe('Session terminate cleanup', () => { + it('shares pending termination and upgrades force without waiting for terminal disposal', async () => { + const session = createSession(); + let finishDisposal = () => {}; + let startedDisposal = () => {}; + const started = new Promise((resolve) => { + startedDisposal = resolve; + }); + session.acpSessionId = 'acp-session-1' as ACPSessionId; + session.terminalManager = createTerminalManager({ + disposeAll: () => { + startedDisposal(); + return new Promise((resolve) => { + finishDisposal = resolve; + }); + }, + }); + let processKilled = () => {}; + const killed = new Promise((resolve) => { + processKilled = resolve; + }); + const handle = createProcessHandle(async (force) => { + expect(force).toBe(true); + handle.child.exitCode = 0; + processKilled(); + }); + (session as unknown as { agentProcess: SessionProcessHandle }).agentProcess = handle; + const first = session.terminate(false); + await started; + const second = session.terminate(true); + expect(second).toBe(first); + await killed; + finishDisposal(); + await first; + expect(session.acpSessionId).toBeNull(); + }); it('disposes ACP terminals before closing the ACP session on graceful terminate', async () => { const disposeAll = vi.fn(async () => {}); const closeSession = vi.fn(async () => true); @@ -113,10 +148,48 @@ describe('Session terminate cleanup', () => { closeSession, } as never; - await expect(session.terminate(false)).resolves.toBeUndefined(); + await expect(session.terminate(false)).rejects.toThrow('Session process termination failed'); expect(disposeAll).toHaveBeenCalledTimes(1); expect(closeSession).toHaveBeenCalledTimes(1); - expect(session.acpSessionId).toBeNull(); + expect(session.acpSessionId).toBe('acp-session-1'); + }); + + it('bounds stalled terminal disposal, still kills processes, and retries without duplicate disposal', async () => { + vi.useFakeTimers(); + try { + const session = createSession(); + let finishDisposal = () => {}; + const pending = new Promise((resolve) => { + finishDisposal = resolve; + }); + const disposeAll = vi.fn(() => pending); + session.terminalManager = createTerminalManager({ disposeAll }); + session.acpSessionId = 'acp-session-1' as ACPSessionId; + const terminateProcess = vi.fn(async () => {}); + // @ts-expect-error - exercising private process ownership + session.agentProcess = createProcessHandle(terminateProcess); + // @ts-expect-error - observing private sandbox lifecycle + const terminateSandbox = vi.spyOn(session.sandbox, 'terminate'); + const terminated = vi.fn(); + session.on('terminated', terminated); + const first = expect(session.terminate(true)).rejects.toThrow( + 'Session process termination failed' + ); + await vi.advanceTimersByTimeAsync(30_000); + await first; + expect(terminateProcess).toHaveBeenCalledWith(true); + expect(terminateSandbox).toHaveBeenCalledWith(true); + expect(terminated).not.toHaveBeenCalled(); + const retry = session.terminate(true); + await Promise.resolve(); + expect(disposeAll).toHaveBeenCalledTimes(1); + finishDisposal(); + await retry; + expect(terminated).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } }); it('skips ACP closeSession during forced terminate', async () => { @@ -150,3 +223,160 @@ describe('Session terminate cleanup', () => { expect(session.agentProcess).toBeNull(); }); }); + +describe('Session bounded process exit', () => { + it('recognizes signal-only exits without sending another termination', async () => { + const session = createSession(); + const handle = createProcessHandle(vi.fn()); + handle.child.signalCode = 'SIGTERM'; + handle.terminate = vi.fn(); + // @ts-expect-error - exercising private bounded process lifecycle + await session.killAndWait(handle, true); + expect(handle.terminate).not.toHaveBeenCalled(); + }); + + it('rejects when a forced process never exits and removes its subscription', async () => { + vi.useFakeTimers(); + try { + const session = createSession(); + const handle = createProcessHandle(vi.fn()); + handle.terminate = vi.fn(async () => {}); + const unsubscribe = vi.fn(); + handle.onExit = vi.fn(() => unsubscribe); + // @ts-expect-error - exercising private bounded process lifecycle + const result = session.killAndWait(handle, true); + const rejected = expect(result).rejects.toThrow('after forced termination'); + await vi.advanceTimersByTimeAsync(5_000); + await rejected; + expect(unsubscribe).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it('bounds graceful waiting, escalates, and clears the expired subscription', async () => { + vi.useFakeTimers(); + try { + const session = createSession(); + const handle = createProcessHandle(vi.fn()); + handle.terminate = vi.fn(async (force) => { + if (force) handle.child.signalCode = 'SIGKILL'; + }); + const unsubscribe = vi.fn(); + handle.onExit = vi.fn(() => unsubscribe); + // @ts-expect-error - exercising private bounded process lifecycle + const result = session.killAndWait(handle, false); + await vi.advanceTimersByTimeAsync(5_000); + await result; + expect(handle.terminate).toHaveBeenNthCalledWith(1, false); + expect(handle.terminate).toHaveBeenNthCalledWith(2, true); + expect(unsubscribe).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it('cleans its deadline and subscription after synchronous exit replay', async () => { + vi.useFakeTimers(); + try { + const session = createSession(); + const handle = createProcessHandle(vi.fn()); + handle.terminate = vi.fn(async () => {}); + const unsubscribe = vi.fn(); + handle.onExit = vi.fn((listener) => { + listener(null, 'SIGTERM'); + return unsubscribe; + }); + // @ts-expect-error - exercising private bounded process lifecycle + await session.killAndWait(handle, false); + expect(unsubscribe).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); +}); + +describe('Session termination failures', () => { + it('preserves stopping ownership after sandbox failure and allows retry', async () => { + const session = createSession(); + const terminated = vi.fn(); + session.on('terminated', terminated); + // @ts-expect-error - observing private sandbox lifecycle + const sandbox = session.sandbox; + const terminate = vi + .spyOn(sandbox, 'terminate') + .mockRejectedValueOnce(new Error('kill failed')) + .mockResolvedValue(undefined); + const cleanup = vi.spyOn(sandbox, 'cleanup').mockResolvedValue(undefined); + await expect(session.terminate(true)).rejects.toThrow('Session process termination failed'); + expect(cleanup).not.toHaveBeenCalled(); + expect(terminated).not.toHaveBeenCalled(); + // @ts-expect-error - observing retained lifecycle state + expect(session.status).toBe('stopping'); + await session.terminate(true); + expect(terminate).toHaveBeenCalledTimes(2); + expect(cleanup).toHaveBeenCalledTimes(1); + expect(terminated).toHaveBeenCalledTimes(1); + }); + + it('attempts the other process and sandbox when one process termination fails', async () => { + const session = createSession(); + const failing = createProcessHandle(async () => { + throw new Error('process failure'); + }); + const terminateOther = vi.fn(async () => {}); + // @ts-expect-error - exercising private process ownership + session.activeProcess = failing; + // @ts-expect-error - exercising private process ownership + session.agentProcess = createProcessHandle(terminateOther); + // @ts-expect-error - observing private sandbox lifecycle + const sandbox = session.sandbox; + const terminateSandbox = vi.spyOn(sandbox, 'terminate').mockResolvedValue(undefined); + const cleanup = vi.spyOn(sandbox, 'cleanup').mockResolvedValue(undefined); + await expect(session.terminate(true)).rejects.toThrow('Session process termination failed'); + expect(terminateOther).toHaveBeenCalledWith(true); + expect(terminateSandbox).toHaveBeenCalledWith(true); + expect(cleanup).not.toHaveBeenCalled(); + // @ts-expect-error - observing retained ownership for retry + expect(session.activeProcess).toBe(failing); + }); +}); + +it('does not hide failed graceful termination when the root exits during the attempt', async () => { + const session = createSession(); + const handle = createProcessHandle(vi.fn()); + handle.terminate = vi.fn(async () => { + handle.child.signalCode = 'SIGTERM'; + throw new Error('tree termination failed'); + }); + // @ts-expect-error - exercising private bounded process lifecycle + await expect(session.killAndWait(handle, false)).rejects.toThrow('tree termination failed'); + expect(handle.terminate).toHaveBeenCalledTimes(1); +}); + +it('retries a refused graceful request forcibly while the root remains live', async () => { + const session = createSession(); + const handle = createProcessHandle(vi.fn()); + handle.terminate = vi.fn(async (force) => { + if (!force) throw new Error('graceful refusal'); + handle.child.signalCode = 'SIGKILL'; + }); + // @ts-expect-error - exercising private bounded process lifecycle + await session.killAndWait(handle, false); + expect(handle.terminate).toHaveBeenNthCalledWith(1, false); + expect(handle.terminate).toHaveBeenNthCalledWith(2, true); +}); + +it('rejects a refused forced retry without claiming completion', async () => { + const session = createSession(); + const handle = createProcessHandle(vi.fn()); + handle.terminate = vi.fn(async (force) => { + throw new Error(force ? 'force refusal' : 'graceful refusal'); + }); + // @ts-expect-error - exercising private bounded process lifecycle + await expect(session.killAndWait(handle, false)).rejects.toThrow('force refusal'); + expect(handle.terminate).toHaveBeenCalledTimes(2); +}); diff --git a/apps/cli/tests/terminal-manager.test.ts b/apps/cli/tests/terminal-manager.test.ts index ee82859a6..5e5a826a4 100644 --- a/apps/cli/tests/terminal-manager.test.ts +++ b/apps/cli/tests/terminal-manager.test.ts @@ -105,3 +105,177 @@ describe('ShellTerminalManager', () => { expect(processHandle.child.kill).not.toHaveBeenCalled(); }); }); + +function releaseFixture(handles: SessionProcessHandle[]) { + const sandbox: SessionSandbox = { + enabled: false, + description: 'test', + applyLimits: async () => {}, + spawn: vi.fn(async () => { + const handle = handles.shift(); + if (!handle) throw new Error('no handle'); + return handle; + }), + terminate: async () => {}, + cleanup: async () => {}, + }; + return new ShellTerminalManager({ + logger: createSilentLogger(), + sessionLabel: 'test', + getActiveAcpSessionId: () => 'acp-1', + resolveWorkdir: () => process.cwd(), + buildEnv: () => ({}), + sandbox, + }); +} + +function observedHandle() { + const handle = createProcessHandle(vi.fn(async () => {})); + let close: (code: number | null, signal: NodeJS.Signals | null) => void = () => {}; + const unsubscribe = vi.fn(); + handle.onClose = (listener) => { + close = listener; + return unsubscribe; + }; + return { + handle, + unsubscribe, + exit: (code = 7) => { + handle.child.exitCode = code; + close(code, null); + }, + }; +} + +it('gates release and existing waiters on the actual exit and coalesces release requests', async () => { + const owned = observedHandle(); + const manager = releaseFixture([owned.handle]); + const id = await manager.createTerminal('acp-1', 'test'); + const observed = vi.fn(); + const waiter = manager.waitForTerminalExit('acp-1', id).then(observed); + const release = manager.releaseTerminal('acp-1', id); + const second = manager.releaseTerminal('acp-1', id); + expect((await manager.terminalOutput('acp-1', id)).exitStatus).toBeNull(); + expect(observed).not.toHaveBeenCalled(); + expect(owned.unsubscribe).not.toHaveBeenCalled(); + owned.exit(23); + await Promise.all([release, second, waiter]); + expect(observed).toHaveBeenCalledWith({ exitCode: 23, signal: undefined }); + expect(owned.unsubscribe).toHaveBeenCalledTimes(1); + expect(owned.handle.terminate).toHaveBeenCalledTimes(1); +}); + +it('preserves failed release state and waiters for retry', async () => { + const owned = observedHandle(); + owned.handle.terminate = vi.fn(async () => { + throw new Error('refused'); + }); + const manager = releaseFixture([owned.handle]); + const id = await manager.createTerminal('acp-1', 'test'); + const observed = vi.fn(); + const waiter = manager.waitForTerminalExit('acp-1', id).then(observed); + await expect(manager.releaseTerminal('acp-1', id)).rejects.toThrow('refused'); + expect((await manager.terminalOutput('acp-1', id)).exitStatus).toBeNull(); + expect(observed).not.toHaveBeenCalled(); + expect(owned.unsubscribe).not.toHaveBeenCalled(); + owned.exit(17); + await waiter; + await manager.releaseTerminal('acp-1', id); + expect(observed).toHaveBeenCalledWith({ exitCode: 17, signal: undefined }); +}); + +it('bounds an unexited process through graceful and forced attempts without synthetic exit', async () => { + vi.useFakeTimers(); + try { + const owned = observedHandle(); + const manager = releaseFixture([owned.handle]); + const id = await manager.createTerminal('acp-1', 'test'); + const release = manager.releaseTerminal('acp-1', id); + const rejected = expect(release).rejects.toThrow('did not report exit'); + await vi.advanceTimersByTimeAsync(10_000); + await rejected; + expect(owned.handle.terminate).toHaveBeenNthCalledWith(1, false); + expect(owned.handle.terminate).toHaveBeenNthCalledWith(2, true); + expect((await manager.terminalOutput('acp-1', id)).exitStatus).toBeNull(); + expect(owned.unsubscribe).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } +}); + +it('aggregates disposal failure after attempting every terminal', async () => { + const failed = observedHandle(); + failed.handle.terminate = vi.fn(async () => { + throw new Error('refused'); + }); + const successful = observedHandle(); + successful.handle.terminate = vi.fn(async () => successful.exit(0)); + const manager = releaseFixture([failed.handle, successful.handle]); + const first = await manager.createTerminal('acp-1', 'first'); + const second = await manager.createTerminal('acp-1', 'second'); + await expect(manager.disposeAll('acp-1')).rejects.toThrow('Terminal disposal failed'); + expect((await manager.terminalOutput('acp-1', first)).exitStatus).toBeNull(); + await expect(manager.terminalOutput('acp-1', second)).rejects.toThrow('already released'); +}); + +it('retains exit observed before the terminal is inserted in the map', async () => { + const owned = observedHandle(); + owned.handle.onClose = (listener) => { + listener(31, null); + return owned.unsubscribe; + }; + const manager = releaseFixture([owned.handle]); + const id = await manager.createTerminal('acp-1', 'test'); + await expect(manager.waitForTerminalExit('acp-1', id)).resolves.toEqual({ + exitCode: 31, + signal: undefined, + }); + await manager.releaseTerminal('acp-1', id); + expect(owned.handle.terminate).not.toHaveBeenCalled(); +}); + +it('forces a refused graceful release while the root is still live', async () => { + const owned = observedHandle(); + owned.handle.terminate = vi.fn(async (force) => { + if (!force) throw new Error('graceful refusal'); + owned.exit(29); + }); + const manager = releaseFixture([owned.handle]); + const id = await manager.createTerminal('acp-1', 'test'); + const waiter = manager.waitForTerminalExit('acp-1', id); + await manager.releaseTerminal('acp-1', id); + expect(owned.handle.terminate).toHaveBeenNthCalledWith(1, false); + expect(owned.handle.terminate).toHaveBeenNthCalledWith(2, true); + await expect(waiter).resolves.toEqual({ exitCode: 29, signal: undefined }); +}); + +it('does not hide tree termination failure when the root exits during the request', async () => { + const owned = observedHandle(); + owned.handle.terminate = vi.fn(async () => { + owned.exit(19); + throw new Error('tree failure'); + }); + const manager = releaseFixture([owned.handle]); + const id = await manager.createTerminal('acp-1', 'test'); + await expect(manager.releaseTerminal('acp-1', id)).rejects.toThrow('tree failure'); + expect(owned.handle.terminate).toHaveBeenCalledTimes(1); + expect(owned.unsubscribe).not.toHaveBeenCalled(); + await expect(manager.waitForTerminalExit('acp-1', id)).resolves.toEqual({ + exitCode: 19, + signal: undefined, + }); +}); + +it('publishes actual close without waiting for hung resource inspection', async () => { + const owned = observedHandle(); + owned.handle.inspectExit = () => new Promise(() => {}); + const manager = releaseFixture([owned.handle]); + const id = await manager.createTerminal('acp-1', 'test'); + const waiter = manager.waitForTerminalExit('acp-1', id); + owned.exit(37); + await expect(waiter).resolves.toEqual({ exitCode: 37, signal: undefined }); + await manager.releaseTerminal('acp-1', id); + expect(owned.unsubscribe).toHaveBeenCalledTimes(1); + expect(owned.handle.terminate).not.toHaveBeenCalled(); +});