diff --git a/src/contexts/acp-connections-context.test.tsx b/src/contexts/acp-connections-context.test.tsx index bd37810a06..d66ae764a8 100644 --- a/src/contexts/acp-connections-context.test.tsx +++ b/src/contexts/acp-connections-context.test.tsx @@ -1,9 +1,11 @@ import { useEffect } from "react" -import { act, render } from "@testing-library/react" +import { act, cleanup, render } from "@testing-library/react" import { useTranslations } from "next-intl" import { afterEach, beforeEach, describe, expect, it, vi } from "vitest" import { AcpConnectionsProvider, + STREAM_FLUSH_FRAME_MS, + STREAM_FLUSH_MAX_MS, useAcpActions, useConnectionStore, } from "@/contexts/acp-connections-context" @@ -2134,6 +2136,605 @@ describe("out-of-turn wire guard + background activity", () => { }) }) +describe("streaming flush window widens with the run it re-renders", () => { + async function mountStreamingOwner() { + h.acpFindConnectionForConversation.mockResolvedValue(null) + await mountProvider() + await act(async () => { + await h.actions!.connect(TAB, "claude_code", "/tmp/x", "sess-1", 42) + }) + const handlers = latestAttachHandlers() + emitAcpEvent(handlers, { + seq: 1, + connection_id: "spawned-conn", + type: "status_changed", + status: "prompting", + }) + return handlers + } + + function liveTextFor(key: string): string { + const content = h.store!.getConnection(key)?.liveMessage?.content ?? [] + return content + .map((block) => (block.type === "text" ? block.text : "")) + .join("") + } + + function liveText(): string { + return liveTextFor(TAB) + } + + /** A second conversation streaming at the same time, on its own connection. */ + const OTHER_TAB = "conv-2-claude_code-43" + + async function mountTwoStreamingOwners() { + h.acpFindConnectionForConversation.mockResolvedValue(null) + h.acpConnect.mockReset() + h.acpConnect + .mockResolvedValueOnce("conn-a") + .mockResolvedValueOnce("conn-b") + .mockResolvedValue("conn-extra") + await mountProvider() + await act(async () => { + await h.actions!.connect(TAB, "claude_code", "/tmp/a", "sess-a", 42) + }) + const a = latestAttachHandlers() + await act(async () => { + await h.actions!.connect(OTHER_TAB, "claude_code", "/tmp/b", "sess-b", 43) + }) + const b = latestAttachHandlers() + for (const [handlers, id] of [ + [a, "conn-a"], + [b, "conn-b"], + ] as const) { + emitAcpEvent(handlers, { + seq: 1, + connection_id: id, + type: "status_changed", + status: "prompting", + }) + } + return { a, b } + } + + it("holds a long run for more frames, and delivers exactly what arrived", async () => { + const handlers = await mountStreamingOwner() + // Mount and connect on real timers (they await the transport); only the + // flush window below is driven by hand. + vi.useFakeTimers() + try { + // First delta of the turn: the live message is empty, so the window is + // the single frame it has always been. + const head = "a".repeat(9000) + emitAcpEvent(handlers, { + seq: 2, + connection_id: "spawned-conn", + type: "content_delta", + text: head, + }) + expect(liveText()).toBe("") + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveText()).toBe(head) + + // The next delta is armed against a 9000-character run, which is past + // the first step: one frame is no longer enough to release it. + emitAcpEvent(handlers, { + seq: 3, + connection_id: "spawned-conn", + type: "content_delta", + text: "b", + }) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveText()).toBe(head) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveText()).toBe(`${head}b`) + + // Whatever the window, every chunk lands once and in order. + let expected = `${head}b` + for (let i = 0; i < 40; i++) { + const text = `-${i}-` + expected += text + emitAcpEvent(handlers, { + seq: 4 + i, + connection_id: "spawned-conn", + type: "content_delta", + text, + }) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_MAX_MS) + }) + } + expect(liveText()).toBe(expected) + } finally { + vi.useRealTimers() + } + }) + + // The window is sized by the RUN the batch appends to, not by how much the + // turn has said in total: a reply that has already written 9 KB and then ran + // a tool is back to rendering a short block, and must not keep paying for + // the prose above it. + it("returns to a single frame when a new run starts", async () => { + const handlers = await mountStreamingOwner() + vi.useFakeTimers() + try { + emitAcpEvent(handlers, { + seq: 2, + connection_id: "spawned-conn", + type: "content_delta", + text: "a".repeat(9000), + }) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + + // A tool call closes the prose run (and flushes the queue itself), so + // the reply that resumes after it starts short again. + emitAcpEvent(handlers, { + seq: 3, + connection_id: "spawned-conn", + type: "tool_call", + tool_call_id: "toolu_1", + title: "Read", + kind: "read", + status: "in_progress", + content: null, + raw_input: null, + raw_output: null, + }) + emitAcpEvent(handlers, { + seq: 4, + connection_id: "spawned-conn", + type: "content_delta", + text: "after", + }) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveText()).toBe(`${"a".repeat(9000)}after`) + + // And it stays there while the new run is short, even though the turn + // now holds more than 9 KB in total. + emitAcpEvent(handlers, { + seq: 5, + connection_id: "spawned-conn", + type: "content_delta", + text: " more", + }) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveText()).toBe(`${"a".repeat(9000)}after more`) + } finally { + vi.useRealTimers() + } + }) + + // An event that flushes the queue mid-window must CANCEL that window, not + // just forget it. A forgotten timer still fires, releases whatever the next + // window had queued, and takes the ref down with it — so the window after it + // is forgotten too. One such event per turn is enough to halve the cadence, + // and a long turn has many (`usage_update`, `tool_call_update`, …), so the + // widening decays back to a flat 16 ms over exactly the turns it is for. + it("cancels the pending window when an event flushes the queue early", async () => { + const handlers = await mountStreamingOwner() + vi.useFakeTimers() + try { + const head = "a".repeat(9000) + emitAcpEvent(handlers, { + seq: 2, + connection_id: "spawned-conn", + type: "content_delta", + text: head, + }) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveText()).toBe(head) + + // Arms a two-frame window against the 9 KB run. + emitAcpEvent(handlers, { + seq: 3, + connection_id: "spawned-conn", + type: "content_delta", + text: "b", + }) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveText()).toBe(head) + + // One frame in, a non-streaming event flushes the queue itself. + emitAcpEvent(handlers, { + seq: 4, + connection_id: "spawned-conn", + type: "usage_update", + used: 1_000, + size: 200_000, + }) + expect(liveText()).toBe(`${head}b`) + + // The next delta arms its own two-frame window from here. The window the + // flush pre-empted must not fire inside it and cut it short. + emitAcpEvent(handlers, { + seq: 5, + connection_id: "spawned-conn", + type: "content_delta", + text: "c", + }) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveText()).toBe(`${head}b`) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveText()).toBe(`${head}bc`) + } finally { + vi.useRealTimers() + } + }) + + // Codeg runs several agents at once by design. A window sized from what one + // conversation is re-rendering must not be charged to another — least of all + // to a background one that costs nothing to flush and gains nothing by + // waiting. + it("never makes one conversation wait on another's long reply", async () => { + const { a, b } = await mountTwoStreamingOwners() + vi.useFakeTimers() + try { + const head = "a".repeat(9000) + emitAcpEvent(a, { + seq: 2, + connection_id: "conn-a", + type: "content_delta", + text: head, + }) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveTextFor(TAB)).toBe(head) + + // A's next delta is armed against its 9 KB run — two frames. B has said + // nothing, so B's is one, and B must get it. + emitAcpEvent(a, { + seq: 3, + connection_id: "conn-a", + type: "content_delta", + text: "A", + }) + emitAcpEvent(b, { + seq: 2, + connection_id: "conn-b", + type: "content_delta", + text: "B", + }) + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveTextFor(OTHER_TAB)).toBe("B") + expect(liveTextFor(TAB)).toBe(head) + + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveTextFor(TAB)).toBe(`${head}A`) + } finally { + vi.useRealTimers() + } + }) + + // The other half of the same rule: flushing one conversation out of turn + // must not release another's window early either. + it("does not let one conversation's event flush another's queue", async () => { + const { a, b } = await mountTwoStreamingOwners() + vi.useFakeTimers() + try { + emitAcpEvent(a, { + seq: 2, + connection_id: "conn-a", + type: "content_delta", + text: "A", + }) + emitAcpEvent(b, { + seq: 2, + connection_id: "conn-b", + type: "content_delta", + text: "B", + }) + // B's tool call flushes B's queue, and only B's. + emitAcpEvent(b, { + seq: 3, + connection_id: "conn-b", + type: "tool_call", + tool_call_id: "toolu_1", + title: "Read", + kind: "read", + status: "in_progress", + content: null, + raw_input: null, + raw_output: null, + }) + // Positive half: the out-of-turn flush really did fire, so A's empty + // reading below is scoping, not a flush that silently does nothing. + expect(liveTextFor(OTHER_TAB)).toBe("B") + expect(liveTextFor(TAB)).toBe("") + + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveTextFor(TAB)).toBe("A") + } finally { + vi.useRealTimers() + } + }) + + // A snapshot REPLACES the live message wholesale. Deltas still coalescing + // when one lands would append to the message it installed — which already + // contains them, because the snapshot is generated at a higher seq — and + // the reply shows the same prose twice. The attach stream re-emits a + // snapshot on RECONNECT, mid-turn, so this is what a dropped WebSocket does + // to a streaming reply, not a corner case. + it("lands coalesced deltas before a mid-turn snapshot replaces the message", async () => { + const handlers = await mountStreamingOwner() + vi.useFakeTimers() + try { + emitAcpEvent(handlers, { + seq: 2, + connection_id: "spawned-conn", + type: "content_delta", + text: "hello ", + }) + expect(liveText()).toBe("") + + // The reconnect snapshot was generated after that delta reached the + // backend, so it already carries the text sitting in our window. Still + // `prompting`: the turn did not stop because the socket did. + h.denormalizeSnapshot.mockReturnValue({ + connectionId: "spawned-conn", + status: "prompting", + sessionId: null, + modes: null, + configOptions: null, + availableCommands: null, + usage: null, + liveMessage: { + id: "live-1", + role: "assistant", + content: [{ type: "text", text: "hello " }], + startedAt: 0, + }, + pendingPermission: null, + pendingAskQuestion: null, + pendingUserMessage: null, + promptCapabilities: null, + selectorsReady: false, + supportsFork: false, + configStale: false, + configStaleKind: null, + lastError: null, + eventSeq: 9, + activeDelegations: [], + }) + hydrateSnapshot(handlers, { + event_seq: 9, + } as unknown as LiveSessionSnapshot) + + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_MAX_MS) + }) + expect(liveText()).toBe("hello ") + } finally { + vi.useRealTimers() + } + }) + + // …and FLUSH is why that is a flush and not a discard. A snapshot behind + // our cursor takes the stale branch, which merges selector fields and + // leaves `liveMessage` alone — so it never redelivers the queued prose, and + // dropping the queue there would lose it outright. Which branch a snapshot + // takes isn't knowable at the call site, so the safe move is the one that + // is correct on both. + it("keeps coalesced deltas a stale snapshot will not redeliver", async () => { + const handlers = await mountStreamingOwner() + vi.useFakeTimers() + try { + emitAcpEvent(handlers, { + seq: 4, + connection_id: "spawned-conn", + type: "content_delta", + text: "hello ", + }) + expect(liveText()).toBe("") + + // eventSeq 2 is behind the cursor the delta above advanced to 4, so + // this hydrate takes the stale branch. Note it carries no live message + // of its own — the stale branch would ignore one anyway. + h.denormalizeSnapshot.mockReturnValue({ + connectionId: "spawned-conn", + status: "connected", + sessionId: null, + modes: null, + configOptions: null, + availableCommands: null, + usage: null, + liveMessage: null, + pendingPermission: null, + pendingAskQuestion: null, + pendingUserMessage: null, + promptCapabilities: null, + selectorsReady: true, + supportsFork: false, + configStale: false, + configStaleKind: null, + lastError: null, + eventSeq: 2, + activeDelegations: [], + }) + hydrateSnapshot(handlers, { + event_seq: 2, + } as unknown as LiveSessionSnapshot) + + // The stale branch merged its latched field and left the turn alone… + expect(h.store!.getConnection(TAB)?.selectorsReady).toBe(true) + expect(h.store!.getConnection(TAB)?.status).toBe("prompting") + // …and the prose that was mid-window is on screen, not dropped. + expect(liveText()).toBe("hello ") + + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_MAX_MS) + }) + expect(liveText()).toBe("hello ") + } finally { + vi.useRealTimers() + } + }) + + // Removing the entry disarms its window: "no connection, no queue". + // + // Asserted on the timer rather than on rendered text, because the text is + // already defended twice over — `status_changed` flushes before it applies + // `prompting`, and the out-of-turn guard drops a batch for a connection + // that isn't prompting — so a leak would have to thread between both to + // show up on screen. The invariant is the thing worth pinning: context keys + // are REUSED (close a tab mid-turn and reopen it and the next connection is + // handed the same `conv---` string), and a window that + // outlives its connection is a dispatch aimed at whoever holds the key up + // to STREAM_FLUSH_MAX_MS later. Cheap to keep impossible; unpleasant to + // rediscover from a duplicated paragraph in someone's reply. + it("disarms a removed connection's flush window", async () => { + const handlers = await mountStreamingOwner() + vi.useFakeTimers() + try { + const idle = vi.getTimerCount() + emitAcpEvent(handlers, { + seq: 2, + connection_id: "spawned-conn", + type: "content_delta", + text: "from the turn that was closed", + }) + expect(liveText()).toBe("") + expect(vi.getTimerCount()).toBe(idle + 1) + + await act(async () => { + await h.actions!.disconnect(TAB) + }) + expect(h.store!.getConnection(TAB)).toBeUndefined() + expect(vi.getTimerCount()).toBe(idle) + } finally { + vi.useRealTimers() + } + }) + + // Settling a connection the backend has forgotten is the one turn-ending + // path that dispatches STATUS_CHANGED directly instead of going through the + // event handler, so nothing else drains the queue — and the moment the + // entry reads `disconnected` the out-of-turn guard drops the batch. The + // last thing the agent managed to say should still be on screen. + it("lands what was mid-window when a connection is settled as gone", async () => { + const handlers = await mountStreamingOwner() + vi.useFakeTimers() + try { + emitAcpEvent(handlers, { + seq: 2, + connection_id: "spawned-conn", + type: "content_delta", + text: "last words", + }) + expect(liveText()).toBe("") + + // Pressing Stop on a connection the backend no longer holds. + h.acpCancel.mockRejectedValueOnce(new Error("Connection not found")) + await act(async () => { + await h.actions!.cancel(TAB) + }) + + expect(h.store!.getConnection(TAB)?.status).toBe("disconnected") + expect(liveText()).toBe("last words") + } finally { + vi.useRealTimers() + } + }) + + // The same rule on a different removal. `DELEGATION_CHILD_DETACH` drops an + // entry too, and it is why the discard is derived from the reducer's result + // rather than from a list of action types: closing the work-task transcript + // dialog on a streaming sub-agent must not leave a window armed either, and + // nobody should have to remember to extend a list to get that. + it("disarms a detached delegation child's flush window", async () => { + const CHILD = "task-conn-1" + await mountProvider() + act(() => { + h.actions!.attachDelegationChild({ + connectionId: CHILD, + parentConnectionId: CHILD, + parentToolUseId: "work-task-9", + agentType: "claude_code", + hydrate: false, + }) + }) + const child = latestAttachHandlers() + emitAcpEvent(child, { + seq: 1, + connection_id: CHILD, + type: "status_changed", + status: "prompting", + }) + + vi.useFakeTimers() + try { + const idle = vi.getTimerCount() + emitAcpEvent(child, { + seq: 2, + connection_id: CHILD, + type: "content_delta", + text: "sub-agent, mid-sentence", + }) + expect(vi.getTimerCount()).toBe(idle + 1) + + act(() => { + h.actions!.detachDelegationChild(CHILD) + }) + expect(h.store!.getConnection(CHILD)).toBeUndefined() + expect(vi.getTimerCount()).toBe(idle) + } finally { + vi.useRealTimers() + } + }) + + // …and the same on unmount. This suite runs the attach transport, which is + // the one whose windows used to survive: the legacy `acp://event` listener + // effect owned the cleanup, and it returns early — before registering any — + // for exactly the transports that stream through attach subscriptions. + it("drops every armed window when the provider unmounts", async () => { + const handlers = await mountStreamingOwner() + vi.useFakeTimers() + try { + const idle = vi.getTimerCount() + emitAcpEvent(handlers, { + seq: 2, + connection_id: "spawned-conn", + type: "content_delta", + text: "mid-sentence", + }) + expect(vi.getTimerCount()).toBe(idle + 1) + + // RTL's `cleanup` unmounts inside its own `act`, and clears its + // registry afterwards — so the suite's auto-cleanup is a no-op here. + cleanup() + expect(vi.getTimerCount()).toBe(idle) + } finally { + vi.useRealTimers() + } + }) +}) + describe("AcpConnectionsProvider Grok cross-agent-type model switch", () => { function grokModelOptions(current: string): SessionConfigOptionInfo[] { return [ @@ -3801,6 +4402,67 @@ describe("connect() teardown races", () => { expect(h.acpConnect).not.toHaveBeenCalled() }) + // Orphan rescue happens MID-TURN, and the reducer drops a `STREAM_BATCH` + // for a key with no connection — so deltas still sitting in the old key's + // flush window are lost text unless they land before the entry moves. Up to + // STREAM_FLUSH_MAX_MS of a live reply. + it("lands the old key's coalesced deltas before rescuing its connection", async () => { + mountDesktop() + await act(async () => {}) + await act(async () => { + await h.actions!.connect(TAB, "claude_code", "/tmp/x", "sess-1") + }) + const onEvent = vi.mocked(subscribe).mock.calls[0]![1] as ( + envelope: EventEnvelope + ) => void + // Hold the clock from here on. The window is only 16 ms, so on real timers + // the rescue's own awaits could outlast it and the delta would land + // because the timer fired — the test would keep passing with the flush + // below deleted. Under fake timers the clock never advances, so the only + // thing that can deliver this text is the explicit flush. + h.acpTouchConnection.mockResolvedValue(true) + vi.useFakeTimers() + try { + act(() => { + onEvent({ + seq: 1, + connection_id: "spawned-conn", + type: "session_started", + session_id: "sess-1", + } as EventEnvelope) + onEvent({ + seq: 2, + connection_id: "spawned-conn", + type: "status_changed", + status: "prompting", + } as EventEnvelope) + // Still inside its flush window when the rescue below fires. + onEvent({ + seq: 3, + connection_id: "spawned-conn", + type: "content_delta", + text: "half a sentence", + } as EventEnvelope) + }) + // Queued, not applied: the window has not elapsed. + expect(h.store!.getConnection(TAB)?.liveMessage?.content).toEqual([]) + + await act(async () => { + await h.actions!.connect(RESCUE_TAB, "claude_code", "/tmp/x", "sess-1") + }) + + const rescued = h.store!.getConnection(RESCUE_TAB) + expect(rescued?.connectionId).toBe("spawned-conn") + expect( + (rescued?.liveMessage?.content ?? []) + .map((block) => (block.type === "text" ? block.text : "")) + .join("") + ).toBe("half a sentence") + } finally { + vi.useRealTimers() + } + }) + it("still connects when the backend GC'd the connection mid-probe", async () => { // Web/attach transport: `onDetached("connection_gone")` drops the entry // outright. That is NOT a rekey — nothing else holds the connection — so diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index 8b8078578f..abea489b41 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -741,6 +741,128 @@ type StreamingAction = parentToolUseId?: string } +/** One display frame: the narrowest window streaming deltas coalesce into. */ +export const STREAM_FLUSH_FRAME_MS = 16 +/** + * The widest — about five batches a second. + * + * Kept well under the 500 ms sample period of the tok/s gauge + * (`useTokenOutputSpeed`), which reads the live message on its own clock: at + * most one window's worth of text can be un-flushed when it samples, so the + * reading stays accurate. Raising this past ~250 ms would make that gauge + * sawtooth, and is not a free knob. + */ +export const STREAM_FLUSH_MAX_MS = 192 +/** Characters of re-rendered live content that buy one more frame. */ +const STREAM_FLUSH_CHARS_PER_FRAME = 8 * 1024 +/** + * What one re-rendered non-prose block costs, in prose-equivalent characters. + * + * Measured in the real component tree (jsdom, React 19), re-rendering a live + * turn the way a batch does — growing prose costs 0.00075 ms/char, while each + * block that re-renders whole costs a FLAT 0.02 ms (a collapsed thinking + * block) to 0.18 ms (a plan card), with a tool card at 0.06 ms. That is 30 to + * 235 prose-equivalent characters; 128 sits inside the range, so 64 cards buy + * one extra frame. + * + * Flat, not proportional to the block's content, because that is what the + * measurement shows: a collapsed thinking block costs the same at 200 and at + * 4000 characters, since the cards render clamped previews and Radix keeps + * closed content unmounted. + */ +const STREAM_FLUSH_BLOCK_CHARS = 128 +/** + * Deltas one connection may coalesce before the window is cut short. A safety + * valve for a burst the timer can't keep up with, not a cadence knob — it + * bounds one connection's unrendered backlog, so it is per connection like the + * window it pre-empts. + */ +const STREAM_QUEUE_CAP = 256 + +/** + * What the next batch will re-render, in prose-equivalent characters, read off + * the live message as of the LAST batch. + * + * Every batch replaces the live message, so the whole turn is re-adapted and + * handed to the renderer again. What that costs is NOT uniform: + * + * - The trailing text/thinking run — the block this batch grows — is + * re-rendered whole: normalized, re-lexed into markdown blocks, + * re-highlighted. Linear in its length, and the dominant term. + * - Settled prose above it is FREE: `TextPart` memoizes on the string by + * value, so an unchanged run bails out before the markdown renderer. + * (Measured: eight extra settled 8 KB blocks cost nothing.) + * - Everything else — tool cards, closed thinking blocks, plan cards, + * steering notes — re-renders on EVERY batch regardless. They memoize on + * the part object, and `createMessageTurnAdapter` refuses to cache a + * streaming turn (`cacheable = !isStreaming && !inProgress`), so each batch + * hands them freshly built objects. Charged a flat + * `STREAM_FLUSH_BLOCK_CHARS` each. + * + * A turn that has run a hundred tools and is now writing its summary has a + * short run and a real per-batch cost; sizing from the run alone would leave + * it on a single frame while it burns a third of every one. + */ +export function liveRerenderChars( + content: readonly LiveContentBlock[] | undefined +): number { + if (!content || content.length === 0) return 0 + let chars = 0 + const lastIndex = content.length - 1 + for (let i = 0; i < content.length; i++) { + const block = content[i] + if (block.type === "text" || block.type === "thinking") { + // Only the trailing run is charged per character — it is the one the + // batch grows, and the only one whose memo the batch invalidates. (A + // trailing thinking run is charged in full on the assumption it is + // expanded while it streams; if it is not, this over-charges by one + // block, which the ceiling bounds. Same for a trailing run that belongs + // to a sub-agent: `parentToolUseId` is deliberately not read here, so a + // delegated run is charged as main prose. It renders inside a capsule + // that may be collapsed, so this errs toward the wider window — the + // direction that costs latency, never correctness.) + if (i === lastIndex) chars += block.text.length + continue + } + chars += STREAM_FLUSH_BLOCK_CHARS + } + return chars +} + +/** + * How long streaming deltas coalesce before one `STREAM_BATCH` lands, given + * what that batch will re-render (see `liveRerenderChars`). + * + * The cost is linear in that, and the window was a flat 16 ms — so the work a + * turn cost grew with the SQUARE of its own output, while the rate it arrived + * at stayed the same. Past a few tens of KB the renderer could no longer keep + * up with the stream, which is #589: at ~300 tok/s the whole UI stops + * responding, on hardware with plenty of headroom. + * + * Measured on a 300 tok/s stream, counting the characters re-rendered across + * a turn: 30 s of output cost 32.5M before and 11.1M after; 120 s cost 518.8M + * before and 59.9M after. + * + * Under 8 KB — nearly every reply — keeps the 16 ms window it has today. Past + * that each further 8 KB buys one more frame, so the cost per second flattens + * out instead of climbing with the answer. + * + * Nothing about WHAT gets delivered changes: the queue merges and dispatches + * exactly as before, every chunk lands once and in order, and every event that + * MUTATES the live message — a tool card, a permission prompt, the end of a + * turn — flushes the queue first, so none of them waits on this window and + * none of them can land out of wire order. (Events that touch nothing the + * transcript renders, such as `permission_resolved` or `async_task`, do not + * flush; that predates this window and is unchanged by it.) + */ +export function streamFlushDelayMs(liveRunChars: number): number { + const frames = Math.max( + 1, + Math.ceil(liveRunChars / STREAM_FLUSH_CHARS_PER_FRAME) + ) + return Math.min(STREAM_FLUSH_MAX_MS, STREAM_FLUSH_FRAME_MS * frames) +} + type ConnectionsMap = Map const MAX_LIVE_TOOL_RAW_OUTPUT_CHARS = 200_000 const MAX_BUFFERED_UNMAPPED_EVENTS_PER_CONNECTION = 64 @@ -3249,8 +3371,14 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // Activity tracking (no re-renders) const lastActivityRef = useRef(new Map()) - const streamingQueueRef = useRef([]) - const flushTimerRef = useRef | null>(null) + // Streaming coalescing queue + its window, PER CONNECTION (see + // `flushStreamingQueue`). Entries are created on the first delta after a + // flush and removed by the flush that drains them, so both maps hold only + // the connections with deltas in flight right now. + const streamingQueuesRef = useRef(new Map()) + const flushTimersRef = useRef( + new Map>() + ) const pendingUnmappedEventsRef = useRef(new Map()) const listenerReadyRef = useRef(false) const listenerReadyWaitersRef = useRef void>>([]) @@ -3281,10 +3409,69 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // ── Dispatch (replaces useReducer dispatch) ── + /** + * Drop ONE connection's queued deltas and its window, without dispatching. + * + * Declared here rather than beside the other streaming helpers because + * `dispatch` below calls it and needs it in scope; it touches only the two + * refs, so there is no cycle. + */ + const discardStreamingKey = useCallback((contextKey: string) => { + const timer = flushTimersRef.current.get(contextKey) + if (timer !== undefined) { + clearTimeout(timer) + flushTimersRef.current.delete(contextKey) + } + streamingQueuesRef.current.delete(contextKey) + }, []) + + /** The same, for every connection at once. */ + const discardStreamingQueues = useCallback(() => { + for (const timer of flushTimersRef.current.values()) clearTimeout(timer) + flushTimersRef.current.clear() + streamingQueuesRef.current.clear() + }, []) + const dispatch = useCallback( (action: Action) => { const prev = storeRef.current.connections const next = connectionsReducer(prev, action) + + // "No entry, no queue." A removed key must not leave deltas armed behind + // it: they land up to `STREAM_FLUSH_MAX_MS` later, and context keys are + // REUSED — the same `conv---` string is handed to the + // next connection that opens on that tab — so a late batch does not + // merely waste a dispatch, it can append a dead turn's prose to a live + // one. + // + // Read off the reducer's OWN result rather than from a list of removal + // actions, so the rule is exactly "the entry is gone" and cannot drift + // from what the reducer decided. It also declines where the reducer + // declines — a rekey onto an occupied key is rejected, and discarding + // for a connection that is still there and still talking would lose its + // trailing prose. + // + // What IS enumerated is the two hot paths, so that the list fails safe: + // forget to add a case here and the cost is a walk over the open + // connections, not a stray window. Listing the removals instead reads + // cheaper and fails the other way — that is how `DELEGATION_CHILD_DETACH` + // went uncovered, and a size check alone would miss the next action + // shaped like `REKEY_CONNECTION`, which removes a key and adds another. + // + // Discard rather than flush: a flush would re-enter `dispatch`, and + // there is no one left to render the result. Callers that DO want the + // deltas landed first call `flushStreamingQueue(key)` before removing — + // `connect()`'s orphan rescue is the one that does, ahead of its rekey. + if ( + next !== prev && + action.type !== "STREAM_BATCH" && + action.type !== "BATCH_TOOL_CALL_UPDATES" + ) { + for (const key of prev.keys()) { + if (!next.has(key)) discardStreamingKey(key) + } + } + if (next === prev) return // no change storeRef.current.connections = next @@ -3337,7 +3524,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { } } }, - [notifyKeyListeners, notifyAllKeyListeners] + [discardStreamingKey, notifyKeyListeners, notifyAllKeyListeners] ) // ── setActiveKey ── @@ -3431,54 +3618,99 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { [dispatch] ) - const flushStreamingQueue = useCallback(() => { - flushTimerRef.current = null - const queued = streamingQueueRef.current - if (queued.length === 0) return - streamingQueueRef.current = [] - - // Merge adjacent deltas by connection key (per-key order preserved), - // reducing reducer work and string copies under high-frequency streams. - const grouped = new Map() - for (const action of queued) { - const list = grouped.get(action.contextKey) - if (!list) { - grouped.set(action.contextKey, [{ ...action }]) - continue + /** + * Drain ONE connection's coalesced deltas into a single `STREAM_BATCH`. + * + * Scheduling is PER CONNECTION. The queue and its window used to be global, + * so the window a batch waited in was whichever connection's delta happened + * to arm the timer. Harmless while that window was a flat 16 ms; once + * `streamFlushDelayMs` sizes it from what is being re-rendered, a long reply + * in one conversation held every OTHER conversation's deltas for up to + * `STREAM_FLUSH_MAX_MS` — including a background conversation with no panel + * mounted, which costs nothing to flush and so bought nothing by waiting. + * Codeg runs several agents at once by design, so that is the normal case, + * not a corner of one. + * + * Connections are independent — own wire, own seq cursor, own + * `ConnectionState` — so there is nothing to coordinate between them, and + * per-key ordering is what the reducer and the out-of-turn guards already + * reason about. Nothing wants "flush everything": teardown discards instead + * (`discardStreamingQueues`), because a batch dispatched into a key that is + * being removed is at best wasted and at worst lands on its successor. + */ + const flushStreamingQueue = useCallback( + (contextKey: string) => { + // CANCEL the pending window, don't just forget it. Most callers are + // event handlers flushing out of turn (a tool card, a permission prompt, + // a usage update), and a timer that is only dropped from the map still + // fires: it releases whatever the NEXT window had queued, early, and + // takes that window's entry with it, so the delta after it arms a third + // timer. One stray timer per out-of-turn flush, each halving the + // cadence — which is how a widened window (`streamFlushDelayMs`) decays + // back to a flat frame over exactly the long turns it exists for. + const timer = flushTimersRef.current.get(contextKey) + if (timer !== undefined) { + clearTimeout(timer) + flushTimersRef.current.delete(contextKey) } - const last = list[list.length - 1] - // Same-type AND same subagent attribution: within one flush window, - // main-thread and parented deltas (or two different subagents') must - // not concatenate — this pre-coalescing runs BEFORE the reducer's - // attribution-aware merge and would otherwise defeat it. - if ( - last && - last.type === action.type && - last.parentToolUseId === action.parentToolUseId - ) { - last.text += action.text - } else { - list.push({ ...action }) + const queued = streamingQueuesRef.current.get(contextKey) + if (queued === undefined) return + streamingQueuesRef.current.delete(contextKey) + if (queued.length === 0) return + + // Merge adjacent deltas (arrival order preserved), reducing reducer work + // and string copies under high-frequency streams. Same-type AND same + // subagent attribution: within one flush window, main-thread and + // parented deltas (or two different subagents') must not concatenate — + // this pre-coalescing runs BEFORE the reducer's attribution-aware merge + // and would otherwise defeat it. + const compacted: StreamingAction[] = [] + for (const action of queued) { + const last = compacted[compacted.length - 1] + if ( + last && + last.type === action.type && + last.parentToolUseId === action.parentToolUseId + ) { + last.text += action.text + } else { + compacted.push({ ...action }) + } } - } - const compacted = Array.from(grouped.values()).flat() - dispatch({ type: "STREAM_BATCH", actions: compacted }) - }, [dispatch]) + dispatch({ type: "STREAM_BATCH", actions: compacted }) + }, + [dispatch] + ) const enqueueStreamingAction = useCallback( (action: StreamingAction) => { - streamingQueueRef.current.push(action) - if (streamingQueueRef.current.length >= 256) { - if (flushTimerRef.current !== null) { - clearTimeout(flushTimerRef.current) - flushTimerRef.current = null - } - flushStreamingQueue() + const { contextKey } = action + let queue = streamingQueuesRef.current.get(contextKey) + if (queue === undefined) { + queue = [] + streamingQueuesRef.current.set(contextKey, queue) + } + queue.push(action) + if (queue.length >= STREAM_QUEUE_CAP) { + // Cap reached — `flushStreamingQueue` clears the pending window itself. + flushStreamingQueue(contextKey) return } - if (flushTimerRef.current === null) { - flushTimerRef.current = setTimeout(flushStreamingQueue, 16) + if (!flushTimersRef.current.has(contextKey)) { + // Size the window from what this batch will re-render, read as of the + // last batch — so it costs one map lookup plus a walk over the live + // turn's blocks, and a fresh turn (empty live message) is back to a + // single frame. See `liveRerenderChars` and `streamFlushDelayMs`. + const delay = streamFlushDelayMs( + liveRerenderChars( + storeRef.current.connections.get(contextKey)?.liveMessage?.content + ) + ) + flushTimersRef.current.set( + contextKey, + setTimeout(() => flushStreamingQueue(contextKey), delay) + ) } }, [flushStreamingQueue] @@ -3658,7 +3890,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { if (!echo) playEventSound(e) switch (e.type) { case "status_changed": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "STATUS_CHANGED", contextKey, status: e.status }) break case "content_delta": @@ -3682,7 +3914,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "claude_sdk_message": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "CLAUDE_API_RETRY", contextKey, @@ -3691,7 +3923,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break case "tool_call": settleRetryIncidentsOnProgress(contextKey) - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "TOOL_CALL", contextKey, @@ -3708,7 +3940,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "tool_call_update": - flushStreamingQueue() + flushStreamingQueue(contextKey) pendingToolCallUpdates.current.push({ contextKey, tool_call_id: e.tool_call_id, @@ -3739,7 +3971,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // as a tool result, never a user message. Those stay in the notes // list above the composer, which is where a reload leaves them too. if (e.item.status !== "delivered") break - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "STEERING_MESSAGE", contextKey, @@ -3782,7 +4014,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // Agent called the blocking `ask_user_question` MCP tool. Flush any // queued streaming so the card renders against current content, then // raise the interactive multiple-choice card above the input box. - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "SET_ASK_QUESTION", contextKey, @@ -3824,7 +4056,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // Grok called `exit_plan_mode`: it's blocked on the user's approval of // the plan. Flush queued streaming so the card renders against current // content, then raise the interactive plan-approval card. - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "SET_PLAN_APPROVAL", contextKey, @@ -3970,7 +4202,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "permission_request": - flushStreamingQueue() + flushStreamingQueue(contextKey) flushPendingToolCallUpdates() dispatch({ type: "PERMISSION_REQUEST", @@ -4001,7 +4233,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { } break case "session_started": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "SESSION_STARTED", contextKey, @@ -4038,7 +4270,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { } break case "session_modes": { - flushStreamingQueue() + flushStreamingQueue(contextKey) // Preferences are applied on the backend during connect (see // `getSavedPrefsForConnect` + `acp_connect`), so `e.modes` already // carries the user's preferred `current_mode_id` — no client-side @@ -4060,7 +4292,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "session_config_options": { - flushStreamingQueue() + flushStreamingQueue(contextKey) // Same as `session_modes`: backend already merged saved prefs // into `current_value` before emitting. dispatch({ @@ -4092,7 +4324,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "session_config_stale": { - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "CONFIG_STALE_CHANGED", contextKey, @@ -4102,7 +4334,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "selectors_ready": { - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "SELECTORS_READY", contextKey, @@ -4119,7 +4351,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "prompt_capabilities": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "PROMPT_CAPABILITIES", contextKey, @@ -4127,7 +4359,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "fork_supported": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "FORK_SUPPORTED", contextKey, @@ -4135,7 +4367,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "mode_changed": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "MODE_CHANGED", contextKey, @@ -4143,7 +4375,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "plan_update": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "PLAN_UPDATE", contextKey, @@ -4190,7 +4422,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // Without this, a delta enqueued just BEFORE the retry arrived lands // just AFTER it and wipes the banner we are about to raise — which pi // reaches routinely, since it retries mid-stream between prose chunks. - flushStreamingQueue() + flushStreamingQueue(contextKey) const retryConn = storeRef.current.connections.get(contextKey) dispatch({ type: "CLAUDE_API_RETRY", @@ -4208,7 +4440,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "turn_complete": { - flushStreamingQueue() + flushStreamingQueue(contextKey) flushPendingToolCallUpdates() // AIR retry warnings settle only at a CLEAN turn end, mirroring the // backend's `apply_event`. A failed turn's terminal failure rides @@ -4281,7 +4513,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "error": { - flushStreamingQueue() + flushStreamingQueue(contextKey) const nc = storeRef.current.connections.get(contextKey) const agentLabel = nc ? getAgentLabel(nc.agentType) @@ -4426,7 +4658,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "session_load_failed": { - flushStreamingQueue() + flushStreamingQueue(contextKey) // Localize via the stable `code` field ("resource_not_found" — // JSON-RPC -32002 — plus "session_unavailable" and // "session_archived", both matched on the wire message). Fall back @@ -4500,7 +4732,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "available_commands": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "AVAILABLE_COMMANDS", contextKey, @@ -4508,7 +4740,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "usage_update": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "USAGE_UPDATE", contextKey, @@ -4677,6 +4909,28 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { let activeSub: EventStreamSubscription | null = null const handlers: AttachHandlers = { onSnapshot: (snapshot) => { + // Land anything still coalescing BEFORE the snapshot replaces the + // live message. This handler also runs on an attach-stream + // RECONNECT, mid-turn, so deltas from before the drop can still be + // queued — and the hydrate would swap `liveMessage` out from under + // them, so the flush that follows would append a run the snapshot + // already contains, duplicating it on screen. + // + // Flush, not discard: `HYDRATE_FROM_SNAPSHOT` has a stale-snapshot + // branch that merges selector fields only and leaves `liveMessage` + // untouched, so discarding would silently drop prose nothing else + // redelivers. Flushing is what the queue's contract asks for anyway + // — every path that reads or replaces `liveMessage` from outside + // should see the same state it would have seen with no coalescing + // at all. + // + // The other three snapshot consumers don't need this: they hydrate + // at attach time, before `bindConnectionRoute` gives the key a + // route, so no delta of theirs can be in flight yet — and a queue + // left by a PREVIOUS connection under a recycled key is discarded + // at `CONNECTION_REMOVED` (see `dispatch`), which is the right + // outcome there, not a flush. + flushStreamingQueue(contextKey) const patch = denormalizeSnapshot(snapshot) dispatch({ type: "HYDRATE_FROM_SNAPSHOT", contextKey, patch }) surfaceSnapshotErrorDetailsRef.current(contextKey, patch) @@ -4744,6 +4998,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { applyMappedEnvelope, captureIdentityBeforeRemoval, dispatch, + flushStreamingQueue, seedDelegationsFromSnapshot, ] ) @@ -4862,18 +5117,24 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { cancelled = true listenerReadyRef.current = false resolveListenerReadyWaiters() - if (flushTimerRef.current !== null) { - clearTimeout(flushTimerRef.current) - flushTimerRef.current = null - } unlisten?.() } - // Every dep here is a `useCallback(..., [])` — the subscription is + // Every dep here is stable for the component's life — each is either a + // `useCallback(..., [])` or, in `dispatch`'s case, a `useCallback` whose + // own deps are all `useCallback(..., [])` — so the subscription is // registered once per mount and torn down only on unmount. The event // handler deliberately isn't a dep; it's reached through // `handleMappedEventRef` so a changing closure can't churn the listener. }, [bufferUnmappedEvent, dispatch, resolveListenerReadyWaiters]) + // Drop every armed window on unmount. Its own effect, because the listener + // effect above returns early on web / remote-desktop transports — before it + // registers any cleanup — and those transports stream through the attach + // subscriptions, which fill these queues just the same. A timer surviving + // the provider fires into a `dispatch` whose store nothing is reading, and + // under a test runner it outlives the test that armed it. + useEffect(() => discardStreamingQueues, [discardStreamingQueues]) + /** * Ask the backend whether it still holds a live connection under this id. * `acp_touch_connection` answers `false` for BOTH "unknown id" and "already @@ -4917,10 +5178,24 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { releaseConnectionRoute(connectionId, contextKey) teardownAttachSubscription(contextKey) pendingUnmappedEventsRef.current.delete(connectionId) + // Land what is still coalescing while the turn is still `prompting`. + // The status change below is dispatched directly rather than through the + // event handler, so nothing else drains the queue — and once the entry + // reads `disconnected` the out-of-turn guard drops the batch, taking the + // last words this connection managed to say with it. Worth a line + // because the window is no longer a frame: `streamFlushDelayMs` can be + // holding up to STREAM_FLUSH_MAX_MS of a live reply when the liveness + // probe settles a connection out from under it. + flushStreamingQueue(contextKey) dispatch({ type: "STATUS_CHANGED", contextKey, status: "disconnected" }) return true }, - [dispatch, releaseConnectionRoute, teardownAttachSubscription] + [ + dispatch, + flushStreamingQueue, + releaseConnectionRoute, + teardownAttachSubscription, + ] ) // ── Backend keepalive + liveness reconciliation timer ── @@ -5435,6 +5710,11 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { } } if (orphanKey && orphanConn) { + // Land the orphan's coalesced deltas while it still HAS an entry: + // the reducer drops a `STREAM_BATCH` for a key with no connection, + // so anything still in its window would be lost text. Up to + // STREAM_FLUSH_MAX_MS of a live reply, and this runs mid-turn. + flushStreamingQueue(orphanKey) // The entry MOVES (REKEY_CONNECTION below deletes `orphanKey`), so // its route has to move with it — a stale orphan-key route would // deliver this connection's events to a contextKey with no entry. @@ -5766,6 +6046,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { connectAsViewer, consumeBufferedEvents, dispatch, + flushStreamingQueue, isConnectionLiveOnBackend, isConnectionReferencedLocally, localOwnerKeyOf, @@ -6052,6 +6333,12 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // didn't visit. reverseMapRef.current.clear() lastActivityRef.current.clear() + // Same reuse hazard as the caches below, on a clock: a delta queued just + // before this would otherwise dispatch up to STREAM_FLUSH_MAX_MS later, + // into whatever now holds its contextKey. `dispatch` repeats this for + // REMOVE_ALL; this call is the one ahead of the await below, which is the + // window a still-armed timer would fire in. + discardStreamingQueues() // Context keys are reused across backends, so a surviving entry here would // suppress the first snapshot alert of an unrelated session. alertedErrorDetailsRef.current.clear() @@ -6061,7 +6348,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { rekeyGenerationRef.current.clear() await Promise.all(promises) dispatch({ type: "REMOVE_ALL" }) - }, [dispatch, teardownAttachSubscription]) + }, [discardStreamingQueues, dispatch, teardownAttachSubscription]) const sendPrompt = useCallback( async ( diff --git a/src/contexts/streaming-flush-cadence.test.ts b/src/contexts/streaming-flush-cadence.test.ts new file mode 100644 index 0000000000..ecaf8541ad --- /dev/null +++ b/src/contexts/streaming-flush-cadence.test.ts @@ -0,0 +1,178 @@ +import { describe, expect, it } from "vitest" +import { + type LiveContentBlock, + STREAM_FLUSH_FRAME_MS, + STREAM_FLUSH_MAX_MS, + type ToolCallInfo, + liveRerenderChars, + streamFlushDelayMs, +} from "@/contexts/acp-connections-context" + +const text = (chars: number): LiveContentBlock => ({ + type: "text", + text: "x".repeat(chars), +}) +const thinking = (chars: number): LiveContentBlock => ({ + type: "thinking", + text: "t".repeat(chars), +}) +const toolCall = (id: string): LiveContentBlock => ({ + type: "tool_call", + info: { + tool_call_id: id, + title: "Bash", + kind: "execute", + status: "completed", + content: null, + raw_input: '{"command":"ls"}', + // The card renders a clamped preview, so its cost does not scale with + // however much output the tool produced. + raw_output_chunks: ["y".repeat(50_000)], + raw_output_total_bytes: 50_000, + locations: null, + meta: null, + images: [], + } satisfies ToolCallInfo, +}) + +describe("liveRerenderChars", () => { + it("charges the trailing run, which is the block a batch grows", () => { + expect(liveRerenderChars([text(4000)])).toBe(4000) + expect(liveRerenderChars([thinking(4000)])).toBe(4000) + }) + + it("charges nothing for prose the batch leaves alone", () => { + // `TextPart` memoizes on the string by value, so a settled run bails out + // before the markdown renderer — measured at zero for eight 8 KB blocks. + expect(liveRerenderChars([text(8192), text(8192), text(100)])).toBe(100) + }) + + it("charges a flat rate per card, whatever the card is holding", () => { + const perBlock = liveRerenderChars([toolCall("a"), text(0)]) + expect(perBlock).toBeGreaterThan(0) + // Independent of the 50 KB of raw output on the block. + expect(liveRerenderChars([toolCall("a"), toolCall("b"), text(0)])).toBe( + 2 * perBlock + ) + // …and it is small next to a run: cards move the window, prose sets it. + expect(perBlock).toBeLessThan(1024) + }) + + it("puts a tool-heavy turn past the first step on its own", () => { + const cards: LiveContentBlock[] = [] + for (let i = 0; i < 100; i++) cards.push(toolCall(`call-${i}`)) + // A hundred tools then a 2 KB summary: the run alone would read as one + // frame, while every one of those cards re-renders on every batch. + expect(streamFlushDelayMs(liveRerenderChars([...cards, text(2048)]))).toBe( + 2 * STREAM_FLUSH_FRAME_MS + ) + expect(streamFlushDelayMs(liveRerenderChars([text(2048)]))).toBe( + STREAM_FLUSH_FRAME_MS + ) + }) + + it("is zero for a turn that has said nothing yet", () => { + expect(liveRerenderChars(undefined)).toBe(0) + expect(liveRerenderChars([])).toBe(0) + }) +}) + +describe("streamFlushDelayMs", () => { + it("leaves an ordinary reply on the single-frame window it has today", () => { + for (const chars of [0, 1, 200, 4096, 8192]) { + expect(streamFlushDelayMs(chars)).toBe(STREAM_FLUSH_FRAME_MS) + } + }) + + it("buys one more frame per further 8 KB, up to the ceiling", () => { + expect(streamFlushDelayMs(8193)).toBe(2 * STREAM_FLUSH_FRAME_MS) + expect(streamFlushDelayMs(16 * 1024)).toBe(2 * STREAM_FLUSH_FRAME_MS) + expect(streamFlushDelayMs(32 * 1024)).toBe(4 * STREAM_FLUSH_FRAME_MS) + expect(streamFlushDelayMs(64 * 1024)).toBe(8 * STREAM_FLUSH_FRAME_MS) + expect(streamFlushDelayMs(1024 * 1024)).toBe(STREAM_FLUSH_MAX_MS) + }) + + it("never returns a window that would stall or spin", () => { + for (const chars of [-1, 0, Number.MAX_SAFE_INTEGER]) { + const delay = streamFlushDelayMs(chars) + expect(delay).toBeGreaterThanOrEqual(STREAM_FLUSH_FRAME_MS) + expect(delay).toBeLessThanOrEqual(STREAM_FLUSH_MAX_MS) + } + }) +}) + +/** + * Replay a 300 tok/s stream (one 4-character chunk per token) through a flush + * schedule and report what it costs. + * + * `charsRendered` is the metric #589 is about: every batch re-renders the prose + * run it appended to, whole, and that cost is linear in the run's length. With + * a flat window the rate is constant while the run keeps growing, so the total + * climbs with the square of the turn's own output. + */ +function replay( + seconds: number, + delayFor: (runChars: number) => number +): { flushes: number; charsRendered: number; delivered: number } { + const CHUNK_CHARS = 4 + const MS_PER_CHUNK = 1000 / 300 + const chunks = seconds * 300 + + let run = 0 + let pending = 0 + let nextFlushAt = delayFor(0) + let flushes = 0 + let charsRendered = 0 + + for (let i = 0; i < chunks; i++) { + pending += CHUNK_CHARS + const now = (i + 1) * MS_PER_CHUNK + if (now < nextFlushAt) continue + run += pending + pending = 0 + flushes++ + charsRendered += run + nextFlushAt = now + delayFor(run) + } + // The turn always ends on an explicit flush (`turn_complete`). + if (pending > 0) { + run += pending + flushes++ + charsRendered += run + } + return { flushes, charsRendered, delivered: run } +} + +const flatWindow = () => STREAM_FLUSH_FRAME_MS + +describe("what the flush cadence costs at 300 tokens/sec", () => { + it("delivers every character either way", () => { + for (const seconds of [10, 30, 120]) { + const expected = seconds * 300 * 4 + expect(replay(seconds, flatWindow).delivered).toBe(expected) + expect(replay(seconds, streamFlushDelayMs).delivered).toBe(expected) + } + }) + + it("stops the re-render cost climbing with the square of the answer", () => { + const flat30 = replay(30, flatWindow) + const flat120 = replay(120, flatWindow) + const now30 = replay(30, streamFlushDelayMs) + const now120 = replay(120, streamFlushDelayMs) + + // A flat window makes four times the output cost sixteen times the work. + expect(flat120.charsRendered / flat30.charsRendered).toBeGreaterThan(12) + + // Backing off keeps that growth near linear… + expect(now120.charsRendered / now30.charsRendered).toBeLessThan(8) + // …and cuts the absolute cost at both lengths. + expect(now30.charsRendered).toBeLessThan(flat30.charsRendered * 0.45) + expect(now120.charsRendered).toBeLessThan(flat120.charsRendered * 0.2) + }) + + it("leaves a short answer on exactly the cadence it has today", () => { + // Five seconds at 300 tok/s is 6 KB — under the first step, so the + // schedule is unchanged for the replies almost every turn produces. + expect(replay(5, streamFlushDelayMs)).toEqual(replay(5, flatWindow)) + }) +})