Repository navigation
perf(think): broadcast transcript deltas at turn boundaries - #2235
Closed
mattzcarey wants to merge 3 commits into
Closed
mattzcarey wants to merge 3 commits into
mattzcarey wants to merge 3 commits into
Conversation
Think's _broadcastMessages() serialised the whole cached transcript (up to the 32 MiB hydration budget) to every socket at least twice per turn: after the incoming persist and after the assistant cutover. Add a cf_agent_chat_messages_delta frame to the shared chat protocol that carries only the messages a turn boundary persisted, plus the epoch of the snapshot it applies to. cf_agent_chat_messages now carries that epoch too. The client (useAgentChat) upserts deltas by id (append for unknown ids) and drops any delta whose epoch differs from the last snapshot it applied, so a delta can never land on the wrong base. Server side, the cache epoch bumps whenever the live cache is wholesale replaced (hydration, branch/regeneration resync, compaction, clear, repair) and every connection records the epoch of the last snapshot it was sent. A delta goes out only when every connection is aligned; otherwise the full snapshot is sent and re-aligns them. Still full: connect/resume, transcript repair, orphan materialisation, stalled/errored partial re-anchors, the post-overflow-retry broadcast (streamed chunks were discarded; only a snapshot reconciles them), and any boundary where an id is not in the cache. Bench (think tests/delta-broadcasts): a turn on a 40-message transcript costs two ~2 KiB deltas instead of two ~86 KiB snapshots.
🦋 Changeset detectedLatest commit: 5c34aab The changes in this PR will be included in the next version bump. This PR includes changesets to release 3 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
Contributor
🟡 agents import sizesMeasured 343 runtime imports as minified bundles. The primary size is gzip; raw minified size is included for diagnosis. An existing import growing by more than 10% is marked red. This report is informational.
Compared Changed imports (103)
All 343 current runtime imports
Reported by agent-think[bot]. |
agents
@cloudflare/ai-chat
@cloudflare/codemode
hono-agents
@cloudflare/shell
@cloudflare/think
@cloudflare/voice
@cloudflare/worker-bundler
commit: |
Review fixes on the delta-broadcast turn boundaries: - Deltas are now negotiated. `useAgentChat` sends `cf_agent_chat_client_capabilities` on every socket open and Think only sends `cf_agent_chat_messages_delta` to connections that declared `transcriptDeltas`. An older `agents` client silently drops the unknown frame — and with it the turn's transcript update — so it keeps getting full snapshots instead. - Rows written while a broadcast was suppressed (`addMessages` mid-turn from a tool `execute`, or `broadcast: false`) are recorded and folded into the next transcript frame. They used to ride the next full snapshot; under deltas they reached no client at all. - `_transcriptNeedsSnapshot` and the deferred rows are only settled by a broadcast that actually reached every connection — one that excluded a live connection still owes it. - Client parity with the snapshot path: the #1778 escape hatch (a later assistant after the protected one clears protection), a delta naming a protected message is held and applied when the stream finishes rather than dropped, and a row matched through a shared `toolCallId` adopts the server's id. - The cache epoch is derived: `_replaceCachedMessages` keeps it when the re-read leaves every row a client holds untouched, so an over-budget windowed transcript (which re-reads storage every incoming turn) stays on deltas instead of falling back to a snapshot each time. - Delta completeness is checked by resolved ids, not row count. Adds a minor changeset for `agents` and `@cloudflare/think`.
Member
Author
|
Closing: Think is moving onto the driver-based ThinkHarness (#2396). This will be revisited there. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
Think._broadcastMessages()serialises the whole cached transcript (up to the 32 MiB hydration budget) to every connected socket, and runs at least twice per turn: after the incoming persist and after the assistant cutover. On a long transcript every turn costs two full snapshots per client.Change
Protocol (shared,
packages/agents/src/chat) — this is a wire change:cf_agent_chat_messages_delta{ epoch, messages }: only the messages a turn boundary persisted (upsert by id; unknown ids append in order).cf_agent_chat_messagesgains an optionalepochnaming the snapshot.cf_agent_chat_client_capabilities{ capabilities: { transcriptDeltas } }:useAgentChatdeclares it on every socket open, and the server sends deltas only to connections that declared it.transcriptDeltas(any olderagentsrelease) keeps receiving full snapshots; older servers never send a delta, and their epoch-less snapshots make the client refuse every delta.Client (
useAgentChat) — stores the epoch of the last snapshot applied and drops any delta whose epoch differs, so a delta arriving before its snapshot, or after the server re-derived its transcript, is never applied out of order. Delta upsert keeps the protected streaming assistant and the cross-tab observed accumulator exactly as the snapshot path does.Server (
think.ts) —_cacheEpochbumps in_replaceCachedMessages(hydration, branch/regeneration resync, compaction, clear, repair). Each connection records the epoch of the last snapshot it was sent (connect, dropped-submit rollback, or broadcast)._broadcastMessagesDelta(ids)sends a delta only when every connection is aligned to the current epoch; otherwise it sends the full frame, which re-aligns everyone.Now deltas: incoming persist (
_handleChatRequest,chat()RPC, programmatic turns),addMessages, and the happy-path assistant cutover (WS and RPC).Still a full snapshot, and why:
_transcriptNeedsSnapshotflag at both comment sites)ai-chat is untouched server-side (still snapshots) but shares the client, so it can adopt the frame later.
Numbers
packages/think/src/tests/delta-broadcasts.test.ts(fails onmain): 40-message transcript, one observer.cf_agent_chat_messagesper turn, ~86 KiB eachcf_agent_chat_messages_delta, one row each (bench asserts < 1/20 of a snapshot)Tests
packages/think/src/react-tests/delta-broadcasts.test.tsx): delta before snapshot dropped; wrong epoch / other instance dropped; matching delta upserts and appends; newer snapshot re-keys the guard; epoch-less snapshot accepts no deltas.think-session,assistant-agent, and the e2ewaitForMessagesBroadcastbarrier to accept delta frames.Suites: think workers 897/897, think react 7/7, ai-chat workers 653/653, agents chat 545/545, agents react 136/136.
Review fixes
addMessages/broadcast: falserows reached no client. Both paths suppress their own broadcast and used to ride the turn's next full snapshot; under deltas that frame names only the assistant row. Suppressed ids are now recorded (_deferTranscriptIds) and folded into the next transcript frame, cleared only once one reached every connection.cf_agent_chat_client_capabilitieson every socket open,_transcriptDeltaCapableper connection on the server, and_everyConnectionAlignedrequires both alignment and capability — a single undeclared client holds everyone to snapshots rather than leaving it frozen.restoreProtectedStreamingAssistantapplies the persisted copy when the stream finishes.toolCallIdnow adopts the server's id instead of keeping the locally minted one._transcriptNeedsSnapshotis no longer cleared by a broadcast that excluded a live connection — that connection is still owed the snapshot._replaceCachedMessageskeeps the epoch when the re-read leaves every row a client holds untouched (identical, or appended at the tail), so an over-budget windowed transcript — which re-reads storage at every incoming turn — stays on deltas instead of falling back to a snapshot every time. The comparison only runs when an aligned, delta-capable connection exists.agentsand@cloudflare/think.New tests: server — old client keeps snapshots, a mixed fleet holds everyone to snapshots until the old client leaves, mid-turn tool
addMessagesrows land on the boundary frame,broadcast: falserows land on the next one, deferred rows stay owed to an excluded connection, and an over-budget windowed transcript stays on deltas. Client — capability declared on open and re-declared on reconnect, server id adopted on atoolCallIdmatch, protected assistant reconciled at stream end, #1778 hatch. PlusparseProtocolMessagecoverage for the new frame.