feat(streams): rollover block log and atomic stream → session cutover

.changeset/streams-block-log-cutover.md

@@ -0,0 +1,13 @@
+---
+"agents": minor
+"@cloudflare/ai-chat": minor
+"@cloudflare/think": minor
+---
+
+feat(streams): rollover block log and an atomic stream → message cutover; no more stream-buffer sweeps.
+
+The Streams chunk log is now mutable rollover blocks: an append grows the open block row (an UPDATE) until it reaches 256 KB, then opens the next. Same one billed row per append as before, but a stream of thousands of chunks is a handful of rows to delete instead of thousands. Existing `cf_agents_stream_chunks` rows are folded into blocks lazily, one stream at a time on first touch, so startup never reads the whole legacy log; the table is dropped once it is empty.
+
+`writer.close({ commit, discard })` (and `error(reason, { … })`) settles the stream, runs the caller's synchronous writes and deletes the stream's rows in one SQLite transaction. `Session.__DO_NOT_USE_WILL_BREAK__sync().upsert()` is the matching synchronous message write; its `after()` dispatches the change feed and auto-compaction once the transaction commits.
+
+Chat hosts (`AIChatAgent`, `Think`) now persist the finished turn's assistant message inside that cutover: the message, the stream's settlement and the deletion of its temporary rows commit together, so a crash leaves either the live stream (recovery rebuilds the message from it) or the message, never neither. `ResumableStream.start()` reclaims anything a crash left behind. The `_cleanupStreamBuffers` alarm is no longer armed (`cleanupStreamBuffers` and `STREAM_CLEANUP_DELAY_SECONDS` are removed from `agents/chat`; the host callback is kept as a no-op so alarms persisted by earlier versions still resolve).

docs/agents/resumable-streaming.md

