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

Fix terminal snapshot flushing

Browse Source
This commit is contained in:
Mohamed Boudra
2026-05-03 13:25:19 +07:00
parent d89ecae615
commit b2247450f9
8 changed files with 445 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/terminal/terminal-manager.ts
+6
View File
@@ -3,6 +3,7 @@ import {
createTerminal,
type CaptureTerminalLinesResult,
type TerminalSession,
type TerminalStateSnapshot,
} from "./terminal.js";
import { resolve, sep, win32, posix } from "node:path";
@@ -33,6 +34,7 @@ export interface TerminalManager {
}): Promise<TerminalSession>;
registerCwdEnv(options: { cwd: string; env: Record<string, string> }): void;
getTerminal(id: string): TerminalSession | undefined;
getTerminalState(id: string): Promise<TerminalStateSnapshot | null>;
killTerminal(id: string): void;
killTerminalAndWait(
id: string,
@@ -210,6 +212,10 @@ export function createTerminalManager(): TerminalManager {
return terminalsById.get(id);
},
async getTerminalState(id: string): Promise<TerminalStateSnapshot | null> {
return terminalsById.get(id)?.getStateSnapshot() ?? null;
},
killTerminal(id: string): void {
removeSessionById(id, { kill: true });
},
packages/server/src/terminal/terminal-session-controller.ts
+66 -15
View File
@@ -26,11 +26,18 @@ import type { TerminalManager, TerminalsChangedEvent } from "./terminal-manager.
const MAX_TERMINAL_STREAM_SLOTS = 256;
interface BufferedTerminalOutput {
data: string;
revision?: number;
}
interface ActiveTerminalStream {
terminalId: string;
slot: number;
unsubscribe: () => void;
needsSnapshot: boolean;
snapshotInFlight: boolean;
bufferedOutputs: BufferedTerminalOutput[];
outputCoalescer: TerminalOutputCoalescer;
}
@@ -480,7 +487,7 @@ export class TerminalSessionController {
const activeStream = this.activeStreams.get(slot);
if (activeStream) {
this.trySendSnapshot(activeStream);
void this.trySendSnapshot(activeStream);
}
}
@@ -613,6 +620,8 @@ export class TerminalSessionController {
slot,
unsubscribe: () => {},
needsSnapshot: true,
snapshotInFlight: false,
bufferedOutputs: [],
outputCoalescer: new TerminalOutputCoalescer({
timers: { setTimeout, clearTimeout },
onFlush: ({ payload }) => {
@@ -640,13 +649,20 @@ export class TerminalSessionController {
if (message.type === "snapshot") {
activeStream.outputCoalescer.flush();
activeStream.needsSnapshot = true;
this.trySendSnapshot(activeStream);
void this.trySendSnapshot(activeStream);
return;
}
if (message.type === "titleChange") {
return;
}
if (activeStream.needsSnapshot || message.data.length === 0) {
if (message.data.length === 0) {
return;
}
if (activeStream.needsSnapshot || activeStream.snapshotInFlight) {
activeStream.bufferedOutputs.push({
data: message.data,
revision: message.revision,
});
return;
}
activeStream.outputCoalescer.handle(message.data);
@@ -654,26 +670,60 @@ export class TerminalSessionController {
return slot;
}
private trySendSnapshot(activeStream: ActiveTerminalStream): void {
if (this.activeStreams.get(activeStream.slot) !== activeStream || !activeStream.needsSnapshot) {
private async trySendSnapshot(activeStream: ActiveTerminalStream): Promise<void> {
if (
this.activeStreams.get(activeStream.slot) !== activeStream ||
!activeStream.needsSnapshot ||
activeStream.snapshotInFlight
) {
return;
}
const terminal = this.terminalManager?.getTerminal(activeStream.terminalId);
if (!terminal) {
if (!this.terminalManager?.getTerminal(activeStream.terminalId)) {
this.detachStream(activeStream.terminalId, { emitExit: true });
return;
}
activeStream.outputCoalescer.flush();
activeStream.needsSnapshot = false;
this.emitBinary(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Snapshot,
slot: activeStream.slot,
payload: encodeTerminalSnapshotPayload(terminal.getState()),
}),
);
activeStream.snapshotInFlight = true;
try {
const snapshot = await this.terminalManager.getTerminalState(activeStream.terminalId);
if (this.activeStreams.get(activeStream.slot) !== activeStream) {
return;
}
if (!snapshot) {
this.detachStream(activeStream.terminalId, { emitExit: true });
return;
}
this.emitBinary(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Snapshot,
slot: activeStream.slot,
payload: encodeTerminalSnapshotPayload(snapshot.state),
}),
);
const bufferedOutputs = activeStream.bufferedOutputs.splice(
0,
activeStream.bufferedOutputs.length,
);
for (const output of bufferedOutputs) {
if (output.revision !== undefined && output.revision <= snapshot.revision) {
continue;
}
activeStream.outputCoalescer.handle(output.data);
}
activeStream.needsSnapshot = false;
} catch (error) {
this.sessionLogger.warn(
{ err: error, terminalId: activeStream.terminalId },
"Failed to pull terminal snapshot",
);
activeStream.needsSnapshot = true;
} finally {
activeStream.snapshotInFlight = false;
}
}
private allocateSlot(): number | null {
@@ -699,6 +749,7 @@ export class TerminalSessionController {
return false;
}
activeStream.outputCoalescer.flush();
activeStream.bufferedOutputs.length = 0;
this.activeStreams.delete(slot);
this.idToSlot.delete(terminalId);
try {
packages/server/src/terminal/terminal-worker-process.ts
+14 -41
View File
@@ -9,7 +9,6 @@ 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 {
@@ -37,11 +36,6 @@ 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) {
@@ -55,40 +49,10 @@ 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,
@@ -150,10 +114,11 @@ async function handleRequest(message: TerminalWorkerRequest): Promise<void> {
case "createTerminal": {
const session = await manager.createTerminal(message.options);
watchTerminal(session);
const initialSnapshot = session.getStateSnapshot();
sendToParent({
type: "terminalCreated",
terminal: toTerminalInfo(session),
state: session.getState(),
state: initialSnapshot.state,
});
sendToParent({
type: "response",
@@ -161,7 +126,7 @@ async function handleRequest(message: TerminalWorkerRequest): Promise<void> {
ok: true,
result: {
terminal: toTerminalInfo(session),
state: session.getState(),
state: initialSnapshot.state,
},
});
return;
@@ -205,6 +170,17 @@ async function handleRequest(message: TerminalWorkerRequest): Promise<void> {
return;
}
case "getTerminalState": {
const session = manager.getTerminal(message.terminalId);
sendToParent({
type: "response",
requestId: message.requestId,
ok: true,
result: session?.getStateSnapshot() ?? null,
});
return;
}
case "listDirectories": {
sendToParent({
type: "response",
@@ -245,9 +221,6 @@ 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
+12 -6
View File
@@ -1,4 +1,9 @@
import type { TerminalExitInfo, ServerMessage, ClientMessage } from "./terminal.js";
import type {
TerminalExitInfo,
ServerMessage,
ClientMessage,
TerminalStateSnapshot,
} from "./terminal.js";
import type { TerminalState } from "../shared/messages.js";
import type { CaptureTerminalLinesResult } from "./terminal.js";
@@ -52,6 +57,11 @@ export type TerminalWorkerRequest =
terminalId: string;
options?: WorkerKillAndWaitOptions;
}
| {
type: "getTerminalState";
requestId: string;
terminalId: string;
}
| {
type: "captureTerminal";
requestId: string;
@@ -105,11 +115,6 @@ export type TerminalWorkerEvent =
terminalId: string;
message: ServerMessage;
}
| {
type: "terminalStateUpdated";
terminalId: string;
state: TerminalState;
}
| {
type: "terminalExit";
terminalId: string;
@@ -136,3 +141,4 @@ export type TerminalWorkerEvent =
export type TerminalWorkerToParentMessage = TerminalWorkerResponse | TerminalWorkerEvent;
export type TerminalWorkerCaptureResult = CaptureTerminalLinesResult;
export type TerminalWorkerStateResult = TerminalStateSnapshot | null;
packages/server/src/terminal/terminal.ts
+32 -11
View File
@@ -27,14 +27,19 @@ export interface TerminalCommandFinishedInfo {
exitCode: number | null;
}
export interface TerminalStateSnapshot {
state: TerminalState;
revision: number;
}
export type ClientMessage =
| { type: "input"; data: string }
| { type: "resize"; rows: number; cols: number }
| { type: "mouse"; row: number; col: number; button: number; action: "down" | "up" | "move" };
export type ServerMessage =
| { type: "output"; data: string }
| { type: "snapshot"; state: TerminalState }
| { type: "output"; data: string; revision?: number }
| { type: "snapshot"; state: TerminalState; revision?: number }
| { type: "titleChange"; title?: string };
export interface TerminalSession {
@@ -48,6 +53,7 @@ export interface TerminalSession {
onTitleChange(listener: (title?: string) => void): () => void;
getSize(): { rows: number; cols: number };
getState(): TerminalState;
getStateSnapshot(): TerminalStateSnapshot;
getTitle(): string | undefined;
getExitInfo(): TerminalExitInfo | null;
kill(): void;
@@ -605,6 +611,7 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
let titleDebounceTimer: ReturnType<typeof setTimeout> | null = null;
let pendingInput = "";
let inputFlushImmediate: ReturnType<typeof setImmediate> | null = null;
let stateRevision = 0;
// Create xterm.js headless terminal
const terminal = new Terminal({
@@ -751,6 +758,18 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
titleChangeListeners.clear();
}
function writeOutputToHeadless(data: string): void {
terminal.write(data, () => {
if (disposed || killed) {
return;
}
stateRevision += 1;
for (const listener of listeners) {
listener({ type: "output", data, revision: stateRevision });
}
});
}
// Pipe PTY output to terminal emulator
ptyProcess.onData((data) => {
if (killed) return;
@@ -758,14 +777,7 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
if (recentOutputText.length > TERMINAL_EXIT_OUTPUT_CHAR_LIMIT) {
recentOutputText = recentOutputText.slice(-TERMINAL_EXIT_OUTPUT_CHAR_LIMIT);
}
terminal.write(data, () => {
if (disposed || killed) {
return;
}
for (const listener of listeners) {
listener({ type: "output", data });
}
});
writeOutputToHeadless(data);
});
ptyProcess.onExit((event) => {
@@ -799,6 +811,13 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
};
}
function getStateSnapshot(): TerminalStateSnapshot {
return {
state: getState(),
revision: stateRevision,
};
}
function getSize(): { rows: number; cols: number } {
return {
rows: terminal.rows,
@@ -846,6 +865,7 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
flushPendingInput();
terminal.resize(msg.cols, msg.rows);
ptyProcess.resize(msg.cols, msg.rows);
stateRevision += 1;
break;
case "mouse":
// Mouse events can be sent as escape sequences if terminal supports it
@@ -859,7 +879,7 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
terminal.write("", () => {
if (!disposed && listeners.has(listener)) {
listener({ type: "snapshot", state: getState() });
listener({ type: "snapshot", ...getStateSnapshot() });
}
});
@@ -999,6 +1019,7 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
onTitleChange,
getSize,
getState,
getStateSnapshot,
getTitle,
getExitInfo,
kill,
packages/server/src/terminal/worker-terminal-manager.test.ts
+21 -10
View File
@@ -20,13 +20,13 @@ function nodeTerminalCommand(script: string): { command: string; args: string[]
}
async function waitForCondition(
predicate: () => boolean,
predicate: () => boolean | Promise<boolean>,
timeoutMs: number,
intervalMs = 25,
): Promise<void> {
const start = Date.now();
while (Date.now() - start < timeoutMs) {
if (predicate()) {
if (await predicate()) {
return;
}
await new Promise((resolve) => setTimeout(resolve, intervalMs));
@@ -35,9 +35,12 @@ async function waitForCondition(
}
function getVisibleText(session: TerminalSession): string {
return session
.getState()
.grid.map((row) =>
return getVisibleTextFromState(session.getState());
}
function getVisibleTextFromState(state: TerminalState): string {
return state.grid
.map((row) =>
row
.map((cell) => cell.char)
.join("")
@@ -173,7 +176,7 @@ it("creates a terminal through the worker and streams output", async () => {
expect(snapshots).toBe(snapshotsBeforeOutput);
});
it("refreshes cached terminal state after worker output", async () => {
it("pulls fresh terminal state from the worker authority", async () => {
const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-state-"));
temporaryDirs.push(cwd);
manager = createWorkerTerminalManager();
@@ -187,9 +190,14 @@ it("refreshes cached terminal state after worker output", async () => {
}),
);
await waitForCondition(() => getVisibleText(session).includes("worker-state-ready"), 10000);
let visibleText = "";
await waitForCondition(async () => {
const snapshot = await manager!.getTerminalState(session.id);
visibleText = snapshot ? getVisibleTextFromState(snapshot.state) : "";
return visibleText.includes("worker-state-ready");
}, 10000);
expect(getVisibleText(session)).toContain("worker-state-ready");
expect(visibleText).toContain("worker-state-ready");
});
it("refreshes cached terminal title after worker title changes", async () => {
@@ -236,9 +244,12 @@ it("captures terminal output from the worker authority", async () => {
session.send({ type: "input", data: "echo hello world\r" });
await waitForCondition(() => getVisibleText(session).includes("hello world"), 10000);
let capture = await manager.captureTerminal(session.id);
await waitForCondition(async () => {
capture = await manager!.captureTerminal(session.id);
return capture.lines.join("\n").includes("hello world");
}, 10000);
const capture = await manager.captureTerminal(session.id);
expect(capture.lines.join("\n")).toContain("hello world");
expect(capture.totalLines).toBeGreaterThan(0);
});
packages/server/src/terminal/worker-terminal-manager.ts
+22 -20
View File
@@ -9,6 +9,7 @@ import type {
TerminalCommandFinishedInfo,
TerminalExitInfo,
TerminalSession,
TerminalStateSnapshot,
} from "./terminal.js";
import type {
TerminalListItem,
@@ -21,6 +22,7 @@ import type {
TerminalWorkerResponse,
TerminalWorkerToParentMessage,
WorkerCreateTerminalOptions,
TerminalWorkerStateResult,
WorkerTerminalInfo,
} from "./terminal-worker-protocol.js";
@@ -186,15 +188,17 @@ export function createWorkerTerminalManager(
return record.info.cwd;
},
send(message: ClientMessage): void {
if (message.type === "resize") {
record.state = {
...record.state,
rows: message.rows,
cols: message.cols,
};
}
sendBestEffortRequest({ type: "send", terminalId: record.info.id, message });
},
subscribe(listener: (msg: ServerMessage) => void): () => void {
record.messageListeners.add(listener);
queueMicrotask(() => {
if (record.messageListeners.has(listener)) {
listener({ type: "snapshot", state: record.state });
}
});
return () => {
record.messageListeners.delete(listener);
};
@@ -237,6 +241,12 @@ export function createWorkerTerminalManager(
getState(): TerminalState {
return record.state;
},
getStateSnapshot(): TerminalStateSnapshot {
return {
state: record.state,
revision: 0,
};
},
getTitle(): string | undefined {
return record.info.title;
},
@@ -297,16 +307,6 @@ export function createWorkerTerminalManager(
}
}
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 {
@@ -400,11 +400,6 @@ export function createWorkerTerminalManager(
return;
}
case "terminalStateUpdated": {
handleTerminalStateUpdatedEvent(message);
return;
}
case "terminalExit": {
handleTerminalExitEvent(message);
return;
@@ -525,6 +520,13 @@ export function createWorkerTerminalManager(
return recordsById.get(id)?.session;
},
async getTerminalState(id: string): Promise<TerminalStateSnapshot | null> {
return (await sendRequest({
type: "getTerminalState",
terminalId: id,
})) as TerminalWorkerStateResult;
},
killTerminal(id: string): void {
void sendRequest({ type: "killTerminal", terminalId: id }).catch(() => {
// no-op; kill is intentionally best-effort and synchronous in the public interface.