From ddc4a93c36e0cdd263d91d01e11e900c8d9ee0ef Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Feb 2026 22:01:52 -0800 Subject: [PATCH] Emit terminal activity events from subprocess polling - Add cross-platform subprocess polling in `TerminalManager` and emit `activity` events when child-process state changes - Update web store to derive `runningTerminalIds` from `activity.hasRunningSubprocess` instead of start/restart events - Extend terminal contracts and tests to include the new `activity` event type Co-authored-by: codex --- apps/server/src/terminalManager.test.ts | 44 +++++- apps/server/src/terminalManager.ts | 191 +++++++++++++++++++++++- apps/web/src/store.test.ts | 29 +++- apps/web/src/store.ts | 4 +- packages/contracts/src/terminal.test.ts | 11 ++ packages/contracts/src/terminal.ts | 6 + 6 files changed, 277 insertions(+), 8 deletions(-) diff --git a/apps/server/src/terminalManager.test.ts b/apps/server/src/terminalManager.test.ts index 9b77d582bc21db4dfd4dee5c600fec64e3904a38..59e8cae453074f71902de5b31b1e32bafdee221e 100644 --- a/apps/server/src/terminalManager.test.ts +++ b/apps/server/src/terminalManager.test.ts @@ -137,7 +137,11 @@ describe("TerminalManager", () => { function makeManager( historyLineLimit = 5, - options: { shellResolver?: () => string } = {}, + options: { + shellResolver?: () => string; + subprocessChecker?: (terminalPid: number) => Promise; + subprocessPollIntervalMs?: number; + } = {}, ) { const logsDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3code-terminal-")); tempDirs.push(logsDir); @@ -147,6 +151,8 @@ describe("TerminalManager", () => { ptyAdapter, historyLineLimit, shellResolver: options.shellResolver ?? (() => "/bin/bash"), + subprocessChecker: options.subprocessChecker, + subprocessPollIntervalMs: options.subprocessPollIntervalMs, }); return { logsDir, ptyAdapter, manager }; } @@ -273,6 +279,42 @@ describe("TerminalManager", () => { manager.dispose(); }); + it("emits subprocess activity events when child-process state changes", async () => { + let hasRunningSubprocess = false; + const { manager } = makeManager(5, { + subprocessChecker: async () => hasRunningSubprocess, + subprocessPollIntervalMs: 20, + }); + const events: TerminalEvent[] = []; + manager.on("event", (event) => { + events.push(event); + }); + + await manager.open(openInput()); + await waitFor(() => events.some((event) => event.type === "started")); + expect(events.some((event) => event.type === "activity")).toBe(false); + + hasRunningSubprocess = true; + await waitFor( + () => + events.some( + (event) => event.type === "activity" && event.hasRunningSubprocess === true, + ), + 1_200, + ); + + hasRunningSubprocess = false; + await waitFor( + () => + events.some( + (event) => event.type === "activity" && event.hasRunningSubprocess === false, + ), + 1_200, + ); + + manager.dispose(); + }); + it("caps persisted history to configured line limit", async () => { const { manager, ptyAdapter } = makeManager(3); await manager.open(openInput()); diff --git a/apps/server/src/terminalManager.ts b/apps/server/src/terminalManager.ts index 8fe6d662e7db7c9e9486936d26525b065aa7bcfd..eb881f3cd011050781e0ca784370477565c24fed 100644 --- a/apps/server/src/terminalManager.ts +++ b/apps/server/src/terminalManager.ts @@ -21,9 +21,13 @@ import { import { createLogger } from "./logger"; import { NodePtyAdapter, type PtyAdapter, type PtyExitEvent, type PtyProcess } from "./ptyAdapter"; +import { runProcess } from "./processRunner"; const DEFAULT_HISTORY_LINE_LIMIT = 5_000; const DEFAULT_PERSIST_DEBOUNCE_MS = 40; +const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000; + +type TerminalSubprocessChecker = (terminalPid: number) => Promise; export interface TerminalManagerEvents { event: [event: TerminalEvent]; @@ -34,6 +38,8 @@ export interface TerminalManagerOptions { historyLineLimit?: number; ptyAdapter?: PtyAdapter; shellResolver?: () => string; + subprocessChecker?: TerminalSubprocessChecker; + subprocessPollIntervalMs?: number; } interface TerminalSessionState { @@ -51,6 +57,7 @@ interface TerminalSessionState { process: PtyProcess | null; unsubscribeData: (() => void) | null; unsubscribeExit: (() => void) | null; + hasRunningSubprocess: boolean; } function defaultShellResolver(): string { @@ -122,6 +129,83 @@ function isRetryableShellSpawnError(error: unknown): boolean { ); } +async function checkWindowsSubprocessActivity(terminalPid: number): Promise { + const command = [ + `$children = Get-CimInstance Win32_Process -Filter "ParentProcessId = ${terminalPid}" -ErrorAction SilentlyContinue`, + "if ($children) { exit 0 }", + "exit 1", + ].join("; "); + try { + const result = await runProcess( + "powershell.exe", + ["-NoProfile", "-NonInteractive", "-Command", command], + { + timeoutMs: 1_500, + allowNonZeroExit: true, + maxBufferBytes: 32_768, + outputMode: "truncate", + }, + ); + return result.code === 0; + } catch { + return false; + } +} + +async function checkPosixSubprocessActivity(terminalPid: number): Promise { + try { + const pgrepResult = await runProcess("pgrep", ["-P", String(terminalPid)], { + timeoutMs: 1_000, + allowNonZeroExit: true, + maxBufferBytes: 32_768, + outputMode: "truncate", + }); + if (pgrepResult.code === 0) { + return pgrepResult.stdout.trim().length > 0; + } + if (pgrepResult.code === 1) { + return false; + } + } catch { + // Fall back to ps when pgrep is unavailable. + } + + try { + const psResult = await runProcess("ps", ["-eo", "pid=,ppid="], { + timeoutMs: 1_000, + allowNonZeroExit: true, + maxBufferBytes: 262_144, + outputMode: "truncate", + }); + if (psResult.code !== 0) { + return false; + } + + for (const line of psResult.stdout.split(/\r?\n/g)) { + const [pidRaw, ppidRaw] = line.trim().split(/\s+/g); + const pid = Number(pidRaw); + const ppid = Number(ppidRaw); + if (!Number.isInteger(pid) || !Number.isInteger(ppid)) continue; + if (ppid === terminalPid) { + return true; + } + } + return false; + } catch { + return false; + } +} + +async function defaultSubprocessChecker(terminalPid: number): Promise { + if (!Number.isInteger(terminalPid) || terminalPid <= 0) { + return false; + } + if (process.platform === "win32") { + return checkWindowsSubprocessActivity(terminalPid); + } + return checkPosixSubprocessActivity(terminalPid); +} + function capHistory(history: string, maxLines: number): string { if (history.length === 0) return history; const hasTrailingNewline = history.endsWith("\n"); @@ -161,6 +245,10 @@ export class TerminalManager extends EventEmitter { private readonly pendingPersistHistory = new Map(); private readonly threadLocks = new Map>(); private readonly persistDebounceMs: number; + private readonly subprocessChecker: TerminalSubprocessChecker; + private readonly subprocessPollIntervalMs: number; + private subprocessPollTimer: ReturnType | null = null; + private subprocessPollInFlight = false; private readonly logger = createLogger("terminal"); constructor(options: TerminalManagerOptions = {}) { @@ -170,6 +258,9 @@ export class TerminalManager extends EventEmitter { this.ptyAdapter = options.ptyAdapter ?? new NodePtyAdapter(); this.shellResolver = options.shellResolver ?? defaultShellResolver; this.persistDebounceMs = DEFAULT_PERSIST_DEBOUNCE_MS; + this.subprocessChecker = options.subprocessChecker ?? defaultSubprocessChecker; + this.subprocessPollIntervalMs = + options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS; fs.mkdirSync(this.logsDir, { recursive: true }); } @@ -198,6 +289,7 @@ export class TerminalManager extends EventEmitter { process: null, unsubscribeData: null, unsubscribeExit: null, + hasRunningSubprocess: false, }; this.sessions.set(sessionKey, session); this.startSession(session, input, "started"); @@ -294,6 +386,7 @@ export class TerminalManager extends EventEmitter { process: null, unsubscribeData: null, unsubscribeExit: null, + hasRunningSubprocess: false, }; this.sessions.set(sessionKey, session); } else { @@ -334,14 +427,17 @@ export class TerminalManager extends EventEmitter { if (input.deleteHistory) { await this.deleteAllHistoryForThread(input.threadId); } + this.updateSubprocessPollingState(); }); } dispose(): void { - for (const session of this.sessions.values()) { - this.stopProcess(session); - } + this.stopSubprocessPolling(); + const sessions = [...this.sessions.values()]; this.sessions.clear(); + for (const session of sessions) { + this.stopProcess(session); + } for (const timer of this.persistTimers.values()) { clearTimeout(timer); } @@ -364,6 +460,7 @@ export class TerminalManager extends EventEmitter { session.rows = input.rows; session.exitCode = null; session.exitSignal = null; + session.hasRunningSubprocess = false; session.updatedAt = new Date().toISOString(); let ptyProcess: PtyProcess | null = null; @@ -413,6 +510,7 @@ export class TerminalManager extends EventEmitter { session.unsubscribeExit = ptyProcess.onExit((event) => { this.onProcessExit(session, event); }); + this.updateSubprocessPollingState(); this.emitEvent({ type: eventType, threadId: session.threadId, @@ -431,7 +529,9 @@ export class TerminalManager extends EventEmitter { session.status = "error"; session.pid = null; session.process = null; + session.hasRunningSubprocess = false; session.updatedAt = new Date().toISOString(); + this.updateSubprocessPollingState(); const message = error instanceof Error ? error.message : "Terminal start failed"; this.emitEvent({ type: "error", @@ -466,6 +566,7 @@ export class TerminalManager extends EventEmitter { this.cleanupProcessHandles(session); session.process = null; session.pid = null; + session.hasRunningSubprocess = false; session.status = "exited"; session.exitCode = Number.isInteger(event.exitCode) ? event.exitCode : null; session.exitSignal = Number.isInteger(event.signal) ? event.signal : null; @@ -478,6 +579,7 @@ export class TerminalManager extends EventEmitter { exitCode: session.exitCode, exitSignal: session.exitSignal, }); + this.updateSubprocessPollingState(); } private stopProcess(session: TerminalSessionState): void { @@ -486,6 +588,7 @@ export class TerminalManager extends EventEmitter { this.cleanupProcessHandles(session); session.process = null; session.pid = null; + session.hasRunningSubprocess = false; session.status = "exited"; session.updatedAt = new Date().toISOString(); try { @@ -498,6 +601,7 @@ export class TerminalManager extends EventEmitter { error: message, }); } + this.updateSubprocessPollingState(); } private cleanupProcessHandles(session: TerminalSessionState): void { @@ -660,6 +764,86 @@ export class TerminalManager extends EventEmitter { } } + private updateSubprocessPollingState(): void { + const hasRunningSessions = [...this.sessions.values()].some( + (session) => session.status === "running" && session.pid !== null, + ); + if (hasRunningSessions) { + this.ensureSubprocessPolling(); + return; + } + this.stopSubprocessPolling(); + } + + private ensureSubprocessPolling(): void { + if (this.subprocessPollTimer) return; + this.subprocessPollTimer = setInterval(() => { + void this.pollSubprocessActivity(); + }, this.subprocessPollIntervalMs); + this.subprocessPollTimer.unref?.(); + void this.pollSubprocessActivity(); + } + + private stopSubprocessPolling(): void { + if (!this.subprocessPollTimer) return; + clearInterval(this.subprocessPollTimer); + this.subprocessPollTimer = null; + } + + private async pollSubprocessActivity(): Promise { + if (this.subprocessPollInFlight) return; + + const runningSessions = [...this.sessions.values()].filter( + (session): session is TerminalSessionState & { pid: number } => + session.status === "running" && Number.isInteger(session.pid), + ); + if (runningSessions.length === 0) { + this.stopSubprocessPolling(); + return; + } + + this.subprocessPollInFlight = true; + try { + await Promise.all( + runningSessions.map(async (session) => { + const terminalPid = session.pid; + let hasRunningSubprocess = false; + try { + hasRunningSubprocess = await this.subprocessChecker(terminalPid); + } catch (error) { + this.logger.warn("failed to check terminal subprocess activity", { + threadId: session.threadId, + terminalId: session.terminalId, + terminalPid, + error: error instanceof Error ? error.message : String(error), + }); + return; + } + + const liveSession = this.sessions.get(toSessionKey(session.threadId, session.terminalId)); + if (!liveSession || liveSession.status !== "running" || liveSession.pid !== terminalPid) { + return; + } + if (liveSession.hasRunningSubprocess === hasRunningSubprocess) { + return; + } + + liveSession.hasRunningSubprocess = hasRunningSubprocess; + liveSession.updatedAt = new Date().toISOString(); + this.emitEvent({ + type: "activity", + threadId: liveSession.threadId, + terminalId: liveSession.terminalId, + createdAt: new Date().toISOString(), + hasRunningSubprocess, + }); + }), + ); + } finally { + this.subprocessPollInFlight = false; + } + } + private async assertValidCwd(cwd: string): Promise { let stats: fs.Stats; try { @@ -686,6 +870,7 @@ export class TerminalManager extends EventEmitter { this.stopProcess(session); this.sessions.delete(key); } + this.updateSubprocessPollingState(); await this.flushPersistQueue(threadId, terminalId); if (deleteHistory) { await this.deleteHistory(threadId, terminalId); diff --git a/apps/web/src/store.test.ts b/apps/web/src/store.test.ts index 3e157b1b25bb4b3c249d043ac1ff78e2a779c8f2..3294cef5030a350594031c8847eb8631b1720155 100644 --- a/apps/web/src/store.test.ts +++ b/apps/web/src/store.test.ts @@ -97,6 +97,17 @@ function makeTerminalStartedEvent(overrides: Partial = {}): Termi }; } +function makeTerminalActivityEvent(overrides: Partial = {}): TerminalEvent { + return { + type: "activity", + threadId: "thread-local-1", + terminalId: DEFAULT_THREAD_TERMINAL_ID, + createdAt: "2026-02-09T00:00:02.000Z", + hasRunningSubprocess: true, + ...overrides, + }; +} + describe("store reducer thread continuity", () => { it("stores codexThreadId from UPDATE_SESSION", () => { const state = makeState( @@ -298,15 +309,27 @@ describe("store reducer thread continuity", () => { ]); }); - it("tracks running terminals from terminal lifecycle events", () => { + it("tracks running terminals from subprocess activity events", () => { const state = makeState(makeThread()); const started = reducer(state, { type: "APPLY_TERMINAL_EVENT", event: makeTerminalStartedEvent(), }); - expect(started.threads[0]?.runningTerminalIds).toEqual([DEFAULT_THREAD_TERMINAL_ID]); + expect(started.threads[0]?.runningTerminalIds).toEqual([]); - const exited = reducer(started, { + const active = reducer(started, { + type: "APPLY_TERMINAL_EVENT", + event: makeTerminalActivityEvent(), + }); + expect(active.threads[0]?.runningTerminalIds).toEqual([DEFAULT_THREAD_TERMINAL_ID]); + + const idle = reducer(active, { + type: "APPLY_TERMINAL_EVENT", + event: makeTerminalActivityEvent({ hasRunningSubprocess: false }), + }); + expect(idle.threads[0]?.runningTerminalIds).toEqual([]); + + const exited = reducer(active, { type: "APPLY_TERMINAL_EVENT", event: { type: "exited", diff --git a/apps/web/src/store.ts b/apps/web/src/store.ts index e7937f4d3b4ee2bd3d6d891a27d0b59f0e012cd5..79cbfa7077ccb4fe4b45081bb83290eb84196f6b 100644 --- a/apps/web/src/store.ts +++ b/apps/web/src/store.ts @@ -678,7 +678,9 @@ export function reducer(state: AppState, action: Action): AppState { action.event.type === "started" || action.event.type === "restarted" ) { - if (action.event.snapshot.status === "running") { + runningTerminalIdSet.delete(action.event.terminalId); + } else if (action.event.type === "activity") { + if (action.event.hasRunningSubprocess) { runningTerminalIdSet.add(action.event.terminalId); } else { runningTerminalIdSet.delete(action.event.terminalId); diff --git a/packages/contracts/src/terminal.test.ts b/packages/contracts/src/terminal.test.ts index 336a0476265b7b45dbf917e654791839f35902b2..8704ee1a53375cadd41d8ca52e0c2991f2e1eb8c 100644 --- a/packages/contracts/src/terminal.test.ts +++ b/packages/contracts/src/terminal.test.ts @@ -139,4 +139,15 @@ describe("terminalEventSchema", () => { }); expect(result.success).toBe(true); }); + + it("accepts activity events", () => { + const result = terminalEventSchema.safeParse({ + type: "activity", + threadId: "thread-1", + terminalId: DEFAULT_TERMINAL_ID, + createdAt: new Date().toISOString(), + hasRunningSubprocess: true, + }); + expect(result.success).toBe(true); + }); }); diff --git a/packages/contracts/src/terminal.ts b/packages/contracts/src/terminal.ts index cec35f5956f2f7172a82902c54ed5da9deb0f498..57d2d9f63c0c1d051a9a836f4c34166f9ead9dd3 100644 --- a/packages/contracts/src/terminal.ts +++ b/packages/contracts/src/terminal.ts @@ -91,6 +91,11 @@ export const terminalRestartedEventSchema = terminalEventBaseSchema.extend({ snapshot: terminalSessionSnapshotSchema, }); +export const terminalActivityEventSchema = terminalEventBaseSchema.extend({ + type: z.literal("activity"), + hasRunningSubprocess: z.boolean(), +}); + export const terminalEventSchema = z.discriminatedUnion("type", [ terminalStartedEventSchema, terminalOutputEventSchema, @@ -98,6 +103,7 @@ export const terminalEventSchema = z.discriminatedUnion("type", [ terminalErrorEventSchema, terminalClearedEventSchema, terminalRestartedEventSchema, + terminalActivityEventSchema, ]); export type TerminalThreadInput = z.input; -- 2.54.0