@@ -70,7 +70,7 @@ function Chat() {
 - Chunks are batched (every 10 chunks) and flushed to SQLite for performance
 - When a client sends `CF_AGENT_STREAM_RESUME_REQUEST`, the server checks for active streams and responds with `CF_AGENT_STREAM_RESUMING`
 - Stale streams (older than 5 minutes) are cleaned up on restore
-- Stream buffers are garbage collected from a scheduled alarm: completed or errored streams are retained for 10 minutes (a brief reconnect-and-replay grace; the assistant message itself is persisted separately), and abandoned in-flight streams are retained for 1 hour after their last chunk before being reclaimed
+- Stream buffers are deleted in the same transaction that persists the assistant message (the cutover), so a finished turn leaves nothing behind and no cleanup alarm is armed. A buffer a crash left behind is reclaimed when the next stream starts: finished streams immediately, abandoned in-flight streams after 1 hour without a chunk (recovery has until then to rebuild the message from it)
 
 ### Client-side (`useAgentChat`)
 

docs/agents/streams.md

@@ -152,16 +152,51 @@ request's signal aborts the tail when the client disconnects.
 `examples/next/streams` is the end-to-end demo. For other transports,
 `read()`/`readBatches()` remain the raw async iterables to pipe yourself.
 
+## Storage: blocks, and the cutover to a message
+
+Chunks are stored as **rollover blocks**: one row per stream holds chunks
+until it reaches 256 KB, then the next append opens a new row. An append
+is one billed row either way (an UPDATE that grows the block, or the INSERT
+of the next one), the same as a row-per-chunk log, but a stream of
+thousands of chunks is a handful of rows, so deleting it is a handful of
+writes instead of thousands. Replay parses one block at a time.
+
+A stream is temporary: once its content has become something else (a
+session message, a report), its rows are dead weight. The **cutover** ends
+the stream, runs your own synchronous writes, and deletes its rows in one
+SQLite transaction:
+
+```ts
+stream.close({
+  commit: () => sessionSync.upsert(message), // synchronous writes only
+  discard: true // delete the stream's rows in the same transaction
+});
+```
+
+Either the message exists and the stream is gone, or `commit` threw, the
+settle rolled back and the stream is still live. Nothing is left for a
+retention sweep. `error(reason, { commit, discard })` is the same for a
+failed producer. `commit` must not await; a Session handle's
+`__DO_NOT_USE_WILL_BREAK__sync().upsert()` is the matching synchronous
+message write, and returns a `notify()` to dispatch the change feed after
+the transaction commits.
+
+Measured on a real Durable Object (400-chunk chat turn, 10 chunks per
+write): the old log paid 42 rows to write and another 42 to sweep; blocks
+pay 42 to write and 3 to cut over.
+
 ## Chat runs on this
 
 `AIChatAgent` and `Think` store their in-flight turn output here:
 `ResumableStream` (from `agents/chat`) is a thin adapter over Streams that
-packs ~10 wire chunks into one stored segment for write economy, maps
-completion/error onto stream settlement, and decides retention in two
-phases: a coarse cutoff on the stream row's own timestamp, verified
-against the newest chunk so an actively appending stream is never swept.
-Existing `cf_ai_chat_stream_*` tables migrate onto the
-capability automatically. The packing pattern is worth copying for any
+packs ~10 wire chunks into one stored segment for write economy, and ends
+every turn with the cutover: the assistant message, the stream's
+settlement and the deletion of its rows commit in one transaction. Nothing
+is swept on an alarm any more. A stream a crash left behind is either
+still `streaming` (recovery rebuilds the message from it) or reclaimed by
+the next stream start, together with in-flight rows abandoned for over an
+hour. Existing `cf_ai_chat_stream_*` tables migrate onto the capability
+automatically. The packing pattern is worth copying for any
 high-frequency producer: buffer what you already hold synchronously, append
 one packed chunk, and unpack on read — durability is unchanged (nothing is
 held across an await at settlement) and rows written drop by ~an order of
@@ -171,8 +206,8 @@ magnitude versus per-token appends.
 
 Live fanout is in-isolate (sufficient: a Durable Object executes in one
 isolate at a time; reconnecting readers replay from their cursor). Retention
-is explicit `delete()` (chat sweeps its own rows on an alarm); age-based
-sweeping in the capability itself, producer-generation fencing on `open()`,
+is explicit: `delete()`, or the cutover's `discard`; age-based sweeping in
+the capability itself, producer-generation fencing on `open()`,
 and transport helpers extracted from chat's resume protocol are future work.
 The design record is
 [`design/rfc-streams.md`](https://github.com/cloudflare/agents/blob/main/design/rfc-streams.md).

experimental/pi-recovery/src/pi-agent.ts

@@ -33,7 +33,6 @@ import {
   ResumableStream,
   createChatStreams,
   bumpChatRecoveryProgress,
-  cleanupStreamBuffers,
   createChatFiberSnapshot,
   readChatRecoveryProgress,
   recordChatTerminal,
@@ -696,7 +695,7 @@ export class PiAgent extends Agent<Env> {
 
   /** Stream-buffer cleanup alarm target (scheduled by ResumableStream cleanup). */
   async _cleanupStreamBuffers(): Promise<void> {
-    await cleanupStreamBuffers(this._resumableStream, async () => {});
+    this._resumableStream.reclaim();
   }
 
   // ── Inspection surface (server stub RPC → e2e assertions) ───────────────────

experimental/tanstack-recovery/src/tanstack-agent.ts

@@ -48,7 +48,6 @@ import {
   ResumeHandshake,
   buildChatRecoveringFrame,
   bumpChatRecoveryProgress,
-  cleanupStreamBuffers,
   createChatFiberSnapshot,
   pendingChatTerminal,
   readChatRecoveryProgress,
@@ -906,7 +905,7 @@ export class TanStackAgent extends Agent<Env> {
 
   /** Stream-buffer cleanup alarm target (scheduled by ResumableStream cleanup). */
   async _cleanupStreamBuffers(): Promise<void> {
-    await cleanupStreamBuffers(this._resumableStream, async () => {});
+    this._resumableStream.reclaim();
   }
 
   // ── Inspection surface (server HTTP → e2e assertions) ───────────────────────

packages/agents/src/chat/index.ts

@@ -48,9 +48,7 @@ export {
 
 export {
   ResumableStream,
-  cleanupStreamBuffers,
   createChatStreams,
-  STREAM_CLEANUP_DELAY_SECONDS,
   type SqlTaggedTemplate
 } from "./resumable-stream";
 export {