perf(server): batch terminal output and bound replay buffer eviction

apps/server/src/features/terminal/host/__tests__/pty-host-runtime.test.ts

@@ -125,6 +125,7 @@ describe("PtyHostProcessRuntime", () => {
       rows: 30,
     });
     pty.emitData("ok\r\n");
+    await vi.advanceTimersByTimeAsync(2);
     expect(pty.write).toHaveBeenCalledWith(Buffer.from("echo ok\r"));
     expect(pty.resize).toHaveBeenCalledWith(100, 30);
     expect(events.map((event) => event.kind)).toContain("output");
@@ -301,4 +302,188 @@ describe("PtyHostProcessRuntime", () => {
     expect(scope.close).toHaveBeenCalledOnce();
     await runtime.dispose();
   });
+
+  const createRunningSession = async (
+    pty: FakePty,
+    events: PtyHostEvent[],
+    queueBytes?: () => number,
+  ): Promise<PtyHostProcessRuntime> => {
+    const runtime = new PtyHostProcessRuntime({
+      platform: "windows",
+      hostRuntime: TEST_HOST_RUNTIME,
+      nativeAbi: "fake-v1",
+      publish: (event) => events.push(event),
+      queueBytes,
+      spawnPty: vi.fn(() => pty),
+      createScope: vi.fn(() => createScope()),
+    });
+    await runtime.receive({
+      contractVersion: 1,
+      kind: "handshake",
+      requestedGeneration: "7",
+      platform: "windows",
+    });
+    await runtime.receive({
+      contractVersion: 1,
+      kind: "create",
+      sessionId: SESSION_ID,
+      hostGeneration: "7",
+      scope: { kind: "workspace", workspaceId: SESSION_ID },
+      executable: "pwsh.exe",
+      arguments: [],
+      cwd: "C:\\repo",
+      cols: 80,
+      rows: 24,
+      env: [],
+    });
+    events.length = 0;
+    return runtime;
+  };
+
+  const outputEvents = (events: readonly PtyHostEvent[]) =>
+    events.filter((event): event is Extract<PtyHostEvent, { kind: "output" }> => event.kind === "output");
+
+  it("coalesces a synchronous output burst into one event", async () => {
+    vi.useFakeTimers();
+    const pty = new FakePty();
+    const events: PtyHostEvent[] = [];
+    const runtime = await createRunningSession(pty, events);
+
+    pty.emitData("a");
+    pty.emitData("b");
+    pty.emitData("c");
+    await vi.advanceTimersByTimeAsync(2);
+
+    const outputs = outputEvents(events);
+    expect(outputs).toHaveLength(1);
+    expect(outputs[0]!.outputSeq).toBe("1");
+    expect(Buffer.from(outputs[0]!.dataBase64, "base64").toString()).toBe("abc");
+    await runtime.dispose();
+    vi.useRealTimers();
+  });
+
+  it("flushes a full-sized chunk without waiting for the batch window", async () => {
+    vi.useFakeTimers();
+    const pty = new FakePty();
+    const events: PtyHostEvent[] = [];
+    const runtime = await createRunningSession(pty, events);
+
+    pty.emitData("x".repeat(65_536 + 10));
+
+    const outputs = outputEvents(events);
+    expect(outputs).toHaveLength(2);
+    expect(Buffer.from(outputs[0]!.dataBase64, "base64").length).toBe(65_536);
+    expect(Buffer.from(outputs[1]!.dataBase64, "base64").length).toBe(10);
+    await runtime.dispose();
+    vi.useRealTimers();
+  });
+
+  it("pauses a flooding PTY while IPC is saturated and resumes after drain", async () => {
+    vi.useFakeTimers();
+    const pty = new FakePty();
+    const events: PtyHostEvent[] = [];
+    let queued = 800 * 1024;
+    const runtime = await createRunningSession(pty, events, () => queued);
+
+    pty.emitData("flood");
+    await vi.advanceTimersByTimeAsync(4);
+    // One pause at spawn, one from backpressure.
+    expect(pty.pause).toHaveBeenCalledTimes(2);
+    expect(outputEvents(events)).toHaveLength(0);
+
+    queued = 0;
+    await vi.advanceTimersByTimeAsync(4);
+    expect(pty.resume).toHaveBeenCalled();
+    const outputs = outputEvents(events);
+    expect(outputs).toHaveLength(1);
+    expect(Buffer.from(outputs[0]!.dataBase64, "base64").toString()).toBe("flood");
+    await runtime.dispose();
+    vi.useRealTimers();
+  });
+
+  it("flushes pending output before the exit event", async () => {
+    vi.useFakeTimers();
+    const pty = new FakePty();
+    const events: PtyHostEvent[] = [];
+    const runtime = await createRunningSession(pty, events);
+
+    pty.emitData("tail");
+    pty.emitExit(3);
+
+    const outputs = outputEvents(events);
+    expect(outputs).toHaveLength(1);
+    const exit = events.at(-1);
+    expect(exit).toMatchObject({ kind: "exit", code: 3, finalOutputSeq: "1" });
+    await runtime.dispose();
+    vi.useRealTimers();
+  });
+
+  it("kills a session whose pending output exceeds the per-session bound", async () => {
+    vi.useFakeTimers();
+    const pty = new FakePty();
+    const events: PtyHostEvent[] = [];
+    const runtime = await createRunningSession(pty, events, () => 900 * 1024);
+
+    pty.emitData("x".repeat(64 * 1024 * 1024 + 1));
+    expect(pty.kill).toHaveBeenCalledOnce();
+    await runtime.dispose();
+    vi.useRealTimers();
+  });
+
+  it("keeps a paused session paused inside the hysteresis band", async () => {
+    vi.useFakeTimers();
+    const pty = new FakePty();
+    const events: PtyHostEvent[] = [];
+    let queued = 800 * 1024;
+    const runtime = await createRunningSession(pty, events, () => queued);
+
+    pty.emitData("flood");
+    await vi.advanceTimersByTimeAsync(4);
+    expect(pty.pause).toHaveBeenCalledTimes(2);
+
+    queued = 600 * 1024;
+    await vi.advanceTimersByTimeAsync(6);
+    expect(pty.resume).toHaveBeenCalledTimes(1);
+    expect(outputEvents(events)).toHaveLength(0);
+
+    queued = 400 * 1024;
+    await vi.advanceTimersByTimeAsync(4);
+    expect(outputEvents(events)).toHaveLength(1);
+    await runtime.dispose();
+    vi.useRealTimers();
+  });
+
+  it("defers the exit event behind pending output while pressured, then publishes in order", async () => {
+    vi.useFakeTimers();
+    const pty = new FakePty();
+    const events: PtyHostEvent[] = [];
+    let queued = 800 * 1024;
+    const runtime = await createRunningSession(pty, events, () => queued);
+
+    pty.emitData("tail");
+    pty.emitExit(3);
+    await vi.advanceTimersByTimeAsync(4);
+    expect(events.map((event) => event.kind)).not.toContain("exit");
+
+    queued = 0;
+    await vi.advanceTimersByTimeAsync(4);
+    expect(events.map((event) => event.kind)).toEqual(["output", "exit"]);
+    expect(events.at(-1)).toMatchObject({ kind: "exit", finalOutputSeq: "1" });
+    await runtime.dispose();
+    vi.useRealTimers();
+  });
+
+  it("publishes a pressured exit by deadline when IPC never drains", async () => {
+    vi.useFakeTimers();
+    const pty = new FakePty();
+    const events: PtyHostEvent[] = [];
+    const runtime = await createRunningSession(pty, events, () => 900 * 1024);
+
+    pty.emitData("dropped");
+    pty.emitExit(0);
+    await vi.advanceTimersByTimeAsync(2_100);
+    expect(events.at(-1)).toMatchObject({ kind: "exit", finalOutputSeq: "0" });
+    await runtime.dispose();
+    vi.useRealTimers();
+  });
 });

