perf: coalesce streamed message_update events to cut remote bandwidth (#375)

app/api/agent/[id]/events/route.ts

@@ -1,8 +1,15 @@
 import { resolveSessionPath } from "@/lib/session-reader";
 import { getRpcSession, startRpcSession, type AgentEvent } from "@/lib/rpc-manager";
+import { createMessageUpdateCoalescer } from "@/lib/event-coalescer";
 
 export const dynamic = "force-dynamic";
 
+// Buffer window for coalescing streamed message_update events. Each update
+// carries the full accumulated message, so forwarding every one amplifies
+// transfer O(n^2) (#375). ~80ms keeps streaming visibly smooth (~12/s) while
+// collapsing bursts of updates into a single send.
+const MESSAGE_UPDATE_FLUSH_MS = 80;
+
 const OMITTED_EVENT_TYPES = new Set(["turn_start", "turn_end", "tool_execution_update"]);
 
 function toClientEvent(event: AgentEvent): AgentEvent | null {
@@ -48,9 +55,23 @@ export async function GET(
       // Send initial connected event
       encode({ type: "connected", sessionId: id });
 
+      // Coalesce streamed message_update events so remote clients don't receive
+      // the full accumulated message on every chunk (#375).
+      const coalescer = createMessageUpdateCoalescer(encode);
+      let flushTimer: ReturnType<typeof setTimeout> | null = null;
+      const scheduleFlush = () => {
+        if (flushTimer) return;
+        flushTimer = setTimeout(() => {
+          flushTimer = null;
+          coalescer.flush();
+        }, MESSAGE_UPDATE_FLUSH_MS);
+      };
+
       const unsubscribe = session.onEvent((event) => {
         const clientEvent = toClientEvent(event);
-        if (clientEvent) encode(clientEvent);
+        if (!clientEvent) return;
+        coalescer.push(clientEvent);
+        if (coalescer.hasPending()) scheduleFlush();
       });
 
       // Heartbeat every 30s to prevent server/proxy timeout (Next.js default ~120-150s)
@@ -65,6 +86,7 @@ export async function GET(
       // Cleanup when client disconnects
       const cleanup = () => {
         clearInterval(heartbeat);
+        if (flushTimer) clearTimeout(flushTimer);
         unsubscribe();
         controller.close();
       };

lib/event-coalescer.test.mjs

@@ -0,0 +1,64 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { createMessageUpdateCoalescer } from "./event-coalescer.ts";
+
+function collector() {
+  const events = [];
+  return { events, emit: (e) => events.push(e) };
+}
+
+const update = (text) => ({ type: "message_update", message: { text } });
+
+test("buffers a message_update until flushed", () => {
+  const { events, emit } = collector();
+  const c = createMessageUpdateCoalescer(emit);
+
+  c.push(update("a"));
+  assert.equal(events.length, 0);
+  assert.equal(c.hasPending(), true);
+
+  c.flush();
+  assert.deepEqual(events, [update("a")]);
+  assert.equal(c.hasPending(), false);
+});
+
+test("coalesces consecutive updates to only the latest", () => {
+  const { events, emit } = collector();
+  const c = createMessageUpdateCoalescer(emit);
+
+  c.push(update("a"));
+  c.push(update("ab"));
+  c.push(update("abc"));
+  c.flush();
+
+  assert.deepEqual(events, [update("abc")]);
+});
+
+test("flushes the pending update before a following non-update event, in order", () => {
+  const { events, emit } = collector();
+  const c = createMessageUpdateCoalescer(emit);
+
+  c.push(update("a"));
+  c.push(update("ab"));
+  c.push({ type: "agent_end" });
+
+  assert.deepEqual(events, [update("ab"), { type: "agent_end" }]);
+  assert.equal(c.hasPending(), false);
+});
+
+test("passes non-update events straight through", () => {
+  const { events, emit } = collector();
+  const c = createMessageUpdateCoalescer(emit);
+
+  c.push({ type: "tool_execution_start", id: "1" });
+  assert.deepEqual(events, [{ type: "tool_execution_start", id: "1" }]);
+  assert.equal(c.hasPending(), false);
+});
+
+test("flush is a no-op when nothing is pending", () => {
+  const { events, emit } = collector();
+  const c = createMessageUpdateCoalescer(emit);
+
+  c.flush();
+  assert.deepEqual(events, []);
+});

lib/event-coalescer.ts

@@ -0,0 +1,56 @@
+import type { AgentEvent } from "./rpc-manager";
+
+/**
+ * Coalesces consecutive `message_update` events for the SSE stream.
+ *
+ * Every `message_update` carries the FULL accumulated message, and the agent
+ * emits one roughly every streamed chunk. Forwarding each to the browser makes
+ * transfer grow O(n^2) with the message size — ~150x amplification over the
+ * actual content on remote/metered connections (issue #375).
+ *
+ * Only the latest pending `message_update` is kept; the owner flushes it on a
+ * short timer (so streaming stays smooth) via {@link flush}. Any other event
+ * flushes the pending update first, so a `message_update` is never reordered
+ * past a later tool/turn/`agent_end` event and the message's final content is
+ * always delivered before the next boundary event.
+ */
+export interface MessageUpdateCoalescer {
+  /** Emit `event`, buffering it when it is a coalescible `message_update`. */
+  push(event: AgentEvent): void;
+  /** Emit the buffered `message_update`, if any. */
+  flush(): void;
+  /** Whether a `message_update` is currently buffered. */
+  hasPending(): boolean;
+}
+
+export function createMessageUpdateCoalescer(
+  emit: (event: AgentEvent) => void,
+): MessageUpdateCoalescer {
+  let pending: AgentEvent | null = null;
+
+  const flush = () => {
+    if (!pending) return;
+    const event = pending;
+    pending = null;
+    emit(event);
+  };
+
+  return {
+    push(event) {
+      if (event.type === "message_update") {
+        // Supersede any earlier pending update — it already carries the full
+        // accumulated message, so only the newest one matters.
+        pending = event;
+        return;
+      }
+      // Preserve ordering: the in-progress message must reach the client before
+      // whatever event follows it.
+      flush();
+      emit(event);
+    },
+    flush,
+    hasPending() {
+      return pending !== null;
+    },
+  };
+}