diff --git a/apps/server/src/main.test.ts b/apps/server/src/main.test.ts index 5ddde930b2c7c21fe04fc49c1037361d5814dc99..9f3f55d4f9cd597375b5997d405f1916f5c8fcff 100644 --- a/apps/server/src/main.test.ts +++ b/apps/server/src/main.test.ts @@ -25,6 +25,7 @@ const serverStart = Effect.acquireRelease( ); const findAvailablePort = vi.fn((preferred: number) => Effect.succeed(preferred)); +// Shared service layer used by this CLI test suite. const testLayer = Layer.mergeAll( Layer.succeed(CliConfig, { cwd: "/tmp/t3-test-workspace", diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 874bdea1d6ff2547c2d731f69c9b25d56812e261..c51e6963aa49baa8eacd24fab48128664ab4ac50 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -219,6 +219,37 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { assert.equal(firstEvent.value.turnId, "turn-1"); }), ); + + it.effect("maps session/closed lifecycle events to canonical session.exited runtime events", () => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + const event: ProviderEvent = { + id: asEventId("evt-session-closed"), + kind: "session", + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + method: "session/closed", + message: "Session stopped", + }; + + lifecycleManager.emit("event", event); + const firstEvent = yield* Fiber.join(firstEventFiber); + + assert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + assert.equal(firstEvent.value.type, "session.exited"); + if (firstEvent.value.type !== "session.exited") { + return; + } + assert.equal(firstEvent.value.sessionId, "sess-1"); + assert.equal(firstEvent.value.message, "Session stopped"); + }), + ); }); afterAll(() => { diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 0a38899302c7bff156dbaf79e2899bb96eafdd9d..35f517a55a9104e10c6e9588eef22e38d70fdc1f 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -333,7 +333,7 @@ function mapToRuntimeEvents(event: ProviderEvent): ReadonlyArray = []; + readonly killSignals: Array = []; private readonly dataListeners = new Set<(data: string) => void>(); private readonly exitListeners = new Set<(event: PtyExitEvent) => void>(); killed = false; @@ -30,8 +31,9 @@ class FakePtyProcess implements PtyProcess { this.resizeCalls.push({ cols, rows }); } - kill(): void { + kill(signal?: string): void { this.killed = true; + this.killSignals.push(signal); } onData(callback: (data: string) => void): () => void { @@ -146,6 +148,8 @@ describe("TerminalManager", () => { shellResolver?: () => string; subprocessChecker?: (terminalPid: number) => Promise; subprocessPollIntervalMs?: number; + processKillGraceMs?: number; + maxRetainedInactiveSessions?: number; } = {}, ) { const logsDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3code-terminal-")); @@ -160,6 +164,10 @@ describe("TerminalManager", () => { ...(options.subprocessPollIntervalMs ? { subprocessPollIntervalMs: options.subprocessPollIntervalMs } : {}), + ...(options.processKillGraceMs ? { processKillGraceMs: options.processKillGraceMs } : {}), + ...(options.maxRetainedInactiveSessions + ? { maxRetainedInactiveSessions: options.maxRetainedInactiveSessions } + : {}), }); return { logsDir, ptyAdapter, manager }; } @@ -437,6 +445,50 @@ describe("TerminalManager", () => { manager.dispose(); }); + it("escalates terminal shutdown to SIGKILL when process does not exit in time", async () => { + const { manager, ptyAdapter } = makeManager(5, { processKillGraceMs: 10 }); + await manager.open(openInput()); + const process = ptyAdapter.processes[0]; + expect(process).toBeDefined(); + if (!process) return; + + await manager.close({ threadId: "thread-1" }); + await waitFor(() => process.killSignals.includes("SIGKILL")); + + expect(process.killSignals[0]).toBe("SIGTERM"); + expect(process.killSignals).toContain("SIGKILL"); + + manager.dispose(); + }); + + it("evicts oldest inactive terminal sessions when retention limit is exceeded", async () => { + const { manager, ptyAdapter } = makeManager(5, { maxRetainedInactiveSessions: 1 }); + + await manager.open(openInput({ threadId: "thread-1" })); + await manager.open(openInput({ threadId: "thread-2" })); + + const first = ptyAdapter.processes[0]; + const second = ptyAdapter.processes[1]; + expect(first).toBeDefined(); + expect(second).toBeDefined(); + if (!first || !second) return; + + first.emitExit({ exitCode: 0, signal: 0 }); + await new Promise((resolve) => setTimeout(resolve, 5)); + second.emitExit({ exitCode: 0, signal: 0 }); + + await waitFor(() => { + const sessions = (manager as unknown as { sessions: Map }).sessions; + return sessions.size === 1; + }); + + const sessions = (manager as unknown as { sessions: Map }).sessions; + const keys = [...sessions.keys()]; + expect(keys).toEqual(["thread-2\u0000default"]); + + manager.dispose(); + }); + it("migrates legacy transcript filenames to terminal-scoped history path on open", async () => { const { manager, logsDir } = makeManager(); const legacyPath = path.join(logsDir, "thread-1.log"); diff --git a/apps/server/src/terminal/Layers/Manager.ts b/apps/server/src/terminal/Layers/Manager.ts index 3c3a9a1322d7dcca930b1c00b67299a5c254261e..ac0d63d8d44d7c910e8f5c2fd1f6184ed2c3faff 100644 --- a/apps/server/src/terminal/Layers/Manager.ts +++ b/apps/server/src/terminal/Layers/Manager.ts @@ -29,6 +29,8 @@ import { const DEFAULT_HISTORY_LINE_LIMIT = 5_000; const DEFAULT_PERSIST_DEBOUNCE_MS = 40; const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000; +const DEFAULT_PROCESS_KILL_GRACE_MS = 1_000; +const DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS = 128; const DEFAULT_OPEN_COLS = 120; const DEFAULT_OPEN_ROWS = 30; const TERMINAL_ENV_BLOCKLIST = new Set(["PORT", "ELECTRON_RENDERER_PORT", "ELECTRON_RUN_AS_NODE"]); @@ -279,6 +281,8 @@ interface TerminalManagerOptions { shellResolver?: () => string; subprocessChecker?: TerminalSubprocessChecker; subprocessPollIntervalMs?: number; + processKillGraceMs?: number; + maxRetainedInactiveSessions?: number; } export class TerminalManagerRuntime extends EventEmitter { @@ -294,8 +298,11 @@ export class TerminalManagerRuntime extends EventEmitter private readonly persistDebounceMs: number; private readonly subprocessChecker: TerminalSubprocessChecker; private readonly subprocessPollIntervalMs: number; + private readonly processKillGraceMs: number; + private readonly maxRetainedInactiveSessions: number; private subprocessPollTimer: ReturnType | null = null; private subprocessPollInFlight = false; + private readonly killEscalationTimers = new Map>(); private readonly logger = createLogger("terminal"); constructor(options: TerminalManagerOptions) { @@ -308,6 +315,9 @@ export class TerminalManagerRuntime extends EventEmitter this.subprocessChecker = options.subprocessChecker ?? defaultSubprocessChecker; this.subprocessPollIntervalMs = options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS; + this.processKillGraceMs = options.processKillGraceMs ?? DEFAULT_PROCESS_KILL_GRACE_MS; + this.maxRetainedInactiveSessions = + options.maxRetainedInactiveSessions ?? DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS; fs.mkdirSync(this.logsDir, { recursive: true }); } @@ -342,6 +352,7 @@ export class TerminalManagerRuntime extends EventEmitter runtimeEnv: normalizedRuntimeEnv(input.env), }; this.sessions.set(sessionKey, session); + this.evictInactiveSessionsIfNeeded(); this.startSession(session, { ...input, cols, rows }, "started"); return this.snapshot(session); } @@ -453,6 +464,7 @@ export class TerminalManagerRuntime extends EventEmitter runtimeEnv: normalizedRuntimeEnv(input.env), }; this.sessions.set(sessionKey, session); + this.evictInactiveSessionsIfNeeded(); } else { this.stopProcess(session); session.cwd = input.cwd; @@ -506,6 +518,10 @@ export class TerminalManagerRuntime extends EventEmitter clearTimeout(timer); } this.persistTimers.clear(); + for (const timer of this.killEscalationTimers.values()) { + clearTimeout(timer); + } + this.killEscalationTimers.clear(); this.pendingPersistHistory.clear(); this.threadLocks.clear(); this.persistQueues.clear(); @@ -584,17 +600,14 @@ export class TerminalManagerRuntime extends EventEmitter }); } catch (error) { if (ptyProcess) { - try { - ptyProcess.kill(); - } catch { - // Ignore kill errors during failed startup cleanup. - } + this.killProcessWithEscalation(ptyProcess, session.threadId, session.terminalId); } session.status = "error"; session.pid = null; session.process = null; session.hasRunningSubprocess = false; session.updatedAt = new Date().toISOString(); + this.evictInactiveSessionsIfNeeded(); this.updateSubprocessPollingState(); const message = error instanceof Error ? error.message : "Terminal start failed"; this.emitEvent({ @@ -627,6 +640,7 @@ export class TerminalManagerRuntime extends EventEmitter } private onProcessExit(session: TerminalSessionState, event: PtyExitEvent): void { + this.clearKillEscalationTimer(session.process); this.cleanupProcessHandles(session); session.process = null; session.pid = null; @@ -643,6 +657,7 @@ export class TerminalManagerRuntime extends EventEmitter exitCode: session.exitCode, exitSignal: session.exitSignal, }); + this.evictInactiveSessionsIfNeeded(); this.updateSubprocessPollingState(); } @@ -655,16 +670,8 @@ export class TerminalManagerRuntime extends EventEmitter session.hasRunningSubprocess = false; session.status = "exited"; session.updatedAt = new Date().toISOString(); - try { - process.kill(); - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - this.logger.warn("failed to kill terminal process", { - threadId: session.threadId, - terminalId: session.terminalId, - error: message, - }); - } + this.killProcessWithEscalation(process, session.threadId, session.terminalId); + this.evictInactiveSessionsIfNeeded(); this.updateSubprocessPollingState(); } @@ -675,6 +682,70 @@ export class TerminalManagerRuntime extends EventEmitter session.unsubscribeExit = null; } + private clearKillEscalationTimer(process: PtyProcess | null): void { + if (!process) return; + const timer = this.killEscalationTimers.get(process); + if (!timer) return; + clearTimeout(timer); + this.killEscalationTimers.delete(process); + } + + private killProcessWithEscalation(process: PtyProcess, threadId: string, terminalId: string): void { + this.clearKillEscalationTimer(process); + try { + process.kill("SIGTERM"); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.logger.warn("failed to kill terminal process", { + threadId, + terminalId, + signal: "SIGTERM", + error: message, + }); + return; + } + + const timer = setTimeout(() => { + this.killEscalationTimers.delete(process); + try { + process.kill("SIGKILL"); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.logger.warn("failed to force-kill terminal process", { + threadId, + terminalId, + signal: "SIGKILL", + error: message, + }); + } + }, this.processKillGraceMs); + timer.unref?.(); + this.killEscalationTimers.set(process, timer); + } + + private evictInactiveSessionsIfNeeded(): void { + const inactiveSessions = [...this.sessions.values()].filter((session) => session.status !== "running"); + if (inactiveSessions.length <= this.maxRetainedInactiveSessions) { + return; + } + + inactiveSessions.sort( + (left, right) => + left.updatedAt.localeCompare(right.updatedAt) || + left.threadId.localeCompare(right.threadId) || + left.terminalId.localeCompare(right.terminalId), + ); + const toEvict = inactiveSessions.length - this.maxRetainedInactiveSessions; + for (const session of inactiveSessions.slice(0, toEvict)) { + const key = toSessionKey(session.threadId, session.terminalId); + this.sessions.delete(key); + this.clearPersistTimer(session.threadId, session.terminalId); + this.pendingPersistHistory.delete(key); + this.persistQueues.delete(key); + this.clearKillEscalationTimer(session.process); + } + } + private queuePersist(threadId: string, terminalId: string, history: string): void { const persistenceKey = toSessionKey(threadId, terminalId); this.pendingPersistHistory.set(persistenceKey, history); diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index eb825f020547b067c4872a0de20940a5c159c095..9e5dd8fba95567f725cf402dd368174859fb7b50 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -593,9 +593,7 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< const unsubscribeTerminalEvents = yield* terminalManager.subscribe( (event) => void Effect.runPromise(onTerminalEvent(event)), ); - yield* Effect.addFinalizer(() => - Effect.all([Effect.sync(() => unsubscribeTerminalEvents()), terminalManager.dispose()]), - ); + yield* Effect.addFinalizer(() => Effect.sync(() => unsubscribeTerminalEvents())); yield* NodeHttpServer.make(() => httpServer, listenOptions);