From 51f7ef3171b339d02370a0679982728e5f4c751c Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 1 May 2026 16:19:25 +0700 Subject: [PATCH] Fix terminal worker state mirroring --- packages/server/src/server/session.ts | 4 +- .../server/src/terminal/terminal-manager.ts | 25 ++- .../src/terminal/terminal-worker-process.ts | 58 ++++++ .../src/terminal/terminal-worker-protocol.ts | 16 ++ .../terminal/worker-terminal-manager.test.ts | 68 +++++++ .../src/terminal/worker-terminal-manager.ts | 176 ++++++++++++------ 6 files changed, 289 insertions(+), 58 deletions(-) diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index ff02930b2..e59944e4a 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -42,7 +42,7 @@ import { type WorkspaceStateBucket, } from "./messages.js"; import type { TerminalManager, TerminalsChangedEvent } from "../terminal/terminal-manager.js"; -import { captureTerminalLines, type TerminalSession } from "../terminal/terminal.js"; +import type { TerminalSession } from "../terminal/terminal.js"; import { TerminalOutputCoalescer } from "../terminal/terminal-output-coalescer.js"; import { TerminalStreamOpcode, @@ -9647,7 +9647,7 @@ export class Session { this.ensureTerminalExitSubscription(session); try { - const capture = captureTerminalLines(session, { + const capture = await this.terminalManager.captureTerminal(msg.terminalId, { start: msg.start, end: msg.end, stripAnsi: msg.stripAnsi, diff --git a/packages/server/src/terminal/terminal-manager.ts b/packages/server/src/terminal/terminal-manager.ts index 336226c48..ec2eb5a91 100644 --- a/packages/server/src/terminal/terminal-manager.ts +++ b/packages/server/src/terminal/terminal-manager.ts @@ -1,4 +1,9 @@ -import { createTerminal, type TerminalSession } from "./terminal.js"; +import { + captureTerminalLines, + createTerminal, + type CaptureTerminalLinesResult, + type TerminalSession, +} from "./terminal.js"; import { resolve, sep, win32, posix } from "node:path"; export interface TerminalListItem { @@ -33,6 +38,10 @@ export interface TerminalManager { id: string, options?: { gracefulTimeoutMs?: number; forceTimeoutMs?: number }, ): Promise; + captureTerminal( + id: string, + options?: { start?: number; end?: number; stripAnsi?: boolean }, + ): Promise; listDirectories(): string[]; killAll(): void; subscribeTerminalsChanged(listener: TerminalsChangedListener): () => void; @@ -220,6 +229,20 @@ export function createTerminalManager(): TerminalManager { } }, + async captureTerminal( + id: string, + options?: { start?: number; end?: number; stripAnsi?: boolean }, + ): Promise { + const session = terminalsById.get(id); + if (!session) { + return { + lines: [], + totalLines: 0, + }; + } + return captureTerminalLines(session, options); + }, + listDirectories(): string[] { return Array.from(terminalsByCwd.keys()); }, diff --git a/packages/server/src/terminal/terminal-worker-process.ts b/packages/server/src/terminal/terminal-worker-process.ts index 672da646e..58d64b11d 100644 --- a/packages/server/src/terminal/terminal-worker-process.ts +++ b/packages/server/src/terminal/terminal-worker-process.ts @@ -1,4 +1,5 @@ import { createTerminalManager } from "./terminal-manager.js"; +import { captureTerminalLines } from "./terminal.js"; import type { TerminalSession } from "./terminal.js"; import type { TerminalWorkerRequest, @@ -8,6 +9,7 @@ import type { const manager = createTerminalManager(); const unsubscribeByTerminalId = new Map void>>(); +const stateFlushTimeoutByTerminalId = new Map>(); let ipcClosing = false; function sendToParent(message: TerminalWorkerToParentMessage): void { @@ -35,6 +37,11 @@ function toTerminalInfo(session: TerminalSession): WorkerTerminalInfo { } function clearTerminalSubscriptions(terminalId: string): void { + const stateFlushTimeout = stateFlushTimeoutByTerminalId.get(terminalId); + if (stateFlushTimeout) { + clearTimeout(stateFlushTimeout); + stateFlushTimeoutByTerminalId.delete(terminalId); + } const subscriptions = unsubscribeByTerminalId.get(terminalId); if (subscriptions) { for (const unsubscribe of subscriptions) { @@ -48,10 +55,40 @@ function clearTerminalSubscriptions(terminalId: string): void { unsubscribeByTerminalId.delete(terminalId); } +function flushTerminalState(session: TerminalSession): void { + const stateFlushTimeout = stateFlushTimeoutByTerminalId.get(session.id); + if (stateFlushTimeout) { + clearTimeout(stateFlushTimeout); + stateFlushTimeoutByTerminalId.delete(session.id); + } + sendToParent({ + type: "terminalStateUpdated", + terminalId: session.id, + state: session.getState(), + }); +} + +function scheduleTerminalStateFlush(session: TerminalSession): void { + const existingTimeout = stateFlushTimeoutByTerminalId.get(session.id); + if (existingTimeout) { + clearTimeout(existingTimeout); + } + stateFlushTimeoutByTerminalId.set( + session.id, + setTimeout(() => { + stateFlushTimeoutByTerminalId.delete(session.id); + flushTerminalState(session); + }, 16), + ); +} + function watchTerminal(session: TerminalSession): void { clearTerminalSubscriptions(session.id); const unsubscribeMessage = session.subscribe((message) => { + if (message.type === "output") { + scheduleTerminalStateFlush(session); + } sendToParent({ type: "terminalMessage", terminalId: session.id, @@ -178,6 +215,24 @@ async function handleRequest(message: TerminalWorkerRequest): Promise { return; } + case "captureTerminal": { + const session = manager.getTerminal(message.terminalId); + const result = session + ? captureTerminalLines(session, { + start: message.start, + end: message.end, + stripAnsi: message.stripAnsi, + }) + : { lines: [], totalLines: 0 }; + sendToParent({ + type: "response", + requestId: message.requestId, + ok: true, + result, + }); + return; + } + case "killAll": { manager.killAll(); for (const terminalId of Array.from(unsubscribeByTerminalId.keys())) { @@ -190,6 +245,9 @@ async function handleRequest(message: TerminalWorkerRequest): Promise { case "send": { const session = manager.getTerminal(message.terminalId); session?.send(message.message); + if (session && message.message.type === "resize") { + flushTerminalState(session); + } sendToParent({ type: "response", requestId: message.requestId, ok: true }); return; } diff --git a/packages/server/src/terminal/terminal-worker-protocol.ts b/packages/server/src/terminal/terminal-worker-protocol.ts index 71f0b3c55..1af598f1f 100644 --- a/packages/server/src/terminal/terminal-worker-protocol.ts +++ b/packages/server/src/terminal/terminal-worker-protocol.ts @@ -1,5 +1,6 @@ import type { TerminalExitInfo, ServerMessage, ClientMessage } from "./terminal.js"; import type { TerminalState } from "../shared/messages.js"; +import type { CaptureTerminalLinesResult } from "./terminal.js"; export interface WorkerTerminalInfo { id: string; @@ -51,6 +52,14 @@ export type TerminalWorkerRequest = terminalId: string; options?: WorkerKillAndWaitOptions; } + | { + type: "captureTerminal"; + requestId: string; + terminalId: string; + start?: number; + end?: number; + stripAnsi?: boolean; + } | { type: "listDirectories"; requestId: string; @@ -96,6 +105,11 @@ export type TerminalWorkerEvent = terminalId: string; message: ServerMessage; } + | { + type: "terminalStateUpdated"; + terminalId: string; + state: TerminalState; + } | { type: "terminalExit"; terminalId: string; @@ -120,3 +134,5 @@ export type TerminalWorkerEvent = }; export type TerminalWorkerToParentMessage = TerminalWorkerResponse | TerminalWorkerEvent; + +export type TerminalWorkerCaptureResult = CaptureTerminalLinesResult; diff --git a/packages/server/src/terminal/worker-terminal-manager.test.ts b/packages/server/src/terminal/worker-terminal-manager.test.ts index a39e4db81..5db396e2e 100644 --- a/packages/server/src/terminal/worker-terminal-manager.test.ts +++ b/packages/server/src/terminal/worker-terminal-manager.test.ts @@ -173,6 +173,74 @@ it("creates a terminal through the worker and streams output", async () => { expect(snapshots).toBe(snapshotsBeforeOutput); }); +it("refreshes cached terminal state after worker output", async () => { + const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-state-")); + temporaryDirs.push(cwd); + manager = createWorkerTerminalManager(); + const session = trackTerminal( + await manager.createTerminal({ + cwd, + ...nodeTerminalCommand(` + process.stdout.write("worker-state-ready\\n"); + setInterval(() => {}, 1000); + `), + }), + ); + + await waitForCondition(() => getVisibleText(session).includes("worker-state-ready"), 10000); + + expect(getVisibleText(session)).toContain("worker-state-ready"); +}); + +it("refreshes cached terminal title after worker title changes", async () => { + const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-title-")); + temporaryDirs.push(cwd); + manager = createWorkerTerminalManager(); + const session = trackTerminal( + await manager.createTerminal({ + cwd, + command: "/bin/sh", + args: ["-lc", "printf '\\033]0;Build Output\\007'; sleep 2"], + }), + ); + + await waitForCondition(() => session.getTitle() === "Build Output", 10000); + + expect(session.getState().title).toBe("Build Output"); +}); + +it("refreshes cached terminal size after worker resize", async () => { + const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-resize-")); + temporaryDirs.push(cwd); + manager = createWorkerTerminalManager(); + const session = trackTerminal(await manager.createTerminal({ cwd })); + + session.send({ type: "resize", rows: 10, cols: 40 }); + + await waitForCondition(() => { + const size = session.getSize(); + return size.rows === 10 && size.cols === 40; + }, 10000); + + expect(session.getState().rows).toBe(10); + expect(session.getState().cols).toBe(40); +}); + +it("captures terminal output from the worker authority", async () => { + const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-capture-")); + temporaryDirs.push(cwd); + manager = createWorkerTerminalManager(); + const session = trackTerminal(await manager.createTerminal({ cwd })); + + session.send({ type: "input", data: "echo hello world\r" }); + + await waitForCondition(() => getVisibleText(session).includes("hello world"), 10000); + + const capture = await manager.captureTerminal(session.id); + expect(capture.lines.join("\n")).toContain("hello world"); + expect(capture.totalLines).toBeGreaterThan(0); +}); + it("does not surface fire-and-forget send timeouts as unhandled rejections", async () => { const worker = new FakeTerminalWorker(); manager = createWorkerTerminalManager({ diff --git a/packages/server/src/terminal/worker-terminal-manager.ts b/packages/server/src/terminal/worker-terminal-manager.ts index 8d771cf26..83860b7cd 100644 --- a/packages/server/src/terminal/worker-terminal-manager.ts +++ b/packages/server/src/terminal/worker-terminal-manager.ts @@ -4,6 +4,7 @@ import { randomUUID } from "node:crypto"; import type { TerminalState } from "../shared/messages.js"; import type { ClientMessage, + CaptureTerminalLinesResult, ServerMessage, TerminalCommandFinishedInfo, TerminalExitInfo, @@ -281,6 +282,103 @@ export function createWorkerTerminalManager( return record; } + function handleTerminalMessageEvent( + message: Extract, + ): void { + const record = recordsById.get(message.terminalId); + if (!record) { + return; + } + if (message.message.type === "snapshot") { + record.state = message.message.state; + } + for (const listener of Array.from(record.messageListeners)) { + listener(message.message); + } + } + + function handleTerminalStateUpdatedEvent( + message: Extract, + ): void { + const record = recordsById.get(message.terminalId); + if (!record) { + return; + } + record.state = message.state; + } + + function handleTerminalExitEvent( + message: Extract, + ): void { + const record = recordsById.get(message.terminalId); + if (!record) { + return; + } + record.exitInfo = message.info; + for (const listener of Array.from(record.exitListeners)) { + listener(message.info); + } + record.exitListeners.clear(); + removeRecord(message.terminalId); + emitTerminalsChanged({ + cwd: record.info.cwd, + terminals: listTerminalItemsForCwd(record.info.cwd), + }); + } + + function handleTerminalTitleChangeEvent( + message: Extract, + ): void { + const record = recordsById.get(message.terminalId); + if (!record) { + return; + } + const nextState = { ...record.state }; + if (message.title) { + nextState.title = message.title; + } else { + delete nextState.title; + } + record.info = { + ...record.info, + ...(message.title ? { title: message.title } : { title: undefined }), + }; + record.state = nextState; + for (const listener of Array.from(record.titleChangeListeners)) { + listener(message.title); + } + emitTerminalsChanged({ + cwd: record.info.cwd, + terminals: listTerminalItemsForCwd(record.info.cwd), + }); + } + + function handleTerminalCommandFinishedEvent( + message: Extract, + ): void { + const record = recordsById.get(message.terminalId); + if (!record) { + return; + } + for (const listener of Array.from(record.commandFinishedListeners)) { + listener(message.info); + } + } + + function handleTerminalsChangedEvent( + message: Extract, + ): void { + emitTerminalsChanged({ + cwd: message.cwd, + terminals: message.terminals.map((terminal) => ({ + id: terminal.id, + name: terminal.name, + cwd: terminal.cwd, + ...(terminal.title ? { title: terminal.title } : {}), + })), + }); + } + function handleWorkerEvent(message: TerminalWorkerToParentMessage): void { switch (message.type) { case "terminalCreated": { @@ -298,77 +396,32 @@ export function createWorkerTerminalManager( } case "terminalMessage": { - const record = recordsById.get(message.terminalId); - if (!record) { - return; - } - if (message.message.type === "snapshot") { - record.state = message.message.state; - } - for (const listener of Array.from(record.messageListeners)) { - listener(message.message); - } + handleTerminalMessageEvent(message); + return; + } + + case "terminalStateUpdated": { + handleTerminalStateUpdatedEvent(message); return; } case "terminalExit": { - const record = recordsById.get(message.terminalId); - if (!record) { - return; - } - record.exitInfo = message.info; - for (const listener of Array.from(record.exitListeners)) { - listener(message.info); - } - record.exitListeners.clear(); - removeRecord(message.terminalId); - emitTerminalsChanged({ - cwd: record.info.cwd, - terminals: listTerminalItemsForCwd(record.info.cwd), - }); + handleTerminalExitEvent(message); return; } case "terminalTitleChange": { - const record = recordsById.get(message.terminalId); - if (!record) { - return; - } - record.info = { - ...record.info, - ...(message.title ? { title: message.title } : { title: undefined }), - }; - for (const listener of Array.from(record.titleChangeListeners)) { - listener(message.title); - } - emitTerminalsChanged({ - cwd: record.info.cwd, - terminals: listTerminalItemsForCwd(record.info.cwd), - }); + handleTerminalTitleChangeEvent(message); return; } case "terminalCommandFinished": { - const record = recordsById.get(message.terminalId); - if (!record) { - return; - } - for (const listener of Array.from(record.commandFinishedListeners)) { - listener(message.info); - } + handleTerminalCommandFinishedEvent(message); return; } case "terminalsChanged": { - emitTerminalsChanged({ - cwd: message.cwd, - terminals: message.terminals.map((terminal) => ({ - id: terminal.id, - name: terminal.name, - cwd: terminal.cwd, - ...(terminal.title ? { title: terminal.title } : {}), - })), - }); + handleTerminalsChangedEvent(message); return; } } @@ -489,6 +542,19 @@ export function createWorkerTerminalManager( }); }, + async captureTerminal( + id: string, + options?: { start?: number; end?: number; stripAnsi?: boolean }, + ): Promise { + return (await sendRequest({ + type: "captureTerminal", + terminalId: id, + ...(options?.start === undefined ? {} : { start: options.start }), + ...(options?.end === undefined ? {} : { end: options.end }), + ...(options?.stripAnsi === undefined ? {} : { stripAnsi: options.stripAnsi }), + })) as CaptureTerminalLinesResult; + }, + listDirectories(): string[] { return Array.from(terminalIdsByCwd.keys()); },