diff --git a/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts b/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts index 6fed9e845..0018e68e8 100644 --- a/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/terminal.e2e.test.ts @@ -380,6 +380,271 @@ function getFrameText(frame: TerminalStreamFrame): string { return ""; } +interface TimedTerminalOutputFrame { + receivedAtMs: number; + payloadBytes: number; + text: string; +} + +interface TerminalOutputCapture { + frames: TimedTerminalOutputFrame[]; + snapshotCount: number; + stop: () => void; +} + +interface TerminalOutputCadenceSummary { + frameCount: number; + maxGapMs: number; + p95GapMs: number; + lastOutputAfterStopMs: number; + payloadBytes: number[]; +} + +interface TerminalRepeatCadenceMeasurement { + enterSummary: TerminalOutputCadenceSummary; + keySummary: TerminalOutputCadenceSummary; + snapshotCount: number; +} + +function startTerminalOutputCapture(ws: WebSocket, slot: number): TerminalOutputCapture { + const decoder = new TextDecoder(); + const frames: TimedTerminalOutputFrame[] = []; + let snapshotCount = 0; + + const onMessage = (raw: WebSocket.RawData, isBinary: boolean) => { + if (!isBinary) { + return; + } + const buffer = toWsBuffer(raw); + if (!buffer) { + return; + } + const frame = decodeTerminalStreamFrame(new Uint8Array(buffer)); + if (!frame || frame.slot !== slot) { + return; + } + if (frame.opcode === TerminalStreamOpcode.Snapshot) { + snapshotCount += 1; + return; + } + if (frame.opcode !== TerminalStreamOpcode.Output || frame.payload.byteLength === 0) return; + frames.push({ + receivedAtMs: performance.now(), + payloadBytes: frame.payload.byteLength, + text: decoder.decode(frame.payload), + }); + }; + + ws.on("message", onMessage); + return { + frames, + get snapshotCount() { + return snapshotCount; + }, + stop: () => { + ws.off("message", onMessage); + }, + }; +} + +function sendRawTerminalInput(ws: WebSocket, slot: number, data: string): void { + ws.send( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Input, + slot, + payload: data, + }), + ); +} + +async function repeatRawTerminalInput(input: { + ws: WebSocket; + slot: number; + data: string; + count: number; + intervalMs: number; +}): Promise<{ + stoppedAtMs: number; +}> { + for (let index = 0; index < input.count; index += 1) { + sendRawTerminalInput(input.ws, input.slot, input.data); + await new Promise((resolve) => setTimeout(resolve, input.intervalMs)); + } + return { + stoppedAtMs: performance.now(), + }; +} + +async function waitForTerminalOutputQuiet( + capture: TerminalOutputCapture, + quietMs: number, + timeoutMs: number, +): Promise { + let lastCount = capture.frames.length; + let quietStartedAtMs = performance.now(); + const startedAtMs = quietStartedAtMs; + + while (performance.now() - startedAtMs < timeoutMs) { + await new Promise((resolve) => setTimeout(resolve, 25)); + if (capture.frames.length !== lastCount) { + lastCount = capture.frames.length; + quietStartedAtMs = performance.now(); + continue; + } + if (performance.now() - quietStartedAtMs >= quietMs) { + return; + } + } +} + +function summarizeTerminalOutputCadence(input: { + frames: TimedTerminalOutputFrame[]; + stoppedAtMs: number; +}): TerminalOutputCadenceSummary { + const gaps = input.frames + .slice(1) + .map((frame, index) => frame.receivedAtMs - input.frames[index]!.receivedAtMs) + .sort((a, b) => a - b); + const p95Index = + gaps.length === 0 ? -1 : Math.min(gaps.length - 1, Math.ceil(gaps.length * 0.95) - 1); + const lastFrame = input.frames.at(-1); + + return { + frameCount: input.frames.length, + maxGapMs: gaps.at(-1) ?? 0, + p95GapMs: p95Index === -1 ? 0 : gaps[p95Index]!, + lastOutputAfterStopMs: lastFrame ? lastFrame.receivedAtMs - input.stoppedAtMs : 0, + payloadBytes: input.frames.map((frame) => frame.payloadBytes), + }; +} + +function getTerminalOutputCadenceFailures( + label: string, + summary: TerminalOutputCadenceSummary, + limits: { + minFrames: number; + maxP95GapMs: number; + maxLastOutputAfterStopMs: number; + }, +): string[] { + const failures: string[] = []; + if (summary.frameCount < limits.minFrames) { + failures.push(`${label}: frameCount ${summary.frameCount} < ${limits.minFrames}`); + } + if (summary.p95GapMs >= limits.maxP95GapMs) { + failures.push(`${label}: p95GapMs ${summary.p95GapMs.toFixed(1)} >= ${limits.maxP95GapMs}`); + } + if (summary.lastOutputAfterStopMs >= limits.maxLastOutputAfterStopMs) { + failures.push( + `${label}: lastOutputAfterStopMs ${summary.lastOutputAfterStopMs.toFixed(1)} >= ${limits.maxLastOutputAfterStopMs}`, + ); + } + return failures; +} + +function expectTerminalOutputCadences(input: { + enterSummary: TerminalOutputCadenceSummary; + keySummary: TerminalOutputCadenceSummary; + snapshotCount: number; +}): void { + const failures = [ + ...getTerminalOutputCadenceFailures("enter-repeat", input.enterSummary, { + minFrames: 20, + maxP95GapMs: 120, + maxLastOutputAfterStopMs: Number.POSITIVE_INFINITY, + }), + ...getTerminalOutputCadenceFailures("key-repeat-after-enter-repeat", input.keySummary, { + minFrames: 40, + maxP95GapMs: 60, + maxLastOutputAfterStopMs: 150, + }), + ]; + if (input.snapshotCount > 0) { + failures.push(`unexpected live snapshot count ${input.snapshotCount} > 0`); + } + if (failures.length === 0) { + return; + } + throw new Error( + `terminal output cadence failed: ${failures.join("; ")}\n${JSON.stringify(input)}`, + ); +} + +async function measureRepeatCadenceForTerminal(input: { + cwd: string; + name: string; + options?: { command?: string; args?: string[] }; + setupInput?: string; +}): Promise { + const created = await ctx.client.createTerminal(input.cwd, input.name, undefined, input.options); + const terminalId = created.terminal!.id; + const ws = await connectRawWebSocket(ctx.daemon.port); + + try { + const slot = await subscribeRawTerminal(ws, terminalId, `sub-repeat-cadence-${input.name}`); + await waitForRawBinaryFrame( + ws, + (frame) => frame.slot === slot && frame.opcode === TerminalStreamOpcode.Snapshot, + 10000, + ); + await new Promise((resolve) => setTimeout(resolve, 1000)); + if (input.setupInput) { + sendRawTerminalInput(ws, slot, input.setupInput); + await new Promise((resolve) => setTimeout(resolve, 1000)); + } + + const enterCapture = startTerminalOutputCapture(ws, slot); + const enterRepeat = await repeatRawTerminalInput({ + ws, + slot, + data: "\r", + count: 120, + intervalMs: 15, + }); + await waitForTerminalOutputQuiet(enterCapture, 700, 10000); + enterCapture.stop(); + const enterSummary = summarizeTerminalOutputCadence({ + frames: enterCapture.frames, + stoppedAtMs: enterRepeat.stoppedAtMs, + }); + + await new Promise((resolve) => setTimeout(resolve, 500)); + + const keyCapture = startTerminalOutputCapture(ws, slot); + const keyRepeat = await repeatRawTerminalInput({ + ws, + slot, + data: "k", + count: 80, + intervalMs: 15, + }); + await waitForTerminalOutputQuiet(keyCapture, 250, 3000); + keyCapture.stop(); + const keySummary = summarizeTerminalOutputCadence({ + frames: keyCapture.frames, + stoppedAtMs: keyRepeat.stoppedAtMs, + }); + + return { + enterSummary, + keySummary, + snapshotCount: enterCapture.snapshotCount + keyCapture.snapshotCount, + }; + } finally { + await closeWebSocket(ws); + } +} + +async function expectRepeatCadenceForTerminal(input: { + cwd: string; + name: string; + options?: { command?: string; args?: string[] }; + setupInput?: string; +}): Promise { + const measurement = await measureRepeatCadenceForTerminal(input); + expectTerminalOutputCadences(measurement); +} + async function subscribeRawTerminal( ws: WebSocket, terminalId: string, @@ -613,6 +878,13 @@ test("client sends input and receives output as raw bytes", async () => { rmSync(cwd, { recursive: true, force: true }); }, 30000); +test("default zsh terminal does not eagerly flush full state during repeat bursts", async () => { + await expectRepeatCadenceForTerminal({ + cwd: process.cwd(), + name: "default-zsh-repeat-cadence", + }); +}, 30000); + test("one client can stream two terminals concurrently", async () => { const cwd = tmpCwd(); const firstCreated = await ctx.client.createTerminal(cwd, "first"); diff --git a/packages/server/src/terminal/terminal-manager.ts b/packages/server/src/terminal/terminal-manager.ts index ec2eb5a91..9741bb139 100644 --- a/packages/server/src/terminal/terminal-manager.ts +++ b/packages/server/src/terminal/terminal-manager.ts @@ -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; registerCwdEnv(options: { cwd: string; env: Record }): void; getTerminal(id: string): TerminalSession | undefined; + getTerminalState(id: string): Promise; killTerminal(id: string): void; killTerminalAndWait( id: string, @@ -210,6 +212,10 @@ export function createTerminalManager(): TerminalManager { return terminalsById.get(id); }, + async getTerminalState(id: string): Promise { + return terminalsById.get(id)?.getStateSnapshot() ?? null; + }, + killTerminal(id: string): void { removeSessionById(id, { kill: true }); }, diff --git a/packages/server/src/terminal/terminal-session-controller.ts b/packages/server/src/terminal/terminal-session-controller.ts index eff63973d..269b41724 100644 --- a/packages/server/src/terminal/terminal-session-controller.ts +++ b/packages/server/src/terminal/terminal-session-controller.ts @@ -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 { + 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 { diff --git a/packages/server/src/terminal/terminal-worker-process.ts b/packages/server/src/terminal/terminal-worker-process.ts index 58d64b11d..42df3ff89 100644 --- a/packages/server/src/terminal/terminal-worker-process.ts +++ b/packages/server/src/terminal/terminal-worker-process.ts @@ -9,7 +9,6 @@ import type { const manager = createTerminalManager(); const unsubscribeByTerminalId = new Map void>>(); -const stateFlushTimeoutByTerminalId = new Map>(); 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 { 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 { ok: true, result: { terminal: toTerminalInfo(session), - state: session.getState(), + state: initialSnapshot.state, }, }); return; @@ -205,6 +170,17 @@ async function handleRequest(message: TerminalWorkerRequest): Promise { 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 { case "send": { const session = manager.getTerminal(message.terminalId); session?.send(message.message); - if (session && message.message.type === "resize") { - flushTerminalState(session); - } sendToParent({ type: "response", requestId: message.requestId, ok: true }); return; } diff --git a/packages/server/src/terminal/terminal-worker-protocol.ts b/packages/server/src/terminal/terminal-worker-protocol.ts index 1af598f1f..7a8e8da71 100644 --- a/packages/server/src/terminal/terminal-worker-protocol.ts +++ b/packages/server/src/terminal/terminal-worker-protocol.ts @@ -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; diff --git a/packages/server/src/terminal/terminal.ts b/packages/server/src/terminal/terminal.ts index 0276ff37b..99dd4cf74 100644 --- a/packages/server/src/terminal/terminal.ts +++ b/packages/server/src/terminal/terminal.ts @@ -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 | null = null; let pendingInput = ""; let inputFlushImmediate: ReturnType | 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 { + 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 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 { 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 boolean, + predicate: () => boolean | Promise, timeoutMs: number, intervalMs = 25, ): Promise { 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); }); diff --git a/packages/server/src/terminal/worker-terminal-manager.ts b/packages/server/src/terminal/worker-terminal-manager.ts index 83860b7cd..3e99371fb 100644 --- a/packages/server/src/terminal/worker-terminal-manager.ts +++ b/packages/server/src/terminal/worker-terminal-manager.ts @@ -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, - ): void { - const record = recordsById.get(message.terminalId); - if (!record) { - return; - } - record.state = message.state; - } - function handleTerminalExitEvent( message: Extract, ): 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 { + 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.