fix(terminal): recover paced replay and viewer geometry

Catch-up must drain through its paced sink rather than reject on the congestion that requested it. Restore measured renderer geometry after replay commits and before live writes. Require an ANSI-capable host explicitly and remove the retired snapshot entry points.
This commit is contained in:
Mohamed Boudra
2026-10-30 10:58:50 +01:00
parent fe3d65ffea
commit d14619e41a
35 changed files with 527 additions and 557 deletions
+2 -2
View File
@@ -54,11 +54,11 @@ without requesting streams. Application caches and which agents are visible rema
## Terminal ANSI replay
`terminal_chunked_replay` is SDK-owned delivery behavior, separate from the daemon's
`features.terminalChunkedReplay`. The daemon sends new binary opcodes only to a source
`terminal_chunked_replay` is SDK-owned delivery behavior. The daemon sends new binary opcodes only to a source
advertising the capability. The SDK keeps the established subscribe request against older
ANSI hosts and normalizes their single Restore frame into begin/data/end. JSON snapshot
wire declarations remain accepted, but JSON snapshot frames have no rendering runtime.
Terminals require `terminal-restore-modes`; older hosts show an update message without attaching.
## Owned observations
@@ -4,7 +4,6 @@ import type {
} from "../terminal/runtime/terminal-emulator-runtime";
import type { Ref } from "react";
import type { ITheme } from "@xterm/xterm";
import type { TerminalState } from "@getpaseo/protocol/messages";
import type { TerminalInputModeState } from "@getpaseo/protocol/terminal-input-mode";
import type { TerminalOutputData } from "../terminal/runtime/terminal-emulator-runtime";
import type {
@@ -20,8 +19,6 @@ export interface TerminalEmulatorHandle {
find?: TerminalFindHandle;
writeOutput: (data: TerminalOutputData) => void;
replay: (event: import("@getpaseo/client").TerminalReplayEvent) => void;
restoreOutput: (data: TerminalOutputData) => void;
renderSnapshot: (state: TerminalState | null) => void;
paste: (text: string) => void;
copySelection: (clipboard: TerminalClipboardWriter) => Promise<string>;
clear: () => void;
@@ -45,7 +42,6 @@ export interface TerminalEmulatorProps {
swipeGesturesEnabled?: boolean;
onSwipeLeft?: () => void;
onSwipeRight?: () => void;
initialSnapshot?: TerminalState | null;
onFindRequest?: () => void;
onFindResult?: (result: TerminalFindResult) => void;
onInput?: (data: string) => Promise<void> | void;
@@ -13,7 +13,6 @@ import {
type ViewStyle,
} from "react-native";
import type { ITheme } from "@xterm/xterm";
import type { TerminalState } from "@getpaseo/protocol/messages";
import {
TerminalInputModeTracker,
terminalInputModeStatesEqual,
@@ -120,43 +119,8 @@ function resolvePanResponderPoint(
};
}
function initialNativeTerminalSize(snapshot: TerminalState | null): TerminalMeasuredSize {
return {
rows: snapshot?.rows ?? MIN_NATIVE_TERMINAL_ROWS,
cols: snapshot?.cols ?? MIN_NATIVE_TERMINAL_COLS,
};
}
function screenStateFromViewport(viewport: TerminalViewportState): TerminalScreenState {
const visibleRows = viewport.grid.length;
const currentViewport = {
firstRow: viewport.firstRow,
lastRow: Math.min(viewport.newestRow, viewport.firstRow + Math.max(1, visibleRows) - 1),
};
return {
viewport,
scroll: {
mode: "following",
firstRow: viewport.firstRow,
visibleRows,
oldestRow: viewport.oldestRow,
newestRow: viewport.newestRow,
currentViewport,
bottomViewport: currentViewport,
},
};
}
function viewportStateFromSnapshot(state: TerminalState): TerminalViewportState {
return {
rows: state.grid.length,
cols: state.cols,
firstRow: 0,
oldestRow: 0,
newestRow: Math.max(0, state.grid.length - 1),
grid: state.grid,
cursor: state.cursor,
};
function initialNativeTerminalSize(): TerminalMeasuredSize {
return { rows: MIN_NATIVE_TERMINAL_ROWS, cols: MIN_NATIVE_TERMINAL_COLS };
}
function NativeTerminalEmulator({
@@ -169,7 +133,6 @@ function NativeTerminalEmulator({
fontSize,
keyboardInset = 0,
isKeyboardVisible = false,
initialSnapshot = null,
onInput,
onTerminalKey,
onResize,
@@ -181,20 +144,18 @@ function NativeTerminalEmulator({
resizeRequestToken = 0,
}: TerminalEmulatorProps) {
const terminalRef = useRef<NativeHeadlessTerminal | null>(null);
const mountInputRef = useRef({ streamKey, initialSnapshot });
const mountInputRef = useRef({ streamKey });
if (
resolveNativeTerminalMountAction({
mountedStreamKey: mountInputRef.current.streamKey,
nextStreamKey: streamKey,
}) === "mount"
) {
mountInputRef.current = { streamKey, initialSnapshot };
mountInputRef.current = { streamKey };
}
const inputRef = useRef<TerminalInputHandle>(null);
const terminalSizeRef = useRef<TerminalMeasuredSize>(initialNativeTerminalSize(initialSnapshot));
const resizePolicyRef = useRef(
createTerminalResizePolicy(initialNativeTerminalSize(initialSnapshot)),
);
const terminalSizeRef = useRef<TerminalMeasuredSize>(initialNativeTerminalSize());
const resizePolicyRef = useRef(createTerminalResizePolicy(initialNativeTerminalSize()));
const activeResizeClaimTokenRef = useRef(0);
const layoutRef = useRef<TerminalMeasuredLayout | null>(null);
const metricsRef = useRef<TerminalGridCellMetrics | null>(null);
@@ -220,11 +181,7 @@ function NativeTerminalEmulator({
...inputModeTrackerRef.current.getState(),
applicationCursorKeys: false,
});
const [terminalScreen, setTerminalScreen] = useState<TerminalScreenState | null>(() => {
return initialSnapshot
? screenStateFromViewport(viewportStateFromSnapshot(initialSnapshot))
: null;
});
const [terminalScreen, setTerminalScreen] = useState<TerminalScreenState | null>(null);
const [selectionRange, setSelectionRange] = useState<TerminalSelectionRange | null>(null);
const selectionRangeRef = useRef<TerminalSelectionRange | null>(null);
const hasNotifiedSelectionRef = useRef(false);
@@ -365,6 +322,7 @@ function NativeTerminalEmulator({
await terminal.write(chunk);
emitInputModeChange();
},
restoreSize: () => terminal.resize(terminalSizeRef.current),
reset: (size) => {
inputModeTrackerRef.current.reset();
if (size) terminal.resize(size);
@@ -434,51 +392,41 @@ function NativeTerminalEmulator({
[schedulePaint],
);
const resetTerminal = useCallback(
(snapshot: TerminalState | null) => {
outputDrainRef.current?.dispose();
outputDrainRef.current = null;
terminalRef.current?.dispose();
screenModelRef.current = null;
latestResolvedScreenRef.current = null;
clearSelection();
const measuredSize = resolveMeasuredNativeTerminalSize({
layout: layoutRef.current,
metrics: metricsRef.current,
});
const size =
measuredSize ??
(snapshot ? { rows: snapshot.rows, cols: snapshot.cols } : terminalSizeRef.current);
terminalSizeRef.current = size;
resizePolicyRef.current = createTerminalResizePolicy(size);
outputDecoderRef.current.decode();
const terminal = createNativeHeadlessTerminal({
rows: size.rows,
cols: size.cols,
scrollbackLines,
});
terminalRef.current = terminal;
screenModelRef.current = createNativeTerminalScreenModel({ terminal });
const outputDrain = createOutputDrain(terminal);
outputDrainRef.current = outputDrain;
const screen = snapshot
? screenStateFromViewport(viewportStateFromSnapshot(snapshot))
: resolveTerminalScreen(terminal);
commitTerminalScreen(screen);
resetInputModeTracker();
if (snapshot) {
outputDrain.restoreSnapshot(snapshot);
}
},
[
commitTerminalScreen,
createOutputDrain,
resetInputModeTracker,
resolveTerminalScreen,
const resetTerminal = useCallback(() => {
outputDrainRef.current?.dispose();
outputDrainRef.current = null;
terminalRef.current?.dispose();
screenModelRef.current = null;
latestResolvedScreenRef.current = null;
clearSelection();
const measuredSize = resolveMeasuredNativeTerminalSize({
layout: layoutRef.current,
metrics: metricsRef.current,
});
const size = measuredSize ?? terminalSizeRef.current;
terminalSizeRef.current = size;
resizePolicyRef.current = createTerminalResizePolicy(size);
outputDecoderRef.current.decode();
const terminal = createNativeHeadlessTerminal({
rows: size.rows,
cols: size.cols,
scrollbackLines,
clearSelection,
],
);
});
terminalRef.current = terminal;
screenModelRef.current = createNativeTerminalScreenModel({ terminal });
const outputDrain = createOutputDrain(terminal);
outputDrainRef.current = outputDrain;
const screen = resolveTerminalScreen(terminal);
commitTerminalScreen(screen);
resetInputModeTracker();
}, [
commitTerminalScreen,
createOutputDrain,
resetInputModeTracker,
resolveTerminalScreen,
scrollbackLines,
clearSelection,
]);
const enqueueOutputText = useCallback((text: string) => {
if (text.length === 0) {
@@ -509,27 +457,6 @@ function NativeTerminalEmulator({
enqueueOutputText(text);
},
replay: (event) => outputDrainRef.current?.replay(event),
restoreOutput: (data: TerminalOutputData) => {
outputDecoderRef.current.decode();
resetInputModeTracker();
const text = outputDecoderRef.current.decode(data, { stream: false });
outputDrainRef.current?.restoreText(text);
},
renderSnapshot: (state: TerminalState | null) => {
outputDecoderRef.current.decode();
resetInputModeTracker();
if (!state) {
resetTerminal(null);
return;
}
clearSelection();
const outputDrain = outputDrainRef.current;
if (!outputDrain) {
resetTerminal(state);
return;
}
outputDrain.restoreSnapshot(state);
},
paste: (text: string) => {
if (text.length === 0) {
return;
@@ -560,7 +487,6 @@ function NativeTerminalEmulator({
screenModelRef.current?.reset();
clearSelection();
resetInputModeTracker();
terminalRef.current?.reset();
schedulePaint();
},
claimSize: claimActiveTerminalSize,
@@ -579,14 +505,13 @@ function NativeTerminalEmulator({
claimActiveTerminalSizeForAction,
clearSelection,
resetInputModeTracker,
resetTerminal,
schedulePaint,
],
);
useEffect(() => {
hasNotifiedMeasuredSizeRef.current = false;
resetTerminal(mountInputRef.current.initialSnapshot);
resetTerminal();
callbacksRef.current.onRendererReadyChange?.({ streamKey, isReady: true });
return () => {
callbacksRef.current.onRendererReadyChange?.({ streamKey, isReady: false });
@@ -19,7 +19,6 @@ import type { DOMProps } from "expo/dom";
import { useDOMImperativeHandle, type DOMImperativeFactory } from "expo/dom";
import "@xterm/xterm/css/xterm.css";
import type { ITheme } from "@xterm/xterm";
import type { TerminalState } from "@getpaseo/protocol/messages";
import type { TerminalInputModeState } from "@getpaseo/protocol/terminal-input-mode";
import type { PendingTerminalModifiers } from "../utils/terminal-keys";
import {
@@ -41,8 +40,6 @@ export interface TerminalEmulatorHandle {
find?: TerminalFindHandle;
writeOutput: (data: TerminalOutputData) => void;
replay: (event: import("@getpaseo/client").TerminalReplayEvent) => void;
restoreOutput: (data: TerminalOutputData) => void;
renderSnapshot: (state: TerminalState | null) => void;
paste: (text: string) => void;
copySelection: (clipboard: TerminalClipboardWriter) => Promise<string>;
clear: () => void;
@@ -109,7 +106,6 @@ interface TerminalEmulatorProps {
swipeGesturesEnabled?: boolean;
onSwipeLeft?: () => void;
onSwipeRight?: () => void;
initialSnapshot?: TerminalState | null;
onFindRequest?: () => void;
onFindResult?: (result: TerminalFindResult) => void;
onInput?: (data: string) => Promise<void> | void;
@@ -148,16 +144,6 @@ declare global {
interface Window {}
}
function isTerminalState(value: unknown): value is TerminalState {
return (
typeof value === "object" &&
value !== null &&
"rows" in value &&
"cols" in value &&
"grid" in value
);
}
export default function TerminalEmulator({
ref,
streamKey,
@@ -173,7 +159,6 @@ export default function TerminalEmulator({
swipeGesturesEnabled = false,
onSwipeLeft,
onSwipeRight,
initialSnapshot = null,
onFindRequest,
onFindResult,
onInput,
@@ -229,8 +214,6 @@ export default function TerminalEmulator({
onOpenLocalFileLink,
onFilePaste,
};
const initialSnapshotRef = useRef(initialSnapshot);
initialSnapshotRef.current = initialSnapshot;
const pendingModifiersRef = useRef(pendingModifiers);
pendingModifiersRef.current = pendingModifiers;
@@ -266,18 +249,6 @@ export default function TerminalEmulator({
cols: typeof event.cols === "number" ? event.cols : undefined,
});
},
restoreOutput: (...args) => {
const data = args[0];
if (data instanceof Uint8Array) runtimeRef.current?.restoreOutput({ data });
},
renderSnapshot: (...args) => {
const state = args[0];
if (state === null) {
runtimeRef.current?.renderSnapshot({ state: null });
} else if (isTerminalState(state)) {
runtimeRef.current?.renderSnapshot({ state });
}
},
paste: (...args) => {
const text = args[0];
if (typeof text === "string" && text.length > 0) {
@@ -313,12 +284,6 @@ export default function TerminalEmulator({
runtimeRef.current?.write({ data });
},
replay: (event) => runtimeRef.current?.replay(event),
restoreOutput: (data: TerminalOutputData) => {
runtimeRef.current?.restoreOutput({ data });
},
renderSnapshot: (state: TerminalState | null) => {
runtimeRef.current?.renderSnapshot({ state });
},
paste: (text: string) => {
pasteText(text);
},
@@ -485,7 +450,6 @@ export default function TerminalEmulator({
runtime.mount({
root,
host,
initialSnapshot: initialSnapshotRef.current,
scrollback: scrollbackLinesRef.current,
theme: mountedThemeRef.current,
fontFamily: fontFamilyRef.current,
+1
View File
@@ -617,6 +617,7 @@ export const ar: TranslationResources = {
terminal: {
hostDisconnected: "Host غير متصل",
updateHost: "حدّث المضيف لاستخدام عارض الطرفية الأصلي.",
restoreRequiresUpdate: "حدّث المضيف لعرض الطرفيات.",
unableToSubscribe: "غير قادر على الاشتراك في المحطة",
fileUploaded: "تم رفع الملف",
pasteFileFailed: "تعذر لصق الملف: {{reason}}",
+1
View File
@@ -614,6 +614,7 @@ export const en = {
terminal: {
hostDisconnected: "Host is not connected",
updateHost: "Update the host to use the native terminal renderer.",
restoreRequiresUpdate: "Update the host to view terminals.",
unableToSubscribe: "Unable to subscribe to terminal",
fileUploaded: "File uploaded",
pasteFileFailed: "Failed to paste file: {{reason}}",
+1
View File
@@ -623,6 +623,7 @@ export const es: TranslationResources = {
terminal: {
hostDisconnected: "Hostno está conectado",
updateHost: "Actualiza el host para usar el renderizador de terminal nativo.",
restoreRequiresUpdate: "Actualiza el host para ver las terminales.",
unableToSubscribe: "No se puede suscribir al terminal",
fileUploaded: "Archivo subido",
pasteFileFailed: "No se pudo pegar el archivo: {{reason}}",
+1
View File
@@ -623,6 +623,7 @@ export const fr: TranslationResources = {
terminal: {
hostDisconnected: "L’hôte n’est pas connecté",
updateHost: "Mettez à jour l’hôte pour utiliser le rendu natif du terminal.",
restoreRequiresUpdate: "Mettez à jour l’hôte pour afficher les terminaux.",
unableToSubscribe: "Impossible de s’abonner au terminal",
fileUploaded: "Fichier envoyé",
pasteFileFailed: "Impossible de coller le fichier : {{reason}}",
+1
View File
@@ -623,6 +623,7 @@ export const ja: TranslationResources = {
terminal: {
hostDisconnected: "ホストが接続されていません",
updateHost: "ネイティブターミナルを使用するにはホストを更新してください。",
restoreRequiresUpdate: "ターミナルを表示するにはホストを更新してください。",
unableToSubscribe: "ターミナルに接続できません",
fileUploaded: "ファイルをアップロードしました",
pasteFileFailed: "ファイルを貼り付けられませんでした: {{reason}}",
+1
View File
@@ -618,6 +618,7 @@ export const ko: TranslationResources = {
terminal: {
hostDisconnected: "호스트가 연결되어 있지 않습니다",
updateHost: "기본 터미널 렌더러를 사용하도록 호스트를 업데이트합니다.",
restoreRequiresUpdate: "터미널을 보려면 호스트를 업데이트하세요.",
unableToSubscribe: "터미널을 구독할 수 없습니다",
fileUploaded: "파일을 업로드했습니다",
pasteFileFailed: "파일을 붙여넣지 못했습니다: {{reason}}",
+1
View File
@@ -622,6 +622,7 @@ export const ptBR: TranslationResources = {
terminal: {
hostDisconnected: "Host não está conectado",
updateHost: "Atualize o host para usar o renderizador de terminal nativo.",
restoreRequiresUpdate: "Atualize o host para visualizar os terminais.",
unableToSubscribe: "Não foi possível assinar o terminal",
fileUploaded: "Arquivo enviado",
pasteFileFailed: "Falha ao colar o arquivo: {{reason}}",
+1
View File
@@ -624,6 +624,7 @@ export const ru: TranslationResources = {
terminal: {
hostDisconnected: "Хост не подключён",
updateHost: "Обновите хост, чтобы использовать нативный рендерер терминала.",
restoreRequiresUpdate: "Обновите хост, чтобы просматривать терминалы.",
unableToSubscribe: "Не удалось подключиться к терминалу",
fileUploaded: "Файл загружен",
pasteFileFailed: "Не удалось вставить файл: {{reason}}",
+1
View File
@@ -617,6 +617,7 @@ export const zhCN: TranslationResources = {
terminal: {
hostDisconnected: "Host 未连接",
updateHost: "请更新主机以使用原生终端渲染器。",
restoreRequiresUpdate: "更新主机以查看终端。",
unableToSubscribe: "无法订阅 Terminal",
fileUploaded: "文件已上传",
pasteFileFailed: "无法粘贴文件:{{reason}}",
@@ -1,3 +1,4 @@
import { createTerminalResizePolicy, updateTerminalResizePolicy } from "./terminal-resize-policy";
import { describe, expect, test } from "vitest";
import {
@@ -11,7 +12,7 @@ import {
NATIVE_TERMINAL_PARSE_CHUNK_CHARS,
} from "./headless-terminal-output-drain";
import { createNativeTerminalScreenModel, type TerminalScreenState } from "./terminal-screen-model";
import { renderTerminalSnapshotToAnsi } from "../runtime/terminal-snapshot";
import { renderTerminalSnapshotToAnsi } from "@getpaseo/protocol/terminal-snapshot";
const ROWS = 12;
const COLS = 80;
@@ -94,6 +95,7 @@ function createScreenRestorer(
const frameRecorder = createFrameRecorder();
let screen: TerminalScreenState | null = null;
const drain = createNativeTerminalOutputDrain({
restoreSize: () => {},
write: (chunk) => terminal.write(chunk),
reset: () => terminal.reset(),
getViewportState: () => {
@@ -107,7 +109,12 @@ function createScreenRestorer(
return {
async restore(snapshot: Awaited<ReturnType<typeof benchmarkSnapshot>>) {
drain.restoreSnapshot(snapshot);
drain.replay({ type: "begin", rows: snapshot.rows, cols: snapshot.cols });
drain.replay({
type: "data",
data: new TextEncoder().encode(renderTerminalSnapshotToAnsi(snapshot)),
});
drain.replay({ type: "end" });
await drain.flush();
frameRecorder.frames[0]?.();
return screen;
@@ -122,6 +129,7 @@ describe("native terminal output drain", () => {
const frameRecorder = createFrameRecorder();
const payload = "x".repeat(NATIVE_TERMINAL_PARSE_CHUNK_CHARS * 2);
const drain = createNativeTerminalOutputDrain({
restoreSize: () => {},
write: async (chunk) => {
writes.push(chunk);
},
@@ -170,6 +178,7 @@ describe("native terminal output drain", () => {
const paintedStates: TerminalViewportState[] = [];
const frameRecorder = createFrameRecorder();
const drain = createNativeTerminalOutputDrain({
restoreSize: () => {},
write: (chunk) => terminal.write(chunk),
reset: () => terminal.reset(),
getViewportState: () => terminal.getViewportState(),
@@ -202,6 +211,7 @@ describe("native terminal output drain", () => {
const scheduledWork: Array<() => void> = [];
const frameRecorder = createFrameRecorder();
const drain = createNativeTerminalOutputDrain({
restoreSize: () => {},
write: (chunk) => terminal.write(chunk),
reset: () => terminal.reset(),
getViewportState: () => terminal.getViewportState(),
@@ -219,7 +229,12 @@ describe("native terminal output drain", () => {
await terminal.write("old-one\r\nold-two\r\nold-three\r\n");
drain.enqueueText("stale-queued\r\n");
drain.restoreText(renderTerminalSnapshotToAnsi(restoredSnapshot));
drain.replay({ type: "begin" });
drain.replay({
type: "data",
data: new TextEncoder().encode(renderTerminalSnapshotToAnsi(restoredSnapshot)),
});
drain.replay({ type: "end" });
await runScheduledWork({ callbacks: scheduledWork, flush: drain.flush() });
const state = terminal.getViewportState();
@@ -273,6 +288,7 @@ test("replays split Unicode and ANSI with one reset then live output", async ()
const terminal = createNativeHeadlessTerminal({ rows: 4, cols: 40, scrollbackLines: 1000 });
let resets = 0;
const drain = createNativeTerminalOutputDrain({
restoreSize: () => {},
write: (text) => terminal.write(text),
reset: (size) => {
resets++;
@@ -330,6 +346,7 @@ test("an empty replacement replay resets after an in-flight write finishes", asy
});
let resets = 0;
const drain = createNativeTerminalOutputDrain({
restoreSize: () => {},
write: async () => {
started();
await blocked;
@@ -354,3 +371,114 @@ test("an empty replacement replay resets after an in-flight write finishes", asy
drain.dispose();
}
});
test("replay geometry returns to measured layout before live writes and unchanged-layout focus", async () => {
const measured = { rows: 24, cols: 40 };
const terminal = createNativeHeadlessTerminal({ ...measured, scrollbackLines: 1000 });
const policy = createTerminalResizePolicy(measured);
const sizes: number[] = [];
const drain = createNativeTerminalOutputDrain({
write: async (text) => {
sizes.push(terminal.getState().cols);
await terminal.write(text);
},
reset: (size) => {
if (size) terminal.resize(size);
terminal.reset();
},
restoreSize: () => terminal.resize(measured),
getViewportState: () => terminal.getViewportState(),
onPaint: () => {},
});
try {
drain.replay({ type: "begin", rows: 30, cols: 120 });
drain.replay({ type: "data", data: new TextEncoder().encode("x".repeat(100) + "\r\n") });
drain.replay({ type: "end" });
drain.enqueueText("LIVE-" + "y".repeat(60));
await drain.flush();
expect(sizes).toEqual([120, 40]);
const focus = updateTerminalResizePolicy(policy, {
source: "claim",
size: measured,
claimToken: 1,
});
expect(focus.measuredSizeChanged).toBe(false);
expect(focus.resizeClaim).toMatchObject(measured);
expect(terminal.getState()).toMatchObject(measured);
expect(terminal.getBufferBounds().cols).toBe(40);
const lines = terminal
.getState()
.grid.map((row) =>
row
.map((cell) => cell.char)
.join("")
.trimEnd(),
)
.join("\n");
expect(lines).toContain("LIVE-" + "y".repeat(35) + "\n" + "y".repeat(25));
} finally {
drain.dispose();
terminal.dispose();
}
});
test.each(["clear", "replacement", "dispose"] as const)(
"cancels queued replay geometry on %s",
async (action) => {
const local = { rows: 24, cols: 40 };
const terminal = createNativeHeadlessTerminal({ ...local, scrollbackLines: 1000 });
let finish!: () => void;
const block = new Promise<void>((resolve) => {
finish = resolve;
});
let started!: () => void;
const writing = new Promise<void>((resolve) => {
started = resolve;
});
let first = true;
let restores = 0;
const drain = createNativeTerminalOutputDrain({
restoreSize: () => {
restores++;
terminal.resize(local);
},
reset: (size) => {
if (size) terminal.resize(size);
terminal.reset();
},
write: async (text) => {
await terminal.write(text);
if (first) {
first = false;
started();
await block;
}
},
getViewportState: () => terminal.getViewportState(),
onPaint: () => {},
});
try {
drain.replay({ type: "begin", rows: 30, cols: 120 });
drain.replay({ type: "data", data: new TextEncoder().encode("old") });
await writing;
drain.replay({ type: "end" });
if (action === "clear") drain.clear();
else if (action === "dispose") {
drain.dispose();
drain.clear();
} else {
drain.replay({ type: "begin", rows: 20, cols: 80 });
drain.replay({ type: "data", data: new TextEncoder().encode("new") });
drain.replay({ type: "end" });
}
finish();
await drain.flush();
expect(restores).toBe(action === "dispose" ? 0 : 1);
if (action !== "dispose") expect(terminal.getState()).toMatchObject(local);
} finally {
finish();
drain.dispose();
terminal.dispose();
}
},
);
@@ -1,5 +1,3 @@
import type { TerminalState } from "@getpaseo/protocol/messages";
import { renderTerminalSnapshotToAnsi } from "../runtime/terminal-snapshot";
import type { TerminalViewportState } from "./headless-terminal-state";
import { nativeTerminalPerformanceNow } from "./terminal-performance";
@@ -16,8 +14,6 @@ export interface NativeTerminalParseSample {
export interface NativeTerminalOutputDrain {
replay(event: import("@getpaseo/client").TerminalReplayEvent): void;
enqueueText(text: string): void;
restoreText(text: string): void;
restoreSnapshot(snapshot: TerminalState): void;
clear(): void;
flush(): Promise<void>;
dispose(): void;
@@ -26,6 +22,7 @@ export interface NativeTerminalOutputDrain {
export interface NativeTerminalOutputDrainOptions {
write: (chunk: string) => Promise<void>;
restoreSize: () => void;
reset: (size?: { rows: number; cols: number }) => void;
getViewportState: () => TerminalViewportState | null;
onPaint: (state: TerminalViewportState) => void;
@@ -69,18 +66,21 @@ export function createNativeTerminalOutputDrain(
const chunkChars = options.chunkChars ?? NATIVE_TERMINAL_PARSE_CHUNK_CHARS;
const chunksPerTurn = options.chunksPerTurn ?? NATIVE_TERMINAL_PARSE_CHUNKS_PER_TURN;
const idleWaiters: Array<() => void> = [];
let pendingText = "";
type Operation =
| { type: "write"; text: string }
| { type: "reset"; size?: { rows: number; cols: number } }
| { type: "restore-size" };
let operations: Operation[] = [];
let queuedChars = 0;
let drainScheduled = false;
let draining = false;
let disposed = false;
let paintFrame: number | null = null;
let generation = 0;
let resetBeforeNextWrite = false;
let replaySize: { rows: number; cols: number } | undefined;
const replayDecoder = new TextDecoder();
function resolveIdleIfReady(): void {
if (drainScheduled || draining || pendingText.length > 0 || resetBeforeNextWrite) {
if (drainScheduled || draining || operations.length > 0) {
return;
}
while (idleWaiters.length > 0) {
@@ -122,15 +122,6 @@ export function createNativeTerminalOutputDrain(
});
}
function replacePendingText(text: string): void {
generation += 1;
pendingText = text;
resetBeforeNextWrite = true;
cancelPendingPaint();
scheduleDrain();
resolveIdleIfReady();
}
async function drainPending(): Promise<void> {
if (disposed || draining) {
return;
@@ -141,18 +132,20 @@ export function createNativeTerminalOutputDrain(
let parsedChunks = 0;
try {
if (resetBeforeNextWrite) {
options.reset(replaySize);
replaySize = undefined;
resetBeforeNextWrite = false;
requestPaint();
}
while (pendingText.length > 0 && parsedChunks < chunksPerTurn) {
if (disposed || activeGeneration !== generation) {
return;
while (operations.length > 0 && parsedChunks < chunksPerTurn) {
if (disposed || activeGeneration !== generation) return;
const operation = operations[0]!;
if (operation.type !== "write") {
operations.shift();
if (operation.type === "reset") options.reset(operation.size);
else options.restoreSize();
requestPaint();
continue;
}
const chunk = pendingText.slice(0, chunkChars);
pendingText = pendingText.slice(chunk.length);
const chunk = operation.text.slice(0, chunkChars);
operation.text = operation.text.slice(chunk.length);
queuedChars -= chunk.length;
if (!operation.text) operations.shift();
const parseStart = nativeTerminalPerformanceNow();
await options.write(chunk);
if (disposed || activeGeneration !== generation) {
@@ -161,14 +154,14 @@ export function createNativeTerminalOutputDrain(
options.onParse?.({
durationMs: nativeTerminalPerformanceNow() - parseStart,
parsedChars: chunk.length,
queuedChars: pendingText.length,
queuedChars,
});
requestPaint();
parsedChunks += 1;
}
} finally {
draining = false;
if (!disposed && (pendingText.length > 0 || resetBeforeNextWrite)) {
if (!disposed && operations.length > 0) {
scheduleDrain();
}
resolveIdleIfReady();
@@ -177,48 +170,48 @@ export function createNativeTerminalOutputDrain(
return {
replay(event): void {
if (disposed) return;
if (event.type === "begin") {
generation++;
replayDecoder.decode();
replaySize = event.rows && event.cols ? { rows: event.rows, cols: event.cols } : undefined;
replacePendingText("");
operations = [
{
type: "reset",
size: event.rows && event.cols ? { rows: event.rows, cols: event.cols } : undefined,
},
];
queuedChars = 0;
cancelPendingPaint();
} else if (event.type === "data") {
this.enqueueText(replayDecoder.decode(event.data, { stream: true }));
} else {
this.enqueueText(replayDecoder.decode());
operations.push({ type: "restore-size" });
}
scheduleDrain();
},
enqueueText(text: string): void {
if (disposed || text.length === 0) {
return;
}
pendingText += text;
if (disposed || !text) return;
const last = operations.at(-1);
if (last?.type === "write") last.text += text;
else operations.push({ type: "write", text });
queuedChars += text.length;
scheduleDrain();
},
restoreText(text: string): void {
if (disposed) {
return;
}
replacePendingText(text);
},
restoreSnapshot(snapshot: TerminalState): void {
if (disposed) {
return;
}
replacePendingText(renderTerminalSnapshotToAnsi(snapshot));
},
clear(): void {
if (disposed) return;
generation += 1;
pendingText = "";
resetBeforeNextWrite = false;
operations = [{ type: "restore-size" }, { type: "reset" }];
queuedChars = 0;
replayDecoder.decode();
cancelPendingPaint();
scheduleDrain();
resolveIdleIfReady();
},
flush(): Promise<void> {
if (!drainScheduled && !draining && pendingText.length === 0 && !resetBeforeNextWrite) {
if (!drainScheduled && !draining && operations.length === 0) {
return Promise.resolve();
}
scheduleDrain();
@@ -230,8 +223,8 @@ export function createNativeTerminalOutputDrain(
dispose(): void {
disposed = true;
generation += 1;
pendingText = "";
resetBeforeNextWrite = false;
operations = [];
queuedChars = 0;
cancelPendingPaint();
while (idleWaiters.length > 0) {
idleWaiters.shift()?.();
@@ -239,7 +232,7 @@ export function createNativeTerminalOutputDrain(
},
getQueuedChars(): number {
return pendingText.length;
return queuedChars;
},
};
}
@@ -166,7 +166,6 @@ function createTerminalHost(input: CreateTerminalHostInput): MountedTerminal {
runtime.mount({
root,
host,
initialSnapshot: null,
scrollback: input.scrollback ?? 10_000,
theme: {
background: "#0b0b0b",
@@ -750,30 +749,11 @@ describe("terminal emulator runtime in a real browser", () => {
const reset = vi.fn(originalReset);
terminal.reset = reset;
mounted.runtime.renderSnapshot({
state: {
rows: terminal.rows,
cols: terminal.cols,
scrollback: [],
grid: [
[
{ char: "p" },
{ char: "r" },
{ char: "o" },
{ char: "m" },
{ char: "p" },
{ char: "t" },
],
],
cursor: {
row: 0,
col: 6,
},
},
});
await nextFrame();
mounted.runtime.replay({ type: "begin", rows: terminal.rows, cols: terminal.cols });
mounted.runtime.replay({ type: "data", data: terminalOutput("prompt") });
mounted.runtime.replay({ type: "end" });
expect(reset).not.toHaveBeenCalled();
await nextFrame();
});
it("opens OSC 8 hyperlinks the same way as plain-text URLs", async () => {
@@ -810,7 +790,6 @@ describe("incremental replay lifecycle", () => {
mounted.runtime.mount({
root: mounted.root,
host: mounted.host,
initialSnapshot: null,
scrollback: 1000,
theme: {},
});
@@ -840,3 +819,20 @@ describe("incremental replay lifecycle", () => {
expect(mounted.inputs[0]?.startsWith("\x1b]4;1;rgb:")).toBe(true);
});
});
it("restores fitted web geometry after replay before live output", async () => {
const mounted = createTerminalHost({ width: 720, height: 360 });
await settleMountRefits();
const local = latestSize(mounted.sizes);
mounted.runtime.replay({ type: "begin", rows: 30, cols: 120 });
mounted.runtime.replay({ type: "data", data: terminalOutput("x".repeat(100)) });
mounted.runtime.replay({ type: "end" });
await new Promise<void>((resolve) =>
mounted.runtime.write({ data: terminalOutput("LIVE"), onCommitted: resolve }),
);
expect(getBrowserTerminal().cols).toBe(latestSize(mounted.sizes).cols);
expect(getBrowserTerminal().cols).toBeLessThan(120);
expect(getBrowserTerminal().rows).toBe(local.rows);
mounted.runtime.resize({ forceClaim: true, shouldClaim: true });
expect(getBrowserTerminal().cols).toBe(latestSize(mounted.sizes).cols);
});
@@ -302,25 +302,14 @@ describe("terminal-emulator-runtime", () => {
expect(committed).toEqual(["first", "second", "clear", "third"]);
});
it("falls back to timeout commit for a barrier op when the gate sentinel never fires", () => {
it("unblocks queued replay writes if a begin barrier never commits", () => {
vi.useFakeTimers();
const { runtime, terminal } = createRuntimeWithTerminal();
const onCommitted = vi.fn();
// restoreOutput is a barrier; if xterm never commits the gate sentinel, the barrier still
// applies (and commits) after the 5s safety timeout instead of stalling forever.
runtime.restoreOutput({
data: terminalOutput("stuck"),
onCommitted,
});
expect(onCommitted).not.toHaveBeenCalled();
expect(terminal.resetCalls).toBe(0);
vi.advanceTimersByTime(5_000);
// The barrier now runs (write submitted) but its own commit still waits on xterm; advance
// again to fire the barrier's own write timeout.
vi.advanceTimersByTime(5_000);
expect(onCommitted).toHaveBeenCalledTimes(1);
const { runtime, writeTexts } = createRuntimeWithTerminal();
runtime.replay({ type: "begin" });
runtime.replay({ type: "data", data: terminalOutput("stuck") });
expect(writeTexts).toEqual(["\x1bc"]);
vi.advanceTimersByTime(5000);
expect(writeTexts).toEqual(["\x1bc", "stuck"]);
});
it("reports input mode changes from terminal output and resets them on snapshots", () => {
@@ -343,18 +332,7 @@ describe("terminal-emulator-runtime", () => {
});
runtime.write({ data: terminalOutput("\x1b[>7u") });
runtime.renderSnapshot({
state: {
rows: 2,
cols: 8,
scrollback: [],
grid: [[{ char: "$" }, { char: " " }]],
cursor: {
row: 0,
col: 2,
},
},
});
runtime.replay({ type: "begin", rows: 2, cols: 8 });
// The plain write reports kitty flags synchronously during drain; the snapshot resets
// them only after its barrier gate (the sentinel write callback) resolves.
expect(inputModeChanges).toEqual([
@@ -410,40 +388,18 @@ describe("terminal-emulator-runtime", () => {
expect(committed).toEqual(["second", "first"]);
});
it("applies a snapshot after the writes ahead of it, never interleaving the reset", () => {
it("keeps replay reset and data after preceding writes, without interleaving", () => {
const { runtime, terminal, writeTexts, writeCallbacks } = createRuntimeWithTerminal();
const committed: string[] = [];
runtime.write({
data: terminalOutput("before-a"),
onCommitted: () => {
committed.push("before-a");
},
});
runtime.write({
data: terminalOutput("before-b"),
onCommitted: () => {
committed.push("before-b");
},
});
runtime.restoreOutput({
data: terminalOutput("snap"),
onCommitted: () => {
committed.push("snap");
},
});
// Plain writes are submitted up front; the snapshot's reset+write has not run yet.
runtime.write({ data: terminalOutput("before-a") });
runtime.write({ data: terminalOutput("before-b") });
runtime.replay({ type: "begin" });
runtime.replay({ type: "data", data: terminalOutput("snap") });
runtime.replay({ type: "end" });
expect(writeTexts).toEqual(["before-a", "before-b"]);
// writeCallbacks: [0]=before-a, [1]=before-b, [2]=barrier gate sentinel.
// The snapshot is gated behind the prior writes; resolving the gate runs it after them.
writeCallbacks[2]?.();
expect(writeTexts).toEqual(["before-a", "before-b", "csnap"]);
// Snapshot commit fires only after its own write callback, strictly after the writes.
expect(writeTexts).toEqual(["before-a", "before-b", "\x1bc"]);
writeCallbacks[3]?.();
expect(committed).toEqual(["snap"]);
expect(writeTexts).toEqual(["before-a", "before-b", "\x1bc", "snap"]);
expect(terminal.resetCalls).toBe(0);
});
@@ -456,7 +412,7 @@ describe("terminal-emulator-runtime", () => {
// No plain writes precede this barrier (mount), so there is nothing to gate: it starts
// at once with no sentinel, flipping suppressInput synchronously.
runtime.restoreOutput({ data: terminalOutput("snapshot") });
runtime.replay({ type: "begin" });
expect(readSuppressInput()).toBe(true);
// writeCallbacks[0] is the barrier's own snapshot write; committing it restores input.
@@ -471,7 +427,7 @@ describe("terminal-emulator-runtime", () => {
// A plain write is now ungated, so the following barrier must wait on the sentinel.
runtime.write({ data: terminalOutput("output") });
runtime.restoreOutput({ data: terminalOutput("snapshot") });
runtime.replay({ type: "begin" });
// The plain write commits first. Only the following sentinel opens the barrier.
expect(readSuppressInput()).toBe(false);
@@ -505,57 +461,40 @@ describe("terminal-emulator-runtime", () => {
expect(onCommittedB).toHaveBeenCalledTimes(1);
});
it("clears ungated writes on unmount so remount barriers apply immediately", () => {
it("clears ungated writes on unmount so remount replay begins immediately", () => {
const { runtime } = createRuntimeWithTerminal();
runtime.write({ data: terminalOutput("before unmount") });
runtime.unmount();
const remounted = attachStubTerminal(runtime);
runtime.restoreOutput({ data: terminalOutput("restored screen") });
expect(remounted.writeTexts).toEqual(["\u001bcrestored screen"]);
expect(remounted.writeCallbacks).toHaveLength(1);
runtime.replay({ type: "begin" });
runtime.replay({ type: "data", data: terminalOutput("restored screen") });
expect(remounted.writeTexts).toEqual(["\x1bc"]);
remounted.writeCallbacks[0]?.();
expect(remounted.writeTexts).toEqual(["\x1bc", "restored screen"]);
});
it("replays snapshots through a single write without first painting a reset terminal", () => {
it("parses replay chunks serially without synchronously resetting the terminal", () => {
const { runtime, terminal, writeTexts, writeCallbacks } = createRuntimeWithTerminal();
runtime.renderSnapshot({
state: {
rows: 2,
cols: 8,
scrollback: [],
grid: [
[{ char: "h" }, { char: "i" }],
[{ char: "$" }, { char: " " }],
],
cursor: {
row: 1,
col: 2,
},
},
});
// The snapshot is a barrier; its write applies after the gate sentinel (writeCallbacks[0]).
writeCallbacks[0]?.();
runtime.replay({ type: "begin", rows: 2, cols: 8 });
runtime.replay({ type: "data", data: terminalOutput("hi") });
runtime.replay({ type: "data", data: terminalOutput("\r\n$") });
runtime.replay({ type: "end" });
expect(terminal.resetCalls).toBe(0);
expect(writeTexts).toHaveLength(1);
expect(writeTexts[0]?.startsWith("\u001bc")).toBe(true);
expect(writeTexts[0]).toContain("hi");
expect(writeTexts).toEqual(["\x1bc"]);
writeCallbacks[0]?.();
expect(writeTexts).toEqual(["\x1bc", "hi"]);
writeCallbacks[1]?.();
expect(writeTexts).toEqual(["\x1bc", "hi", "\r\n$"]);
});
it("restores server-rendered ANSI snapshots through the snapshot write path", () => {
it("restores worker-rendered ANSI through the replay lifecycle", () => {
const { runtime, terminal, writeTexts, writeCallbacks } = createRuntimeWithTerminal();
runtime.restoreOutput({ data: terminalOutput("restored screen") });
// restoreOutput is a barrier (suppressInput); it applies after the gate sentinel.
runtime.replay({ type: "begin" });
runtime.replay({ type: "data", data: terminalOutput("restored screen") });
runtime.replay({ type: "end" });
writeCallbacks[0]?.();
expect(terminal.resetCalls).toBe(0);
expect(writeTexts).toEqual(["\u001bcrestored screen"]);
expect(writeTexts).toEqual(["\x1bc", "restored screen"]);
});
it("forces a refit when resize is requested", () => {
@@ -7,7 +7,6 @@ import { WebLinksAddon } from "@xterm/addon-web-links";
import { WebglAddon } from "@xterm/addon-webgl";
import { LigaturesAddon } from "@xterm/addon-ligatures/lib/addon-ligatures.mjs";
import { Terminal, type ITheme, type IMarker } from "@xterm/xterm";
import type { TerminalState } from "@getpaseo/protocol/messages";
import {
type TerminalInputModeState,
TerminalInputModeTracker,
@@ -24,7 +23,6 @@ import {
resolveMacTerminalEditingShortcut,
shouldInterceptDomTerminalKey,
} from "@/utils/terminal-keys";
import { renderTerminalSnapshotToAnsi } from "./terminal-snapshot";
import {
createTerminalLocalFileLinkProvider,
type TerminalLocalFileLinkSource,
@@ -55,7 +53,6 @@ const FIND_HIGHLIGHT_LIMIT = 20_000;
export interface TerminalEmulatorRuntimeMountInput {
root: HTMLDivElement;
host: HTMLDivElement;
initialSnapshot: TerminalState | null;
scrollback: number;
theme: ITheme;
fontFamily?: string;
@@ -134,7 +131,7 @@ interface TerminalEmulatorRuntimeDisposables {
}
interface TerminalOutputOperation {
type: "write" | "clear" | "snapshot";
type: "write" | "clear" | "snapshot" | "fit";
data: TerminalOutputData;
rows?: number;
cols?: number;
@@ -777,10 +774,6 @@ export class TerminalEmulatorRuntime {
fitAndEmitResize({ forceRefresh: true, shouldClaim: false });
}, 0);
if (input.initialSnapshot) {
this.renderSnapshot({ state: input.initialSnapshot });
}
this.processOutputQueue();
const disposables: TerminalEmulatorRuntimeDisposables = {
@@ -898,28 +891,18 @@ export class TerminalEmulatorRuntime {
onFilePaste?.(file);
}
renderSnapshot(input: { state: TerminalState | null; onCommitted?: () => void }): void {
if (!input.state) {
this.clear(input);
return;
}
this.restoreOutput({
data: encodeTerminalOutput(renderTerminalSnapshotToAnsi(input.state)),
rows: input.state.rows,
cols: input.state.cols,
...(input.onCommitted ? { onCommitted: input.onCommitted } : {}),
});
}
replay(event: import("@getpaseo/client").TerminalReplayEvent): void {
if (event.type === "begin") {
this.restoreOutput({ data: EMPTY_TERMINAL_OUTPUT, rows: event.rows, cols: event.cols });
} else if (event.type === "data") {
this.write({ data: event.data, suppressInput: true });
} else {
this.outputOperations.push({ type: "fit", data: EMPTY_TERMINAL_OUTPUT });
this.processOutputQueue();
}
}
restoreOutput(input: {
private restoreOutput(input: {
data: TerminalOutputData;
rows?: number;
cols?: number;
@@ -1184,6 +1167,17 @@ export class TerminalEmulatorRuntime {
this.processOutputQueue();
};
if (operation.type === "fit") {
// xterm resize flushes its write buffer synchronously. Leave the parser's
// commit callback before fitting, or that flush consumes the next write.
queueMicrotask(() => {
if (this.inFlightOutputOperation !== operation || this.terminal !== terminal) return;
this.fitAndEmitResize?.({ shouldClaim: false });
finalizeOperation(operation);
});
return;
}
if (operation.type === "clear") {
this.inputModeDecoder.decode();
this.inputModeTracker.reset();
@@ -31,7 +31,6 @@ function mount(options?: { isMac: boolean }) {
runtime.mount({
root,
host,
initialSnapshot: null,
scrollback: 1000,
theme: { background: "#ffffff", foreground: "#111111" },
});
@@ -38,7 +38,6 @@ function mountTerminal(left: number): TerminalView {
runtime.mount({
root,
host,
initialSnapshot: null,
scrollback: 1000,
fontFamily: "monospace",
fontSize: 16,
@@ -1,3 +1,4 @@
import { renderTerminalSnapshotToAnsi } from "@getpaseo/protocol/terminal-snapshot";
import { page } from "@vitest/browser/context";
import { afterEach, describe, expect, it } from "vitest";
import type { TerminalState } from "@getpaseo/protocol/messages";
@@ -57,7 +58,6 @@ function mount(width: number, height: number): Mounted {
runtime.mount({
root,
host,
initialSnapshot: null,
scrollback: 10_000,
theme: { background: "#0b0b0b", foreground: "#e6e6e6", cursor: "#e6e6e6" },
});
@@ -90,9 +90,15 @@ function someRowContains(term: InspectableTerminal, text: string): boolean {
return dumpRows(term).some((r) => r.text.includes(text));
}
function renderSnapshotCommitted(runtime: TerminalEmulatorRuntime, state: TerminalState) {
function replayCommitted(runtime: TerminalEmulatorRuntime, state: TerminalState) {
return new Promise<void>((resolve) => {
runtime.renderSnapshot({ state, onCommitted: resolve });
runtime.replay({ type: "begin", rows: state.rows, cols: state.cols });
runtime.replay({
type: "data",
data: new TextEncoder().encode(renderTerminalSnapshotToAnsi(state)),
});
runtime.replay({ type: "end" });
runtime.write({ data: new Uint8Array([0]), onCommitted: resolve });
});
}
@@ -149,7 +155,7 @@ describe("terminal resize reflow repro (Paseo terminal)", () => {
// 1) Mid-stream snapshot arrives (server overflowed 256KB). It carries the
// long line wrapped at the server width as separate grid rows.
await renderSnapshotCommitted(m.runtime, buildWrappedSnapshot(narrowCols));
await replayCommitted(m.runtime, buildWrappedSnapshot(narrowCols));
await waitFor(() => someRowContains(m.terminal, "[15:30:00.123]"));
// 2) Normal post-snapshot streaming resumes (autowrap on -> soft-wrapped).
@@ -4,14 +4,14 @@ import { resolveTerminalRestoreOptions } from "./terminal-restore-options";
describe("terminal restore options", () => {
it("omits restore options for daemons without terminal restore modes", () => {
expect(
expect(() =>
resolveTerminalRestoreOptions({
scrollbackLines: 10000,
supportsTerminalRestoreModes: false,
canClaimSize: true,
size: { rows: 24, cols: 80 },
}),
).toBeUndefined();
).toThrow("Update the host to view terminals.");
});
it("requests visible snapshot restore with configured scrollback for capable daemons", () => {
@@ -1,3 +1,4 @@
import { i18n } from "@/i18n/i18next";
import type { SubscribeTerminalRequest } from "@getpaseo/protocol/messages";
export interface ResolveTerminalRestoreOptionsInput {
@@ -9,9 +10,9 @@ export interface ResolveTerminalRestoreOptionsInput {
export function resolveTerminalRestoreOptions(
input: ResolveTerminalRestoreOptionsInput,
): SubscribeTerminalRequest["restore"] | undefined {
): NonNullable<SubscribeTerminalRequest["restore"]> {
if (!input.supportsTerminalRestoreModes) {
return undefined;
throw new Error(i18n.t("workspace.terminal.restoreRequiresUpdate"));
}
return {
@@ -2,7 +2,7 @@ import { Terminal as HeadlessTerminal } from "@xterm/headless";
import { Terminal as ClientTerminal } from "@xterm/xterm";
import { describe, expect, it } from "vitest";
import { renderTerminalSnapshotToAnsi } from "./terminal-snapshot";
import { renderTerminalSnapshotToAnsi } from "@getpaseo/protocol/terminal-snapshot";
interface SnapshotCell {
char: string;
@@ -1 +0,0 @@
export { renderTerminalSnapshotToAnsi } from "@getpaseo/protocol/terminal-snapshot";
@@ -1,3 +1,4 @@
import { resolveTerminalRestoreOptions } from "./terminal-restore-options";
import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
import { describe, expect, it } from "vitest";
@@ -283,3 +284,28 @@ it("reports observation failure without declaring the PTY exited, then allows re
expect(h.statuses.at(-1)?.error).toBe(null);
h.controller.dispose();
});
it("does not resize or attach on a JSON-era host and reports the required update", () => {
const client = new FakeTerminalStreamClient();
const statuses: TerminalStreamControllerStatus[] = [];
const controller = new TerminalStreamController({
client,
getPreferredSize: () => ({ rows: 24, cols: 40 }),
onOutput: () => {},
getRestoreOptions: () =>
resolveTerminalRestoreOptions({
supportsTerminalRestoreModes: false,
scrollbackLines: 10000,
canClaimSize: true,
size: { rows: 24, cols: 40 },
}),
onStatusChange: (status) => statuses.push(status),
});
controller.setTerminal({ terminalId: "old-host-terminal" });
expect(client.subscribeCalls).toEqual([]);
expect(client.resizeCalls).toEqual([]);
expect(statuses.at(-1)).toMatchObject({
isAttaching: false,
error: "Update the host to view terminals.",
});
});
@@ -66,10 +66,10 @@ export class TerminalStreamController {
this.options.onStatusChange?.({ terminalId: null, isAttaching: false, error: null });
return;
}
const restore = this.options.getRestoreOptions?.();
this.options.onStatusChange?.({ terminalId: nextTerminalId, isAttaching: true, error: null });
let subscription: ReturnType<DaemonClient["observeTerminal"]>;
try {
const restore = this.options.getRestoreOptions?.();
const preferredSize = restore?.size ?? this.options.getPreferredSize();
if (preferredSize)
this.options.client.sendTerminalInput(nextTerminalId, {
@@ -3,7 +3,6 @@ import { describe, expect, it } from "vitest";
import {
applyTerminalRendererReadyChange,
resolveTerminalStreamTarget,
shouldReplayTerminalSnapshotForRenderer,
shouldShowTerminalLoadingOverlay,
} from "./terminal-renderer-readiness";
@@ -74,27 +73,6 @@ describe("terminal-renderer-readiness", () => {
expect(current).toBeNull();
});
it("replays snapshots only for ready events from the current renderer", () => {
expect(
shouldReplayTerminalSnapshotForRenderer({
change: { streamKey: "scope:terminal-1", isReady: true },
terminalStreamKey: "scope:terminal-1",
}),
).toBe(true);
expect(
shouldReplayTerminalSnapshotForRenderer({
change: { streamKey: "scope:terminal-1", isReady: false },
terminalStreamKey: "scope:terminal-1",
}),
).toBe(false);
expect(
shouldReplayTerminalSnapshotForRenderer({
change: { streamKey: "scope:terminal-1", isReady: true },
terminalStreamKey: "scope:terminal-2",
}),
).toBe(false);
});
it("does not attach the stream before the matching renderer is ready", () => {
expect(
resolveTerminalStreamTarget({
@@ -14,13 +14,6 @@ export function applyTerminalRendererReadyChange(
return currentReadyStreamKey === change.streamKey ? null : currentReadyStreamKey;
}
export function shouldReplayTerminalSnapshotForRenderer(input: {
change: TerminalRendererReadyChange;
terminalStreamKey: string;
}): boolean {
return input.change.isReady && input.change.streamKey === input.terminalStreamKey;
}
export function resolveTerminalStreamTarget(input: {
terminalId: string;
terminalStreamKey: string;
-1
View File
@@ -3740,7 +3740,6 @@ export const ServerInfoStatusPayloadSchema = z
resumeOnStartupSetting: z.boolean().optional(),
// COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23.
"terminal-restore-modes": z.boolean().optional(),
terminalChunkedReplay: z.boolean().optional(),
// COMPAT(terminalInputModeReplay): added in v0.2.6, remove gate after 2027-02-02.
"terminal-input-mode-replay": z.boolean().optional(),
// COMPAT(terminalSizeOwnership): added in v0.2.6, remove gate after 2027-02-02.
@@ -11,7 +11,6 @@ import {
type WSOutboundMessage,
} from "@getpaseo/protocol/messages";
import {
decodeTerminalSnapshotPayload,
decodeTerminalStreamFrame,
encodeTerminalStreamFrame,
TerminalStreamOpcode,
@@ -423,17 +422,24 @@ async function collectRawBinaryFrames(
});
}
async function decodeRestoreState(
frame: TerminalStreamFrame,
terminalId: string,
): Promise<TerminalState> {
const size = ctx.daemon.daemon.terminalManager!.getTerminal(terminalId)!.getSize();
const terminal = new HeadlessTerminal({ ...size, scrollback: 1000, allowProposedApi: true });
try {
await new Promise<void>((resolve) => terminal.write(frame.payload, resolve));
return readHeadlessState(terminal);
} finally {
terminal.dispose();
}
}
function getFrameText(frame: TerminalStreamFrame): string {
if (frame.opcode === TerminalStreamOpcode.Output) {
return new TextDecoder().decode(frame.payload);
}
if (frame.opcode === TerminalStreamOpcode.Snapshot) {
const state = decodeTerminalSnapshotPayload(frame.payload);
if (!state) {
return "";
}
return extractStateText(state);
}
if (frame.opcode === TerminalStreamOpcode.Restore) {
return new TextDecoder().decode(frame.payload);
}
@@ -483,7 +489,7 @@ function startTerminalOutputCapture(ws: WebSocket, slot: number): TerminalOutput
if (!frame || frame.slot !== slot) {
return;
}
if (frame.opcode === TerminalStreamOpcode.Snapshot) {
if (frame.opcode === TerminalStreamOpcode.Restore) {
snapshotCount += 1;
return;
}
@@ -671,7 +677,7 @@ async function measureRepeatCadenceForTerminal(input: {
const slot = await subscribeRawTerminal(ws, terminalId, `sub-repeat-cadence-${input.name}`);
await waitForRawBinaryFrame(
ws,
(frame) => frame.slot === slot && frame.opcode === TerminalStreamOpcode.Snapshot,
(frame) => frame.slot === slot && frame.opcode === TerminalStreamOpcode.Restore,
10000,
);
await new Promise((resolve) => setTimeout(resolve, 1000));
@@ -854,13 +860,7 @@ test("live terminal restore skips the initial snapshot", async () => {
expect(outputText).toContain("after-live");
expect(outputText).not.toContain("before-live");
expect(
outputFrames.some(
(frame) =>
frame.opcode === TerminalStreamOpcode.Snapshot ||
frame.opcode === TerminalStreamOpcode.Restore,
),
).toBe(false);
expect(outputFrames.some((frame) => frame.opcode === TerminalStreamOpcode.Restore)).toBe(false);
} finally {
await closeWebSocket(ws);
}
@@ -938,15 +938,12 @@ test("propagates debounced terminal titles through list responses and snapshots"
}
expect(listedTitle).toBe("Build Output");
const snapshotPromise = waitForTerminalSnapshot(
ctx.client,
terminalId,
(state) => state.title === "Build Output",
const attached = await ctx.client.subscribeTerminal(terminalId);
expect(attached.error).toBeNull();
const afterAttach = await ctx.client.listTerminals(cwd);
expect(afterAttach.terminals.find((terminal) => terminal.id === terminalId)?.title).toBe(
"Build Output",
);
await ctx.client.subscribeTerminal(terminalId);
const snapshot = await snapshotPromise;
expect(snapshot.title).toBe("Build Output");
rmSync(cwd, { recursive: true, force: true });
}, 30000);
@@ -1010,13 +1007,10 @@ test("subscribe response is sent before the initial snapshot frame", async () =>
return;
}
const frame = decodeTerminalStreamFrame(new Uint8Array(buffer));
if (frame?.opcode !== TerminalStreamOpcode.Snapshot) {
return;
}
const state = decodeTerminalSnapshotPayload(frame.payload);
if (!state || !extractStateText(state).includes("hello-ordering")) {
if (frame?.opcode !== TerminalStreamOpcode.Restore) {
return;
}
if (!getFrameText(frame).includes("hello-ordering")) return;
events.push("snapshot");
maybeResolve();
};
@@ -1150,7 +1144,7 @@ test("reconnected terminal streams replay active input modes after the snapshot"
const firstSlot = await subscribeRawTerminal(firstWs, terminalId, "sub-input-mode-first");
await waitForRawBinaryFrame(
firstWs,
(frame) => frame.slot === firstSlot && frame.opcode === TerminalStreamOpcode.Snapshot,
(frame) => frame.slot === firstSlot && frame.opcode === TerminalStreamOpcode.Restore,
);
sendRawTerminalInput(firstWs, firstSlot, "e");
await waitForRawBinaryFrame(
@@ -1177,7 +1171,7 @@ test("reconnected terminal streams replay active input modes after the snapshot"
const secondSlot = await subscribeRawTerminal(secondWs, terminalId, "sub-input-mode-second");
await waitForRawBinaryFrame(
secondWs,
(frame) => frame.slot === secondSlot && frame.opcode === TerminalStreamOpcode.Snapshot,
(frame) => frame.slot === secondSlot && frame.opcode === TerminalStreamOpcode.Restore,
);
const replay = await waitForRawBinaryFrame(
secondWs,
@@ -1209,7 +1203,7 @@ test("reconnected terminal streams do not replay input modes before they are ena
const slot = await subscribeRawTerminal(ws, terminalId, "sub-input-mode-inactive");
await waitForRawBinaryFrame(
ws,
(frame) => frame.slot === slot && frame.opcode === TerminalStreamOpcode.Snapshot,
(frame) => frame.slot === slot && frame.opcode === TerminalStreamOpcode.Restore,
);
await waitForNoRawBinaryFrame(ws, 300);
} finally {
@@ -1225,7 +1219,7 @@ test("fast output to a slow websocket client falls back to a snapshot", async ()
const ws = await connectRawWebSocket(ctx.daemon.port);
await subscribeRawTerminal(ws, terminalId, "sub-raw");
await waitForRawBinaryFrame(ws, (frame) => frame.opcode === TerminalStreamOpcode.Snapshot, 10000);
await waitForRawBinaryFrame(ws, (frame) => frame.opcode === TerminalStreamOpcode.Restore, 10000);
const rawSocket = (ws as WebSocket & { _socket?: { pause: () => void; resume: () => void } })
._socket;
@@ -1240,12 +1234,12 @@ test("fast output to a slow websocket client falls back to a snapshot", async ()
await new Promise((resolve) => setTimeout(resolve, 750));
const catchUpFramePromise = waitForRawBinaryFrame(
ws,
(frame) => frame.opcode === TerminalStreamOpcode.Snapshot,
(frame) => frame.opcode === TerminalStreamOpcode.Restore,
15000,
);
rawSocket!.resume();
const catchUpFrame = await catchUpFramePromise;
expect(catchUpFrame.opcode).toBe(TerminalStreamOpcode.Snapshot);
expect(catchUpFrame.opcode).toBe(TerminalStreamOpcode.Restore);
await closeWebSocket(ws);
rmSync(cwd, { recursive: true, force: true });
@@ -1440,10 +1434,10 @@ test("websocket terminate then new connection gets snapshot with all prior outpu
await subscribeRawTerminal(firstSocket, terminalId, "sub-drop-a");
const initialSnapshot = await waitForRawBinaryFrame(
firstSocket,
(frame) => frame.opcode === TerminalStreamOpcode.Snapshot,
(frame) => frame.opcode === TerminalStreamOpcode.Restore,
10000,
);
expect(initialSnapshot.opcode).toBe(TerminalStreamOpcode.Snapshot);
expect(initialSnapshot.opcode).toBe(TerminalStreamOpcode.Restore);
ctx.client.sendTerminalInput(terminalId, {
type: "input",
@@ -1479,9 +1473,9 @@ test("websocket terminate then new connection gets snapshot with all prior outpu
try {
await subscribeRawTerminal(secondSocket, terminalId, "sub-drop-b");
const reconnectFrame = await waitForRawBinaryFrame(secondSocket, () => true, 10000);
expect(reconnectFrame.opcode).toBe(TerminalStreamOpcode.Snapshot);
expect(reconnectFrame.opcode).toBe(TerminalStreamOpcode.Restore);
const state = decodeTerminalSnapshotPayload(reconnectFrame.payload);
const state = await decodeRestoreState(reconnectFrame, terminalId);
expect(state).not.toBeNull();
expect(extractStateText(state!)).toContain("before-drop");
expect(extractStateText(state!)).toContain("while-dead");
@@ -1550,11 +1544,11 @@ test("snapshot fidelity through websocket decode preserves dimensions and visibl
await subscribeRawTerminal(ws, terminalId, "sub-fidelity");
const snapshotFrame = await waitForRawBinaryFrame(
ws,
(frame) => frame.opcode === TerminalStreamOpcode.Snapshot,
(frame) => frame.opcode === TerminalStreamOpcode.Restore,
10000,
);
const state = decodeTerminalSnapshotPayload(snapshotFrame.payload);
const state = await decodeRestoreState(snapshotFrame, terminalId);
expect(state).not.toBeNull();
expect(state).toMatchObject({
rows: 24,
@@ -1582,7 +1576,7 @@ test("terminal exit prevents resubscribe and sends no frames after exit", async
await subscribeRawTerminal(ws, terminalId, "sub-exit");
await waitForRawBinaryFrame(
ws,
(frame) => frame.opcode === TerminalStreamOpcode.Snapshot,
(frame) => frame.opcode === TerminalStreamOpcode.Restore,
10000,
);
@@ -1598,8 +1592,7 @@ test("terminal exit prevents resubscribe and sends no frames after exit", async
const exitMessage = await exitMessagePromise;
expect(exitMessage.message.type).toBe("terminal_stream_exit");
const subscribeResult = await ctx.client.subscribeTerminal(terminalId);
expect(subscribeResult.error).toBe("Terminal not found");
await expect(ctx.client.subscribeTerminal(terminalId)).rejects.toThrow("Terminal not found");
await waitForNoRawBinaryFrame(ws, 750);
} finally {
await closeWebSocket(ws);
@@ -1617,7 +1610,7 @@ test("empty input frame does not crash the server", async () => {
const slot = await subscribeRawTerminal(ws, terminalId, "sub-empty-input");
await waitForRawBinaryFrame(
ws,
(frame) => frame.opcode === TerminalStreamOpcode.Snapshot,
(frame) => frame.opcode === TerminalStreamOpcode.Restore,
10000,
);
@@ -1656,7 +1649,7 @@ test("1MB output burst keeps frames decodable and terminal usable afterward", as
await subscribeRawTerminal(ws, terminalId, "sub-large-output");
await waitForRawBinaryFrame(
ws,
(frame) => frame.opcode === TerminalStreamOpcode.Snapshot,
(frame) => frame.opcode === TerminalStreamOpcode.Restore,
10000,
);
@@ -1673,7 +1666,7 @@ test("1MB output burst keeps frames decodable and terminal usable afterward", as
.reduce((sum, frame) => sum + frame.payload.byteLength, 0);
return (
outputBytes >= 1024 * 1024 ||
collectedFrames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)
collectedFrames.some((frame) => frame.opcode === TerminalStreamOpcode.Restore)
);
},
20000,
@@ -1683,7 +1676,7 @@ test("1MB output burst keeps frames decodable and terminal usable afterward", as
for (const frame of frames) {
expect(
frame.opcode === TerminalStreamOpcode.Output ||
frame.opcode === TerminalStreamOpcode.Snapshot,
frame.opcode === TerminalStreamOpcode.Restore,
).toBe(true);
}
@@ -822,86 +822,100 @@ test("same-workspace terminal directory observers do not share a teardown slot",
}
});
test("terminal output has separate server IDs and binary slots, including in a shared session", async () => {
const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
const admin = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.8.0" });
const peers: SubscriptionPeer[] = [];
try {
await admin.connect();
const workspace = (
await admin.createWorkspace({ source: { kind: "directory", path: daemon.staticDir } })
).workspace!;
const result = await admin.createTerminal(daemon.staticDir, "Output ownership", undefined, {
workspaceId: workspace.id,
command: process.execPath,
args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"],
test.each([false, true])(
"terminal output has separate server IDs and binary slots, including in a shared session (chunked=%s)",
async (chunked) => {
const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
const admin = new DaemonClient({
url: `ws://127.0.0.1:${daemon.port}/ws`,
appVersion: "0.8.0",
});
expect(result.error).toBe(null);
const idle = await SubscriptionPeer.connect(daemon.port, "terminal-output-session");
peers.push(idle);
const peer = await SubscriptionPeer.connect(daemon.port, "terminal-output-session");
peers.push(peer);
for (const requestId of ["output-a", "output-b"]) {
await peer.request({
type: "subscribe_terminal_request",
requestId,
terminalId: result.terminal!.id,
const peers: SubscriptionPeer[] = [];
try {
await admin.connect();
const workspace = (
await admin.createWorkspace({ source: { kind: "directory", path: daemon.staticDir } })
).workspace!;
const result = await admin.createTerminal(daemon.staticDir, "Output ownership", undefined, {
workspaceId: workspace.id,
command: process.execPath,
args: ["-e", "process.stdin.on('data', data => process.stdout.write(data))"],
});
}
const replies = peer.frames.flatMap((frame) =>
frame.type === "session" &&
frame.message.type === "subscribe_terminal_response" &&
frame.message.payload.error === null
? [frame.message.payload]
: [],
);
expect(replies).toHaveLength(2);
const ids = replies.map(
(reply) => (reply as typeof reply & { subscriptionId: string }).subscriptionId,
);
expect(ids.every((id) => typeof id === "string")).toBe(true);
expect(ids[0]).not.toBe(ids[1]);
expect(replies[0].slot).not.toBe(replies[1].slot);
await expect
.poll(() =>
peer.binaryFrames
.map(decodeTerminalStreamFrame)
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Snapshot)
.map((frame) => frame!.slot),
)
.toEqual(expect.arrayContaining(replies.map((reply) => reply.slot)));
expect(idle.binaryFrames).toEqual([]);
await peer.request({
type: "subscription.release.request",
requestId: "release-output-a",
subscriptionId: ids[0],
});
peer.binaryFrames.length = 0;
admin.sendTerminalInput(result.terminal!.id, { type: "input", data: "remaining-observer\n" });
await expect
.poll(
() =>
expect(result.error).toBe(null);
const idle = await SubscriptionPeer.connect(daemon.port, "terminal-output-session", {
terminal_chunked_replay: chunked,
});
peers.push(idle);
const peer = await SubscriptionPeer.connect(daemon.port, "terminal-output-session", {
terminal_chunked_replay: chunked,
});
peers.push(peer);
for (const requestId of ["output-a", "output-b"]) {
await peer.request({
type: "subscribe_terminal_request",
requestId,
terminalId: result.terminal!.id,
});
}
const replies = peer.frames.flatMap((frame) =>
frame.type === "session" &&
frame.message.type === "subscribe_terminal_response" &&
frame.message.payload.error === null
? [frame.message.payload]
: [],
);
expect(replies).toHaveLength(2);
const ids = replies.map(
(reply) => (reply as typeof reply & { subscriptionId: string }).subscriptionId,
);
expect(ids.every((id) => typeof id === "string")).toBe(true);
expect(ids[0]).not.toBe(ids[1]);
expect(replies[0].slot).not.toBe(replies[1].slot);
await expect
.poll(() =>
peer.binaryFrames
.map(decodeTerminalStreamFrame)
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
.map((frame) => Buffer.from(frame!.payload).toString())
.join(""),
{ timeout: 5000 },
)
.toContain("remaining-observer");
expect(
peer.binaryFrames
.map(decodeTerminalStreamFrame)
.every((frame) => frame?.slot === replies[1].slot),
).toBe(true);
expect(idle.binaryFrames).toEqual([]);
await admin.killTerminal(result.terminal!.id);
} finally {
for (const peer of peers) peer.close();
await admin.close();
await daemon.close();
}
});
.filter(
(frame) =>
frame?.opcode ===
(chunked ? TerminalStreamOpcode.ReplayEnd : TerminalStreamOpcode.Restore),
)
.map((frame) => frame!.slot),
)
.toEqual(expect.arrayContaining(replies.map((reply) => reply.slot)));
expect(idle.binaryFrames).toEqual([]);
await peer.request({
type: "subscription.release.request",
requestId: "release-output-a",
subscriptionId: ids[0],
});
peer.binaryFrames.length = 0;
admin.sendTerminalInput(result.terminal!.id, { type: "input", data: "remaining-observer\n" });
await expect
.poll(
() =>
peer.binaryFrames
.map(decodeTerminalStreamFrame)
.filter((frame) => frame?.opcode === TerminalStreamOpcode.Output)
.map((frame) => Buffer.from(frame!.payload).toString())
.join(""),
{ timeout: 5000 },
)
.toContain("remaining-observer");
expect(
peer.binaryFrames
.map(decodeTerminalStreamFrame)
.every((frame) => frame?.slot === replies[1].slot),
).toBe(true);
expect(idle.binaryFrames).toEqual([]);
await admin.killTerminal(result.terminal!.id);
} finally {
for (const peer of peers) peer.close();
await admin.close();
await daemon.close();
}
},
);
test("a modern connection and pure reads retain no client observation producers", async () => {
const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
@@ -2016,7 +2016,6 @@ export class VoiceAssistantWebSocketServer {
resumeOnStartupSetting: true,
// COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23.
"terminal-restore-modes": true,
terminalChunkedReplay: true,
// COMPAT(terminalInputModeReplay): added in v0.2.6, remove gate after 2027-02-02.
"terminal-input-mode-replay": true,
// COMPAT(terminalSizeOwnership): added in v0.2.6, remove gate after 2027-02-02.
@@ -43,7 +43,9 @@ function replayFixture(
}
function createController(
options: TerminalSessionControllerOptions & { emitBinary(frame: Uint8Array): void },
options: TerminalSessionControllerOptions & {
emitBinary(frame: Uint8Array): void | Promise<void>;
},
) {
const source = {};
const ownership = new SessionDelivery(
@@ -676,7 +678,12 @@ describe("terminal-session-controller workspace-scoped subscriptions", () => {
});
describe("terminal-session-controller backpressure snapshot fallback", () => {
async function setup(getClientBufferedAmount: () => number | null): Promise<{
async function setup(
getClientBufferedAmount: () => number | null,
chunked = false,
pace?: () => Promise<void>,
): Promise<{
messages: SessionOutboundMessage[];
pushOutput: (data: string) => void;
frames: TerminalStreamFrame[];
}> {
@@ -726,13 +733,18 @@ describe("terminal-session-controller backpressure snapshot fallback", () => {
};
const frames: TerminalStreamFrame[] = [];
const messages: SessionOutboundMessage[] = [];
const controller = createController({
terminalManager,
emit: vi.fn(),
emit: (message) => {
messages.push(message);
},
supportsChunkedReplay: () => chunked,
emitBinary: (bytes) => {
const frame = decodeTerminalStreamFrame(bytes);
if (frame) {
frames.push(frame);
if (frame.opcode === TerminalStreamOpcode.ReplayBegin) return pace?.();
}
},
hasBinaryChannel: () => true,
@@ -754,6 +766,7 @@ describe("terminal-session-controller backpressure snapshot fallback", () => {
return {
pushOutput: (data) => terminalListener?.({ type: "output", data, revision: 2 }),
frames,
messages,
};
}
@@ -818,6 +831,32 @@ describe("terminal-session-controller backpressure snapshot fallback", () => {
expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Restore)).toBe(true);
});
test("paces capable catch-up while the direct socket drains instead of ending the stream", async () => {
let buffered = 0;
let release!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
const { pushOutput, frames, messages } = await setup(
() => buffered,
true,
() => (buffered ? gate : Promise.resolve()),
);
buffered = 8 * 1024 * 1024;
pushOutput("y".repeat(300 * 1024));
await waitForCoalescerFlush();
expect(messages.filter((message) => message.type === "terminal_stream_exit")).toEqual([]);
expect(frames.map((frame) => frame.opcode)).toEqual([TerminalStreamOpcode.ReplayBegin]);
buffered = 0;
release();
await waitForCoalescerFlush();
expect(frames.map((frame) => frame.opcode)).toContain(TerminalStreamOpcode.ReplayEnd);
frames.length = 0;
buffered = 8 * 1024 * 1024; // Small output must stay live after the replay resets its byte counter.
pushOutput("live-after-recovery");
await waitForCoalescerFlush();
expect(frames.map((frame) => frame.opcode)).toEqual([TerminalStreamOpcode.Output]);
});
});
function exitFixture(readReplay: () => Promise<ReplayFixture | null>, backedUp = false) {
@@ -1102,25 +1102,6 @@ export class TerminalSessionController {
snapshotOptions,
async (event) => {
if (this.activeStreams.get(activeStream.slot) !== activeStream) return;
const buffered = this.options.getClientBufferedAmount?.(activeStream.owner.source);
if (
chunked &&
buffered !== null &&
buffered !== undefined &&
buffered > MAX_CLIENT_BUFFERED_BYTES
) {
activeStream.owner.emit({
type: "terminal_stream_exit",
payload: {
terminalId: activeStream.terminalId,
error: "Terminal replay transport overflow; attach again",
},
});
this.detachRegistration(activeStream);
void activeStream.owner.release();
return;
}
await deliver(event);
},
);