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

Align terminal stream with snapshot-only attach

Browse Source
This commit is contained in:
Mohamed Boudra
2026-03-22 22:49:05 +07:00
parent 138c774297
commit babc724220
14 changed files with 262 additions and 550 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/client/daemon-client.test.ts
-4
View File
@@ -1257,7 +1257,6 @@ describe("DaemonClient", () => {
type: "subscribe_terminal_response",
payload: {
terminalId: "term-1",
state: null,
error: null,
requestId: "sub-1",
},
@@ -1308,7 +1307,6 @@ describe("DaemonClient", () => {
type: "subscribe_terminal_response",
payload: {
terminalId: "term-1",
state: null,
error: null,
requestId: "sub-2",
},
@@ -1356,7 +1354,6 @@ describe("DaemonClient", () => {
type: "subscribe_terminal_response",
payload: {
terminalId: "term-1",
state: null,
error: null,
requestId: "sub-3",
},
@@ -1418,7 +1415,6 @@ describe("DaemonClient", () => {
type: "subscribe_terminal_response",
payload: {
terminalId: "term-1",
state: null,
error: null,
requestId: "sub-4",
},
packages/server/src/server/daemon-e2e/terminal.e2e.test.ts
+25 -37
View File
@@ -4,6 +4,11 @@ import { tmpdir } from "os";
import path from "path";
import WebSocket from "ws";
import { DaemonClient } from "../../client/daemon-client.js";
import {
WSOutboundMessageSchema,
type TerminalState,
type WSOutboundMessage,
} from "../../shared/messages.js";
import {
decodeTerminalStreamFrame,
TerminalStreamOpcode,
@@ -11,6 +16,8 @@ import {
} from "../../shared/terminal-stream-protocol.js";
import { createDaemonTestContext, type DaemonTestContext } from "../test-utils/index.js";
type RawSessionEnvelope = Extract<WSOutboundMessage, { type: "session" }>;
function tmpCwd(): string {
return mkdtempSync(path.join(tmpdir(), "daemon-terminal-e2e-"));
}
@@ -24,10 +31,7 @@ function createLogger() {
};
}
function extractStateText(state: {
grid: Array<Array<{ char: string }>>;
scrollback: Array<Array<{ char: string }>>;
}): string {
function extractStateText(state: Pick<TerminalState, "grid" | "scrollback">): string {
return [...state.scrollback, ...state.grid]
.map((row) =>
row
@@ -57,21 +61,9 @@ async function waitForCondition(
async function waitForTerminalSnapshot(
client: DaemonClient,
terminalId: string,
predicate: (state: {
rows: number;
cols: number;
grid: Array<Array<{ char: string }>>;
scrollback: Array<Array<{ char: string }>>;
cursor: { row: number; col: number };
}) => boolean,
predicate: (state: TerminalState) => boolean,
timeout = 10000,
): Promise<{
rows: number;
cols: number;
grid: Array<Array<{ char: string }>>;
scrollback: Array<Array<{ char: string }>>;
cursor: { row: number; col: number };
}> {
): Promise<TerminalState> {
return new Promise((resolve, reject) => {
const timeoutHandle = setTimeout(() => {
unsubscribe();
@@ -156,9 +148,7 @@ async function connectRawWebSocket(port: number): Promise<WebSocket> {
const helloReady = waitForRawSessionMessage(
ws,
(message) =>
message.type === "session" &&
message.message?.type === "status" &&
message.message.payload?.status === "server_info",
message.message.type === "status" && message.message.payload.status === "server_info",
10000,
);
@@ -198,15 +188,9 @@ async function closeWebSocket(ws: WebSocket, timeout = 5000): Promise<void> {
async function waitForRawSessionMessage(
ws: WebSocket,
predicate: (message: {
type?: string;
message?: { type?: string; payload?: Record<string, any> };
}) => boolean,
predicate: (message: RawSessionEnvelope) => boolean,
timeout = 10000,
): Promise<{
type?: string;
message?: { type?: string; payload?: Record<string, any> };
}> {
): Promise<RawSessionEnvelope> {
return new Promise((resolve, reject) => {
const timeoutHandle = setTimeout(() => {
cleanup();
@@ -220,10 +204,11 @@ async function waitForRawSessionMessage(
return;
}
try {
const parsed = JSON.parse(text) as {
type?: string;
message?: { type?: string; payload?: Record<string, any> };
};
const parsedResult = WSOutboundMessageSchema.safeParse(JSON.parse(text));
if (!parsedResult.success || parsedResult.data.type !== "session") {
return;
}
const parsed = parsedResult.data;
if (!predicate(parsed)) {
return;
}
@@ -294,9 +279,8 @@ async function subscribeRawTerminal(ws: WebSocket, terminalId: string, requestId
const ready = waitForRawSessionMessage(
ws,
(message) =>
message.type === "session" &&
message.message?.type === "subscribe_terminal_response" &&
message.message.payload?.requestId === requestId,
message.message.type === "subscribe_terminal_response" &&
message.message.payload.requestId === requestId,
10000,
);
@@ -311,7 +295,11 @@ async function subscribeRawTerminal(ws: WebSocket, terminalId: string, requestId
}),
);
await ready;
const message = await ready;
if (message.message.type !== "subscribe_terminal_response") {
throw new Error("Expected subscribe_terminal_response");
}
expect(message.message.payload).not.toHaveProperty("state");
}
describe("daemon E2E terminal", () => {
packages/server/src/server/session.ts
+55 -114
View File
@@ -279,8 +279,7 @@ type ActiveTerminalStream = {
terminalId: string;
unsubscribe: () => void;
needsSnapshot: boolean;
primed: boolean;
snapshotTimer: ReturnType<typeof setTimeout> | null;
snapshotRetryTimer: ReturnType<typeof setTimeout> | null;
};
export type SessionRuntimeMetrics = {
@@ -1817,8 +1816,8 @@ export class Session {
if (!resize) {
return;
}
activeStream.needsSnapshot = true;
terminal.send({ type: "resize", rows: resize.rows, cols: resize.cols });
this.queueTerminalSnapshot(activeStream, terminal);
return;
}
@@ -7666,7 +7665,6 @@ export class Session {
type: "subscribe_terminal_response",
payload: {
terminalId: msg.terminalId,
state: null,
error: "Terminal manager not available",
requestId: msg.requestId,
},
@@ -7680,7 +7678,6 @@ export class Session {
type: "subscribe_terminal_response",
payload: {
terminalId: msg.terminalId,
state: null,
error: "Terminal not found",
requestId: msg.requestId,
},
@@ -7689,23 +7686,16 @@ export class Session {
}
this.ensureTerminalExitSubscription(session);
const activeStream = this.bindActiveTerminalStream(session);
// Send initial state
this.emit({
type: "subscribe_terminal_response",
payload: {
terminalId: msg.terminalId,
state: session.getState(),
error: null,
requestId: msg.requestId,
},
});
if (activeStream) {
this.sendTerminalSnapshot(activeStream, session);
activeStream.primed = true;
}
this.bindActiveTerminalStream(session);
}
private handleUnsubscribeTerminalRequest(msg: UnsubscribeTerminalRequest): void {
@@ -7803,126 +7793,77 @@ export class Session {
const activeStream: ActiveTerminalStream = {
terminalId: terminal.id,
unsubscribe: () => {},
needsSnapshot: false,
primed: false,
snapshotTimer: null,
needsSnapshot: true,
snapshotRetryTimer: null,
};
const trySendSnapshot = () => {
if (this.activeTerminalStream !== activeStream || !activeStream.needsSnapshot) {
return;
}
if (this.getCurrentBinaryBufferedAmount() > TERMINAL_STREAM_LOW_WATER_BYTES) {
if (!activeStream.snapshotRetryTimer) {
activeStream.snapshotRetryTimer = setTimeout(() => {
activeStream.snapshotRetryTimer = null;
trySendSnapshot();
}, 33);
}
return;
}
if (activeStream.snapshotRetryTimer) {
clearTimeout(activeStream.snapshotRetryTimer);
activeStream.snapshotRetryTimer = null;
}
activeStream.needsSnapshot = false;
this.emitBinary(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Snapshot,
payload: encodeTerminalSnapshotPayload(terminal.getState()),
}),
);
};
activeStream.unsubscribe = terminal.subscribe((message) => {
if (this.activeTerminalStream !== activeStream) {
return;
}
if (message.type === "output") {
this.handleActiveTerminalOutput(activeStream, terminal, message.data);
if (message.type === "snapshot") {
trySendSnapshot();
return;
}
if (message.type === "snapshot") {
this.maybeSendQueuedTerminalSnapshot(activeStream, terminal);
if (activeStream.needsSnapshot || message.data.length === 0) {
return;
}
if (this.getCurrentBinaryBufferedAmount() >= TERMINAL_STREAM_HIGH_WATER_BYTES) {
activeStream.needsSnapshot = true;
trySendSnapshot();
return;
}
this.emitBinary(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Output,
payload: new Uint8Array(Buffer.from(message.data, "utf8")),
}),
);
if (this.getCurrentBinaryBufferedAmount() >= TERMINAL_STREAM_HIGH_WATER_BYTES) {
activeStream.needsSnapshot = true;
trySendSnapshot();
}
});
this.activeTerminalStream = activeStream;
return activeStream;
}
private handleActiveTerminalOutput(
activeStream: ActiveTerminalStream,
terminal: TerminalSession,
data: string,
): void {
if (!activeStream.primed || data.length === 0) {
return;
}
if (activeStream.needsSnapshot) {
this.maybeSendQueuedTerminalSnapshot(activeStream, terminal);
return;
}
if (this.getCurrentBinaryBufferedAmount() > TERMINAL_STREAM_HIGH_WATER_BYTES) {
activeStream.needsSnapshot = true;
this.scheduleTerminalSnapshotRetry(activeStream, terminal);
return;
}
this.emitBinary(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Output,
payload: new Uint8Array(Buffer.from(data, "utf8")),
}),
);
if (this.getCurrentBinaryBufferedAmount() > TERMINAL_STREAM_HIGH_WATER_BYTES) {
activeStream.needsSnapshot = true;
this.scheduleTerminalSnapshotRetry(activeStream, terminal);
}
}
private queueTerminalSnapshot(
activeStream: ActiveTerminalStream,
terminal: TerminalSession,
): void {
if (this.getCurrentBinaryBufferedAmount() > TERMINAL_STREAM_HIGH_WATER_BYTES) {
activeStream.needsSnapshot = true;
this.scheduleTerminalSnapshotRetry(activeStream, terminal);
return;
}
this.sendTerminalSnapshot(activeStream, terminal);
}
private maybeSendQueuedTerminalSnapshot(
activeStream: ActiveTerminalStream,
terminal: TerminalSession,
): void {
if (!activeStream.needsSnapshot) {
return;
}
if (this.getCurrentBinaryBufferedAmount() >= TERMINAL_STREAM_LOW_WATER_BYTES) {
this.scheduleTerminalSnapshotRetry(activeStream, terminal);
return;
}
this.sendTerminalSnapshot(activeStream, terminal);
}
private sendTerminalSnapshot(
activeStream: ActiveTerminalStream,
terminal: TerminalSession,
): void {
this.clearTerminalSnapshotRetry(activeStream);
activeStream.needsSnapshot = false;
this.emitBinary(
encodeTerminalStreamFrame({
opcode: TerminalStreamOpcode.Snapshot,
payload: encodeTerminalSnapshotPayload(terminal.getState()),
}),
);
}
private scheduleTerminalSnapshotRetry(
activeStream: ActiveTerminalStream,
terminal: TerminalSession,
): void {
if (activeStream.snapshotTimer) {
return;
}
activeStream.snapshotTimer = setTimeout(() => {
activeStream.snapshotTimer = null;
if (this.activeTerminalStream !== activeStream || !activeStream.needsSnapshot) {
return;
}
this.maybeSendQueuedTerminalSnapshot(activeStream, terminal);
}, 33);
}
private clearTerminalSnapshotRetry(activeStream: ActiveTerminalStream): void {
if (!activeStream.snapshotTimer) {
return;
}
clearTimeout(activeStream.snapshotTimer);
activeStream.snapshotTimer = null;
}
private detachActiveTerminalStream(options?: { emitExit: boolean }): boolean {
const activeStream = this.activeTerminalStream;
if (!activeStream) {
return false;
}
this.activeTerminalStream = null;
this.clearTerminalSnapshotRetry(activeStream);
if (activeStream.snapshotRetryTimer) {
clearTimeout(activeStream.snapshotRetryTimer);
activeStream.snapshotRetryTimer = null;
}
try {
activeStream.unsubscribe();
} catch (error) {
packages/server/src/shared/messages.ts
+3 -4
View File
@@ -2104,15 +2104,15 @@ export const TerminalCellSchema = z.object({
bold: z.boolean().optional(),
italic: z.boolean().optional(),
underline: z.boolean().optional(),
});
}).strict();
export const TerminalStateSchema = z.object({
rows: z.number(),
cols: z.number(),
grid: z.array(z.array(TerminalCellSchema)),
scrollback: z.array(z.array(TerminalCellSchema)),
cursor: z.object({ row: z.number(), col: z.number() }),
});
cursor: z.object({ row: z.number(), col: z.number() }).strict(),
}).strict();
export const ListTerminalsResponseSchema = z.object({
type: z.literal("list_terminals_response"),
@@ -2144,7 +2144,6 @@ export const SubscribeTerminalResponseSchema = z.object({
type: z.literal("subscribe_terminal_response"),
payload: z.object({
terminalId: z.string(),
state: TerminalStateSchema.nullable(),
error: z.string().nullable(),
requestId: z.string(),
}),
packages/server/src/shared/terminal-stream-protocol.test.ts
+46
View File
@@ -56,4 +56,50 @@ describe("terminal stream protocol", () => {
it("rejects unknown opcodes", () => {
expect(decodeTerminalStreamFrame(new Uint8Array([0xff, 0x01]))).toBeNull();
});
it("rejects malformed JSON payloads", () => {
const malformed = new TextEncoder().encode("{");
expect(decodeTerminalResizePayload(malformed)).toBeNull();
expect(decodeTerminalSnapshotPayload(malformed)).toBeNull();
});
it("rejects invalid resize and snapshot shapes", () => {
expect(
decodeTerminalResizePayload(new TextEncoder().encode(JSON.stringify({ rows: "24", cols: 80 }))),
).toBeNull();
expect(
decodeTerminalSnapshotPayload(
new TextEncoder().encode(
JSON.stringify({
rows: 1,
cols: 1,
grid: [[{ char: "A" }]],
scrollback: [],
}),
),
),
).toBeNull();
});
it("rejects unknown fields in resize and snapshot payloads", () => {
expect(
decodeTerminalResizePayload(
new TextEncoder().encode(JSON.stringify({ rows: 24, cols: 80, extra: true })),
),
).toBeNull();
expect(
decodeTerminalSnapshotPayload(
new TextEncoder().encode(
JSON.stringify({
rows: 1,
cols: 1,
grid: [[{ char: "A", extra: true }]],
scrollback: [],
cursor: { row: 0, col: 1 },
}),
),
),
).toBeNull();
});
});
packages/server/src/shared/terminal-stream-protocol.ts
+7 -3
View File
@@ -4,7 +4,7 @@ import { TerminalStateSchema } from "./messages.js";
export const TerminalStreamResizeSchema = z.object({
rows: z.number().int().positive(),
cols: z.number().int().positive(),
});
}).strict();
export const TerminalStreamOpcode = {
Output: 0x01,
@@ -115,6 +115,10 @@ function encodeJsonPayload(value: unknown): Uint8Array {
}
function decodeJsonPayload(bytes: Uint8Array): unknown {
const text = new TextDecoder().decode(bytes);
return JSON.parse(text);
try {
const text = new TextDecoder().decode(bytes);
return JSON.parse(text);
} catch {
return null;
}
}
packages/server/src/terminal/terminal.test.ts
+25
View File
@@ -438,6 +438,31 @@ describe("Terminal", () => {
unsubscribe();
});
it("emits output only after getState reflects the new data", async () => {
const session = trackSession(
await createTerminal({
cwd: "/tmp",
shell: "/bin/sh",
env: { PS1: "$ " },
}),
);
await waitForLines(session, ["$"]);
const outputSeenInState = new Promise<boolean>((resolve) => {
const unsubscribe = session.subscribe((message) => {
if (message.type !== "output" || !message.data.includes("state-after-output")) {
return;
}
unsubscribe();
const stateText = getLines(session.getState()).join("\n");
resolve(stateText.includes("state-after-output"));
});
});
session.send({ type: "input", data: "echo state-after-output\r" });
expect(await outputSeenInState).toBe(true);
});
it("unsubscribe stops receiving messages", async () => {
const session = trackSession(
await createTerminal({
packages/server/src/terminal/terminal.ts
+9 -5
View File
@@ -322,10 +322,13 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
// Pipe PTY output to terminal emulator
ptyProcess.onData((data) => {
if (killed) return;
for (const listener of listeners) {
listener({ type: "output", data });
}
terminal.write(data, () => {
if (disposed || killed) {
return;
}
for (const listener of listeners) {
listener({ type: "output", data });
}
scheduleStateBroadcast();
});
});
@@ -359,6 +362,7 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
case "resize":
terminal.resize(msg.cols, msg.rows);
ptyProcess.resize(msg.cols, msg.rows);
scheduleStateBroadcast();
break;
case "mouse":
// Mouse events can be sent as escape sequences if terminal supports it
@@ -370,8 +374,8 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
function subscribe(listener: (msg: ServerMessage) => void): () => void {
listeners.add(listener);
queueMicrotask(() => {
if (listeners.has(listener)) {
terminal.write("", () => {
if (!disposed && listeners.has(listener)) {
listener({ type: "snapshot", state: getState() });
}
});