fix: smooth terminal streaming

packages/server/src/server/session.ts

@@ -40,6 +40,7 @@ import {
 } from "./messages.js";
 import type { TerminalManager, TerminalsChangedEvent } from "../terminal/terminal-manager.js";
 import { captureTerminalLines, type TerminalSession } from "../terminal/terminal.js";
+import { TerminalOutputCoalescer } from "../terminal/terminal-output-coalescer.js";
 import {
   TerminalStreamOpcode,
   encodeTerminalSnapshotPayload,
@@ -329,6 +330,7 @@ type ActiveTerminalStream = {
   slot: number;
   unsubscribe: () => void;
   needsSnapshot: boolean;
+  outputCoalescer: TerminalOutputCoalescer;
 };
 
 export type SessionRuntimeMetrics = {
@@ -8876,6 +8878,21 @@ export class Session {
       slot,
       unsubscribe: () => {},
       needsSnapshot: true,
+      outputCoalescer: new TerminalOutputCoalescer({
+        timers: { setTimeout, clearTimeout },
+        onFlush: ({ payload }) => {
+          if (this.activeTerminalStreams.get(slot) !== activeStream) {
+            return;
+          }
+          this.emitBinary(
+            encodeTerminalStreamFrame({
+              opcode: TerminalStreamOpcode.Output,
+              slot,
+              payload,
+            }),
+          );
+        },
+      }),
     };
 
     this.activeTerminalStreams.set(slot, activeStream);
@@ -8885,21 +8902,19 @@ export class Session {
       if (this.activeTerminalStreams.get(slot) !== activeStream) {
         return;
       }
-      if (message.type === "snapshot" || message.type === "titleChange") {
+      if (message.type === "snapshot") {
+        activeStream.outputCoalescer.flush();
         activeStream.needsSnapshot = true;
         this.trySendTerminalSnapshot(activeStream);
         return;
       }
+      if (message.type === "titleChange") {
+        return;
+      }
       if (activeStream.needsSnapshot || message.data.length === 0) {
         return;
       }
-      this.emitBinary(
-        encodeTerminalStreamFrame({
-          opcode: TerminalStreamOpcode.Output,
-          slot,
-          payload: new Uint8Array(Buffer.from(message.data, "utf8")),
-        }),
-      );
+      activeStream.outputCoalescer.handle(message.data);
     });
     return slot;
   }
@@ -8918,6 +8933,7 @@ export class Session {
       return;
     }
 
+    activeStream.outputCoalescer.flush();
     activeStream.needsSnapshot = false;
     this.emitBinary(
       encodeTerminalStreamFrame({
@@ -8950,6 +8966,7 @@ export class Session {
       this.terminalIdToSlot.delete(terminalId);
       return false;
     }
+    activeStream.outputCoalescer.flush();
     this.activeTerminalStreams.delete(slot);
     this.terminalIdToSlot.delete(terminalId);
     try {

packages/server/src/terminal/terminal-output-coalescer.test.ts

@@ -0,0 +1,79 @@
+import { describe, expect, test } from "vitest";
+import { TerminalOutputCoalescer } from "./terminal-output-coalescer.js";
+
+function createHarness() {
+  let nextTimer = 1;
+  const scheduled = new Map<number, { callback: () => void; delayMs: number }>();
+  const flushes: Array<{ payload: string; chars: number; bytes: number }> = [];
+  const coalescer = new TerminalOutputCoalescer({
+    timers: {
+      setTimeout: ((callback: () => void, delayMs?: number) => {
+        const id = nextTimer;
+        nextTimer += 1;
+        scheduled.set(id, { callback, delayMs: delayMs ?? 0 });
+        return id;
+      }) as typeof setTimeout,
+      clearTimeout: ((id: number) => {
+        scheduled.delete(id);
+      }) as typeof clearTimeout,
+    },
+    onFlush: ({ payload, chars, bytes }) => {
+      flushes.push({ payload: payload.toString("utf8"), chars, bytes });
+    },
+  });
+
+  return {
+    coalescer,
+    flushes,
+    scheduled,
+    runScheduled() {
+      const callbacks = Array.from(scheduled.values());
+      scheduled.clear();
+      for (const { callback } of callbacks) {
+        callback();
+      }
+    },
+  };
+}
+
+describe("TerminalOutputCoalescer", () => {
+  test("coalesces output into one 5ms flush window", () => {
+    const { coalescer, flushes, runScheduled, scheduled } = createHarness();
+
+    coalescer.handle("a");
+    coalescer.handle("b");
+    coalescer.handle("é");
+
+    expect(scheduled.size).toBe(1);
+    expect(Array.from(scheduled.values())).toEqual([
+      { callback: expect.any(Function), delayMs: 5 },
+    ]);
+    expect(flushes).toEqual([]);
+
+    runScheduled();
+
+    expect(flushes).toEqual([{ payload: "abé", chars: 3, bytes: 4 }]);
+  });
+
+  test("manual flush drains pending output and cancels the scheduled flush", () => {
+    const { coalescer, flushes, runScheduled, scheduled } = createHarness();
+
+    coalescer.handle("hello");
+    coalescer.flush();
+    runScheduled();
+
+    expect(scheduled.size).toBe(0);
+    expect(flushes).toEqual([{ payload: "hello", chars: 5, bytes: 5 }]);
+  });
+
+  test("dispose drops pending output", () => {
+    const { coalescer, flushes, runScheduled, scheduled } = createHarness();
+
+    coalescer.handle("pending");
+    coalescer.dispose();
+    runScheduled();
+
+    expect(scheduled.size).toBe(0);
+    expect(flushes).toEqual([]);
+  });
+});

packages/server/src/terminal/terminal-output-coalescer.ts

@@ -0,0 +1,87 @@
+export type TerminalOutputCoalescerTimers = {
+  setTimeout: typeof setTimeout;
+  clearTimeout: typeof clearTimeout;
+};
+
+export type TerminalOutputCoalescerFlush = {
+  payload: Buffer;
+  chars: number;
+  bytes: number;
+};
+
+export type TerminalOutputCoalescerOptions = {
+  timers: TerminalOutputCoalescerTimers;
+  flushDelayMs?: number;
+  onFlush: (payload: TerminalOutputCoalescerFlush) => void;
+};
+
+const DEFAULT_FLUSH_DELAY_MS = 5;
+
+export class TerminalOutputCoalescer {
+  private readonly timers: TerminalOutputCoalescerTimers;
+  private readonly flushDelayMs: number;
+  private readonly onFlush: (payload: TerminalOutputCoalescerFlush) => void;
+  private chunks: Buffer[] = [];
+  private bytes = 0;
+  private chars = 0;
+  private flushTimer: ReturnType<typeof setTimeout> | null = null;
+
+  constructor(options: TerminalOutputCoalescerOptions) {
+    this.timers = options.timers;
+    this.flushDelayMs = options.flushDelayMs ?? DEFAULT_FLUSH_DELAY_MS;
+    this.onFlush = options.onFlush;
+  }
+
+  handle(data: string): void {
+    if (data.length === 0) {
+      return;
+    }
+
+    const chunk = Buffer.from(data, "utf8");
+    this.chunks.push(chunk);
+    this.bytes += chunk.byteLength;
+    this.chars += data.length;
+
+    if (!this.flushTimer) {
+      this.flushTimer = this.timers.setTimeout(() => {
+        this.flushTimer = null;
+        this.flush();
+      }, this.flushDelayMs);
+    }
+  }
+
+  flush(): void {
+    this.clearTimer();
+
+    if (this.chunks.length === 0) {
+      return;
+    }
+
+    const payload =
+      this.chunks.length === 1 ? this.chunks[0]! : Buffer.concat(this.chunks, this.bytes);
+    const bytes = this.bytes;
+    const chars = this.chars;
+    this.clearPending();
+
+    this.onFlush({ payload, bytes, chars });
+  }
+
+  dispose(): void {
+    this.clearTimer();
+    this.clearPending();
+  }
+
+  private clearTimer(): void {
+    if (!this.flushTimer) {
+      return;
+    }
+    this.timers.clearTimeout(this.flushTimer);
+    this.flushTimer = null;
+  }
+
+  private clearPending(): void {
+    this.chunks = [];
+    this.bytes = 0;
+    this.chars = 0;
+  }
+}

packages/server/src/terminal/terminal.ts

@@ -560,6 +560,8 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
   let title: string | undefined;
   let pendingTitle: string | undefined;
   let titleDebounceTimer: ReturnType<typeof setTimeout> | null = null;
+  let pendingInput = "";
+  let inputFlushImmediate: ReturnType<typeof setImmediate> | null = null;
 
   // Create xterm.js headless terminal
   const terminal = new Terminal({
@@ -689,6 +691,11 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
       return;
     }
     disposed = true;
+    pendingInput = "";
+    if (inputFlushImmediate) {
+      clearImmediate(inputFlushImmediate);
+      inputFlushImmediate = null;
+    }
     if (titleDebounceTimer) {
       clearTimeout(titleDebounceTimer);
       titleDebounceTimer = null;
@@ -757,14 +764,44 @@ export async function createTerminal(options: CreateTerminalOptions): Promise<Te
     };
   }
 
+  function writeInputToPty(data: string): void {
+    ptyProcess.write(data);
+  }
+
+  function flushPendingInput(): void {
+    if (inputFlushImmediate) {
+      clearImmediate(inputFlushImmediate);
+      inputFlushImmediate = null;
+    }
+    const data = pendingInput;
+    pendingInput = "";
+    if (!data || killed || disposed) {
+      return;
+    }
+    writeInputToPty(data);
+  }
+
+  function scheduleInputFlush(): void {
+    if (inputFlushImmediate) {
+      return;
+    }
+    inputFlushImmediate = setImmediate(() => {
+      inputFlushImmediate = null;
+      flushPendingInput();
+    });
+  }
+
   function send(msg: ClientMessage): void {
     if (killed) return;
 
     switch (msg.type) {
-      case "input":
-        ptyProcess.write(msg.data);
+      case "input": {
+        pendingInput += msg.data;
+        scheduleInputFlush();
         break;
+      }
       case "resize":
+        flushPendingInput();
         terminal.resize(msg.cols, msg.rows);
         ptyProcess.resize(msg.cols, msg.rows);
         break;