Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .changeset/recovery-progress-from-streams.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
---
"agents": patch
"@cloudflare/ai-chat": patch
"@cloudflare/think": patch
---

perf(chat): derive the recovery forward-progress marker from the stream log instead of bumping a KV counter per credited chunk. `ResumableStream.progressMarker()` counts durably flushed segments — live streams from their log tails, deleted streams from a retired total folded in as their rows are removed — so the marker stays monotonic across cutover and reclaim, never moves on a reconnect replay or a recovery re-persist, and ignores compaction. A parent forwarding a sub-agent's output credits it explicitly through `creditProgress()`. Nothing is written per chunk any more; one row is written per stream retired. The old KV counter is read once per isolate and seeded into the marker so an in-flight incident never sees it drop, and the hosts mirror the marker's durable part back to that key per stream retired, so a rollback reads no lower either. The Streams sync aperture gains an `onDelete` hook so a chat row deleted through the public capability is retired like any other. `AIChatAgent` now flushes a settled tool result to SQLite the moment it is stored, as `Think` already did, so it is durable before the next packed flush and counts as progress immediately. The work budget's unit is now the durable segment, and `DEFAULT_CHAT_RECOVERY_MAX_WORK` moves from 1000 to 10000 to stay as generous as before for delta-heavy turns. Two cutover fixes ride along: a Think agent-tool child now keeps its stream rows for the parent to tail after completion, as ai-chat already did, and the Streams capability re-derives its legacy-table flag after a rolled-back cutover.
36 changes: 36 additions & 0 deletions design/rfc-chat-recovery-work-budget.md
Original file line number Diff line number Diff line change
Expand Up @@ -281,3 +281,39 @@ time, the breaker stops the platform retry storm and the bill, but the agent is
only usable again once the stored footprint shrinks (the #1710 windowed-hydration
/ media-eviction lever). The breaker bounds the blast radius; it does not shrink
the working set.

## The marker is derived from the stream log

