authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-26 08:39:05-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log9b8184df7e1a5fd1837b21eaa225d06f614b574b
tree274dce5f888f4e0c94eaa3dd42e0bcd418f6ff2e
parentaa5a144d04c5067b9344289c6bc6f28a8acb9765

fix memory leaks

6 files changed, 172 insertions(+), 19 deletions(-)

apps/server/src/main.test.ts+1
...@@ -25,6 +25,7 @@ const serverStart = Effect.acquireRelease(...@@ -25,6 +25,7 @@ const serverStart = Effect.acquireRelease(
25);25);
26const findAvailablePort = vi.fn((preferred: number) => Effect.succeed(preferred));26const findAvailablePort = vi.fn((preferred: number) => Effect.succeed(preferred));
2727
28// Shared service layer used by this CLI test suite.
28const testLayer = Layer.mergeAll(29const testLayer = Layer.mergeAll(
29 Layer.succeed(CliConfig, {30 Layer.succeed(CliConfig, {
30 cwd: "/tmp/t3-test-workspace",31 cwd: "/tmp/t3-test-workspace",
apps/server/src/provider/Layers/CodexAdapter.test.ts+31
...@@ -219,6 +219,37 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {...@@ -219,6 +219,37 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
219 assert.equal(firstEvent.value.turnId, "turn-1");219 assert.equal(firstEvent.value.turnId, "turn-1");
220 }),220 }),
221 );221 );
222
223 it.effect("maps session/closed lifecycle events to canonical session.exited runtime events", () =>
224 Effect.gen(function* () {
225 const adapter = yield* CodexAdapter;
226 const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild);
227
228 const event: ProviderEvent = {
229 id: asEventId("evt-session-closed"),
230 kind: "session",
231 provider: "codex",
232 sessionId: asSessionId("sess-1"),
233 createdAt: new Date().toISOString(),
234 method: "session/closed",
235 message: "Session stopped",
236 };
237
238 lifecycleManager.emit("event", event);
239 const firstEvent = yield* Fiber.join(firstEventFiber);
240
241 assert.equal(firstEvent._tag, "Some");
242 if (firstEvent._tag !== "Some") {
243 return;
244 }
245 assert.equal(firstEvent.value.type, "session.exited");
246 if (firstEvent.value.type !== "session.exited") {
247 return;
248 }
249 assert.equal(firstEvent.value.sessionId, "sess-1");
250 assert.equal(firstEvent.value.message, "Session stopped");
251 }),
252 );
222});253});
223254
224afterAll(() => {255afterAll(() => {
apps/server/src/provider/Layers/CodexAdapter.ts+1-1
...@@ -333,7 +333,7 @@ function mapToRuntimeEvents(event: ProviderEvent): ReadonlyArray<ProviderRuntime...@@ -333,7 +333,7 @@ function mapToRuntimeEvents(event: ProviderEvent): ReadonlyArray<ProviderRuntime
333 ];333 ];
334 }334 }
335335
336 if (event.method === "session/exited") {336 if (event.method === "session/exited" || event.method === "session/closed") {
337 return [337 return [
338 {338 {
339 type: "session.exited",339 type: "session.exited",
apps/server/src/terminal/Layers/Manager.test.ts+53-1
...@@ -16,6 +16,7 @@ import { Encoding } from "effect";...@@ -16,6 +16,7 @@ import { Encoding } from "effect";
16class FakePtyProcess implements PtyProcess {16class FakePtyProcess implements PtyProcess {
17 readonly writes: string[] = [];17 readonly writes: string[] = [];
18 readonly resizeCalls: Array<{ cols: number; rows: number }> = [];18 readonly resizeCalls: Array<{ cols: number; rows: number }> = [];
19 readonly killSignals: Array<string | undefined> = [];
19 private readonly dataListeners = new Set<(data: string) => void>();20 private readonly dataListeners = new Set<(data: string) => void>();
20 private readonly exitListeners = new Set<(event: PtyExitEvent) => void>();21 private readonly exitListeners = new Set<(event: PtyExitEvent) => void>();
21 killed = false;22 killed = false;
...@@ -30,8 +31,9 @@ class FakePtyProcess implements PtyProcess {...@@ -30,8 +31,9 @@ class FakePtyProcess implements PtyProcess {
30 this.resizeCalls.push({ cols, rows });31 this.resizeCalls.push({ cols, rows });
31 }32 }
3233
33 kill(): void {34 kill(signal?: string): void {
34 this.killed = true;35 this.killed = true;
36 this.killSignals.push(signal);
35 }37 }
3638
37 onData(callback: (data: string) => void): () => void {39 onData(callback: (data: string) => void): () => void {
...@@ -146,6 +148,8 @@ describe("TerminalManager", () => {...@@ -146,6 +148,8 @@ describe("TerminalManager", () => {
146 shellResolver?: () => string;148 shellResolver?: () => string;
147 subprocessChecker?: (terminalPid: number) => Promise<boolean>;149 subprocessChecker?: (terminalPid: number) => Promise<boolean>;
148 subprocessPollIntervalMs?: number;150 subprocessPollIntervalMs?: number;
151 processKillGraceMs?: number;
152 maxRetainedInactiveSessions?: number;
149 } = {},153 } = {},
150 ) {154 ) {
151 const logsDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3code-terminal-"));155 const logsDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3code-terminal-"));
...@@ -160,6 +164,10 @@ describe("TerminalManager", () => {...@@ -160,6 +164,10 @@ describe("TerminalManager", () => {
160 ...(options.subprocessPollIntervalMs164 ...(options.subprocessPollIntervalMs
161 ? { subprocessPollIntervalMs: options.subprocessPollIntervalMs }165 ? { subprocessPollIntervalMs: options.subprocessPollIntervalMs }
162 : {}),166 : {}),
167 ...(options.processKillGraceMs ? { processKillGraceMs: options.processKillGraceMs } : {}),
168 ...(options.maxRetainedInactiveSessions
169 ? { maxRetainedInactiveSessions: options.maxRetainedInactiveSessions }
170 : {}),
163 });171 });
164 return { logsDir, ptyAdapter, manager };172 return { logsDir, ptyAdapter, manager };
165 }173 }
...@@ -437,6 +445,50 @@ describe("TerminalManager", () => {...@@ -437,6 +445,50 @@ describe("TerminalManager", () => {
437 manager.dispose();445 manager.dispose();
438 });446 });
439447
448 it("escalates terminal shutdown to SIGKILL when process does not exit in time", async () => {
449 const { manager, ptyAdapter } = makeManager(5, { processKillGraceMs: 10 });
450 await manager.open(openInput());
451 const process = ptyAdapter.processes[0];
452 expect(process).toBeDefined();
453 if (!process) return;
454
455 await manager.close({ threadId: "thread-1" });
456 await waitFor(() => process.killSignals.includes("SIGKILL"));
457
458 expect(process.killSignals[0]).toBe("SIGTERM");
459 expect(process.killSignals).toContain("SIGKILL");
460
461 manager.dispose();
462 });
463
464 it("evicts oldest inactive terminal sessions when retention limit is exceeded", async () => {
465 const { manager, ptyAdapter } = makeManager(5, { maxRetainedInactiveSessions: 1 });
466
467 await manager.open(openInput({ threadId: "thread-1" }));
468 await manager.open(openInput({ threadId: "thread-2" }));
469
470 const first = ptyAdapter.processes[0];
471 const second = ptyAdapter.processes[1];
472 expect(first).toBeDefined();
473 expect(second).toBeDefined();
474 if (!first || !second) return;
475
476 first.emitExit({ exitCode: 0, signal: 0 });
477 await new Promise((resolve) => setTimeout(resolve, 5));
478 second.emitExit({ exitCode: 0, signal: 0 });
479
480 await waitFor(() => {
481 const sessions = (manager as unknown as { sessions: Map<string, unknown> }).sessions;
482 return sessions.size === 1;
483 });
484
485 const sessions = (manager as unknown as { sessions: Map<string, unknown> }).sessions;
486 const keys = [...sessions.keys()];
487 expect(keys).toEqual(["thread-2\u0000default"]);
488
489 manager.dispose();
490 });
491
440 it("migrates legacy transcript filenames to terminal-scoped history path on open", async () => {492 it("migrates legacy transcript filenames to terminal-scoped history path on open", async () => {
441 const { manager, logsDir } = makeManager();493 const { manager, logsDir } = makeManager();
442 const legacyPath = path.join(logsDir, "thread-1.log");494 const legacyPath = path.join(logsDir, "thread-1.log");
apps/server/src/terminal/Layers/Manager.ts+85-14
...@@ -29,6 +29,8 @@ import {...@@ -29,6 +29,8 @@ import {
29const DEFAULT_HISTORY_LINE_LIMIT = 5_000;29const DEFAULT_HISTORY_LINE_LIMIT = 5_000;
30const DEFAULT_PERSIST_DEBOUNCE_MS = 40;30const DEFAULT_PERSIST_DEBOUNCE_MS = 40;
31const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000;31const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000;
32const DEFAULT_PROCESS_KILL_GRACE_MS = 1_000;
33const DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS = 128;
32const DEFAULT_OPEN_COLS = 120;34const DEFAULT_OPEN_COLS = 120;
33const DEFAULT_OPEN_ROWS = 30;35const DEFAULT_OPEN_ROWS = 30;
34const TERMINAL_ENV_BLOCKLIST = new Set(["PORT", "ELECTRON_RENDERER_PORT", "ELECTRON_RUN_AS_NODE"]);36const TERMINAL_ENV_BLOCKLIST = new Set(["PORT", "ELECTRON_RENDERER_PORT", "ELECTRON_RUN_AS_NODE"]);
...@@ -279,6 +281,8 @@ interface TerminalManagerOptions {...@@ -279,6 +281,8 @@ interface TerminalManagerOptions {
279 shellResolver?: () => string;281 shellResolver?: () => string;
280 subprocessChecker?: TerminalSubprocessChecker;282 subprocessChecker?: TerminalSubprocessChecker;
281 subprocessPollIntervalMs?: number;283 subprocessPollIntervalMs?: number;
284 processKillGraceMs?: number;
285 maxRetainedInactiveSessions?: number;
282}286}
283287
284export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents> {288export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents> {
...@@ -294,8 +298,11 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>...@@ -294,8 +298,11 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>
294 private readonly persistDebounceMs: number;298 private readonly persistDebounceMs: number;
295 private readonly subprocessChecker: TerminalSubprocessChecker;299 private readonly subprocessChecker: TerminalSubprocessChecker;
296 private readonly subprocessPollIntervalMs: number;300 private readonly subprocessPollIntervalMs: number;
301 private readonly processKillGraceMs: number;
302 private readonly maxRetainedInactiveSessions: number;
297 private subprocessPollTimer: ReturnType<typeof setInterval> | null = null;303 private subprocessPollTimer: ReturnType<typeof setInterval> | null = null;
298 private subprocessPollInFlight = false;304 private subprocessPollInFlight = false;
305 private readonly killEscalationTimers = new Map<PtyProcess, ReturnType<typeof setTimeout>>();
299 private readonly logger = createLogger("terminal");306 private readonly logger = createLogger("terminal");
300307
301 constructor(options: TerminalManagerOptions) {308 constructor(options: TerminalManagerOptions) {
...@@ -308,6 +315,9 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>...@@ -308,6 +315,9 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>
308 this.subprocessChecker = options.subprocessChecker ?? defaultSubprocessChecker;315 this.subprocessChecker = options.subprocessChecker ?? defaultSubprocessChecker;
309 this.subprocessPollIntervalMs =316 this.subprocessPollIntervalMs =
310 options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS;317 options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS;
318 this.processKillGraceMs = options.processKillGraceMs ?? DEFAULT_PROCESS_KILL_GRACE_MS;
319 this.maxRetainedInactiveSessions =
320 options.maxRetainedInactiveSessions ?? DEFAULT_MAX_RETAINED_INACTIVE_SESSIONS;
311 fs.mkdirSync(this.logsDir, { recursive: true });321 fs.mkdirSync(this.logsDir, { recursive: true });
312 }322 }
313323
...@@ -342,6 +352,7 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>...@@ -342,6 +352,7 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>
342 runtimeEnv: normalizedRuntimeEnv(input.env),352 runtimeEnv: normalizedRuntimeEnv(input.env),
343 };353 };
344 this.sessions.set(sessionKey, session);354 this.sessions.set(sessionKey, session);
355 this.evictInactiveSessionsIfNeeded();
345 this.startSession(session, { ...input, cols, rows }, "started");356 this.startSession(session, { ...input, cols, rows }, "started");
346 return this.snapshot(session);357 return this.snapshot(session);
347 }358 }
...@@ -453,6 +464,7 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>...@@ -453,6 +464,7 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>
453 runtimeEnv: normalizedRuntimeEnv(input.env),464 runtimeEnv: normalizedRuntimeEnv(input.env),
454 };465 };
455 this.sessions.set(sessionKey, session);466 this.sessions.set(sessionKey, session);
467 this.evictInactiveSessionsIfNeeded();
456 } else {468 } else {
457 this.stopProcess(session);469 this.stopProcess(session);
458 session.cwd = input.cwd;470 session.cwd = input.cwd;
...@@ -506,6 +518,10 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>...@@ -506,6 +518,10 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>
506 clearTimeout(timer);518 clearTimeout(timer);
507 }519 }
508 this.persistTimers.clear();520 this.persistTimers.clear();
521 for (const timer of this.killEscalationTimers.values()) {
522 clearTimeout(timer);
523 }
524 this.killEscalationTimers.clear();
509 this.pendingPersistHistory.clear();525 this.pendingPersistHistory.clear();
510 this.threadLocks.clear();526 this.threadLocks.clear();
511 this.persistQueues.clear();527 this.persistQueues.clear();
...@@ -584,17 +600,14 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>...@@ -584,17 +600,14 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>
584 });600 });
585 } catch (error) {601 } catch (error) {
586 if (ptyProcess) {602 if (ptyProcess) {
587 try {603 this.killProcessWithEscalation(ptyProcess, session.threadId, session.terminalId);
588 ptyProcess.kill();
589 } catch {
590 // Ignore kill errors during failed startup cleanup.
591 }
592 }604 }
593 session.status = "error";605 session.status = "error";
594 session.pid = null;606 session.pid = null;
595 session.process = null;607 session.process = null;
596 session.hasRunningSubprocess = false;608 session.hasRunningSubprocess = false;
597 session.updatedAt = new Date().toISOString();609 session.updatedAt = new Date().toISOString();
610 this.evictInactiveSessionsIfNeeded();
598 this.updateSubprocessPollingState();611 this.updateSubprocessPollingState();
599 const message = error instanceof Error ? error.message : "Terminal start failed";612 const message = error instanceof Error ? error.message : "Terminal start failed";
600 this.emitEvent({613 this.emitEvent({
...@@ -627,6 +640,7 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>...@@ -627,6 +640,7 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>
627 }640 }
628641
629 private onProcessExit(session: TerminalSessionState, event: PtyExitEvent): void {642 private onProcessExit(session: TerminalSessionState, event: PtyExitEvent): void {
643 this.clearKillEscalationTimer(session.process);
630 this.cleanupProcessHandles(session);644 this.cleanupProcessHandles(session);
631 session.process = null;645 session.process = null;
632 session.pid = null;646 session.pid = null;
...@@ -643,6 +657,7 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>...@@ -643,6 +657,7 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>
643 exitCode: session.exitCode,657 exitCode: session.exitCode,
644 exitSignal: session.exitSignal,658 exitSignal: session.exitSignal,
645 });659 });
660 this.evictInactiveSessionsIfNeeded();
646 this.updateSubprocessPollingState();661 this.updateSubprocessPollingState();
647 }662 }
648663
...@@ -655,24 +670,80 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>...@@ -655,24 +670,80 @@ export class TerminalManagerRuntime extends EventEmitter<TerminalManagerEvents>
655 session.hasRunningSubprocess = false;670 session.hasRunningSubprocess = false;
656 session.status = "exited";671 session.status = "exited";
657 session.updatedAt = new Date().toISOString();672 session.updatedAt = new Date().toISOString();
673 this.killProcessWithEscalation(process, session.threadId, session.terminalId);
674 this.evictInactiveSessionsIfNeeded();
675 this.updateSubprocessPollingState();
676 }
677
678 private cleanupProcessHandles(session: TerminalSessionState): void {
679 session.unsubscribeData?.();
680 session.unsubscribeData = null;
681 session.unsubscribeExit?.();
682 session.unsubscribeExit = null;
683 }
684
685 private clearKillEscalationTimer(process: PtyProcess | null): void {
686 if (!process) return;
687 const timer = this.killEscalationTimers.get(process);
688 if (!timer) return;
689 clearTimeout(timer);
690 this.killEscalationTimers.delete(process);
691 }
692
693 private killProcessWithEscalation(process: PtyProcess, threadId: string, terminalId: string): void {
694 this.clearKillEscalationTimer(process);
658 try {695 try {
659 process.kill();696 process.kill("SIGTERM");
660 } catch (error) {697 } catch (error) {
661 const message = error instanceof Error ? error.message : String(error);698 const message = error instanceof Error ? error.message : String(error);
662 this.logger.warn("failed to kill terminal process", {699 this.logger.warn("failed to kill terminal process", {
663 threadId: session.threadId,700 threadId,
664 terminalId: session.terminalId,701 terminalId,
702 signal: "SIGTERM",
665 error: message,703 error: message,
666 });704 });
705 return;
667 }706 }
668 this.updateSubprocessPollingState();707
708 const timer = setTimeout(() => {
709 this.killEscalationTimers.delete(process);
710 try {
711 process.kill("SIGKILL");
712 } catch (error) {
713 const message = error instanceof Error ? error.message : String(error);
714 this.logger.warn("failed to force-kill terminal process", {
715 threadId,
716 terminalId,
717 signal: "SIGKILL",
718 error: message,
719 });
720 }
721 }, this.processKillGraceMs);
722 timer.unref?.();
723 this.killEscalationTimers.set(process, timer);
669 }724 }
670725
671 private cleanupProcessHandles(session: TerminalSessionState): void {726 private evictInactiveSessionsIfNeeded(): void {
672 session.unsubscribeData?.();727 const inactiveSessions = [...this.sessions.values()].filter((session) => session.status !== "running");
673 session.unsubscribeData = null;728 if (inactiveSessions.length <= this.maxRetainedInactiveSessions) {
674 session.unsubscribeExit?.();729 return;
675 session.unsubscribeExit = null;730 }
731
732 inactiveSessions.sort(
733 (left, right) =>
734 left.updatedAt.localeCompare(right.updatedAt) ||
735 left.threadId.localeCompare(right.threadId) ||
736 left.terminalId.localeCompare(right.terminalId),
737 );
738 const toEvict = inactiveSessions.length - this.maxRetainedInactiveSessions;
739 for (const session of inactiveSessions.slice(0, toEvict)) {
740 const key = toSessionKey(session.threadId, session.terminalId);
741 this.sessions.delete(key);
742 this.clearPersistTimer(session.threadId, session.terminalId);
743 this.pendingPersistHistory.delete(key);
744 this.persistQueues.delete(key);
745 this.clearKillEscalationTimer(session.process);
746 }
676 }747 }
677748
678 private queuePersist(threadId: string, terminalId: string, history: string): void {749 private queuePersist(threadId: string, terminalId: string, history: string): void {
apps/server/src/wsServer.ts+1-3
...@@ -593,9 +593,7 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<...@@ -593,9 +593,7 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<
593 const unsubscribeTerminalEvents = yield* terminalManager.subscribe(593 const unsubscribeTerminalEvents = yield* terminalManager.subscribe(
594 (event) => void Effect.runPromise(onTerminalEvent(event)),594 (event) => void Effect.runPromise(onTerminalEvent(event)),
595 );595 );
596 yield* Effect.addFinalizer(() =>596 yield* Effect.addFinalizer(() => Effect.sync(() => unsubscribeTerminalEvents()));
597 Effect.all([Effect.sync(() => unsubscribeTerminalEvents()), terminalManager.dispose()]),
598 );
599597
600 yield* NodeHttpServer.make(() => httpServer, listenOptions);598 yield* NodeHttpServer.make(() => httpServer, listenOptions);
601599