jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Fix terminal worker state mirroring

Browse Source
This commit is contained in:
Mohamed Boudra
2026-05-01 16:19:25 +07:00
parent af38c6e135
commit 51f7ef3171
6 changed files with 289 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/session.ts
+2 -2
View File
@@ -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,
packages/server/src/terminal/terminal-manager.ts
+24 -1
View File
@@ -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<void>;
captureTerminal(
id: string,
options?: { start?: number; end?: number; stripAnsi?: boolean },
): Promise<CaptureTerminalLinesResult>;
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<CaptureTerminalLinesResult> {
const session = terminalsById.get(id);
if (!session) {
return {
lines: [],
totalLines: 0,
};
}
return captureTerminalLines(session, options);
},
listDirectories(): string[] {
return Array.from(terminalsByCwd.keys());
},
packages/server/src/terminal/terminal-worker-process.ts
+58
View File
@@ -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<string, Array<() => void>>();
const stateFlushTimeoutByTerminalId = new Map<string, ReturnType<typeof setTimeout>>();
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<void> {
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<void> {
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;
}
packages/server/src/terminal/terminal-worker-protocol.ts
+16
View File
@@ -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;
packages/server/src/terminal/worker-terminal-manager.test.ts
+68
View File
@@ -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({
packages/server/src/terminal/worker-terminal-manager.ts
+121 -55
View File
@@ -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<TerminalWorkerToParentMessage, { type: "terminalMessage" }>,
): 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<TerminalWorkerToParentMessage, { type: "terminalStateUpdated" }>,
): void {
const record = recordsById.get(message.terminalId);
if (!record) {
return;
}
record.state = message.state;
}
function handleTerminalExitEvent(
message: Extract<TerminalWorkerToParentMessage, { type: "terminalExit" }>,
): 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<TerminalWorkerToParentMessage, { type: "terminalTitleChange" }>,
): 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<TerminalWorkerToParentMessage, { type: "terminalCommandFinished" }>,
): 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<TerminalWorkerToParentMessage, { type: "terminalsChanged" }>,
): 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<CaptureTerminalLinesResult> {
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());
},