perf(streams,tasks,chat): cut storage row writes to legacy parity on the streaming hot path

.changeset/lean-stream-writes.md

@@ -0,0 +1,17 @@
+---
+"agents": patch
+---
+
+Cut storage row writes across Streams, the chat adapter, and Tasks — the streaming hot path now writes exactly what the pre-capability chat pattern wrote.
+
+Streams: the append fence is a read instead of a guarded UPDATE (a Durable Object executes one synchronous block at a time, so state-check + tail-read + INSERT is exactly as atomic), removing one stream-row write per append. The stream row is written only at open and settle; settlement stamps the final cursor, and live cursors/liveness derive from the chunk log's tail. `readBatches` termination and the reader liveness checks moved to narrow reads.
+
+Chat adapter: the retention sweep decides abandonment in two phases (coarse row cutoff, then one indexed chunk-tail read per candidate) so an actively appending stream is never swept; the legacy migration imports rows complete (final count and last-activity stamped up front, chunk imports are bare INSERTs — 1+N writes instead of 1+2N); `destroy()` no longer flushes chunks it deletes in the same call; the cleanup alarm no longer scans the table twice; dead `_segmentIndex` state removed.
+
+Tasks: claim refreshes amortize to one row write per half claim-slack of wall time instead of one per step; already-elapsed sleeps journal born-completed in one INSERT; duplicate status messages skip their write; startup reconcile skips job-queue upserts that already match; a parked-run cancel settles in one row write; settle paths only re-sync the wake mirror when their write actually landed.
+
+Replay memory is bounded: the chat adapter's chunk replay iterates the stored log in pages (a generator over paged reads) instead of materializing the whole turn per reconnecting client.
+
+Schema: the hot-write capability tables (stream chunks, task runs, task steps, jobs — none released) are now WITHOUT ROWID. Cloudflare bills index maintenance as rows written, and an ordinary rowid table's PRIMARY KEY is a hidden UNIQUE index — so every chunk append was billing 2 rows despite being one table write. WITHOUT ROWID makes it exactly 1. The stream metadata table deliberately stays a rowid table: rowid is the insertion-order tiebreak that keeps newest-first deterministic for same-millisecond rows, at one billed row per stream open. The task runs table also drops its `(state, next_at)` index, which taxed every claim/refresh/settle write to speed one startup scan.
+
+The in-suite storage-ops benchmark now pins adapter/legacy write parity exactly (12 table rows per 100-chunk turn, ~8.5× under naive per-chunk appends), models the two-phase sweep, and a write-accounting test pins the billed model per statement (a 100-chunk turn bills 14 rows vs the legacy schema's 33).

design/rfc-streams.md

@@ -41,8 +41,12 @@ const status = await this.streams.status("reply:123");
 
 - **Cursors are monotonic sequence numbers** (0-based; `from` is
   inclusive; `status().cursor` is the next to be assigned). The append
-  fence — a count-bump UPDATE that succeeds only while the stream is live —
-  assigns them and rejects writes to settled streams.
+  fence is a read: `#append` is one synchronous block (state check, chunk-log
+  tail read, chunk INSERT), and a Durable Object executes one synchronous
+  block at a time, so the check cannot interleave with a settle or another
+  append. Settled streams reject writes exactly as before, but an append
+  costs one row write (the chunk), not two — the stream row is written only
+  at open and settle, where settlement stamps the final cursor.
 - **`open()` is idempotent on the id**: reopening a live stream returns a
   writer at its cursor; reopening a settled stream throws; settling twice
   is a no-op so recovery callers stay idempotent.
@@ -99,8 +103,10 @@ duplicate-free sequence.
 `AIChatAgent` and `Think` — is a thin adapter over this capability:
 producer-side coalescing (~10 wire chunks packed per stored segment, for
 storage-op economy), the chat wire protocol and replay handshake, and
-retention policy (10-minute completed grace, 1-hour abandoned window keyed
-off the stream row's `updated_at`, swept on chat's own schedule). Chat
+retention policy (10-minute completed grace; a 1-hour abandoned window
+decided in two phases — a coarse cutoff on the stream row's `updated_at`,
+then one indexed chunk-tail read per candidate to confirm the producer
+really stopped — swept on chat's own schedule). Chat
 streams carry their request id as the indexed `tag`; legacy
 `cf_ai_chat_stream_*` tables migrate wholesale on first construction —
 an in-flight stream survives the upgrade — then are dropped. The adapter
@@ -112,10 +118,17 @@ public API, so live readers and diagnostics observe chat streams like any
 other stream.
 
 Storage-op accounting (benchmarked in-suite on real DO SQLite): the packed
-adapter writes within 2× of the legacy pattern (the fence per segment,
-buying settled-write rejection, the cursor, and `updated_at`), ~9× under
-naive per-chunk appends, and retention sweeps read only stream rows —
-down ~6× and no longer proportional to stored chunks.
+adapter writes exactly the legacy pattern's rows — one stream-row insert,
+one chunk insert per segment, one settle update — since the read fence
+removed the per-segment row write; ~8.5× under naive per-chunk appends,
+and retention sweeps read the stream rows plus at most one indexed
+chunk-tail row per stale live candidate — down ~6× from the legacy
+correlated-subquery shape and no longer proportional to stored chunks.
+In _billed_ terms the gap is wider still: Cloudflare counts index
+maintenance in `rowsWritten`, the chunk table is WITHOUT ROWID so a chunk
+append bills exactly one row, and the legacy schema's per-chunk hidden-PK
+and stream indexes billed three per chunk insert (14 vs 33 billed rows
+for a 100-chunk turn — pinned by the write-accounting test).
 
 ## How the design evolved
 
@@ -132,6 +145,33 @@ cursor })` and a `recover` callback read `status()` as evidence. When
    protocol's own `Last-Event-ID` rather than inventing an offset header.
 3. **The chat replatform landed in the same PR** rather than as a
    follow-up, with the chat suites as the parity ratchet.
+4. **The append fence moved from a guarded UPDATE to a read, and live
+   authority moved from the stream row to the chunk log.** The original
+   fence was a count-bump UPDATE — one extra row write per append buying
+   settled-write rejection, the cursor, and `updated_at`. All three are
+   derivable: the isolate's single-threaded execution makes a synchronous
+   read-check-insert exactly as atomic, the chunk log's tail is the live
+   cursor and liveness signal (rows read cost ~1/1000th of rows written on
+   DO SQLite), and settlement stamps the row exact for terminal reads.
+   While a stream is live its row's `chunk_count`/`updated_at` are
+   deliberately stale; every consumer derives through one helper
+   (`Streams.#tail`), and the storage-ops benchmark pins the resulting
+   write parity with the pre-capability chat pattern.
+5. **The tables went WITHOUT ROWID once the billed metric was measured.**
+   Cloudflare bills `rowsWritten`, which counts index maintenance — and an
+   ordinary rowid table's PRIMARY KEY is a hidden UNIQUE index, so every
+   chunk INSERT was billing 2 rows while `total_changes()` (the parity
+   benchmark's metric) reported 1. WITHOUT ROWID makes the PK the table:
+   a chunk append bills exactly one row, pinned by the write-accounting
+   test. The same treatment covers the task run/step and job-queue
+   tables. The stream _metadata_ table is the deliberate exception: it
+   stays a rowid table because rowid is the insertion-order tiebreak that
+   keeps "newest first" deterministic when same-tag rows share a
+   created_at millisecond (ids are random nanoids), and its hidden-index
+   cost lands once per stream open, never per chunk. The task runs table
+   dropped its `(state, next_at)` index — a per-write tax on every claim,
+   refresh, and settle, paid only to speed the startup reconcile's one
+   scan.
 
 ## Alternatives considered
 

docs/agents/streams.md

@@ -157,8 +157,10 @@ request's signal aborts the tail when the client disconnects.
 `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 reads `updated_at` as the
-retention signal. Existing `cf_ai_chat_stream_*` tables migrate onto the
+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
 high-frequency producer: buffer what you already hold synchronously, append
 one packed chunk, and unpack on read — durability is unchanged (nothing is

packages/agents/src/chat/resumable-stream.ts

@@ -9,7 +9,8 @@
  * - Chunk buffering (packed segments — batched writes for storage-op economy)
  * - Stream lifecycle (start, complete, error) mapped onto Streams settlement
  * - Chunk replay for reconnecting clients (framing in `replay-frames.ts`)
- * - Stale stream cleanup (row-level retention, no chunk-table scans)
+ * - Stale stream cleanup (row-level retention; at most one indexed
+ *   chunk-tail read per stale live candidate, never a chunk-table scan)
  * - Active stream restoration after agent restart
  * - One-time migration of legacy `cf_ai_chat_stream_*` tables
  *
@@ -40,6 +41,11 @@ const CHUNK_BUFFER_MAX_SIZE = 100;
  * segment.
  */
 const SEGMENT_MAX_BYTES = 512_000;
+/**
+ * Stored segments per page when replaying a stream's chunk log. Bounds
+ * replay memory to one page of segment bodies rather than the whole turn.
+ */
+const REPLAY_PAGE_SEGMENTS = 10;
 /** Default cleanup interval for old streams (ms) - every 10 minutes */
 const CLEANUP_INTERVAL_MS = 10 * 60 * 1000;
 /**
@@ -60,9 +66,11 @@ const COMPLETED_RETENTION_MS = 10 * 60 * 1000;
  * Generous relative to {@link COMPLETED_RETENTION_MS}: an interrupted turn must
  * have ample time to be resumed by a reconnecting client or healed by task
  * replay before its buffer is reaped. Only a stream that has produced no
- * chunk for this long is treated as truly dead. Keyed off last activity (the
- * stream row's `updated_at`, bumped by every append) so a long but still-active
- * stream is never swept mid-flight.
+ * chunk for this long is treated as truly dead. Last activity is decided in
+ * two phases — a coarse cutoff on the stream row's `updated_at` (stamped at
+ * open, not per append), then one indexed read of the newest chunk's
+ * timestamp for rows past it — so a long but still-active stream is never
+ * swept mid-flight.
  */
 const ABANDONED_STREAM_RETENTION_MS = 60 * 60 * 1000;
 /** Shared encoder for UTF-8 byte length measurement */
@@ -168,9 +176,6 @@ export type SqlTaggedTemplate = {
 export class ResumableStream {
   private _activeStreamId: string | null = null;
   private _activeRequestId: string | null = null;
-  /** Monotonic segment ordering index — the backing stream's cursor. */
-  private _segmentIndex = 0;
-
   /**
    * Whether the active stream was started in this instance (true) or
    * restored from SQLite after hibernation/restart (false). An orphaned
@@ -269,12 +274,18 @@ export class ResumableStream {
           createdAt
         );
 
+        // The row is imported complete: final chunk count and last-activity
+        // timestamp up front, because importChunk is a bare log INSERT that
+        // never touches the stream row. A terminal row must carry its exact
+        // cursor at rest (nothing stamps it later), and a live row's
+        // updated_at seeds the sweep's coarse cutoff until real appends
+        // resume.
         this.ops.importStream({
           streamId,
           state,
           tag: String(row.request_id),
           metadata,
-          chunkCount: 0,
+          chunkCount: chunkRows.length,
           createdAt,
           updatedAt: closedAt ?? lastChunkAt,
           closedAt
@@ -290,9 +301,6 @@ export class ResumableStream {
           } catch {
             // Opaque body string.
           }
-          // importChunk bumps updated_at to each chunk's own timestamp, so
-          // after the loop the row's last activity is the newest chunk's —
-          // exactly the legacy sweep's semantics.
           this.ops.importChunk(streamId, value, Number(chunk.created_at));
         }
       }
@@ -342,7 +350,6 @@ export class ResumableStream {
     const streamId = nanoid();
     this._activeStreamId = streamId;
     this._activeRequestId = requestId;
-    this._segmentIndex = 0;
     this._isLive = true;
     this._activeIsContinuation = options.continuation ?? false;
 
@@ -376,7 +383,6 @@ export class ResumableStream {
     this.ops.settle(streamId, "completed", null);
     this._activeStreamId = null;
     this._activeRequestId = null;
-    this._segmentIndex = 0;
     this._isLive = false;
     this._activeIsContinuation = false;
 
@@ -394,7 +400,6 @@ export class ResumableStream {
     this.ops.settle(streamId, "errored", null);
     this._activeStreamId = null;
     this._activeRequestId = null;
-    this._segmentIndex = 0;
     this._isLive = false;
     this._activeIsContinuation = false;
   }
@@ -478,7 +483,7 @@ export class ResumableStream {
           : chunks.map((chunk) => chunk.body);
 
       try {
-        this._segmentIndex = this.ops.append(streamId, segment) + 1;
+        this.ops.append(streamId, segment);
       } catch {
         // The stream settled or was deleted while chunks were buffered (a
         // late writer after markError/cleanup); the chunks are dropped.
@@ -490,13 +495,23 @@ export class ResumableStream {
 
   // ── Chunk replay ───────────────────────────────────────────────────
 
-  /** Stored chunk bodies for one stream, packed segments expanded, in order. */
-  private _storedBodies(streamId: string): string[] {
-    const bodies: string[] = [];
-    for (const row of this.ops.readAll(streamId)) {
-      bodies.push(...unpackSegment(row.chunk));
+  /**
+   * Stored chunk bodies for one stream, packed segments expanded, in order.
+   * A generator over paged reads, so replaying a large turn holds one page
+   * of segments in memory instead of the whole stored stream; iteration is
+   * synchronous end to end (WebSocket sends don't await), so the pages see
+   * a consistent log.
+   */
+  private *_storedBodies(streamId: string): Generator<string> {
+    let next = 0;
+    for (;;) {
+      const rows = this.ops.readChunks(streamId, next, REPLAY_PAGE_SEGMENTS);
+      for (const row of rows) {
+        next = row.seq + 1;
+        yield* unpackSegment(row.chunk);
+      }
+      if (rows.length < REPLAY_PAGE_SEGMENTS) return;
     }
-    return bodies;
   }
 
   /**
@@ -685,8 +700,6 @@ export class ResumableStream {
       // replayed after hibernation still carries `continuation: true` on
       // its frames (#1733).
       this._activeIsContinuation = row.chat.isContinuation === 1;
-      // Resume the segment ordering index past the stored cursor.
-      this._segmentIndex = row.chunk_count;
     }
   }
 
@@ -700,17 +713,19 @@ export class ResumableStream {
     this.ops.deleteMany(this._chatRows().map((row) => row.stream_id));
     this._activeStreamId = null;
     this._activeRequestId = null;
-    this._segmentIndex = 0;
     this._activeIsContinuation = false;
   }
 
   /**
    * Remove all chat stream data (called on destroy). The backing tables
    * belong to the Streams capability and are shared with other producers,
-   * so this deletes chat's rows rather than dropping tables.
+   * so this deletes chat's rows rather than dropping tables. Buffered
+   * chunks are dropped (clearAll resets the buffer), not flushed: they
+   * belong to a chat-owned stream this very call deletes, so writing them
+   * first would only pay row writes for rows that die in the same
+   * synchronous block.
    */
   destroy() {
-    this.flushBuffer();
     this.clearAll();
   }
 
@@ -719,10 +734,13 @@ export class ResumableStream {
    * gate used by {@link _maybeCleanupOldStreams}. Intended to be driven by an
    * alarm so idle/hibernated chat DOs still reclaim buffers even when no
    * further stream ever completes to trigger the lazy path.
+   *
+   * @returns How many chat stream rows survive the sweep — the re-arm
+   * signal, so the alarm body needs no second scan of the table.
    */
-  cleanup(now: number = Date.now()): void {
+  cleanup(now: number = Date.now()): number {
     this._lastCleanupTime = now;
-    this._sweepOldStreams(now);
+    return this._sweepOldStreams(now);
   }
 
   /**
@@ -750,19 +768,28 @@ export class ResumableStream {
    *  different retentions: a completed buffer is redundant with the persisted
    *  message and needs only a brief replay grace, whereas an in-flight buffer
    *  must outlive resume/replay before it is presumed dead. Abandonment is
-   *  keyed off the stream row's `updated_at` (bumped by every append), so the
-   *  sweep never scans the chunk table. */
-  private _sweepOldStreams(now: number) {
+   *  decided in two phases: the stream row's `updated_at` (stamped at open,
+   *  not per append — appends write only the chunk log) is the coarse
+   *  cutoff, and only rows past it pay one indexed read of the newest
+   *  chunk's timestamp to confirm the producer really stopped. An actively
+   *  appending stream is never swept, and a quiet sweep still reads no
+   *  chunk rows at all.
+   *  @returns How many chat stream rows survive the sweep. */
+  private _sweepOldStreams(now: number): number {
     const completedCutoff = now - COMPLETED_RETENTION_MS;
     const abandonedCutoff = now - ABANDONED_STREAM_RETENTION_MS;
-    const reclaimable = this._chatRows()
+    const rows = this._chatRows();
+    const reclaimable = rows
       .filter((row) =>
         row.state === "streaming"
-          ? row.updated_at < abandonedCutoff
+          ? row.updated_at < abandonedCutoff &&
+            (this.ops.lastChunkAt(row.stream_id) ?? row.updated_at) <
+              abandonedCutoff
           : (row.closed_at ?? row.updated_at) < completedCutoff
       )
       .map((row) => row.stream_id);
     this.ops.deleteMany(reclaimable);
+    return rows.length - reclaimable.length;
   }
 
   // ── Test helpers (matching old AIChatAgent test API) ────────────────
@@ -776,7 +803,7 @@ export class ResumableStream {
   getStreamChunks(
     streamId: string
   ): Array<{ body: string; chunk_index: number }> {
-    return this._storedBodies(streamId).map((body, chunk_index) => ({
+    return [...this._storedBodies(streamId)].map((body, chunk_index) => ({
       body,
       chunk_index
     }));
@@ -827,9 +854,10 @@ export class ResumableStream {
   }
 
   /**
-   * Append a chunk to a stream dated `ageMs` in the past. Used to exercise the
-   * last-activity sweep threshold: a long-running streaming row with a *recent*
-   * chunk must survive even when its start time is older than the cutoff.
+   * Append a chunk to a stream dated `ageMs` in the past. Used to exercise
+   * the sweep's phase-2 verification: a long-running streaming row with a
+   * *recent* chunk must survive even when its row `updated_at` (stamped at
+   * open, not per append) is older than the coarse cutoff.
    * @internal For testing only
    */
   insertChunkAt(streamId: string, body: string, ageMs: number): void {
@@ -850,11 +878,10 @@ export class ResumableStream {
  * `@internal`
  */
 export async function cleanupStreamBuffers(
-  stream: Pick<ResumableStream, "cleanup" | "hasReclaimableStreams">,
+  stream: Pick<ResumableStream, "cleanup">,
   rearm: () => Promise<void>
 ): Promise<void> {
-  stream.cleanup();
-  if (stream.hasReclaimableStreams()) {
+  if (stream.cleanup() > 0) {
     await rearm();
   }
 }