From 8a25483b1aef5dbcddc88ad7f24f4cc0e227c197 Mon Sep 17 00:00:00 2001 From: Adam Dalloul <47503782+Adam-Dalloul@users.noreply.github.com> Date: Thu, 10 Sep 2026 08:47:51 -0700 Subject: [PATCH 1/6] perf(chat): widen the streaming flush window with the reply it re-renders At around 300 tokens a second the whole UI stops responding (#589), on hardware with plenty of headroom. Measured where the time goes, per streaming batch, driving a realistic chunk stream through the real code. The store and timeline are not it: `setLiveMessage` plus `computeTimeline` costs 0.03 to 0.12 ms a batch, and running the message adapter on top brings that to 0.15 to 1.0 ms. Rendering the reply is three orders of magnitude above that. Each batch replaces the live message, so the prose run it appended to is handed to the markdown renderer again whole: normalized, re-lexed into blocks (marked, 0.7 ms at 4 KB rising to 6.9 ms at 64 KB), re-highlighted, re-rendered. In the test renderer that is 3.1 ms a batch at 4 KB and 27 ms at 64 KB. An unchanged string costs 0.075 ms, so the whole of it is the run having grown. The window those batches landed in was a flat 16 ms whatever the reply had grown to, so the work a turn costs rose with the square of its own output while the rate it arrived at stayed put. Replaying 300 tok/s and counting the characters re-rendered across the turn: 30 seconds of output cost 32.5M, 120 seconds cost 518.8M: sixteen times the work for four times the answer. The window now scales with the run being re-rendered. Under 8 KB, which is nearly every reply, it is the same 16 ms as today; past that each further 8 KB buys one more frame, up to 192 ms. Same replay: 30 s falls to 11.1M and 120 s to 59.9M, and the growth goes from quadratic to near linear. It is sized from the run rather than the whole message, so a reply that has already written 9 KB and then ran a tool is back to a single frame for the block it starts next. Nothing about what gets delivered changes. The queue merges and dispatches exactly as before, every chunk lands once and in order, and every non-streaming event still flushes it immediately, so a tool card, a permission prompt or the end of a turn never waits on this window. What is left is the per-batch cost itself: the run is still re-lexed and re-rendered whole each time. Splitting a streaming reply at block boundaries so only the tail is rebuilt would remove that, but not without changing how markdown spanning the split renders. --- src/contexts/acp-connections-context.test.tsx | 146 ++++++++++++++++++ src/contexts/acp-connections-context.tsx | 55 ++++++- src/contexts/streaming-flush-cadence.test.ts | 106 +++++++++++++ 3 files changed, 306 insertions(+), 1 deletion(-) create mode 100644 src/contexts/streaming-flush-cadence.test.ts diff --git a/src/contexts/acp-connections-context.test.tsx b/src/contexts/acp-connections-context.test.tsx index bd37810a06..e98dcf23e7 100644 --- a/src/contexts/acp-connections-context.test.tsx +++ b/src/contexts/acp-connections-context.test.tsx @@ -4,6 +4,8 @@ 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,150 @@ 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 liveText(): string { + const content = h.store!.getConnection(TAB)?.liveMessage?.content ?? [] + return content + .map((block) => (block.type === "text" ? block.text : "")) + .join("") + } + + 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() + } + }) +}) + describe("AcpConnectionsProvider Grok cross-agent-type model switch", () => { function grokModelOptions(current: string): SessionConfigOptionInfo[] { return [ diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index 8b8078578f..d22bdfd53f 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -741,6 +741,44 @@ 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. */ +export const STREAM_FLUSH_MAX_MS = 192 +/** Characters of live prose that buy one more frame of coalescing. */ +const STREAM_FLUSH_CHARS_PER_FRAME = 8 * 1024 + +/** + * How long streaming deltas coalesce before one `STREAM_BATCH` lands, given + * the length of the prose run the batch will grow. + * + * Each batch replaces the live message, and the run it appended to is + * re-rendered whole: normalized, re-lexed into markdown blocks, re-highlighted. + * That is linear in the run's length, 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. + * + * A run 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, and + * every non-streaming event still flushes it immediately, so a tool card, a + * permission prompt or the end of a turn never waits on this window. + */ +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 @@ -3478,7 +3516,22 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { return } if (flushTimerRef.current === null) { - flushTimerRef.current = setTimeout(flushStreamingQueue, 16) + // Size the window from the prose run this batch will grow — the block + // the reducer will append to, which is what gets re-rendered whole. + // Read as of the last batch, so it costs one map lookup, and a fresh + // turn (empty live message, or a run that just restarted after a tool + // call) is back to a single frame. See `streamFlushDelayMs`. + const content = storeRef.current.connections.get(action.contextKey) + ?.liveMessage?.content + const last = content?.[content.length - 1] + const runChars = + last && (last.type === "text" || last.type === "thinking") + ? last.text.length + : 0 + flushTimerRef.current = setTimeout( + flushStreamingQueue, + streamFlushDelayMs(runChars) + ) } }, [flushStreamingQueue] diff --git a/src/contexts/streaming-flush-cadence.test.ts b/src/contexts/streaming-flush-cadence.test.ts new file mode 100644 index 0000000000..b9e867c3d0 --- /dev/null +++ b/src/contexts/streaming-flush-cadence.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it } from "vitest" +import { + STREAM_FLUSH_FRAME_MS, + STREAM_FLUSH_MAX_MS, + streamFlushDelayMs, +} from "@/contexts/acp-connections-context" + +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)) + }) +}) From f2f86d83402b6b32252007947b9bbe5ef4056824 Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 17 Sep 2026 09:05:27 +0800 Subject: [PATCH 2/6] fix(chat): cancel the pending flush window when an event flushes early MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `flushStreamingQueue` nulled `flushTimerRef` without clearing the timer, so every out-of-turn flush (a tool card, a permission prompt, a usage update) left a stray `setTimeout` behind. That timer still fires: it releases whatever the NEXT window had queued, early, and nulls the ref out from under that window, so the delta after it arms a third timer. One stray per out-of-turn flush, each halving the effective cadence. Invisible while the window was a flat 16 ms — a stray timer only fired a batch a few milliseconds early. With a window that widens to 192 ms with the run it re-renders, it is the mechanism decaying back to a flat frame over exactly the long, tool-heavy turns it exists for. Co-Authored-By: Claude Opus 5 (1M context) --- src/contexts/acp-connections-context.test.tsx | 65 +++++++++++++++++++ src/contexts/acp-connections-context.tsx | 18 +++-- 2 files changed, 78 insertions(+), 5 deletions(-) diff --git a/src/contexts/acp-connections-context.test.tsx b/src/contexts/acp-connections-context.test.tsx index e98dcf23e7..998971d691 100644 --- a/src/contexts/acp-connections-context.test.tsx +++ b/src/contexts/acp-connections-context.test.tsx @@ -2278,6 +2278,71 @@ describe("streaming flush window widens with the run it re-renders", () => { 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() + } + }) }) describe("AcpConnectionsProvider Grok cross-agent-type model switch", () => { diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index d22bdfd53f..0de0e78ab0 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -3470,7 +3470,18 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { ) const flushStreamingQueue = useCallback(() => { - flushTimerRef.current = null + // 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 detached from the ref still fires: + // it releases whatever the NEXT window had queued, early, and nulls the + // ref out from under that window 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. + if (flushTimerRef.current !== null) { + clearTimeout(flushTimerRef.current) + flushTimerRef.current = null + } const queued = streamingQueueRef.current if (queued.length === 0) return streamingQueueRef.current = [] @@ -3508,10 +3519,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { (action: StreamingAction) => { streamingQueueRef.current.push(action) if (streamingQueueRef.current.length >= 256) { - if (flushTimerRef.current !== null) { - clearTimeout(flushTimerRef.current) - flushTimerRef.current = null - } + // Cap reached — `flushStreamingQueue` clears the pending window itself. flushStreamingQueue() return } From 61e1c470d68029780e8bbff32df931ce238684cc Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 17 Sep 2026 10:23:20 +0800 Subject: [PATCH 3/6] perf(chat): size the streaming flush window per connection, and by what re-renders MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two corrections to the window `streamFlushDelayMs` introduced. Both are about what the window is measured from, not about what it does. 1. Scheduling is now PER CONNECTION. The coalescing queue and its timer were global, so the window a batch waited in was whichever connection's delta happened to arm it. Harmless while that window was a flat 16 ms; once it is sized from what is being re-rendered, a long reply in one conversation held every OTHER conversation's deltas for up to 192 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. Queue and timer are keyed by contextKey now, and every event handler flushes its own connection rather than all of them. Connections are independent — own wire, own seq cursor, own ConnectionState — so there was never anything to coordinate; per-key ordering is what the reducer and the out-of-turn guards already reason about. Two things fall out of having the per-key handle: - Orphan rescue lands the old key's coalesced deltas before the entry moves. The reducer drops a STREAM_BATCH for a key with no connection, so anything still in its window was lost text — up to 192 ms of a live reply, mid-turn. - disconnectAll discards pending windows instead of letting them dispatch into whatever next holds a recycled contextKey. 2. The window is sized by everything the batch re-renders, not just the run. Measured in the real component tree (jsdom, React 19), re-rendering a live turn the way a batch does: growing prose 0.00075 ms/char (0.77 ms/KB) settled text block 0 ms — TextPart's by-value memo holds tool card 0.060 ms flat closed thinking block 0.023 ms flat, same at 200 and 4000 chars plan card 0.176 ms flat Cards are flat because they render clamped previews and Radix keeps closed content unmounted — so charging them by content would be wrong, and charging them nothing leaves a turn that ran a hundred tools and is now writing its summary on a single frame while it burns a third of every one. They are charged 128 prose-equivalent characters each, inside the measured 30–235 range; 64 cards buy one extra frame. Also records on STREAM_FLUSH_MAX_MS that it must stay under the 500 ms sample period of the tok/s gauge, and corrects the claim that EVERY non-streaming event flushes the queue — the ones that mutate the live message do, which is what ordering needs; permission_resolved and async_task do not, and never did. Co-Authored-By: Claude Opus 5 (1M context) --- src/contexts/acp-connections-context.test.tsx | 177 ++++++++- src/contexts/acp-connections-context.tsx | 357 ++++++++++++------ src/contexts/streaming-flush-cadence.test.ts | 72 ++++ 3 files changed, 496 insertions(+), 110 deletions(-) diff --git a/src/contexts/acp-connections-context.test.tsx b/src/contexts/acp-connections-context.test.tsx index 998971d691..1ffe71dfbb 100644 --- a/src/contexts/acp-connections-context.test.tsx +++ b/src/contexts/acp-connections-context.test.tsx @@ -2153,13 +2153,50 @@ describe("streaming flush window widens with the run it re-renders", () => { return handlers } - function liveText(): string { - const content = h.store!.getConnection(TAB)?.liveMessage?.content ?? [] + 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 @@ -2343,6 +2380,91 @@ describe("streaming flush window widens with the run it re-renders", () => { 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", + }) + // B's tool call flushes B's queue, and only B's. + emitAcpEvent(b, { + seq: 2, + 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, + }) + expect(liveTextFor(TAB)).toBe("") + + act(() => { + vi.advanceTimersByTime(STREAM_FLUSH_FRAME_MS) + }) + expect(liveTextFor(TAB)).toBe("A") + } finally { + vi.useRealTimers() + } + }) }) describe("AcpConnectionsProvider Grok cross-agent-type model switch", () => { @@ -4012,6 +4134,57 @@ 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 + 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([]) + + h.acpTouchConnection.mockResolvedValue(true) + 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") + }) + 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 0de0e78ab0..c62093164f 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -743,33 +743,113 @@ type StreamingAction = /** One display frame: the narrowest window streaming deltas coalesce into. */ export const STREAM_FLUSH_FRAME_MS = 16 -/** The widest — about five batches a second. */ +/** + * 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 live prose that buy one more frame of coalescing. */ +/** 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.) + 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 - * the length of the prose run the batch will grow. + * what that batch will re-render (see `liveRerenderChars`). * - * Each batch replaces the live message, and the run it appended to is - * re-rendered whole: normalized, re-lexed into markdown blocks, re-highlighted. - * That is linear in the run's length, 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. + * 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. * - * A run 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, and - * every non-streaming event still flushes it immediately, so a tool card, a - * permission prompt or the end of a turn never waits on this window. + * 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( @@ -3287,8 +3367,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>>([]) @@ -3469,76 +3555,119 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { [dispatch] ) - const flushStreamingQueue = useCallback(() => { - // 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 detached from the ref still fires: - // it releases whatever the NEXT window had queued, early, and nulls the - // ref out from under that window 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. - if (flushTimerRef.current !== null) { - clearTimeout(flushTimerRef.current) - 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 queue into a single `STREAM_BATCH`. */ + const flushStreamingKey = 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] + ) + + /** + * Flush one connection's coalesced deltas, or every connection's when + * `contextKey` is omitted (teardown only). + * + * 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. + */ + const flushStreamingQueue = useCallback( + (contextKey?: string) => { + if (contextKey !== undefined) { + flushStreamingKey(contextKey) + return + } + // Snapshot the keys: `flushStreamingKey` mutates the map it iterates. + for (const key of Array.from(streamingQueuesRef.current.keys())) { + flushStreamingKey(key) + } + }, + [flushStreamingKey] + ) + + /** Drop every queued delta and its window, without dispatching. */ + const discardStreamingQueues = useCallback(() => { + for (const timer of flushTimersRef.current.values()) clearTimeout(timer) + flushTimersRef.current.clear() + streamingQueuesRef.current.clear() + }, []) const enqueueStreamingAction = useCallback( (action: StreamingAction) => { - streamingQueueRef.current.push(action) - if (streamingQueueRef.current.length >= 256) { + 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() + flushStreamingQueue(contextKey) return } - if (flushTimerRef.current === null) { - // Size the window from the prose run this batch will grow — the block - // the reducer will append to, which is what gets re-rendered whole. - // Read as of the last batch, so it costs one map lookup, and a fresh - // turn (empty live message, or a run that just restarted after a tool - // call) is back to a single frame. See `streamFlushDelayMs`. - const content = storeRef.current.connections.get(action.contextKey) - ?.liveMessage?.content - const last = content?.[content.length - 1] - const runChars = - last && (last.type === "text" || last.type === "thinking") - ? last.text.length - : 0 - flushTimerRef.current = setTimeout( - flushStreamingQueue, - streamFlushDelayMs(runChars) + 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) ) } }, @@ -3719,7 +3848,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": @@ -3743,7 +3872,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "claude_sdk_message": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "CLAUDE_API_RETRY", contextKey, @@ -3752,7 +3881,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break case "tool_call": settleRetryIncidentsOnProgress(contextKey) - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "TOOL_CALL", contextKey, @@ -3769,7 +3898,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, @@ -3800,7 +3929,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, @@ -3843,7 +3972,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, @@ -3885,7 +4014,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, @@ -4031,7 +4160,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "permission_request": - flushStreamingQueue() + flushStreamingQueue(contextKey) flushPendingToolCallUpdates() dispatch({ type: "PERMISSION_REQUEST", @@ -4062,7 +4191,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { } break case "session_started": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "SESSION_STARTED", contextKey, @@ -4099,7 +4228,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 @@ -4121,7 +4250,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({ @@ -4153,7 +4282,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "session_config_stale": { - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "CONFIG_STALE_CHANGED", contextKey, @@ -4163,7 +4292,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "selectors_ready": { - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "SELECTORS_READY", contextKey, @@ -4180,7 +4309,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "prompt_capabilities": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "PROMPT_CAPABILITIES", contextKey, @@ -4188,7 +4317,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "fork_supported": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "FORK_SUPPORTED", contextKey, @@ -4196,7 +4325,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "mode_changed": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "MODE_CHANGED", contextKey, @@ -4204,7 +4333,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "plan_update": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "PLAN_UPDATE", contextKey, @@ -4251,7 +4380,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", @@ -4269,7 +4398,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 @@ -4342,7 +4471,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) @@ -4487,7 +4616,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 @@ -4561,7 +4690,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { break } case "available_commands": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "AVAILABLE_COMMANDS", contextKey, @@ -4569,7 +4698,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { }) break case "usage_update": - flushStreamingQueue() + flushStreamingQueue(contextKey) dispatch({ type: "USAGE_UPDATE", contextKey, @@ -4923,17 +5052,19 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { cancelled = true listenerReadyRef.current = false resolveListenerReadyWaiters() - if (flushTimerRef.current !== null) { - clearTimeout(flushTimerRef.current) - flushTimerRef.current = null - } + discardStreamingQueues() unlisten?.() } // Every dep here is a `useCallback(..., [])` — 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]) + }, [ + bufferUnmappedEvent, + discardStreamingQueues, + dispatch, + resolveListenerReadyWaiters, + ]) /** * Ask the backend whether it still holds a live connection under this id. @@ -5496,6 +5627,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. @@ -5827,6 +5963,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { connectAsViewer, consumeBufferedEvents, dispatch, + flushStreamingQueue, isConnectionLiveOnBackend, isConnectionReferencedLocally, localOwnerKeyOf, @@ -6113,6 +6250,10 @@ 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. + 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() @@ -6122,7 +6263,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 index b9e867c3d0..ecaf8541ad 100644 --- a/src/contexts/streaming-flush-cadence.test.ts +++ b/src/contexts/streaming-flush-cadence.test.ts @@ -1,10 +1,82 @@ 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]) { From 9bbe799a140f60d609e966977577c76c0016e904 Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 17 Sep 2026 13:18:04 +0800 Subject: [PATCH 4/6] perf(chat): make "no connection, no queue" an invariant of the flush window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the per-connection rework, closing the review notes left on it. The coalescing queue is meant to be invisible: any path that reads or replaces a connection's `liveMessage` from outside should see the state it would have seen with no coalescing at all. Two places didn't. - A snapshot REPLACES the live message, and the attach stream re-emits one on reconnect, mid-turn. Deltas still coalescing then appended to the message the snapshot installed — which already contained them, since the snapshot is generated at a higher seq — and the reply showed the same prose twice. Flush before hydrating, rather than discard: the stale-snapshot branch leaves `liveMessage` untouched, so discarding would drop prose nothing else redelivers. - A removed key kept its window armed. Context keys are reused, so a stray timer is a dispatch aimed at whoever holds the key up to STREAM_FLUSH_MAX_MS later. Discard at the one place every removal funnels through — `dispatch`, ahead of the reducer so a no-op removal is covered too — so the invariant holds for removal sites added later. The rendered text was 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 this is the invariant, not a fix for a reachable duplication; the test asserts the disarmed timer accordingly. Unmount cleanup moves to its own effect. It lived in the legacy `acp://event` listener effect, which returns early — before registering any cleanup — for exactly the web / remote-desktop transports that stream through attach subscriptions and fill these queues just the same. Also collapses `flushStreamingKey` into `flushStreamingQueue` now that its optional-key branch has no callers (teardown discards instead of flushing), and notes that `liveRerenderChars` charges a trailing sub-agent run as main prose — erring toward the wider window, which costs latency, never correctness. Co-Authored-By: Claude Opus 5 (1M context) --- src/contexts/acp-connections-context.test.tsx | 213 +++++++++++++++--- src/contexts/acp-connections-context.tsx | 168 +++++++++----- 2 files changed, 293 insertions(+), 88 deletions(-) diff --git a/src/contexts/acp-connections-context.test.tsx b/src/contexts/acp-connections-context.test.tsx index 1ffe71dfbb..98914827ca 100644 --- a/src/contexts/acp-connections-context.test.tsx +++ b/src/contexts/acp-connections-context.test.tsx @@ -1,5 +1,5 @@ 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 { @@ -2442,10 +2442,16 @@ describe("streaming flush window widens with the run it re-renders", () => { type: "content_delta", text: "A", }) - // B's tool call flushes B's queue, and only B's. 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", @@ -2455,6 +2461,9 @@ describe("streaming flush window widens with the run it re-renders", () => { 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(() => { @@ -2465,6 +2474,128 @@ describe("streaming flush window widens with the run it re-renders", () => { 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() + } + }) + + // 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() + } + }) + + // …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) + + act(() => { + cleanup() + }) + expect(vi.getTimerCount()).toBeLessThan(idle + 1) + } finally { + vi.useRealTimers() + } + }) }) describe("AcpConnectionsProvider Grok cross-agent-type model switch", () => { @@ -4147,42 +4278,52 @@ describe("connect() teardown races", () => { const onEvent = vi.mocked(subscribe).mock.calls[0]![1] as ( envelope: EventEnvelope ) => void - 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([]) - + // 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) - await act(async () => { - await h.actions!.connect(RESCUE_TAB, "claude_code", "/tmp/x", "sess-1") - }) + 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([]) - 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") + 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 () => { diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index c62093164f..baf1c6db7f 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -816,7 +816,11 @@ export function liveRerenderChars( // 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.) + // 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 } @@ -3405,8 +3409,57 @@ 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) => { + // A removed key must not leave deltas armed behind it. They would 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 appends a dead turn's prose to a live one. + // + // Done here, at the one place every removal funnels through, rather than + // at the six-plus sites that dispatch these: this is the invariant + // ("no entry, no queue"), and it holds for removals added later too. + // + // 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. + // + // Ahead of the reducer, because a no-op removal (the entry is already + // gone) returns early below and would skip the cleanup — and "the entry + // is gone" is exactly when a stray queue must not survive. + if (action.type === "CONNECTION_REMOVED") { + discardStreamingKey(action.contextKey) + } else if (action.type === "REKEY_CONNECTION") { + discardStreamingKey(action.fromKey) + } else if (action.type === "REMOVE_ALL") { + discardStreamingQueues() + } + const prev = storeRef.current.connections const next = connectionsReducer(prev, action) if (next === prev) return // no change @@ -3461,7 +3514,12 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { } } }, - [notifyKeyListeners, notifyAllKeyListeners] + [ + discardStreamingKey, + discardStreamingQueues, + notifyKeyListeners, + notifyAllKeyListeners, + ] ) // ── setActiveKey ── @@ -3555,8 +3613,27 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { [dispatch] ) - /** Drain ONE connection's queue into a single `STREAM_BATCH`. */ - const flushStreamingKey = useCallback( + /** + * 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, @@ -3601,46 +3678,6 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { [dispatch] ) - /** - * Flush one connection's coalesced deltas, or every connection's when - * `contextKey` is omitted (teardown only). - * - * 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. - */ - const flushStreamingQueue = useCallback( - (contextKey?: string) => { - if (contextKey !== undefined) { - flushStreamingKey(contextKey) - return - } - // Snapshot the keys: `flushStreamingKey` mutates the map it iterates. - for (const key of Array.from(streamingQueuesRef.current.keys())) { - flushStreamingKey(key) - } - }, - [flushStreamingKey] - ) - - /** Drop every queued delta and its window, without dispatching. */ - const discardStreamingQueues = useCallback(() => { - for (const timer of flushTimersRef.current.values()) clearTimeout(timer) - flushTimersRef.current.clear() - streamingQueuesRef.current.clear() - }, []) - const enqueueStreamingAction = useCallback( (action: StreamingAction) => { const { contextKey } = action @@ -4867,6 +4904,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) @@ -4934,6 +4993,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { applyMappedEnvelope, captureIdentityBeforeRemoval, dispatch, + flushStreamingQueue, seedDelegationsFromSnapshot, ] ) @@ -5052,19 +5112,21 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { cancelled = true listenerReadyRef.current = false resolveListenerReadyWaiters() - discardStreamingQueues() unlisten?.() } // Every dep here is a `useCallback(..., [])` — 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, - discardStreamingQueues, - dispatch, - resolveListenerReadyWaiters, - ]) + }, [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. @@ -6252,7 +6314,9 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { 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. + // 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. From a9254722aba56abec93629f183562448413aeed9 Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 17 Sep 2026 13:38:29 +0800 Subject: [PATCH 5/6] perf(chat): derive the queue discard from the reducer, not a list of actions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up. The enumerated form missed a removal: `detachDelegationChild` dispatches `DELEGATION_CHILD_DETACH`, which deletes the entry, with no flush and no discard ahead of it — so closing the work-task transcript dialog on a streaming sub-agent left a window armed. The previous commit claimed the invariant held "for removals added later too" while missing one that already existed, which is the argument against writing it as a list at all. Read it off the reducer's own result instead: discard for any key that was in the map before and is not in it after. That cannot drift from what the reducer decided, and it declines where the reducer declines — a rekey onto an occupied key is rejected, and discarding for a connection still there and still talking would lose its trailing prose. The gate is two property reads on the hot `STREAM_BATCH` path. Also from the review: - `markConnectionGone` dispatches STATUS_CHANGED directly, so nothing drains the queue, and once the entry reads `disconnected` the out-of-turn guard drops the batch. Pre-existing, but the window is no longer a frame — this change made it up to twelve times as much of a live reply. Flush first. - The stale-snapshot branch now has a test. "Flush, not discard" at the hydrate was argued in a comment and pinned by nothing: swapping in a discard left the whole file green, because the only branch where it matters is the one that leaves `liveMessage` untouched. - Corrected a comment that justified the discard's placement with a case the reducer cannot produce (`CONNECTION_REMOVED` always returns a fresh map, so the no-op early return never fires for it), and one that claimed every dep of the listener effect is a `useCallback(..., [])` now that `dispatch` has deps of its own. Dropped a redundant `act()` around RTL's `cleanup`, which unmounts inside its own. Co-Authored-By: Claude Opus 5 (1M context) --- src/contexts/acp-connections-context.test.tsx | 115 +++++++++++++++++- src/contexts/acp-connections-context.tsx | 74 ++++++----- 2 files changed, 156 insertions(+), 33 deletions(-) diff --git a/src/contexts/acp-connections-context.test.tsx b/src/contexts/acp-connections-context.test.tsx index 98914827ca..fc83d0543e 100644 --- a/src/contexts/acp-connections-context.test.tsx +++ b/src/contexts/acp-connections-context.test.tsx @@ -2535,6 +2535,67 @@ describe("streaming flush window widens with the run it re-renders", () => { } }) + // …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 @@ -2571,6 +2632,52 @@ describe("streaming flush window widens with the run it re-renders", () => { } }) + // 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 — @@ -2588,10 +2695,10 @@ describe("streaming flush window widens with the run it re-renders", () => { }) expect(vi.getTimerCount()).toBe(idle + 1) - act(() => { - cleanup() - }) - expect(vi.getTimerCount()).toBeLessThan(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() } diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index baf1c6db7f..b1735e61c5 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -3434,34 +3434,39 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { const dispatch = useCallback( (action: Action) => { - // A removed key must not leave deltas armed behind it. They would 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 appends a dead turn's prose to a live one. + 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. // - // Done here, at the one place every removal funnels through, rather than - // at the six-plus sites that dispatch these: this is the invariant - // ("no entry, no queue"), and it holds for removals added later too. + // 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. Four actions drop entries today + // (`CONNECTION_REMOVED`, `REMOVE_ALL`, `REKEY_CONNECTION`, + // `DELEGATION_CHILD_DETACH`), three of them conditionally — a rekey onto + // an occupied key is declined, and discarding for a connection that is + // still there and still talking would lose its trailing prose. A fifth + // added later is covered without touching this. // + // The gate is two property reads on the hot `STREAM_BATCH` path: only a + // removal shrinks the map, and a rekey is the one removal that doesn't + // (it swaps one key for another). + if (next.size < prev.size || action.type === "REKEY_CONNECTION") { + for (const key of prev.keys()) { + if (!next.has(key)) discardStreamingKey(key) + } + } + // 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. - // - // Ahead of the reducer, because a no-op removal (the entry is already - // gone) returns early below and would skip the cleanup — and "the entry - // is gone" is exactly when a stray queue must not survive. - if (action.type === "CONNECTION_REMOVED") { - discardStreamingKey(action.contextKey) - } else if (action.type === "REKEY_CONNECTION") { - discardStreamingKey(action.fromKey) - } else if (action.type === "REMOVE_ALL") { - discardStreamingQueues() - } - const prev = storeRef.current.connections - const next = connectionsReducer(prev, action) if (next === prev) return // no change storeRef.current.connections = next @@ -3514,12 +3519,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { } } }, - [ - discardStreamingKey, - discardStreamingQueues, - notifyKeyListeners, - notifyAllKeyListeners, - ] + [discardStreamingKey, notifyKeyListeners, notifyAllKeyListeners] ) // ── setActiveKey ── @@ -5114,7 +5114,9 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { resolveListenerReadyWaiters() 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. @@ -5171,10 +5173,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 ── From 7810182197fde163ab58dcc5c6d287ef9a8fef61 Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 17 Sep 2026 13:56:54 +0800 Subject: [PATCH 6/6] perf(chat): make the queue-discard gate fail safe, and pin the two edits that weren't MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second review pass. Both remaining changes are about the cost of being wrong later rather than about anything broken now. The gate on the discard walk enumerated the removals (a size shrink, plus `REKEY_CONNECTION` for the one removal that swaps rather than shrinks). That is the cheaper read and the wrong failure mode: an action added later that is shaped like a rekey escapes silently, which is exactly how `DELEGATION_CHILD_DETACH` went uncovered. Enumerate the two hot paths instead — `STREAM_BATCH` and `BATCH_TOOL_CALL_UPDATES`, the only actions where a walk over the open connections would be worth avoiding — so a forgotten case costs that walk and not a stray window. `markConnectionGone`'s flush was the riskiest line in the previous commit and the only one nothing pinned: deleting it left the whole file green. It has a test now — Stop pressed on a connection the backend has forgotten, with a reply still mid-window, keeps its last words on screen. Also drops a sentence that claimed three of the four removal actions are conditional (`REMOVE_ALL` has no guard, and `CONNECTION_REMOVED` always runs its delete), and moves the "discard rather than flush" note back above the block it explains. Co-Authored-By: Claude Opus 5 (1M context) --- src/contexts/acp-connections-context.test.tsx | 30 ++++++++++++++++ src/contexts/acp-connections-context.tsx | 35 +++++++++++-------- 2 files changed, 50 insertions(+), 15 deletions(-) diff --git a/src/contexts/acp-connections-context.test.tsx b/src/contexts/acp-connections-context.test.tsx index fc83d0543e..d66ae764a8 100644 --- a/src/contexts/acp-connections-context.test.tsx +++ b/src/contexts/acp-connections-context.test.tsx @@ -2632,6 +2632,36 @@ describe("streaming flush window widens with the run it re-renders", () => { } }) + // 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 diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index b1735e61c5..abea489b41 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -3446,26 +3446,31 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // // 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. Four actions drop entries today - // (`CONNECTION_REMOVED`, `REMOVE_ALL`, `REKEY_CONNECTION`, - // `DELEGATION_CHILD_DETACH`), three of them conditionally — a rekey onto - // an occupied key is declined, and discarding for a connection that is - // still there and still talking would lose its trailing prose. A fifth - // added later is covered without touching this. + // 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. // - // The gate is two property reads on the hot `STREAM_BATCH` path: only a - // removal shrinks the map, and a rekey is the one removal that doesn't - // (it swaps one key for another). - if (next.size < prev.size || action.type === "REKEY_CONNECTION") { - for (const key of prev.keys()) { - if (!next.has(key)) discardStreamingKey(key) - } - } - // 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