Both progress-keyed bounds — the no-progress window and the work meter — read
one monotonic number. It began as a count of assistant messages, which
compaction could lower (#1628), then became a KV counter bumped per credited
chunk, which was a Durable Object KV get and put on the streaming hot path:
two per tool call, one per text or reasoning segment start, one per five
seconds of deltas, and one per five seconds of forwarded sub-agent output.

The chunk log the hosts already flush to is the durable record of produced
content, so the marker is now read from it (`ResumableStream.progressMarker`):

- a live stream contributes its log tail, a settled stream its final cursor;
- when a stream's rows are deleted — the cutover, the reclaim on the next
`start()`, `clearAll()` — its segments are folded into a retired total
first, in the same synchronous block, so the sum never moves when rows go
away;
- a parent forwarding a sub-agent's output credits one unit explicitly
(`creditProgress()`), throttled as before, because that output goes to
clients and not to the parent's own log.

The unit changed from "credited chunk" to "durable segment": one per ten
packed streaming chunks, one per settled tool result, one per explicit
credit. For delta-heavy turns that is finer than the old counter, which
credited a long text segment once per five seconds, so the default
`maxRecoveryWork` moved from 1 000 to 10 000 to stay as generous in the new
unit while remaining finite. Nothing is written per chunk; one row is written
per stream retired and per explicit credit.

The old KV counter (`CHAT_RECOVERY_PROGRESS_KEY`) is read once per isolate and
folded into its own column of the progress row by max, beside the retired
total, so an incident whose recorded high-water mark predates the upgrade
never sees the marker read lower, a seed can never swallow segments retired
before it, and a repeated seed is idempotent. The key is not deleted: a
rollback resumes from a value no lower than it left.
2 changes: 1 addition & 1 deletion packages/agents/src/chat/lifecycle.ts
Original file line number Diff line number Diff line change
Expand Up @@ -224,7 +224,7 @@ export type ChatRecoveryProgressContext = {
* Configuration for durable chat recovery, which is always enabled for chat
* turns. `true` uses these defaults: `maxAttempts: 10`,
* `stableTimeoutMs: 10_000`, `noProgressTimeoutMs: 300_000` (5 min),
* `maxRecoveryWork: 1000`, and a generic terminal message. Assign an object to
* `maxRecoveryWork: 10000`, and a generic terminal message. Assign an object to
* tune recovery without disabling its durable bookkeeping.
*
* **Apply this as a class field or in the constructor — never assign it in
Expand Down
11 changes: 6 additions & 5 deletions packages/agents/src/chat/recovery-codec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -87,11 +87,12 @@ export interface ProgressCreditThrottle {
}

/**
* The single, host-agnostic rule for crediting recovery forward progress from a
* stored stream chunk — the convergence of what `AIChatAgent` and `Think`
* previously each decided on their own (ai-chat keyed on chunk type only; Think
* keyed on its flush cadence). Both hosts now call this at chunk-store time so
* the bump TIMING is identical:
* The host-agnostic rule for crediting recovery forward progress from a stored
* stream chunk, from when the marker was a counter bumped per chunk. Neither
* host bumps a counter any more: the marker is derived from the stream log
* (`ResumableStream.progressMarker`), so a chunk counts once its segment is
* durably flushed. The rule stays exported for consumers that still keep a
* counter of their own:
*
* - a **milestone** ({@link ChatRecoveryCodec.isProgressChunk}) always credits;
* - **streaming content** ({@link ChatRecoveryCodec.isStreamingContentChunk})
Expand Down
33 changes: 25 additions & 8 deletions packages/agents/src/chat/recovery-incident.ts
Original file line number Diff line number Diff line change
Expand Up @@ -105,10 +105,14 @@ export type ChatRecoveryIncident = {

export const CHAT_RECOVERY_INCIDENT_KEY_PREFIX = "cf:chat-recovery:incident:";
/**
* Durable, monotonic forward-progress counter for recovery budget resets.
* Bumped at production time when new content is streamed, so it reflects
* genuinely new content and is immune to reconnects/re-persists; never
* recomputed from the (compactable) transcript.
* The pre-derivation forward-progress counter: a KV integer bumped per
* credited chunk. The marker is now derived from the stream log
* (`ResumableStream.progressMarker`), and this key is only read — once per
* isolate, to seed the derived marker so it never reads lower than the
* high-water mark an incident recorded before the upgrade. Current code
* writes it only as a mirror of the derived marker's durable part — one put
* per stream retired, none per chunk — so a build rolled back to the
* counter never reads a marker lower than an incident recorded here.
*/
export const CHAT_RECOVERY_PROGRESS_KEY = "cf:chat-recovery:progress";
/**
Expand Down Expand Up @@ -155,8 +159,17 @@ export const DEFAULT_CHAT_RECOVERY_MAX_ATTEMPTS = 10;
* the incident is deleted). A very long agentic turn under heavy interruption
* that legitimately needs more should raise `maxRecoveryWork` (or set it to
* `Infinity` to restore the pre-#1825 unbounded behavior).
*
* The unit is one durable stream segment — about ten packed streaming chunks,
* or one settled tool result, which is flushed on its own — plus one per
* explicit credit for forwarded sub-agent output. The marker is derived from
* the stream log (`ResumableStream.progressMarker`), so that is what it can
* count. The earlier KV counter credited per milestone chunk and per five
* seconds of deltas, a coarser measure of streamed text; 10 000 segments
* (on the order of 100 000 chunks of re-run output) keeps the budget as
* generous as 1 000 credits was for delta-heavy turns, and still finite.
*/
export const DEFAULT_CHAT_RECOVERY_MAX_WORK = 1000;
export const DEFAULT_CHAT_RECOVERY_MAX_WORK = 10_000;
/**
* Tight, OOM-specific retry budget (#1825). A Durable Object memory-limit reset
* (`isDurableObjectMemoryLimitReset`) is usually deterministic — the turn's
Expand Down Expand Up @@ -384,9 +397,13 @@ export async function readChatRecoveryProgress(
}

/**
* Advance the durable recovery-progress counter by one. Called when genuinely new
* content is durably flushed (real, reconnect-immune forward progress); shared by
* `AIChatAgent` and `Think`.
* Advance the KV progress counter by one.
*
* @deprecated Hosts no longer bump a counter per credited chunk: the marker
* is derived from the stream log (`ResumableStream.progressMarker`) and
* explicit credits go through `ResumableStream.creditProgress`. Kept for
* code that still maintains the KV counter; a value written here is folded
* into the derived marker on the next seed.
*/
export async function bumpChatRecoveryProgress(
storage: Pick<DurableObjectStorage, "get" | "put">
Expand Down
Loading
Loading