diff --git a/app/api/agent/[id]/events/route.ts b/app/api/agent/[id]/events/route.ts index 1a6f95710..dd329e0bc 100644 --- a/app/api/agent/[id]/events/route.ts +++ b/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 | 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(); }; diff --git a/lib/event-coalescer.test.mjs b/lib/event-coalescer.test.mjs new file mode 100644 index 000000000..979a6cbaf --- /dev/null +++ b/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, []); +}); diff --git a/lib/event-coalescer.ts b/lib/event-coalescer.ts new file mode 100644 index 000000000..36204dc9b --- /dev/null +++ b/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; + }, + }; +}