authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-12 22:01:52-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-12 22:20:01-08:00
logddc4a93c36e0cdd263d91d01e11e900c8d9ee0ef
tree852a401425f6c019316050d00e9dcb3b6494749c
parent901579233cf0c846abcd9622f1443db6d3bb5834

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 <codex@users.noreply.github.com>

6 files changed, 276 insertions(+), 7 deletions(-)

apps/server/src/terminalManager.test.ts+43-1
......@@ -137,7 +137,11 @@ describe("TerminalManager", () => {
137137
138138 function makeManager(
139139 historyLineLimit = 5,
140 options: { shellResolver?: () => string } = {},
140 options: {
141 shellResolver?: () => string;
142 subprocessChecker?: (terminalPid: number) => Promise<boolean>;
143 subprocessPollIntervalMs?: number;
144 } = {},
141145 ) {
142146 const logsDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3code-terminal-"));
143147 tempDirs.push(logsDir);
......@@ -147,6 +151,8 @@ describe("TerminalManager", () => {
147151 ptyAdapter,
148152 historyLineLimit,
149153 shellResolver: options.shellResolver ?? (() => "/bin/bash"),
154 subprocessChecker: options.subprocessChecker,
155 subprocessPollIntervalMs: options.subprocessPollIntervalMs,
150156 });
151157 return { logsDir, ptyAdapter, manager };
152158 }
......@@ -273,6 +279,42 @@ describe("TerminalManager", () => {
273279 manager.dispose();
274280 });
275281
282 it("emits subprocess activity events when child-process state changes", async () => {
283 let hasRunningSubprocess = false;
284 const { manager } = makeManager(5, {
285 subprocessChecker: async () => hasRunningSubprocess,
286 subprocessPollIntervalMs: 20,
287 });
288 const events: TerminalEvent[] = [];
289 manager.on("event", (event) => {
290 events.push(event);
291 });
292
293 await manager.open(openInput());
294 await waitFor(() => events.some((event) => event.type === "started"));
295 expect(events.some((event) => event.type === "activity")).toBe(false);
296
297 hasRunningSubprocess = true;
298 await waitFor(
299 () =>
300 events.some(
301 (event) => event.type === "activity" && event.hasRunningSubprocess === true,
302 ),
303 1_200,
304 );
305
306 hasRunningSubprocess = false;
307 await waitFor(
308 () =>
309 events.some(
310 (event) => event.type === "activity" && event.hasRunningSubprocess === false,
311 ),
312 1_200,
313 );
314
315 manager.dispose();
316 });
317
276318 it("caps persisted history to configured line limit", async () => {
277319 const { manager, ptyAdapter } = makeManager(3);
278320 await manager.open(openInput());
apps/server/src/terminalManager.ts+187-2
......@@ -21,9 +21,13 @@ import {
2121
2222import { createLogger } from "./logger";
2323import { NodePtyAdapter, type PtyAdapter, type PtyExitEvent, type PtyProcess } from "./ptyAdapter";
24import { runProcess } from "./processRunner";
2425
2526const DEFAULT_HISTORY_LINE_LIMIT = 5_000;
2627const DEFAULT_PERSIST_DEBOUNCE_MS = 40;
28const DEFAULT_SUBPROCESS_POLL_INTERVAL_MS = 1_000;
29
30type TerminalSubprocessChecker = (terminalPid: number) => Promise<boolean>;
2731
2832export interface TerminalManagerEvents {
2933 event: [event: TerminalEvent];
......@@ -34,6 +38,8 @@ export interface TerminalManagerOptions {
3438 historyLineLimit?: number;
3539 ptyAdapter?: PtyAdapter;
3640 shellResolver?: () => string;
41 subprocessChecker?: TerminalSubprocessChecker;
42 subprocessPollIntervalMs?: number;
3743}
3844
3945interface TerminalSessionState {
......@@ -51,6 +57,7 @@ interface TerminalSessionState {
5157 process: PtyProcess | null;
5258 unsubscribeData: (() => void) | null;
5359 unsubscribeExit: (() => void) | null;
60 hasRunningSubprocess: boolean;
5461}
5562
5663function defaultShellResolver(): string {
......@@ -122,6 +129,83 @@ function isRetryableShellSpawnError(error: unknown): boolean {
122129 );
123130}
124131
132async function checkWindowsSubprocessActivity(terminalPid: number): Promise<boolean> {
133 const command = [
134 `$children = Get-CimInstance Win32_Process -Filter "ParentProcessId = ${terminalPid}" -ErrorAction SilentlyContinue`,
135 "if ($children) { exit 0 }",
136 "exit 1",
137 ].join("; ");
138 try {
139 const result = await runProcess(
140 "powershell.exe",
141 ["-NoProfile", "-NonInteractive", "-Command", command],
142 {
143 timeoutMs: 1_500,
144 allowNonZeroExit: true,
145 maxBufferBytes: 32_768,
146 outputMode: "truncate",
147 },
148 );
149 return result.code === 0;
150 } catch {
151 return false;
152 }
153}
154
155async function checkPosixSubprocessActivity(terminalPid: number): Promise<boolean> {
156 try {
157 const pgrepResult = await runProcess("pgrep", ["-P", String(terminalPid)], {
158 timeoutMs: 1_000,
159 allowNonZeroExit: true,
160 maxBufferBytes: 32_768,
161 outputMode: "truncate",
162 });
163 if (pgrepResult.code === 0) {
164 return pgrepResult.stdout.trim().length > 0;
165 }
166 if (pgrepResult.code === 1) {
167 return false;
168 }
169 } catch {
170 // Fall back to ps when pgrep is unavailable.
171 }
172
173 try {
174 const psResult = await runProcess("ps", ["-eo", "pid=,ppid="], {
175 timeoutMs: 1_000,
176 allowNonZeroExit: true,
177 maxBufferBytes: 262_144,
178 outputMode: "truncate",
179 });
180 if (psResult.code !== 0) {
181 return false;
182 }
183
184 for (const line of psResult.stdout.split(/\r?\n/g)) {
185 const [pidRaw, ppidRaw] = line.trim().split(/\s+/g);
186 const pid = Number(pidRaw);
187 const ppid = Number(ppidRaw);
188 if (!Number.isInteger(pid) || !Number.isInteger(ppid)) continue;
189 if (ppid === terminalPid) {
190 return true;
191 }
192 }
193 return false;
194 } catch {
195 return false;
196 }
197}
198
199async function defaultSubprocessChecker(terminalPid: number): Promise<boolean> {
200 if (!Number.isInteger(terminalPid) || terminalPid <= 0) {
201 return false;
202 }
203 if (process.platform === "win32") {
204 return checkWindowsSubprocessActivity(terminalPid);
205 }
206 return checkPosixSubprocessActivity(terminalPid);
207}
208
125209function capHistory(history: string, maxLines: number): string {
126210 if (history.length === 0) return history;
127211 const hasTrailingNewline = history.endsWith("\n");
......@@ -161,6 +245,10 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
161245 private readonly pendingPersistHistory = new Map<string, string>();
162246 private readonly threadLocks = new Map<string, Promise<void>>();
163247 private readonly persistDebounceMs: number;
248 private readonly subprocessChecker: TerminalSubprocessChecker;
249 private readonly subprocessPollIntervalMs: number;
250 private subprocessPollTimer: ReturnType<typeof setInterval> | null = null;
251 private subprocessPollInFlight = false;
164252 private readonly logger = createLogger("terminal");
165253
166254 constructor(options: TerminalManagerOptions = {}) {
......@@ -170,6 +258,9 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
170258 this.ptyAdapter = options.ptyAdapter ?? new NodePtyAdapter();
171259 this.shellResolver = options.shellResolver ?? defaultShellResolver;
172260 this.persistDebounceMs = DEFAULT_PERSIST_DEBOUNCE_MS;
261 this.subprocessChecker = options.subprocessChecker ?? defaultSubprocessChecker;
262 this.subprocessPollIntervalMs =
263 options.subprocessPollIntervalMs ?? DEFAULT_SUBPROCESS_POLL_INTERVAL_MS;
173264 fs.mkdirSync(this.logsDir, { recursive: true });
174265 }
175266
......@@ -198,6 +289,7 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
198289 process: null,
199290 unsubscribeData: null,
200291 unsubscribeExit: null,
292 hasRunningSubprocess: false,
201293 };
202294 this.sessions.set(sessionKey, session);
203295 this.startSession(session, input, "started");
......@@ -294,6 +386,7 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
294386 process: null,
295387 unsubscribeData: null,
296388 unsubscribeExit: null,
389 hasRunningSubprocess: false,
297390 };
298391 this.sessions.set(sessionKey, session);
299392 } else {
......@@ -334,14 +427,17 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
334427 if (input.deleteHistory) {
335428 await this.deleteAllHistoryForThread(input.threadId);
336429 }
430 this.updateSubprocessPollingState();
337431 });
338432 }
339433
340434 dispose(): void {
341 for (const session of this.sessions.values()) {
435 this.stopSubprocessPolling();
436 const sessions = [...this.sessions.values()];
437 this.sessions.clear();
438 for (const session of sessions) {
342439 this.stopProcess(session);
343440 }
344 this.sessions.clear();
345441 for (const timer of this.persistTimers.values()) {
346442 clearTimeout(timer);
347443 }
......@@ -364,6 +460,7 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
364460 session.rows = input.rows;
365461 session.exitCode = null;
366462 session.exitSignal = null;
463 session.hasRunningSubprocess = false;
367464 session.updatedAt = new Date().toISOString();
368465
369466 let ptyProcess: PtyProcess | null = null;
......@@ -413,6 +510,7 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
413510 session.unsubscribeExit = ptyProcess.onExit((event) => {
414511 this.onProcessExit(session, event);
415512 });
513 this.updateSubprocessPollingState();
416514 this.emitEvent({
417515 type: eventType,
418516 threadId: session.threadId,
......@@ -431,7 +529,9 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
431529 session.status = "error";
432530 session.pid = null;
433531 session.process = null;
532 session.hasRunningSubprocess = false;
434533 session.updatedAt = new Date().toISOString();
534 this.updateSubprocessPollingState();
435535 const message = error instanceof Error ? error.message : "Terminal start failed";
436536 this.emitEvent({
437537 type: "error",
......@@ -466,6 +566,7 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
466566 this.cleanupProcessHandles(session);
467567 session.process = null;
468568 session.pid = null;
569 session.hasRunningSubprocess = false;
469570 session.status = "exited";
470571 session.exitCode = Number.isInteger(event.exitCode) ? event.exitCode : null;
471572 session.exitSignal = Number.isInteger(event.signal) ? event.signal : null;
......@@ -478,6 +579,7 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
478579 exitCode: session.exitCode,
479580 exitSignal: session.exitSignal,
480581 });
582 this.updateSubprocessPollingState();
481583 }
482584
483585 private stopProcess(session: TerminalSessionState): void {
......@@ -486,6 +588,7 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
486588 this.cleanupProcessHandles(session);
487589 session.process = null;
488590 session.pid = null;
591 session.hasRunningSubprocess = false;
489592 session.status = "exited";
490593 session.updatedAt = new Date().toISOString();
491594 try {
......@@ -498,6 +601,7 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
498601 error: message,
499602 });
500603 }
604 this.updateSubprocessPollingState();
501605 }
502606
503607 private cleanupProcessHandles(session: TerminalSessionState): void {
......@@ -660,6 +764,86 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
660764 }
661765 }
662766
767 private updateSubprocessPollingState(): void {
768 const hasRunningSessions = [...this.sessions.values()].some(
769 (session) => session.status === "running" && session.pid !== null,
770 );
771 if (hasRunningSessions) {
772 this.ensureSubprocessPolling();
773 return;
774 }
775 this.stopSubprocessPolling();
776 }
777
778 private ensureSubprocessPolling(): void {
779 if (this.subprocessPollTimer) return;
780 this.subprocessPollTimer = setInterval(() => {
781 void this.pollSubprocessActivity();
782 }, this.subprocessPollIntervalMs);
783 this.subprocessPollTimer.unref?.();
784 void this.pollSubprocessActivity();
785 }
786
787 private stopSubprocessPolling(): void {
788 if (!this.subprocessPollTimer) return;
789 clearInterval(this.subprocessPollTimer);
790 this.subprocessPollTimer = null;
791 }
792
793 private async pollSubprocessActivity(): Promise<void> {
794 if (this.subprocessPollInFlight) return;
795
796 const runningSessions = [...this.sessions.values()].filter(
797 (session): session is TerminalSessionState & { pid: number } =>
798 session.status === "running" && Number.isInteger(session.pid),
799 );
800 if (runningSessions.length === 0) {
801 this.stopSubprocessPolling();
802 return;
803 }
804
805 this.subprocessPollInFlight = true;
806 try {
807 await Promise.all(
808 runningSessions.map(async (session) => {
809 const terminalPid = session.pid;
810 let hasRunningSubprocess = false;
811 try {
812 hasRunningSubprocess = await this.subprocessChecker(terminalPid);
813 } catch (error) {
814 this.logger.warn("failed to check terminal subprocess activity", {
815 threadId: session.threadId,
816 terminalId: session.terminalId,
817 terminalPid,
818 error: error instanceof Error ? error.message : String(error),
819 });
820 return;
821 }
822
823 const liveSession = this.sessions.get(toSessionKey(session.threadId, session.terminalId));
824 if (!liveSession || liveSession.status !== "running" || liveSession.pid !== terminalPid) {
825 return;
826 }
827 if (liveSession.hasRunningSubprocess === hasRunningSubprocess) {
828 return;
829 }
830
831 liveSession.hasRunningSubprocess = hasRunningSubprocess;
832 liveSession.updatedAt = new Date().toISOString();
833 this.emitEvent({
834 type: "activity",
835 threadId: liveSession.threadId,
836 terminalId: liveSession.terminalId,
837 createdAt: new Date().toISOString(),
838 hasRunningSubprocess,
839 });
840 }),
841 );
842 } finally {
843 this.subprocessPollInFlight = false;
844 }
845 }
846
663847 private async assertValidCwd(cwd: string): Promise<void> {
664848 let stats: fs.Stats;
665849 try {
......@@ -686,6 +870,7 @@ export class TerminalManager extends EventEmitter<TerminalManagerEvents> {
686870 this.stopProcess(session);
687871 this.sessions.delete(key);
688872 }
873 this.updateSubprocessPollingState();
689874 await this.flushPersistQueue(threadId, terminalId);
690875 if (deleteHistory) {
691876 await this.deleteHistory(threadId, terminalId);
apps/web/src/store.test.ts+26-3
......@@ -97,6 +97,17 @@ function makeTerminalStartedEvent(overrides: Partial<TerminalEvent> = {}): Termi
9797 };
9898}
9999
100function makeTerminalActivityEvent(overrides: Partial<TerminalEvent> = {}): TerminalEvent {
101 return {
102 type: "activity",
103 threadId: "thread-local-1",
104 terminalId: DEFAULT_THREAD_TERMINAL_ID,
105 createdAt: "2026-02-09T00:00:02.000Z",
106 hasRunningSubprocess: true,
107 ...overrides,
108 };
109}
110
100111describe("store reducer thread continuity", () => {
101112 it("stores codexThreadId from UPDATE_SESSION", () => {
102113 const state = makeState(
......@@ -298,15 +309,27 @@ describe("store reducer thread continuity", () => {
298309 ]);
299310 });
300311
301 it("tracks running terminals from terminal lifecycle events", () => {
312 it("tracks running terminals from subprocess activity events", () => {
302313 const state = makeState(makeThread());
303314 const started = reducer(state, {
304315 type: "APPLY_TERMINAL_EVENT",
305316 event: makeTerminalStartedEvent(),
306317 });
307 expect(started.threads[0]?.runningTerminalIds).toEqual([DEFAULT_THREAD_TERMINAL_ID]);
318 expect(started.threads[0]?.runningTerminalIds).toEqual([]);
319
320 const active = reducer(started, {
321 type: "APPLY_TERMINAL_EVENT",
322 event: makeTerminalActivityEvent(),
323 });
324 expect(active.threads[0]?.runningTerminalIds).toEqual([DEFAULT_THREAD_TERMINAL_ID]);
325
326 const idle = reducer(active, {
327 type: "APPLY_TERMINAL_EVENT",
328 event: makeTerminalActivityEvent({ hasRunningSubprocess: false }),
329 });
330 expect(idle.threads[0]?.runningTerminalIds).toEqual([]);
308331
309 const exited = reducer(started, {
332 const exited = reducer(active, {
310333 type: "APPLY_TERMINAL_EVENT",
311334 event: {
312335 type: "exited",
apps/web/src/store.ts+3-1
......@@ -678,7 +678,9 @@ export function reducer(state: AppState, action: Action): AppState {
678678 action.event.type === "started" ||
679679 action.event.type === "restarted"
680680 ) {
681 if (action.event.snapshot.status === "running") {
681 runningTerminalIdSet.delete(action.event.terminalId);
682 } else if (action.event.type === "activity") {
683 if (action.event.hasRunningSubprocess) {
682684 runningTerminalIdSet.add(action.event.terminalId);
683685 } else {
684686 runningTerminalIdSet.delete(action.event.terminalId);
packages/contracts/src/terminal.test.ts+11
......@@ -139,4 +139,15 @@ describe("terminalEventSchema", () => {
139139 });
140140 expect(result.success).toBe(true);
141141 });
142
143 it("accepts activity events", () => {
144 const result = terminalEventSchema.safeParse({
145 type: "activity",
146 threadId: "thread-1",
147 terminalId: DEFAULT_TERMINAL_ID,
148 createdAt: new Date().toISOString(),
149 hasRunningSubprocess: true,
150 });
151 expect(result.success).toBe(true);
152 });
142153});
packages/contracts/src/terminal.ts+6
......@@ -91,6 +91,11 @@ export const terminalRestartedEventSchema = terminalEventBaseSchema.extend({
9191 snapshot: terminalSessionSnapshotSchema,
9292});
9393
94export const terminalActivityEventSchema = terminalEventBaseSchema.extend({
95 type: z.literal("activity"),
96 hasRunningSubprocess: z.boolean(),
97});
98
9499export const terminalEventSchema = z.discriminatedUnion("type", [
95100 terminalStartedEventSchema,
96101 terminalOutputEventSchema,
......@@ -98,6 +103,7 @@ export const terminalEventSchema = z.discriminatedUnion("type", [
98103 terminalErrorEventSchema,
99104 terminalClearedEventSchema,
100105 terminalRestartedEventSchema,
106 terminalActivityEventSchema,
101107]);
102108
103109export type TerminalThreadInput = z.input<typeof terminalThreadInputSchema>;