diff --git a/apps/cli/src/agent/AGENTS.md b/apps/cli/src/agent/AGENTS.md index bd930c053..047170f77 100644 --- a/apps/cli/src/agent/AGENTS.md +++ b/apps/cli/src/agent/AGENTS.md @@ -81,8 +81,15 @@ 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 retained-child-handle cleanup utility, and reports termination failure; protocol + session-close failure must still proceed to process cleanup. + Windows termination must never reopen a cached PID. Observed child exit is the + utility's guarantee; descendant teardown requires spawn-time Job Object ownership. + Protocol authentication cleanup failure must return an error and retain its child + and provider slot until observed exit. Cancellation must permit cleanup retry. + 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..f9fe4f90e 100644 --- a/apps/cli/src/agent/acp-authentication.test.ts +++ b/apps/cli/src/agent/acp-authentication.test.ts @@ -9,6 +9,13 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import type { Logger } from '@/utils/logger'; import { createStdinWritableStream, createStdoutReadableStream } from '@/utils/stream'; import { AcpAuthenticationManager, probeBuiltinAuthentication } from './acp-authentication'; +import * as acpRunner from './acp-runner'; + +vi.mock('@/utils/windows-child-process', () => ({ + terminateWindowsChildProcess: async (child: ChildProcess) => { + if (child.exitCode == null && child.signalCode == null) child.kill('SIGKILL'); + }, +})); const createSilentLogger = (): Logger => ({ info: () => {}, @@ -45,6 +52,52 @@ function createDeferred() { } describe('AcpAuthenticationManager', () => { + it.each(['success', 'failure'] as const)( + 'retains authentication ownership after root exit until %s cleanup settles', + async (outcome) => { + const child = createFakeChild(); + const cleanup = createDeferred(); + const failure = new Error('client shutdown failed'); + const shutdown = vi + .spyOn(acpRunner, 'shutdownLocalAcpAgent') + .mockImplementationOnce(async () => { + await cleanup.promise; + if (outcome === 'failure') throw failure; + }) + .mockResolvedValue(undefined); + const spawnProcess = vi.fn(() => child); + const manager = new AcpAuthenticationManager(createSilentLogger(), { + spawnProcess: spawnProcess as never, + resolveLoginShellEnv: async () => ({}), + }); + const input = { + cliType: 'builtin' as const, + agentType: 'codex', + runtimeOverrides: { codexPath: '/test/codex' }, + }; + const authentication = manager.authenticate({ requestId: 'owned', ...input }); + await vi.waitFor(() => expect(spawnProcess).toHaveBeenCalledOnce()); + manager.cancel('owned'); + child.exitCode = 0; + child.emit('exit', 0, null); + await expect(authentication).resolves.toMatchObject({ disposition: 'cancelled' }); + expect(manager.getAgentType('owned')).toBe('codex'); + await expect(manager.authenticate({ requestId: 'overlap', ...input })).resolves.toMatchObject( + { success: false, error: 'Codex authentication is already running' } + ); + manager.cancel('owned'); + expect(shutdown).toHaveBeenCalledOnce(); + cleanup.resolve(); + await new Promise((resolve) => setImmediate(resolve)); + if (outcome === 'failure') { + expect(manager.getAgentType('owned')).toBe('codex'); + manager.cancel('owned'); + } + await vi.waitFor(() => expect(manager.getAgentType('owned')).toBeUndefined()); + expect(shutdown).toHaveBeenCalledTimes(outcome === 'failure' ? 2 : 1); + } + ); + afterEach(() => { vi.useRealTimers(); vi.restoreAllMocks(); @@ -316,8 +369,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, @@ -325,6 +382,74 @@ describe('AcpAuthenticationManager', () => { }); }); + it.each(['direct', 'npx'] as const)( + 'reports %s protocol cleanup failure and retains the login owner until cancellation retry exits', + async (launcher) => { + const child = createFakeChild(); + const originalKill = child.kill; + child.kill = vi.fn(() => { + throw new Error('cleanup failed'); + }); + const stdin = new PassThrough(); + const stdout = new PassThrough(); + child.stdin = stdin; + child.stdout = stdout; + child.stderr = new PassThrough(); + acp + .agent({ name: 'cleanup-test' }) + .onRequest(acp.methods.agent.initialize, async ({ params }) => ({ + protocolVersion: params.protocolVersion, + authMethods: [{ id: 'oauth', name: 'OAuth' }], + })) + .onRequest(acp.methods.agent.authenticate, async () => { + child.stderr?.emit('data', 'Error: Cannot find module auth-startup-dependency'); + return {}; + }) + .connect( + acp.ndJsonStream(createStdinWritableStream(stdout), createStdoutReadableStream(stdin)) + ); + const spawnProcess = vi.fn(() => child); + const manager = new AcpAuthenticationManager(createSilentLogger(), { + spawnProcess: spawnProcess as never, + resolveLoginShellEnv: async () => ({}), + terminationGraceMs: 2, + }); + const input = { + cliType: 'custom' as const, + agentType: 'cleanup-test', + customAcp: + launcher === 'npx' + ? { command: 'npx', args: ['--yes', '@lody-test/auth-cleanup-incomplete@0.0.0'] } + : { command: '/test/custom-acp', args: [] }, + }; + const progress = vi.fn(); + await expect( + manager.authenticate({ requestId: 'cleanup-1', ...input, onProgress: progress }) + ).resolves.toMatchObject({ success: false, disposition: 'error', error: 'cleanup failed' }); + expect(progress).not.toHaveBeenCalledWith({ status: 'authenticated' }); + expect(manager.getAgentType('cleanup-1')).toBe('cleanup-test'); + await expect( + manager.authenticate({ requestId: 'cleanup-2', ...input }) + ).resolves.toMatchObject({ + success: false, + error: 'cleanup-test authentication is already running', + }); + expect(spawnProcess).toHaveBeenCalledOnce(); + manager.cancel('cleanup-1'); + await vi.waitFor(() => expect(child.kill).toHaveBeenCalledTimes(2)); + // Allow the rejected teardown's finally to make the same owner retryable. + await new Promise((resolve) => setImmediate(resolve)); + expect(manager.getAgentType('cleanup-1')).toBe('cleanup-test'); + child.kill = originalKill; + manager.cancel('cleanup-1'); + await vi.waitFor(() => expect(manager.getAgentType('cleanup-1')).toBeUndefined()); + expect(originalKill).toHaveBeenCalled(); + expect(manager.cancel('cleanup-1')).toEqual({ success: true, disposition: 'not-running' }); + stdin.destroy(); + stdout.destroy(); + child.stderr.destroy(); + } + ); it('bridges request-scoped ACP form elicitation for a custom provider', async () => { const child = createFakeChild(); const stdin = new PassThrough(); @@ -531,7 +656,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-authentication.ts b/apps/cli/src/agent/acp-authentication.ts index ea4dd10b7..4486c6ee6 100644 --- a/apps/cli/src/agent/acp-authentication.ts +++ b/apps/cli/src/agent/acp-authentication.ts @@ -136,6 +136,9 @@ type RunningAuthentication = { cancelled: boolean; timedOut: boolean; terminating: boolean; + cleanupPromise?: Promise; + cleanupFailed?: boolean; + workflowFinished?: boolean; acceptsAuthorizationCode: boolean; authorizationCodeSubmitted: boolean; abortController: AbortController; @@ -622,7 +625,7 @@ export class AcpAuthenticationManager { detached: process.platform !== 'win32', windowsHide: true, }); - running.child = child; + this.trackAuthenticationChild(options.agentType, running, child); child.stdin?.on('error', (error: unknown) => { this.logger.debug( `[acp-auth] ${displayName} authorization input failed: ${formatErrorMessage(error)}` @@ -684,9 +687,8 @@ export class AcpAuthenticationManager { if (timeoutHandle) { clearTimeout(timeoutHandle); } - if (this.runningByAgentType.get(options.agentType) === running) { - this.runningByAgentType.delete(options.agentType); - } + running.workflowFinished = true; + this.releaseFinishedAuthentication(options.agentType, running); } } @@ -872,8 +874,16 @@ export class AcpAuthenticationManager { logger: this.logger, logPrefix: '[acp-auth]', getStderrTail: () => lastStderrTail, + shouldRetryError: () => !running.abortController.signal.aborted && !running.cleanupFailed, attempt: async ({ args }) => { running.abortController.signal.throwIfAborted(); + if ( + running.cleanupFailed || + running.terminating || + (running.child && running.child.exitCode == null && running.child.signalCode == null) + ) { + throw new Error('Previous authentication process cleanup is incomplete'); + } lastStderrTail = ''; options.onProgress?.({ status: 'starting' }); const child = spawnAcpProcess({ @@ -887,7 +897,7 @@ export class AcpAuthenticationManager { args: [...args], spawnImpl: this.spawnProcess, }); - running.child = child; + this.trackAuthenticationChild(options.agentType, running, child); running.terminating = false; child.stderr?.setEncoding('utf8'); const authorizationParser = new AcpAgentAuthorizationOutputParser(); @@ -1089,17 +1099,7 @@ export class AcpAuthenticationManager { running.pendingInteraction?.resolve({ action: 'cancel' }); running.pendingInteraction = undefined; startupMonitor.dispose(); - await shutdownLocalAcpAgent({ - agentProcess: child, - logger: this.logger, - sessionLabel: `acp-auth:${options.agentType}:protocol`, - exitTimeoutMs: this.terminationGraceMs, - }).catch((error: unknown) => { - this.logger.debug( - `[acp-auth] Failed to terminate protocol authentication process: ${formatErrorMessage(error)}` - ); - }); - if (running.child === child) running.child = undefined; + await this.cleanupAuthentication(options.agentType, running, 'protocol'); } }, }) @@ -1113,6 +1113,27 @@ export class AcpAuthenticationManager { return { success: true, disposition: 'authenticated' }; } + private releaseFinishedAuthentication(agentType: string, running: RunningAuthentication): void { + const child = running.child; + if ( + running.workflowFinished && + !running.terminating && + !running.cleanupFailed && + (!child || child.exitCode != null || child.signalCode != null) && + this.runningByAgentType.get(agentType) === running + ) { + this.runningByAgentType.delete(agentType); + } + } + + private trackAuthenticationChild( + agentType: string, + running: RunningAuthentication, + child: ChildProcess + ): void { + running.child = child; + child.once('exit', () => this.releaseFinishedAuthentication(agentType, running)); + } private terminateAuthentication( agentType: string, running: RunningAuthentication, @@ -1121,17 +1142,42 @@ export class AcpAuthenticationManager { // Protocol authentication spans launch preparation, a JSON-RPC wait, and // possibly a second process, so the signal is raised even with no child yet. running.abortController.abort(); - if (running.terminating || !running.child) return; - running.terminating = true; - void shutdownLocalAcpAgent({ - agentProcess: running.child, - logger: this.logger, - sessionLabel: `acp-auth:${agentType}:${reason}`, - exitTimeoutMs: this.terminationGraceMs, - }).catch((error: unknown) => { + void this.cleanupAuthentication(agentType, running, reason).catch((error: unknown) => { this.logger.debug( `[acp-auth] Failed to terminate authentication process: ${formatErrorMessage(error)}` ); }); } + + private cleanupAuthentication( + agentType: string, + running: RunningAuthentication, + reason: string + ): Promise { + if (running.cleanupPromise) return running.cleanupPromise; + const child = running.child; + if (!child) return Promise.resolve(); + running.terminating = true; + const cleanup = shutdownLocalAcpAgent({ + agentProcess: child, + logger: this.logger, + sessionLabel: `acp-auth:${agentType}:${reason}`, + exitTimeoutMs: this.terminationGraceMs, + }) + .then(() => { + running.cleanupFailed = false; + if (running.child === child) running.child = undefined; + }) + .catch((error: unknown) => { + running.cleanupFailed = true; + throw error; + }) + .finally(() => { + running.terminating = false; + running.cleanupPromise = undefined; + this.releaseFinishedAuthentication(agentType, running); + }); + running.cleanupPromise = cleanup; + return cleanup; + } } diff --git a/apps/cli/src/agent/acp-npx-startup-policy.ts b/apps/cli/src/agent/acp-npx-startup-policy.ts index f2595d3b6..11ea7c22c 100644 --- a/apps/cli/src/agent/acp-npx-startup-policy.ts +++ b/apps/cli/src/agent/acp-npx-startup-policy.ts @@ -30,6 +30,8 @@ export type RunNpxStartupWithRecoveryOptions = { logPrefix: string; attempt(input: NpxStartupAttemptInput): Promise; getStderrTail(): string; + /** Stop before error classification/cache recovery when the caller cannot safely retry. */ + shouldRetryError?: (error: unknown) => boolean; cleanupFailedAttempt?: () => Promise; startupTimeouts?: AcpStartupTimeoutOptions; coldInitTimeoutMs?: number; @@ -147,6 +149,7 @@ export async function runNpxStartupWithRecovery( startupTimeouts, }); } catch (error) { + if (options.shouldRetryError?.(error) === false) throw error; if (attempt >= maxAttempts) { throw error; } diff --git a/apps/cli/src/agent/acp-runner.test.ts b/apps/cli/src/agent/acp-runner.test.ts index d4c49a33e..eca509eab 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 childCleanup = vi.hoisted(() => vi.fn<() => Promise>()); +vi.mock('@/utils/windows-child-process', () => ({ terminateWindowsChildProcess: childCleanup })); +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' }); + childCleanup.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,113 @@ 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 child cleanup despite protocol close failure', async () => { + Object.defineProperty(process, 'platform', { value: 'win32' }); + const child = createFakeChildProcess({ pid: 1234 }); + let finish: (() => void) | undefined; + childCleanup.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(childCleanup).toHaveBeenCalledTimes(1); + child.signalCode = 'SIGKILL'; + finish?.(); + await Promise.all([first, second]); + expect(settled).toBe(true); + expect(child.kill).not.toHaveBeenCalled(); + }); + + it('reports Windows child failure and permits retry', async () => { + Object.defineProperty(process, 'platform', { value: 'win32' }); + const child = createFakeChildProcess({ pid: 1234 }); + childCleanup + .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 cleanup return with wrapper exit', async () => { + Object.defineProperty(process, 'platform', { value: 'win32' }); + vi.useFakeTimers(); + childCleanup.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 child 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..534222b17 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 { terminateWindowsChildProcess } from '@/utils/windows-child-process'; 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 terminateWindowsChildProcess(child, true, { timeoutMs: exitTimeoutMs }); + if (!(await waitForChildProcessExit(child, exitTimeoutMs))) { + throw new Error( + `[${sessionLabel}] ACP agent process did not exit after Windows child 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/agent/npx-cache.test.ts b/apps/cli/src/agent/npx-cache.test.ts index fcb00c38d..b64b79be4 100644 --- a/apps/cli/src/agent/npx-cache.test.ts +++ b/apps/cli/src/agent/npx-cache.test.ts @@ -1,6 +1,6 @@ import { homedir } from 'node:os'; import { basename, dirname, join } from 'node:path'; -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { AcpTimeoutError } from './agent-client'; import { @@ -358,6 +358,36 @@ describe('inspectNpxInstallState', () => { }); describe('runNpxStartupWithRecovery', () => { + it.each(['cleanup failed', 'cancelled'])( + 'preserves nonretryable %s without reading stale stderr or retrying', + async (message) => { + const failure = new Error(message); + const attempt = vi.fn(async () => { + throw failure; + }); + const stderr = vi.fn(() => 'Cannot find module'); + const cleanup = vi.fn(); + await expect( + runNpxStartupWithRecovery({ + command: 'npx', + args: npxArgs(), + env: { npm_config_cache: '/cache' }, + logger, + logPrefix: '[test]', + npxCacheIo: makeIo({}), + npxCacheRoots: ['/cache/_npx'], + attempt, + getStderrTail: stderr, + cleanupFailedAttempt: cleanup, + shouldRetryError: () => false, + }) + ).rejects.toBe(failure); + expect(attempt).toHaveBeenCalledOnce(); + expect(stderr).not.toHaveBeenCalled(); + expect(cleanup).not.toHaveBeenCalled(); + } + ); + it('uses the cold npx init timeout when the install is missing', async () => { const attempts: NpxStartupAttemptInput[] = []; const io = makeIo({}); 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/AGENTS.md b/apps/cli/src/lib/AGENTS.md index ebda8d393..1eb42195d 100644 --- a/apps/cli/src/lib/AGENTS.md +++ b/apps/cli/src/lib/AGENTS.md @@ -15,6 +15,20 @@ control-plane path is DEPRECATED; do not add functionality to it. NDJSON so ACP runtime/auth progress crosses the daemon socket immediately; legacy clients keep the buffered JSON envelope. `MachineRuntime` may collect responses for completion, but it must also forward each response to the streaming observer as it is sent. +- Stopping local message admission must reject both new requests and queued requests + whose handlers have not started. Discarded handlers must never run after cleanup; + active handlers retain their actual completion/error result because side effects may + already have happened. `MessageProcessor` owns this boundary and `MachineRuntime` + forwards admission/discard errors to each waiting local control response. +- Fleet shutdown always attempts independent terminal PTY cleanup and collects local + endpoint errors, even when a workspace cleanup fails. Retain failed workspace owners + and their shared watcher/cloud dependencies for retry; dispose shared services only + once no workspace owner remains, attempting every eligible disposer before rejecting. + Close terminal endpoint admission and await its admitted opens before snapshotting the + PTY pool: destroying the client socket does not cancel an awaited workdir resolution. + Pool admission closes synchronously and is rechecked after workdir resolution. Forced + cleanup kills existing PTYs independently of that drain, attempts every PTY despite + kill errors, and retains failed records for retry; a late resolver must never spawn. - `cloud-cli-port.ts` is the sole official-build composition root for cloud clients, endpoint-derived adapters, and their lifecycle. `start.ts` validates identity/deployment configuration once and injects the resulting `CloudPort` diff --git a/apps/cli/src/lib/local-terminal-server.ts b/apps/cli/src/lib/local-terminal-server.ts index 8bf26c444..47ffe2f66 100644 --- a/apps/cli/src/lib/local-terminal-server.ts +++ b/apps/cli/src/lib/local-terminal-server.ts @@ -29,6 +29,9 @@ type TerminalSocketState = { let terminalServer: net.Server | null = null; let activeSocketPath: string | null = null; let terminalServerStart: Promise | null = null; +let terminalServerStop: Promise | null = null; +let acceptingMessages = false; +const pendingMessages = new Set>(); // Tracks live client connections so shutdown can destroy them immediately. // `net.Server.close()` only stops accepting new connections and otherwise waits // for every open connection to end on its own; the Electron terminal relay holds @@ -236,6 +239,7 @@ async function handleMessage( } export async function startLocalTerminalServer(config: LocalTerminalServerConfig): Promise { + if (terminalServerStop) await terminalServerStop; if (terminalServer) { return; } @@ -243,6 +247,7 @@ export async function startLocalTerminalServer(config: LocalTerminalServerConfig return await terminalServerStart; } + acceptingMessages = true; terminalServerStart = startLocalTerminalServerInner(config).finally(() => { terminalServerStart = null; }); @@ -257,6 +262,10 @@ async function startLocalTerminalServerInner(config: LocalTerminalServerConfig): await removeStaleUnixSocket(socketPath, 'local_terminal_socket_in_use'); const server = net.createServer((socket) => { + if (!acceptingMessages) { + socket.destroy(); + return; + } let buffer = ''; // One decoder per connection: a pasted multi-byte character can land on a // socket chunk boundary, and per-chunk `toString('utf8')` would turn it into @@ -269,6 +278,7 @@ async function startLocalTerminalServerInner(config: LocalTerminalServerConfig): }); socket.on('data', (chunk) => { + if (!acceptingMessages) return; buffer += decodeChunk(chunk); // Compare char length (O(1)) rather than re-scanning the whole buffer with // Buffer.byteLength on every chunk (O(n²) across a large multi-chunk paste). @@ -312,7 +322,12 @@ async function startLocalTerminalServerInner(config: LocalTerminalServerConfig): continue; } - void handleMessage(config, socket, state, parsed.data); + const pending = handleMessage(config, socket, state, parsed.data); + pendingMessages.add(pending); + void pending.then( + () => pendingMessages.delete(pending), + () => pendingMessages.delete(pending) + ); } newlineIndex = buffer.indexOf('\n'); } @@ -358,10 +373,25 @@ async function startLocalTerminalServerInner(config: LocalTerminalServerConfig): }); } -export async function stopLocalTerminalServer(): Promise { - if (!terminalServer) { - return; - } +export function stopLocalTerminalServer(): Promise { + acceptingMessages = false; + if (terminalServerStop) return terminalServerStop; + const stop = stopLocalTerminalServerInner(); + terminalServerStop = stop; + void stop.then( + () => { + if (terminalServerStop === stop) terminalServerStop = null; + }, + () => { + if (terminalServerStop === stop) terminalServerStop = null; + } + ); + return stop; +} + +async function stopLocalTerminalServerInner(): Promise { + // A shutdown that races listen must also close that newly created listener. + await terminalServerStart?.catch(() => undefined); const server = terminalServer; const socketPath = activeSocketPath; @@ -378,8 +408,12 @@ export async function stopLocalTerminalServer(): Promise { activeClientSockets.clear(); await new Promise((resolve) => { - server.close(() => resolve()); + if (server) server.close(() => resolve()); + else resolve(); }); + // Destroying a socket does not cancel an already admitted open awaiting its + // workdir. Drain those handlers before the fleet snapshots and closes PTYs. + await Promise.allSettled(pendingMessages); if (socketPath && process.platform !== 'win32' && fs.existsSync(socketPath)) { fs.unlinkSync(socketPath); } diff --git a/apps/cli/src/lib/lody-fleet.ts b/apps/cli/src/lib/lody-fleet.ts index 7fc0c957c..6a2561f55 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,9 +530,41 @@ export class LodyFleet { }); } - async shutdown(): Promise { - if (this.stopped) return; + async forceTerminateSessions(): Promise { + this.stopped = true; + this.terminalPtyService.stopAdmission(); + const terminalFailures: unknown[] = []; + // A workdir resolver can remain pending indefinitely. Existing PTYs must + // still receive forced cleanup without waiting for the endpoint drain. + try { + this.terminalPtyService.closeAll(); + } catch (error) { + terminalFailures.push(error); + } + const results = await Promise.allSettled( + Array.from(this.runtimes.values(), (runtime) => runtime.lody.forceTerminateSessions()) + ); + const failures = [ + ...terminalFailures, + ...results.flatMap((result) => (result.status === 'rejected' ? [result.reason] : [])), + ]; + if (failures.length > 0) + throw new AggregateError(failures, 'Forced fleet 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.terminalPtyService.stopAdmission(); this.stopRuntimeStateLoop(); this.memoryPressure.stop(); if (this.prStatusPoller) { @@ -541,9 +574,9 @@ export class LodyFleet { // Stop accepting local work before draining workspace runtimes. Endpoint // teardown must not sit behind slow agent/session cleanup, and the owning // Host lease remains held until this shutdown barrier completes. + const localTerminalShutdown = Promise.allSettled([stopLocalTerminalServer()]); const localServicesStopped = Promise.allSettled([ stopLocalIpcSocketServers(), - stopLocalTerminalServer(), stopLocalLoroDataPlaneServer(), stopLodyMcpHttpServer(), ]); @@ -560,34 +593,69 @@ 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] : [] + ); + const workspaceCleanupFailed = cleanupFailures.length > 0; + const terminalShutdownResults = await localTerminalShutdown; + // PTYs are independent of workspace document flush/retry. A retained failed + // runtime must not keep these processes alive just because its cleanup failed. + try { + this.terminalPtyService.closeAll(); + } catch (error) { + cleanupFailures.push(error); + } + + // Failed workspace owners may still need shared services on the next cleanup + // attempt. Release those dependencies only after every owner was removed. + if (this.runtimes.size === 0) { + for (const dispose of [ + () => this.workspaceWatchCoordinator.dispose(), + () => this.cloudPort.dispose(), + ]) { + try { + await dispose(); + } catch (error) { + cleanupFailures.push(error); + } } } - await this.workspaceWatchCoordinator.dispose(); - await this.cloudPort.dispose(); - for (const result of await localServicesStopped) { + for (const result of [...terminalShutdownResults, ...(await localServicesStopped)]) { if (result.status === 'rejected') { + cleanupFailures.push(result.reason); this.logger.debug( `[fleet] Failed to stop a local service: ${formatErrorMessage(result.reason)}` ); } } - this.terminalPtyService.closeAll(); + if (cleanupFailures.length > 0) { + throw new AggregateError( + cleanupFailures, + workspaceCleanupFailed ? 'Workspace cleanup failed' : 'Fleet cleanup failed' + ); + } } private async applyWorkspaceList( 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..04782d1cd 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; @@ -267,7 +274,7 @@ export class MachineRuntime { ); return await new Promise((resolve, reject) => { - this.messageProcessor.enqueue(message, async (nextMessage) => { + const processMessage = async (nextMessage: LocalSessionControlRequestValidated) => { const responses: LocalSessionControlResponse[] = []; let settled = false; const resolveOnce = () => { @@ -328,7 +335,8 @@ export class MachineRuntime { rejectOnce(error); throw error; } - }); + }; + this.messageProcessor.enqueue(message, processMessage, reject); }); } diff --git a/apps/cli/src/lib/message-processor.ts b/apps/cli/src/lib/message-processor.ts index 4492f7a17..d119fd409 100644 --- a/apps/cli/src/lib/message-processor.ts +++ b/apps/cli/src/lib/message-processor.ts @@ -6,6 +6,13 @@ import { ConcurrentQueue } from './concurrent-queue'; type QueuedControlMessage = LocalSessionControlRequestValidated; type MessageQueueKey = string; +export class MessageProcessorStoppedError extends Error { + constructor() { + super('Local control message processor is stopping'); + this.name = 'MessageProcessorStoppedError'; + } +} + interface ProcessorEvents { 'message:processed': (message: QueuedControlMessage) => void; 'message:error': (error: Error, message: QueuedControlMessage) => void; @@ -22,6 +29,7 @@ interface ProcessorEvents { export class MessageProcessor extends EventEmitter { private readonly queue: ConcurrentQueue; private isStopped = false; + private readonly pendingDiscards = new Set<() => void>(); private static readonly QUEUE_WAIT_WARNING_MS = 10_000; private static readonly PROCESSING_WARNING_MS = 30_000; @@ -38,10 +46,13 @@ export class MessageProcessor extends EventEmitter { */ enqueue( message: QueuedControlMessage, - handler: (msg: QueuedControlMessage) => Promise + handler: (msg: QueuedControlMessage) => Promise, + onDiscard?: (error: MessageProcessorStoppedError) => void ): void { if (this.isStopped) { - this.logger.debug('MessageProcessor is stopped, ignoring new message'); + const error = new MessageProcessorStoppedError(); + onDiscard?.(error); + this.emit('message:error', error, message); return; } @@ -62,6 +73,15 @@ export class MessageProcessor extends EventEmitter { ); }, MessageProcessor.QUEUE_WAIT_WARNING_MS); waitWarning.unref?.(); + let discarded = false; + const discard = () => { + discarded = true; + clearInterval(waitWarning); + const error = new MessageProcessorStoppedError(); + onDiscard?.(error); + this.emit('message:error', error, message); + }; + this.pendingDiscards.add(discard); this.logger.debug( `Enqueued message type=${message.type} sessionId=${sessionId || 'N/A'} active=${ @@ -70,6 +90,8 @@ export class MessageProcessor extends EventEmitter { ); void this.queue.enqueue(queueKey, async () => { + this.pendingDiscards.delete(discard); + if (discarded) return; const startTime = Date.now(); started = true; clearInterval(waitWarning); @@ -140,6 +162,11 @@ export class MessageProcessor extends EventEmitter { */ stop(): void { this.isStopped = true; + // Reject only requests whose handlers have not started. Active handlers may + // already have committed side effects and must report their actual result. + const pending = [...this.pendingDiscards]; + this.pendingDiscards.clear(); + for (const discard of pending) discard(); this.logger.debug('MessageProcessor stopped'); } diff --git a/apps/cli/src/lib/terminal-pty-service.ts b/apps/cli/src/lib/terminal-pty-service.ts index 8a60dc50b..de5c07d96 100644 --- a/apps/cli/src/lib/terminal-pty-service.ts +++ b/apps/cli/src/lib/terminal-pty-service.ts @@ -63,6 +63,7 @@ export interface TerminalPtyServiceApi { resize(terminalId: string, cols: number, rows: number): void; close(terminalId: string): void; closeSession(sessionId: string): void; + stopAdmission(): void; closeAll(): void; onEvent(handler: (event: TerminalServerEvent) => void): () => void; } @@ -147,6 +148,7 @@ class TerminalPtyServiceImpl implements TerminalPtyServiceApi { private readonly records = new Map(); private readonly sessionIndex = new Map>(); private readonly pendingSessionOpens = new Map(); + private admissionClosed = false; private readonly handlers = new Set<(event: TerminalServerEvent) => void>(); constructor(options: TerminalPtyServiceOptions) { @@ -171,10 +173,12 @@ class TerminalPtyServiceImpl implements TerminalPtyServiceApi { } async open(params: TerminalOpenParams): Promise { + if (this.admissionClosed) throw new Error('terminal_service_stopping'); const sessionId = params.sessionId as SessionId; this.reserveSessionOpen(params.sessionId); try { const cwd = await this.resolveSessionWorkdir(sessionId); + if (this.admissionClosed) throw new Error('terminal_service_stopping'); const terminalId = randomUUID(); const shell = resolveShellCommand(); const terminal = loadPty().spawn(shell.file, shell.args, { @@ -273,11 +277,22 @@ class TerminalPtyServiceImpl implements TerminalPtyServiceApi { } } + stopAdmission(): void { + this.admissionClosed = true; + } + closeAll(): void { const ids = [...this.records.keys()]; + const failures: unknown[] = []; for (const terminalId of ids) { - this.closeIfPresent(terminalId); + try { + this.records.get(terminalId)?.pty.kill(); + } catch (error) { + // Retain ownership for retry, and still attempt every other terminal. + failures.push(error); + } } + if (failures.length) throw new AggregateError(failures, 'Terminal PTY cleanup failed'); } onEvent(handler: (event: TerminalServerEvent) => void): () => void { @@ -304,7 +319,8 @@ class TerminalPtyServiceImpl implements TerminalPtyServiceApi { this.logger.debug( `[terminal] failed to close terminalId=${terminalId}: ${formatErrorMessage(error)}` ); - this.removeRecord(terminalId); + // Session-termination callbacks must not throw, but failed kills still + // need an owner so fleet closeAll/forced cleanup can retry them. } } 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..9a8aca131 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 { terminateWindowsChildProcess } from '@/utils/windows-child-process'; 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< @@ -329,8 +330,8 @@ class NoopSessionSandbox implements SessionSandbox { child, async () => null, async (force) => { - if (typeof child.pid === 'number' && child.pid > 0) { - await this.terminateProcessTree(child.pid, force, detached); + if (this.deps.platform === 'win32' || (typeof child.pid === 'number' && child.pid > 0)) { + await this.terminateProcessTree(child, force, detached); return; } await terminateChildProcessDirectly(child, force); @@ -338,23 +339,30 @@ 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); }; child.once('exit', cleanupTrackedProcess); child.once('close', cleanupTrackedProcess); - child.once('error', cleanupTrackedProcess); + // A kill error is retryable; only observed process exit releases ownership. await configureExecutionProcessBestEffort(child.pid, this.deps, this.logger); } return processHandle; } 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 terminateWindowsChildProcess(child, force); 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..f6549b5a4 100644 --- a/apps/cli/src/session/session.ts +++ b/apps/cli/src/session/session.ts @@ -110,6 +110,12 @@ export class Session extends EventEmitter implements ISession { private readonly startedAtMs = getServerNow(); private activeProcess: SessionProcessHandle | null = null; private agentProcess: SessionProcessHandle | null = null; + private agentCreationInProgress = false; + 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 +243,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 +298,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 +313,58 @@ 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. + // On Windows the OS signal terminates immediately; closeSession above is + // the graceful ACP phase. Kill both processes and observe their actual exit. // 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 +382,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(() => { - unsubscribe(); - resolve(); - }); - // Guard: if the process exited between the check above and - // registering the listener, resolve immediately. - if (child.exitCode !== null) { + 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(); - } + 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 +595,27 @@ export class Session extends EventEmitter implements ISession { } async createAgent(callbacks: CreateAgentConfig): Promise { + if (this.agentCreationInProgress || this.agentProcess) { + throw new Error('Previous agent ownership must be released before another launch'); + } + this.agentCreationInProgress = true; + try { + return await this.createAgentOnce(callbacks); + } finally { + this.agentCreationInProgress = false; + } + } + + private async createAgentOnce(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, @@ -502,20 +638,13 @@ export class Session extends EventEmitter implements ISession { if (!handle) { return; } - try { - await this.killAndWait(handle, true); - } catch (error) { - this.logger.debug( - `[${ - this.sessionId - }] Failed to terminate ACP startup attempt before retry: ${formatErrorMessage(error)}` - ); - } finally { - if (this.agentProcess === handle) { - this.agentProcess = null; - } - lastAgentProcessHandle = null; + // Recovery may start another child only after cleanup is confirmed. + // Keep ownership intact when termination fails so shutdown can retry. + await this.killAndWait(handle, true); + if (this.agentProcess === handle) { + this.agentProcess = null; } + lastAgentProcessHandle = null; }; const attemptCreateAgent = async ( @@ -532,6 +661,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 +672,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; @@ -554,7 +688,7 @@ export class Session extends EventEmitter implements ISession { this.logger.debug( `[${this.sessionId}] ACP agent process exited with code ${code} signal ${signal}` ); - this.agentProcess = null; + if (this.agentProcess === agentProcessHandle) this.agentProcess = null; void agentProcessHandle .inspectExit(code, signal) .then((violation) => { diff --git a/apps/cli/src/session/terminal-manager.ts b/apps/cli/src/session/terminal-manager.ts index 8ccf1c0a1..1b07f84b0 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 { @@ -60,6 +62,8 @@ const DEFAULT_TERMINAL_BYTE_LIMIT = 1024 * 1024; // 1MB of retained output abstract class BaseTerminalManager implements TerminalManager { protected terminals = new Map>(); + private admissionClosed = false; + private readonly pendingStarts = new Set>(); protected readonly logger: Logger; protected readonly sessionLabel: string; private readonly getActiveSessionId: () => string | null; @@ -82,6 +86,7 @@ abstract class BaseTerminalManager implements TerminalManager { ): Promise { this.ensureValidSession(acpSessionId); + if (this.admissionClosed) throw new Error('Terminal manager is shutting down'); const terminalId = randomUUID(); const state: TerminalState = { id: terminalId, @@ -91,6 +96,7 @@ abstract class BaseTerminalManager implements TerminalManager { truncated: false, exitStatus: null, waiters: [], + disposed: false, }; const hooks: TerminalHooks = { @@ -101,20 +107,31 @@ abstract class BaseTerminalManager implements TerminalManager { }, }; - state.handle = await this.startProcess( - { - terminalId, - command, - args: args ?? [], - cwd, - env, - }, - hooks - ); + let finishStart = () => {}; + const pendingStart = new Promise((resolve) => { + finishStart = resolve; + }); + this.pendingStarts.add(pendingStart); + try { + state.handle = await this.startProcess( + { + terminalId, + command, + args: args ?? [], + cwd, + env, + }, + hooks + ); - this.terminals.set(terminalId, state); - this.logger.debug(`[${this.sessionLabel}] Terminal ${terminalId} started: ${command}`); - return terminalId; + this.terminals.set(terminalId, state); + this.logger.debug(`[${this.sessionLabel}] Terminal ${terminalId} started: ${command}`); + if (this.admissionClosed) throw new Error('Terminal launch cancelled by shutdown'); + return terminalId; + } finally { + this.pendingStarts.delete(pendingStart); + finishStart(); + } } async terminalOutput(acpSessionId: string, terminalId: string) { @@ -130,20 +147,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 { @@ -164,15 +218,21 @@ abstract class BaseTerminalManager implements TerminalManager { async disposeAll(acpSessionId: string): Promise { this.ensureValidSession(acpSessionId); + this.admissionClosed = true; + await Promise.all(this.pendingStarts); const terminalIds = Array.from(this.terminals.keys()); 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 +246,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 +286,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 +365,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 +380,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 +400,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-child-process.real-process.test.ts b/apps/cli/src/utils/windows-child-process.real-process.test.ts new file mode 100644 index 000000000..e0ee60938 --- /dev/null +++ b/apps/cli/src/utils/windows-child-process.real-process.test.ts @@ -0,0 +1,45 @@ +import { spawn } from 'node:child_process'; +import { once } from 'node:events'; +import { randomUUID } from 'node:crypto'; +import { describe, expect, it } from 'vitest'; +import { terminateWindowsChildProcess } from './windows-child-process'; + +describe.skipIf(process.platform !== 'win32')('Windows child handle integration', () => { + it('terminates the retained child handle without consulting its cached PID', async () => { + const child = spawn( + process.execPath, + ['-e', "process.stdout.write('ready');setInterval(()=>{},1000)"], + { + stdio: ['ignore', 'pipe', 'ignore'], + windowsHide: true, + } + ); + const exited = once(child, 'exit', { signal: AbortSignal.timeout(10_000) }); + void exited.catch(() => {}); + try { + if (!child.stdout) throw new Error('Fixture stdout unavailable'); + await once(child.stdout, 'data', { signal: AbortSignal.timeout(10_000) }); + Object.defineProperty(child, 'pid', { + get: () => { + throw new Error('Cached PID was read'); + }, + }); + await terminateWindowsChildProcess(child, true); + await exited; + expect(child.exitCode !== null || child.signalCode !== null).toBe(true); + } finally { + if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL'); + await exited; + } + }, 15_000); + it('accepts a queued spawn failure without mistaking it for a failed kill', async () => { + const child = spawn(`lody-missing-${randomUUID()}.exe`, [], { windowsHide: true }); + const failure = once(child, 'error', { signal: AbortSignal.timeout(5000) }); + const termination = terminateWindowsChildProcess(child, true, { timeoutMs: 5000 }); + await expect(termination).resolves.toBeUndefined(); + const [error] = await failure; + expect(error).toMatchObject({ code: 'ENOENT' }); + expect(child.exitCode).toBeLessThan(0); + await expect(terminateWindowsChildProcess(child, true)).resolves.toBeUndefined(); + }); +}); diff --git a/apps/cli/src/utils/windows-child-process.test.ts b/apps/cli/src/utils/windows-child-process.test.ts new file mode 100644 index 000000000..98ad846ec --- /dev/null +++ b/apps/cli/src/utils/windows-child-process.test.ts @@ -0,0 +1,124 @@ +import { EventEmitter } from 'events'; +import type { ChildProcess } from 'child_process'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { terminateWindowsChildProcess } from './windows-child-process'; + +function fixture(): ChildProcess { + const child = new EventEmitter() as ChildProcess; + Object.defineProperty(child, 'pid', { + get: () => { + throw new Error('Cached PID was read'); + }, + }); + child.exitCode = null; + child.signalCode = null; + child.kill = vi.fn(() => true); + return child; +} + +afterEach(() => vi.useRealTimers()); + +describe('terminateWindowsChildProcess', () => { + it.each([false, true])( + 'uses the retained handle and observes exit without waiting for close (force=%s)', + async (force) => { + const child = fixture(); + const result = terminateWindowsChildProcess(child, force); + expect(child.kill).toHaveBeenCalledWith(force ? 'SIGKILL' : 'SIGTERM'); + child.emit('exit', 0, null); + await result; + expect(child.eventNames()).toEqual([]); + } + ); + + it.each(['exitCode', 'signalCode'] as const)( + 'skips an already exited child (%s)', + async (field) => { + const child = fixture(); + if (field === 'exitCode') child.exitCode = 0; + else child.signalCode = 'SIGTERM'; + await terminateWindowsChildProcess(child, true); + expect(child.kill).not.toHaveBeenCalled(); + } + ); + + it('handles exit emitted synchronously during kill', async () => { + const child = fixture(); + child.kill = vi.fn(() => { + child.emit('exit', 0, null); + return true; + }); + await terminateWindowsChildProcess(child, true); + expect(child.eventNames()).toEqual([]); + }); + + it('waits for exit when kill returns false during an OS exit race', async () => { + const child = fixture(); + child.kill = vi.fn(() => false); + const settled = vi.fn(); + const result = terminateWindowsChildProcess(child, true).then(settled); + await Promise.resolve(); + expect(settled).not.toHaveBeenCalled(); + child.emit('exit', 0, null); + await result; + expect(settled).toHaveBeenCalledOnce(); + }); + + it.each([false, true])( + 'times out without retrying a PID or accepting close (kill=%s)', + async (accepted) => { + vi.useFakeTimers(); + const child = fixture(); + child.kill = vi.fn(() => accepted); + const result = terminateWindowsChildProcess(child, true, { timeoutMs: 25 }); + const rejected = expect(result).rejects.toThrow('timed out'); + child.emit('close', 0, null); + await vi.advanceTimersByTimeAsync(25); + await rejected; + expect(child.kill).toHaveBeenCalledOnce(); + expect(child.eventNames()).toEqual([]); + expect(vi.getTimerCount()).toBe(0); + } + ); + + it.each(['throw', 'event'])('redacts a kill %s and permits a fresh attempt', async (kind) => { + const child = fixture(); + child.kill = vi.fn(() => { + if (kind === 'throw') throw new Error('secret'); + child.emit('error', new Error('secret')); + return false; + }); + await expect(terminateWindowsChildProcess(child, true)).rejects.toThrow( + 'Windows child termination failed' + ); + child.kill = vi.fn(() => { + child.emit('exit', 0, null); + return true; + }); + await terminateWindowsChildProcess(child, true); + expect(child.eventNames()).toEqual([]); + }); + + it('does not accept EINVAL without subsequent terminal lifecycle evidence', async () => { + vi.useFakeTimers(); + const child = fixture(); + child.kill = vi.fn(() => { + throw Object.assign(new Error('invalid handle'), { code: 'EINVAL' }); + }); + const termination = terminateWindowsChildProcess(child, true, { timeoutMs: 25 }); + const rejected = expect(termination).rejects.toThrow('timed out'); + await vi.advanceTimersByTimeAsync(25); + await rejected; + expect(child.eventNames()).toEqual([]); + }); + it.each([0, -1, NaN, Infinity, 2_147_483_648])( + 'rejects invalid timeout %s before signaling', + async (timeoutMs) => { + const child = fixture(); + await expect(terminateWindowsChildProcess(child, true, { timeoutMs })).rejects.toThrow( + 'bounded' + ); + expect(child.kill).not.toHaveBeenCalled(); + } + ); +}); diff --git a/apps/cli/src/utils/windows-child-process.ts b/apps/cli/src/utils/windows-child-process.ts new file mode 100644 index 000000000..495359bb9 --- /dev/null +++ b/apps/cli/src/utils/windows-child-process.ts @@ -0,0 +1,68 @@ +import type { ChildProcess } from 'child_process'; + +export interface WindowsChildProcessOptions { + timeoutMs?: number; +} + +/** + * Terminate through Node's retained Windows process handle, never a cached PID. + * This confirms root exit only. Descendant teardown requires spawn-time job ownership. + * Windows treats both SIGTERM and SIGKILL as immediate termination; callers must + * complete any graceful protocol shutdown before invoking this utility. + */ +export async function terminateWindowsChildProcess( + child: ChildProcess, + force: boolean, + options: WindowsChildProcessOptions = {} +): Promise { + if (child.exitCode != null || child.signalCode != null) return; + const timeoutMs = options.timeoutMs ?? 5_000; + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0 || timeoutMs > 2_147_483_647) { + throw new Error('Windows child termination timeout must be a positive bounded number'); + } + await new Promise((resolve, reject) => { + let settled = false; + const finish = (error?: Error) => { + if (settled) return; + settled = true; + clearTimeout(timer); + child.off('error', onError); + child.off('exit', onExit); + if (error) reject(error); + else resolve(); + }; + const onError = (error?: NodeJS.ErrnoException) => { + // Node records a negative spawn exit code before emitting spawn failure. + // No process exists in this case, and no exit event will follow. + if ( + child.exitCode != null && + child.exitCode < 0 && + (error?.syscall === 'spawn' || error?.syscall?.startsWith('spawn ')) + ) { + finish(); + return; + } + finish(new Error('Windows child termination failed')); + }; + const onExit = () => finish(); + const timer = setTimeout( + () => finish(new Error('Windows child termination timed out')), + timeoutMs + ); + child.once('error', onError); + child.once('exit', onExit); + try { + // false can mean the OS process exited before Node delivered its exit event. + // Continue waiting for that event; neither a boolean nor a PID proves exit. + child.kill(force ? 'SIGKILL' : 'SIGTERM'); + } catch (error) { + // A not-yet-delivered spawn failure has no valid native handle: kill + // returns EINVAL. Await its queued error; the deadline still rejects + // if no terminal lifecycle evidence arrives. + if ( + !(typeof error === 'object' && error !== null && 'code' in error && error.code === 'EINVAL') + ) + onError(); + } + }); +} diff --git a/apps/cli/tests/local-terminal-server-shutdown.test.ts b/apps/cli/tests/local-terminal-server-shutdown.test.ts new file mode 100644 index 000000000..430019366 --- /dev/null +++ b/apps/cli/tests/local-terminal-server-shutdown.test.ts @@ -0,0 +1,92 @@ +import net from 'node:net'; +import { afterEach, expect, it, vi } from 'vitest'; +import { + startLocalTerminalServer, + stopLocalTerminalServer, +} from '../src/lib/local-terminal-server'; +import type { TerminalPtyServiceApi } from '../src/lib/terminal-pty-service'; +import type { Logger } from '../src/utils/logger'; + +const socketPath = vi.hoisted(() => + process.platform === 'win32' + ? `\\\\.\\pipe\\lody-terminal-shutdown-test-${process.pid}-${Date.now()}` + : `/tmp/lody-terminal-shutdown-test-${process.pid}-${Date.now()}.sock` +); +vi.mock('@lody/shared/node/local-terminal', () => ({ + getLocalTerminalSocketPath: () => socketPath, +})); +vi.mock('@lody/shared/node/local-ipc', () => ({ ensureLocalDaemonRunDir() {} })); +const logger: Logger = { + info() {}, + warn() {}, + error() {}, + success() {}, + debug() {}, + setLevel() {}, + child: () => logger, + close: async () => {}, +}; +afterEach(async () => { + await stopLocalTerminalServer(); +}); + +it.each(['resolve', 'reject'] as const)( + 'drains an admitted open that later %ss after its real socket closes', + async (outcome) => { + let completeOpen = () => {}; + let failOpen: (error: Error) => void = () => {}; + const opening = new Promise((resolve, reject) => { + completeOpen = resolve; + failOpen = reject; + }); + const liveTerminals = new Set(); + const open = vi.fn(async () => { + await opening; + liveTerminals.add('terminal-1'); + return { terminalId: 'terminal-1' }; + }); + const closeAll = vi.fn(() => liveTerminals.clear()); + const service: TerminalPtyServiceApi = { + open, + closeAll, + stopAdmission() {}, + list: () => [], + attach: () => ({ title: '', scrollback: '' }), + input() {}, + resize() {}, + close() {}, + closeSession() {}, + onEvent: () => () => {}, + }; + await startLocalTerminalServer({ logger, terminalPtyService: service }); + const socket = net.createConnection(socketPath); + await new Promise((resolve, reject) => { + socket.once('connect', resolve); + socket.once('error', reject); + }); + const closed = new Promise((resolve) => socket.once('close', () => resolve())); + socket.write( + `${JSON.stringify({ type: 'open', requestId: 'open-1', sessionId: 'session-1', cols: 80, rows: 24 })}\n` + ); + await vi.waitFor(() => expect(open).toHaveBeenCalledOnce()); + const stopped = vi.fn(); + const stopping = stopLocalTerminalServer(); + const shutdown = stopping.then(stopped); + try { + expect(stopLocalTerminalServer()).toBe(stopping); + await closed; + expect(stopped).not.toHaveBeenCalled(); + if (outcome === 'resolve') completeOpen(); + else failOpen(new Error('workdir unavailable')); + await shutdown; + expect(liveTerminals.size).toBe(outcome === 'resolve' ? 1 : 0); + closeAll(); + expect(liveTerminals.size).toBe(0); + expect(open).toHaveBeenCalledOnce(); + } finally { + completeOpen(); + socket.destroy(); + await stopping; + } + } +); 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..14fd6f3fa --- /dev/null +++ b/apps/cli/tests/lody-fleet-shutdown.test.ts @@ -0,0 +1,225 @@ +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'; +import { stopLocalTerminalServer } from '../src/lib/local-terminal-server'; + +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 workspaceWatchCoordinator = { dispose: vi.fn(async () => {}) }; + const cloudPort = { dispose: vi.fn(async () => {}) }; + const terminalPtyService = { closeAll: vi.fn(), stopAdmission: vi.fn() }; + 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, + cloudPort, + terminalPtyService, + }); + return { fleet, runtimes, workspaceWatchCoordinator, cloudPort, terminalPtyService }; +} + +describe('fleet process shutdown ownership', () => { + it('aggregates forced PTY failure after also attempting workspace cleanup', async () => { + const entry = runtime('force-failures'); + const ptyFailure = new Error('PTY kill refused'); + const workspaceFailure = new Error('workspace force refused'); + entry.lody.forceTerminateSessions.mockRejectedValueOnce(workspaceFailure); + const { fleet, terminalPtyService } = fixture([entry]); + terminalPtyService.closeAll.mockImplementationOnce(() => { + throw ptyFailure; + }); + await expect(fleet.forceTerminateSessions()).rejects.toMatchObject({ + errors: [ptyFailure, workspaceFailure], + }); + expect(terminalPtyService.stopAdmission).toHaveBeenCalledOnce(); + expect(entry.lody.forceTerminateSessions).toHaveBeenCalledOnce(); + await fleet.forceTerminateSessions(); + expect(terminalPtyService.closeAll).toHaveBeenCalledTimes(2); + }); + + 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 fleet 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); +}); + +it('closes independent PTYs after workspace failure while retaining shared retry dependencies', async () => { + const entry = runtime('retry-dependencies'); + const workspaceFailure = new Error('workspace producer cleanup failed'); + entry.lody.cleanup.mockRejectedValueOnce(workspaceFailure).mockResolvedValue(undefined); + const { fleet, runtimes, terminalPtyService, workspaceWatchCoordinator, cloudPort } = fixture([ + entry, + ]); + await expect(fleet.shutdown()).rejects.toMatchObject({ errors: [workspaceFailure] }); + expect(terminalPtyService.closeAll).toHaveBeenCalledTimes(1); + expect(runtimes.get(entry.workspace.id)).toBe(entry); + expect(workspaceWatchCoordinator.dispose).not.toHaveBeenCalled(); + expect(cloudPort.dispose).not.toHaveBeenCalled(); + entry.lody.cleanup.mockImplementation(async () => { + expect(workspaceWatchCoordinator.dispose).not.toHaveBeenCalled(); + expect(cloudPort.dispose).not.toHaveBeenCalled(); + }); + await fleet.shutdown(); + expect(runtimes.size).toBe(0); + expect(terminalPtyService.closeAll).toHaveBeenCalledTimes(2); + expect(workspaceWatchCoordinator.dispose).toHaveBeenCalledTimes(1); + expect(cloudPort.dispose).toHaveBeenCalledTimes(1); +}); + +it('reports independent PTY and workspace failures together and permits retry', async () => { + const entry = runtime('multiple-failures'); + const workspaceFailure = new Error('workspace failure'); + const ptyFailure = new Error('PTY failure'); + entry.lody.cleanup.mockRejectedValueOnce(workspaceFailure); + const { fleet, runtimes, terminalPtyService } = fixture([entry]); + terminalPtyService.closeAll.mockImplementationOnce(() => { + throw ptyFailure; + }); + await expect(fleet.shutdown()).rejects.toMatchObject({ errors: [workspaceFailure, ptyFailure] }); + expect(runtimes.get(entry.workspace.id)).toBe(entry); + await fleet.shutdown(); + expect(entry.lody.cleanup).toHaveBeenCalledTimes(2); + expect(runtimes.size).toBe(0); + expect(terminalPtyService.closeAll).toHaveBeenCalledTimes(2); +}); + +it('waits for admitted terminal opens before the PTY close snapshot even if a workspace fails', async () => { + let finishEndpoint = () => {}; + const endpoint = new Promise((resolve) => { + finishEndpoint = resolve; + }); + vi.mocked(stopLocalTerminalServer).mockReturnValueOnce(endpoint); + const entry = runtime('pending-terminal'); + entry.lody.cleanup.mockRejectedValueOnce(new Error('workspace failure')); + const { fleet, terminalPtyService } = fixture([entry]); + const shutdown = expect(fleet.shutdown()).rejects.toThrow('Workspace cleanup failed'); + await vi.waitFor(() => expect(entry.lody.cleanup).toHaveBeenCalledOnce()); + expect(terminalPtyService.closeAll).not.toHaveBeenCalled(); + finishEndpoint(); + await shutdown; + expect(terminalPtyService.closeAll).toHaveBeenCalledOnce(); +}); + +it('attempts every eligible shared disposer even when an earlier disposer fails', async () => { + const { fleet, terminalPtyService, workspaceWatchCoordinator, cloudPort } = fixture([]); + const watcherFailure = new Error('watcher failure'); + const cloudFailure = new Error('cloud failure'); + workspaceWatchCoordinator.dispose.mockRejectedValueOnce(watcherFailure); + cloudPort.dispose.mockRejectedValueOnce(cloudFailure); + await expect(fleet.shutdown()).rejects.toMatchObject({ errors: [watcherFailure, cloudFailure] }); + expect(terminalPtyService.closeAll).toHaveBeenCalledTimes(1); + expect(cloudPort.dispose).toHaveBeenCalledTimes(1); + await fleet.shutdown(); + expect(workspaceWatchCoordinator.dispose).toHaveBeenCalledTimes(2); + expect(cloudPort.dispose).toHaveBeenCalledTimes(2); +}); + +it('reports local endpoint stop failure after closing independent and shared resources', async () => { + const endpointFailure = new Error('terminal endpoint failure'); + vi.mocked(stopLocalTerminalServer).mockRejectedValueOnce(endpointFailure); + const { fleet, terminalPtyService, workspaceWatchCoordinator, cloudPort } = fixture([]); + await expect(fleet.shutdown()).rejects.toMatchObject({ errors: [endpointFailure] }); + expect(terminalPtyService.closeAll).toHaveBeenCalledTimes(1); + expect(workspaceWatchCoordinator.dispose).toHaveBeenCalledTimes(1); + expect(cloudPort.dispose).toHaveBeenCalledTimes(1); + await fleet.shutdown(); +}); diff --git a/apps/cli/tests/machine-runtime-stop-admission.test.ts b/apps/cli/tests/machine-runtime-stop-admission.test.ts new file mode 100644 index 000000000..8ed2da613 --- /dev/null +++ b/apps/cli/tests/machine-runtime-stop-admission.test.ts @@ -0,0 +1,150 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { LocalSessionControlRequestValidated, SessionId } from '@lody/shared'; +import { SessionPreviewRevokeRequestSchema } from '@lody/shared'; +import { MachineRuntime } from '../src/lib/machine-runtime'; +import { MessageProcessor, MessageProcessorStoppedError } from '../src/lib/message-processor'; +import type { Logger } from '../src/utils/logger'; + +function deferred() { + let release = () => {}; + const promise = new Promise((resolve) => { + release = resolve; + }); + return { promise, release }; +} + +const logger: Logger = { + info: () => {}, + warn: () => {}, + error: () => {}, + success: () => {}, + debug: () => {}, + setLevel: () => {}, + setDebug: () => {}, + child: () => logger, + close: async () => {}, +}; +const message: LocalSessionControlRequestValidated = { + type: 'session/cancel', + sessionId: 'synthetic-stop-session' as SessionId, +}; + +function runtimeWithHandler(handleMessage: () => Promise) { + const runtime = new MachineRuntime({ + sessionManagerFactory: () => { + throw new Error('Not used by dispatch test'); + }, + workspaceDocument: { clearMachineMonitorProvider: () => {} } as never, + handlerConfig: {} as never, + memoryPressure: { + getLatest: async () => { + throw new Error('No OS probes in dispatch test'); + }, + refresh: async () => { + throw new Error('No OS probes in dispatch test'); + }, + }, + logger, + }); + Object.assign(runtime, { + handler: { + handleMessage, + cancelPendingPermissionRequests: () => {}, + cleanup: async () => {}, + }, + }); + return runtime; +} + +describe('local control shutdown admission', () => { + it('rejects requests arriving after force termination begins', async () => { + const handler = vi.fn(async () => {}); + const runtime = runtimeWithHandler(handler); + await runtime.forceTerminateSessions(); + await expect(runtime.dispatchLocalMessageForResponse(message)).rejects.toBeInstanceOf( + MessageProcessorStoppedError + ); + expect(handler).not.toHaveBeenCalled(); + await runtime.cleanup(); + }); + + it('rejects queued dispatch immediately, but preserves active completion and skips queued side effects', async () => { + const entered = deferred(); + const finish = deferred(); + const handler = vi.fn(async () => { + entered.release(); + await finish.promise; + }); + const runtime = runtimeWithHandler(handler); + const active = runtime.dispatchLocalMessageForResponse(message); + await entered.promise; + const queued = runtime.dispatchLocalMessageForResponse(message); + const rejected = expect(queued).rejects.toBeInstanceOf(MessageProcessorStoppedError); + await runtime.forceTerminateSessions(); + // This must settle while the active handler is still blocked. + await rejected; + expect(handler).toHaveBeenCalledTimes(1); + finish.release(); + await expect(active).resolves.toEqual([]); + await runtime.cleanup(); + expect(handler).toHaveBeenCalledTimes(1); + }); + + it('does not replace an active handler failure with a shutdown error', async () => { + const entered = deferred(); + const finish = deferred(); + const failure = new Error('Actual handler failure'); + const runtime = runtimeWithHandler(async () => { + entered.release(); + await finish.promise; + throw failure; + }); + const result = runtime.dispatchLocalMessageForResponse(message); + const rejected = expect(result).rejects.toBe(failure); + await entered.promise; + await runtime.forceTerminateSessions(); + finish.release(); + await rejected; + await runtime.cleanup(); + }); + + it('discards entries waiting for global capacity exactly once across repeated stop calls', async () => { + const processor = new MessageProcessor(logger, 1); + const entered = deferred(); + const finish = deferred(); + const activeDiscard = vi.fn(); + processor.enqueue( + message, + async () => { + entered.release(); + await finish.promise; + }, + activeDiscard + ); + await entered.promise; + const queuedHandler = vi.fn(async () => {}); + const discard = vi.fn(); + // Preview requests have a separate lane, so this waits on global capacity. + processor.enqueue( + SessionPreviewRevokeRequestSchema.parse({ + type: 'session/preview-revoke', + sessionId: message.sessionId, + machineId: 'synthetic-machine', + workspaceId: 'synthetic-workspace', + requestedByUserId: 'synthetic-user', + }), + queuedHandler, + discard + ); + await Promise.resolve(); + expect(processor.getQueueSize()).toBe(1); + processor.stop(); + processor.stop(); + expect(discard).toHaveBeenCalledTimes(1); + expect(discard).toHaveBeenCalledWith(expect.any(MessageProcessorStoppedError)); + expect(activeDiscard).not.toHaveBeenCalled(); + finish.release(); + await processor.drain(); + expect(queuedHandler).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/cli/tests/session-sandbox.test.ts b/apps/cli/tests/session-sandbox.test.ts index 93120ee14..300839189 100644 --- a/apps/cli/tests/session-sandbox.test.ts +++ b/apps/cli/tests/session-sandbox.test.ts @@ -1,5 +1,6 @@ import { EventEmitter } from 'events'; import path from 'path'; +import { randomUUID } from 'node:crypto'; import { describe, expect, it, vi } from 'vitest'; import type { ChildProcess } from 'child_process'; @@ -35,6 +36,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 +180,114 @@ class FakeCgroupFs { } describe('session sandbox', () => { + it.skipIf(process.platform !== 'win32')( + 'cleans a buffered real spawn failure without signaling a nonexistent child', + async () => { + const factory = createSessionSandboxFactory({ + logger: createSilentLogger(), + deps: { + platform: 'win32', + spawnProcess: realSpawn, + configureExecutionProcess: vi.fn(async () => {}), + }, + }); + const sandbox = await factory('windows-spawn-failure' as SessionId); + const handle = await sandbox.spawn(`lody-missing-${randomUUID()}.exe`, [], { + cwd: process.cwd(), + stdio: 'ignore', + }); + const failure = await new Promise((resolve) => handle.onError(resolve)); + expect(failure).toMatchObject({ code: 'ENOENT' }); + const kill = vi.spyOn(handle.child, 'kill'); + await handle.terminate(false); + await sandbox.terminate(true); + expect(kill).not.toHaveBeenCalled(); + expect(await sandbox.readResourceAccounting()).toMatchObject({ rootPids: [] }); + } + ); + it('awaits Windows child exit through the retained handle', 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-child' as SessionId); + const handle = await sandbox.spawn('node', [], { + cwd: process.cwd(), + env: {}, + stdio: 'ignore', + }); + const settled = vi.fn(); + const termination = handle.terminate(true).then(settled); + await Promise.resolve(); + expect(settled).not.toHaveBeenCalled(); + expect(child.kill).toHaveBeenCalledWith('SIGKILL'); + expect(spawnProcess).toHaveBeenCalledTimes(1); + child.exitCode = 0; + child.emit('exit', 0, null); + await termination; + }); + + it('never signals an exited Windows child', 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(child.kill).not.toHaveBeenCalled(); + }); + + it('attempts every Windows root and retains a failed handle for retry', async () => { + const first = new FakeChildProcess(1234); + const second = new FakeChildProcess(2345); + first.kill.mockImplementation(() => { + first.emit('error', new Error('kill failed')); + return false; + }); + second.kill.mockImplementation(() => { + second.exitCode = 0; + second.emit('exit', 0, null); + return true; + }); + const spawnProcess = vi.fn( + (command: string) => (command === 'first' ? first : second) 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(first.kill).toHaveBeenCalledOnce(); + expect(second.kill).toHaveBeenCalledOnce(); + expect(spawnProcess).toHaveBeenCalledTimes(2); + expect(await sandbox.readResourceAccounting()).toMatchObject({ rootPids: [1234] }); + first.kill.mockImplementation(() => { + first.exitCode = 0; + first.emit('exit', 0, null); + return true; + }); + await sandbox.terminate(true); + expect(first.kill).toHaveBeenCalledTimes(2); + expect(await sandbox.readResourceAccounting()).toMatchObject({ rootPids: [] }); + }); it('applies process resource profiles on Linux', async () => { const setPriority = vi.fn(); const writeFile = vi.fn(async () => {}); @@ -639,7 +749,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..e5aef3c0c 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,228 @@ 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); +}); + +it('propagates failed startup cleanup and retains ownership for shutdown', async () => { + const session = createSession(); + const cleanupError = new Error('startup cleanup refused'); + const handle = createProcessHandle(async () => { + throw cleanupError; + }); + // No stdin causes startup to fail after the session takes ownership. + // @ts-expect-error - injecting the owned sandbox boundary + const spawn = vi.spyOn(session.sandbox, 'spawn').mockResolvedValue(handle); + await expect( + session.createAgent({ + cliType: 'registry', + agentType: 'opencode', + command: 'opencode', + args: ['acp'], + } as Parameters[0]) + ).rejects.toBe(cleanupError); + expect(spawn).toHaveBeenCalledTimes(1); + // @ts-expect-error - verifying retained ownership after failure + expect(session.agentProcess).toBe(handle); + await expect( + session.createAgent({ + cliType: 'registry', + agentType: 'opencode', + command: 'opencode', + args: ['acp'], + } as Parameters[0]) + ).rejects.toThrow('Previous agent ownership'); + expect(spawn).toHaveBeenCalledTimes(1); + // @ts-expect-error - the failed retry cannot replace the original live owner + expect(session.agentProcess).toBe(handle); + handle.terminate = vi.fn(async () => { + handle.child.exitCode = 0; + }); + await session.terminate(true); + expect(handle.terminate).toHaveBeenCalledWith(true); +}); + +it('serializes agent launch admission before the sandbox publishes a handle', async () => { + const session = createSession(); + let entered = () => {}; + const spawning = new Promise((resolve) => { + entered = resolve; + }); + let release = (_handle: SessionProcessHandle) => {}; + // @ts-expect-error - injecting the async owned spawn boundary + const spawn = vi.spyOn(session.sandbox, 'spawn').mockImplementation(() => { + entered(); + return new Promise((resolve) => { + release = resolve; + }); + }); + const callbacks = { + cliType: 'registry', + agentType: 'opencode', + command: 'opencode', + args: ['acp'], + } as Parameters[0]; + const first = session.createAgent(callbacks); + const rejected = expect(first).rejects.toThrow(); + await spawning; + await expect(session.createAgent(callbacks)).rejects.toThrow('Previous agent ownership'); + expect(spawn).toHaveBeenCalledOnce(); + release(createProcessHandle(async () => {})); + await rejected; + await session.terminate(true); +}); diff --git a/apps/cli/tests/terminal-manager.test.ts b/apps/cli/tests/terminal-manager.test.ts index ee82859a6..681a3cc40 100644 --- a/apps/cli/tests/terminal-manager.test.ts +++ b/apps/cli/tests/terminal-manager.test.ts @@ -105,3 +105,242 @@ 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(); +}); + +it('closes admission and drains pending terminal starts before disposal returns', async () => { + const owned = observedHandle(); + owned.handle.terminate = vi.fn(async () => owned.exit(0)); + let finishSpawn: (handle: SessionProcessHandle) => void = () => {}; + const pending = new Promise((resolve) => { + finishSpawn = resolve; + }); + const sandbox: SessionSandbox = { + enabled: false, + description: 'test', + applyLimits: async () => {}, + spawn: vi.fn(() => pending), + terminate: async () => {}, + cleanup: async () => {}, + }; + const manager = new ShellTerminalManager({ + logger: createSilentLogger(), + sessionLabel: 'test', + getActiveAcpSessionId: () => 'acp-1', + resolveWorkdir: () => process.cwd(), + buildEnv: () => ({}), + sandbox, + }); + const start = expect(manager.createTerminal('acp-1', 'test')).rejects.toThrow( + 'cancelled by shutdown' + ); + const disposed = vi.fn(); + const disposal = manager.disposeAll('acp-1').then(disposed); + await expect(manager.createTerminal('acp-1', 'late')).rejects.toThrow('shutting down'); + await Promise.resolve(); + expect(disposed).not.toHaveBeenCalled(); + finishSpawn(owned.handle); + await Promise.all([start, disposal]); + expect(owned.handle.terminate).toHaveBeenCalledTimes(1); + expect(owned.unsubscribe).toHaveBeenCalledTimes(1); + expect(sandbox.spawn).toHaveBeenCalledTimes(1); +}); + +it('finishes disposal when a pending terminal launch rejects', async () => { + let failSpawn: (error: Error) => void = () => {}; + const pending = new Promise((_resolve, reject) => { + failSpawn = reject; + }); + const sandbox: SessionSandbox = { + enabled: false, + description: 'test', + applyLimits: async () => {}, + spawn: () => pending, + terminate: async () => {}, + cleanup: async () => {}, + }; + const manager = new ShellTerminalManager({ + logger: createSilentLogger(), + sessionLabel: 'test', + getActiveAcpSessionId: () => 'acp-1', + resolveWorkdir: () => process.cwd(), + buildEnv: () => ({}), + sandbox, + }); + const start = expect(manager.createTerminal('acp-1', 'test')).rejects.toThrow('launch failed'); + const disposal = manager.disposeAll('acp-1'); + failSpawn(new Error('launch failed')); + await Promise.all([start, disposal]); +}); diff --git a/apps/cli/tests/terminal-pty-service-shutdown.test.ts b/apps/cli/tests/terminal-pty-service-shutdown.test.ts new file mode 100644 index 000000000..b50c1655f --- /dev/null +++ b/apps/cli/tests/terminal-pty-service-shutdown.test.ts @@ -0,0 +1,113 @@ +import { beforeEach, expect, it, vi } from 'vitest'; +import { makeTerminalPtyService } from '../src/lib/terminal-pty-service'; +import { LodyFleet } from '../src/lib/lody-fleet'; +import type { Logger } from '../src/utils/logger'; + +const native = vi.hoisted(() => ({ spawn: vi.fn() })); +vi.mock('node:module', async (original) => { + const actual = await original(); + return { + createRequire: (...args: Parameters) => { + const require = actual.createRequire(...args); + return Object.assign( + (id: string) => (id === '@lydell/node-pty' ? native : require(id)), + require + ); + }, + }; +}); +const logger: Logger = { + info() {}, + warn() {}, + error() {}, + success() {}, + debug() {}, + setLevel() {}, + child: () => logger, + close: async () => {}, +}; +const params = { sessionId: 'session-1', cols: 80, rows: 24 }; +beforeEach(() => { + native.spawn.mockReset(); +}); + +it('forced fleet cleanup closes existing PTYs while a workdir read hangs and prevents its late spawn', async () => { + const kill = vi.fn(); + native.spawn.mockReturnValue({ kill, onData() {}, onExit() {} }); + let resolveWorkdir: (cwd: string) => void = () => {}; + const pendingWorkdir = new Promise((resolve) => { + resolveWorkdir = resolve; + }); + const resolveSessionWorkdir = vi + .fn() + .mockResolvedValueOnce(process.cwd()) + .mockReturnValue(pendingWorkdir); + const service = makeTerminalPtyService({ logger, resolveSessionWorkdir }); + await service.open(params); + const pending = service.open(params); + const settled = vi.fn(); + void pending.then(settled, settled); + const rejected = expect(pending).rejects.toThrow('terminal_service_stopping'); + const fleet: LodyFleet = Object.assign(Object.create(LodyFleet.prototype), { + runtimes: new Map(), + terminalPtyService: service, + }); + await fleet.forceTerminateSessions(); + expect(settled).not.toHaveBeenCalled(); + expect(kill).toHaveBeenCalledOnce(); + expect(native.spawn).toHaveBeenCalledOnce(); + await expect(service.open(params)).rejects.toThrow('terminal_service_stopping'); + expect(resolveSessionWorkdir).toHaveBeenCalledTimes(2); + resolveWorkdir(process.cwd()); + await rejected; + expect(native.spawn).toHaveBeenCalledOnce(); +}); + +it('attempts every PTY kill and retains failed ownership for a retry', async () => { + const firstKill = vi.fn().mockImplementationOnce(() => { + throw new Error('kill refused'); + }); + const secondKill = vi.fn(); + native.spawn + .mockReturnValueOnce({ kill: firstKill, onData() {}, onExit() {} }) + .mockReturnValueOnce({ kill: secondKill, onData() {}, onExit() {} }); + const service = makeTerminalPtyService({ + logger, + resolveSessionWorkdir: async () => process.cwd(), + }); + await service.open(params); + await service.open(params); + expect(() => service.closeAll()).toThrow('Terminal PTY cleanup failed'); + expect(secondKill).toHaveBeenCalledOnce(); + expect(service.list(params.sessionId)).toHaveLength(2); + service.closeAll(); + expect(firstKill).toHaveBeenCalledTimes(2); +}); + +it('retains a failed session-close kill for fleet retry and releases only on observed exit', async () => { + const kill = vi.fn().mockImplementationOnce(() => { + throw new Error('session close refused'); + }); + let exit: (event: { exitCode: number }) => void = () => {}; + native.spawn.mockReturnValue({ + kill, + onData() {}, + onExit: (listener: typeof exit) => { + exit = listener; + }, + }); + const service = makeTerminalPtyService({ + logger, + resolveSessionWorkdir: async () => process.cwd(), + }); + const opened = await service.open(params); + expect(() => service.closeSession(params.sessionId)).not.toThrow(); + expect(service.list(params.sessionId).map((terminal) => terminal.terminalId)).toEqual([ + opened.terminalId, + ]); + service.closeAll(); + expect(kill).toHaveBeenCalledTimes(2); + expect(service.list(params.sessionId)).toHaveLength(1); + exit({ exitCode: 0 }); + expect(service.list(params.sessionId)).toHaveLength(0); +});