apps/server/src/features/terminal/host/pty-host-process.ts

@@ -3,7 +3,6 @@ import type { HostRuntime } from "@mcode/shared/node/host-runtime";
 import {
   PTY_HOST_MAX_MESSAGE_BYTES,
   PTY_HOST_MAX_RETAINED_RECORDS,
-  PtyHostEventSchema,
   type PtyHostEvent,
 } from "./pty-host-protocol.js";
 import { PtyHostProcessRuntime } from "./pty-host-runtime.js";
@@ -85,17 +84,17 @@ export function runPtyHostProcess(hostRuntime: HostRuntime): PtyHostProcessRunti
     process.exitCode = 1;
     process.disconnect?.();
   };
+  // The server revalidates every event on receipt; the host trusts its own producer shapes.
   const publish = (event: PtyHostEvent): void => {
-    const validated = PtyHostEventSchema().parse(event);
     if (!process.connected || !process.send) {
       throw new Error("PTY host IPC channel is unavailable");
     }
-    const eventBytes = NodeBuffer.Buffer.byteLength(JSON.stringify(validated), "utf8");
+    const eventBytes = NodeBuffer.Buffer.byteLength(JSON.stringify(event), "utf8");
     if (outboundBytes + eventBytes > MAX_IPC_QUEUE_BYTES) {
       throw new Error("PTY host event queue exceeds 1 MiB");
     }
     outboundBytes += eventBytes;
-    process.send(validated, (error) => {
+    process.send(event, (error) => {
       outboundBytes = Math.max(0, outboundBytes - eventBytes);
       if (error) void failHost(error);
     });