feat(streams): durable incremental output as a Lifecycle capability (agents/streams) - #2173
Open
mattzcarey wants to merge 28 commits into
Open
feat(streams): durable incremental output as a Lifecycle capability (agents/streams)#2173mattzcarey wants to merge 28 commits into
mattzcarey wants to merge 28 commits into
Conversation
One Fibers capability per Lifecycle Object owns named definitions created with fibers.create(name, run). Runs replay from the top on every attempt: journaled step results, durable sleeps with first-deadline authority, per-step retry/timeout policy, a status live gate, cooperative cancellation, and generation-fenced claims. Deadlines live in cf_fiber_runs.next_at and reach the shared physical alarm only through the Lifecycle alarm-contribution model, exactly as the Scheduler does.
FiberHarnessObject drives the capability with real Lifecycle startup, real SQLite, and real platform alarms; instance counters separate real step execution from journal hits to prove replay memoization. Covers acceptance dedup, retry parking, interrupted-attempt reclaim from a seeded dead generation, sleep deadline authority, the status live gate, cancellation, timeouts, divergence, missing definitions, Scheduler alarm coexistence, and the bounded alarm batch.
Replace imperative fibers.create() handles and the startup registry lock with a Scheduler-style constructor definitions map. The map is the registry, rebuilt on every wake, so recovery of in-flight runs is correct by construction — nothing to register at the right moment and no lock to trip over. Runs start with the typed fibers.run(name, input, options); fibers.handle(name) is a pure typed lens scoped to one definition; framework internals attach through an internal composition-root resolver aperture mirroring the Scheduler's callback-name resolver. Also adds the examples/next/fibers example (smoke-tested end to end under wrangler dev) and records the amendment in the RFC.
A definition may pair its handler with a recover callback that owns unclean
interruptions instead of automatic replay. It receives the run input,
metadata, and the interrupted step — name, attempt, stable idempotency key,
and the last checkpoint() the lost attempt wrote through its step attempt
context — and decides replay (immediate or deferred), complete, fail, or
cancel. Clean step failures never reach recovery; the retry policy owns
them. Recovery runs under its own 'recovering' claim with the same
generation fencing and claim-deadline backstop, survives its own
interruption (callbacks must be idempotent), and a throwing recover retries
on exponential backoff with a bounded budget before the run fails. The
internal composition-root resolver accepts { run, recover } entries so
framework definitions (the Think replatform) get the same recovery seam.
… messengers Agent installs the capability automatically as experimental this.fibers. Subclass definitions go on the overridable fiberDefinitions field (rebuilt every wake, resolved lazily so field order never matters); framework definitions attach through a composition-root aperture mirroring the Scheduler's callback resolver; handlers run in the Agent's invocation boundary; and due runs dispatch at the same startup point as the legacy fiber scan — before the user's onStart. Think chat turns, AIChatAgent chat turns, and Think messenger replies now execute on the capability: each live closure runs as one journaled step in the caller's invocation context with checkpoint-backed stash(), and an unclean interruption synthesizes the legacy FiberRecoveryContext and routes through the unchanged _handleInternalFiberRecovery -> ChatRecoveryEngine seam (messenger recovery likewise, with re-entry checkpoints persisted in host storage). The recovery brain did not move, which is why the full think suite passes untouched: agents 1,938/1,938, think 887/887, ai-chat 655/655. Legacy runFiber()/startFiber() user APIs are unchanged and still recovered by their own scan; facet-hosted turns stay on the legacy engine until routed Fibers land.
…ll e2e
Split the workers-pool harness so Fibers is proven both ways:
FiberHarnessObject now installs Fibers as its ONLY capability, and the new
FiberSchedulerCoexistObject installs Fibers and the Scheduler together for
the shared-alarm arbitration test.
Add fibers-capability-eviction.test.ts on the existing e2e kill harness:
real wrangler dev with persisted state, SIGKILL mid-step, restart. Step
executions are recorded in the host's own SQLite, proving completed steps
replayed from the journal without re-executing while the interrupted step
ran again; the { run, recover } variant proves the recovery callback
receives the interrupted step and its last checkpoint() across a real
process death and settles the run by decision.
Rename the new durable-execution capability from Fibers to Tasks: agents/fibers -> agents/tasks, class Tasks, taskDefinitions on Agent, Task* types/errors, task:* events on a new agents:task diagnostics channel, and cf_agents_task_runs / cf_agents_task_steps tables (all unreleased). 'Fiber' now unambiguously means the legacy engine (runFiber/startFiber and their vocabulary, all untouched) — the rename removes the two-meanings problem the constructor-map redesign created, and lines up with the future MCP Tasks adapter story. Fixtures, suites, the kill e2e, docs, the example, and the RFC record all follow; the Agent schema DDL snapshot now includes the capability's tables. Add design/rfc-streams.md (proposed): a Streams capability owning the durable chunk log, cursor, and replay-then-tail reads — the incremental- output half of the pattern the Think migration validated — composed with Tasks through checkpointed cursors and status() evidence, populated by extracting chat's resumable-stream store. Verified on the renamed engine: agents 1,938/1,938, think 887/887, ai-chat 655/655, tasks SIGKILL e2e 2/2.
One Streams instance per Durable Object owns an ordered, durable chunk log
per stream with a monotonic cursor: idempotent open(), synchronous durable
append() that wakes live readers, close()/error() settlement (no-op when
already terminal, so recovery callers stay idempotent), replay-then-tail
read({ from, signal }), and status() reporting state and cursor. Reads are
independent of producer liveness; the capability consumes only storage and
events — no alarm — so it also works on facets.
This is the incremental-output half of the pattern the Tasks migration
validated, composed without coupling: a task step appends to a stream and
checkpoints { streamId, cursor }, and its recover callback reads
streams.status() as durable interruption evidence. Producers that resume
from stream.cursor never duplicate a chunk.
Proven on real DOs in both fixture shapes (StreamHarnessObject standalone,
TaskStreamComposeObject composed, 13 tests) and across a real SIGKILL
(streams-capability-eviction e2e): the chunks appended before death survive
exactly, and recovery finalizes the stream at precisely that cursor. Ships
with examples/next/streams (SSE serving with cursor reconnects),
docs/agents/streams.md, and the accepted design/rfc-streams.md. The chat
resumable-stream migration follows separately with the chat suites as the
parity ratchet.
🦋 Changeset detectedLatest commit: 9f40fdf The changes in this PR will be included in the next version bump. This PR includes changesets to release 4 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 |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
agents
@cloudflare/ai-chat
@cloudflare/codemode
hono-agents
@cloudflare/shell
@cloudflare/think
@cloudflare/voice
@cloudflare/worker-bundler
commit: |
readBatches(streamId, { from, signal, batchSize }) yields non-empty arrays
of consecutive chunks with the same lifecycle as read(): replay yields up
to batchSize chunks per array (default 100), and a live tail yields
everything that accumulated since the last wakeup as one array — so a
consumer paying per write (an SSE flush, an RPC hop, a history append)
pays once per backlog, not once per chunk.
read() now delegates to readBatches with per-chunk abort checks, so the
existing suite exercises the shared core; two new tests pin the batch
boundaries (batchSize slicing from a cursor, one-array-per-wakeup
coalescing on a live tail).
…ility ResumableStream becomes chat's producer-side coalescing and wire-protocol adapter over agents/streams: in-flight turn output lives in the shared durable chunk log (cf_agents_streams / cf_agents_stream_chunks), written through the capability's fenced append — which also wakes live streams.read() consumers and emits agents:stream events — with completion/error mapped onto stream settlement. The packed-segment write policy (~10 wire chunks per stored chunk) stays in the adapter, so write economy is preserved; retention keeps its 10m/1h windows but keys off the stream row's updated_at, so sweeps never scan the chunk table. Legacy cf_ai_chat_stream_* tables (both on-disk generations) migrate wholesale on first construction — an in-flight stream keeps its id, chunks, and last-activity across the upgrade — then are dropped. The adapter's surface is synchronous and constructed before the Lifecycle starts, so it runs on a loud-named internal sync aperture (Streams.__DO_NOT_USE_WILL_BREAK__sync()) whose invariant-bearing writes go through the same private methods as the public API. StreamStatus gains updatedAt (last write activity). AIChatAgent, Think, and the experimental recovery agents install the backing capability as readonly streams (createChatStreams() raises maxChunkBytes for packed segments); the recovery engines' stream-evidence lookups move from raw legacy-table SQL onto adapter methods. A storage-ops benchmark (storage-ops-bench.test.ts, real DO SQLite via total_changes()) pins the cost model: packed adapter writes 440 rows vs the legacy pattern's 240 (the fence per segment) and 4040 for naive per-chunk appends, with retention-sweep reads down from 239 to 40 and no longer proportional to stored chunks. Parity ratchet, assertions unchanged: think 887/887 (+2 react), ai-chat 737/737 and e2e 11/11 (real wrangler SIGKILL recovery on the new store — also fixes the nightly-only hasFiberRows helper stale since the chat turns moved to the Tasks capability), agents 1954/1954.
- open(id, { tag }): an indexed, deliberately non-unique application lookup
key, fixed at creation (a reopen naming a different tag throws — config
conflict, not resume). list({ tag }) composes with the state filter,
newest first, so 'latest stream of this operation' is
list({ tag, limit: 1 })[0]. Part of the initial schema — the tables are
unreleased, so no migration.
- readBatches onUpToDate: fires once when the reader first reaches the
durable tail. Caught-up is distinct from ended — a live stream is up to
date while tailing. Useful to flush replayed UI or flip a live indicator.
- sseResponse(streams, id, { request }): one-call SSE serving. Each chunk's
seq rides the SSE id: field, so a reconnecting EventSource resumes via
Last-Event-ID with zero client code (?from= works too); control events
mark up-to-date and done/error (carrying the recorded reason); heartbeat
comments keep idle proxies alive; request.signal aborts the tail; 404 for
missing streams. examples/next/streams now serves through it.
Producer epoch fencing and sliding-TTL retention stay named follow-ups;
both need design.
Merges the Lifecycle work-queue rework (#2175) and the WebSockets capability extraction (#2169), porting this branch's capabilities to the new pattern: - Tasks no longer implements the removed alarm-contribution surface (getNextAlarm/onAlarm/alarms.rearm). Every non-terminal run's authoritative next_at is mirrored as one Lifecycle queue job (id = run id, so a retime is a same-id push); wakes dispatch through onJob, whose outcome is derived from the run row after execution — the single source of truth that supersedes any same-id push made mid-drive. Startup reconcile mirrors every non-terminal run, which also covers seeded rows. - The maxRunsPerAlarm batching option, the batch harness, and its test are gone: dispatch pacing (due ordering, backlog warnings, the memory-limit breaker) is the queue driver's job now. - The boot-recovery aperture (__DO_NOT_USE_WILL_BREAK__dispatchDueRuns) and Agent's startup call are gone: interrupted runs' mirror jobs are overdue after a crash and re-fire on the post-startup alarm derivation. - Streams needs no port: it has no time-based behavior (appends happen in the producer's invocation; readers wake by append or reconnect), pushes zero jobs, and keeps working on facets. - Test seed helpers mirror seeded runs into cf_agents_jobs and backdate both rows, matching what acceptance does. - Agent installs tasks alongside main's _webSockets in the capability chain; docs and changesets now describe the queue model.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…play
The custom-recovery surface is gone: { run, recover } definitions,
TaskInterruption/TaskRecoveryDecision, the recovering state and its backoff
budget, step checkpoint(), and the engine's recovery claim path (~350
lines). Definitions are plain handlers again. An unclean interruption
replays the handler on the next wake; replay safety comes from step
idempotency keys (external writes deduplicate) and durable evidence read at
the top of the work — a producer that starts at stream.cursor resumes
instead of redoing. task:attempt:interrupted still reports the step a lost
attempt left mid-execution.
Two things made recover unnecessary once the API was used end-to-end: the
Streams capability turned interruption evidence into durable state a
replayed handler reads directly, and the chat replatform showed its one
real consumer — the ChatRecoveryEngine — expresses the same decision as a
branch at handler entry when the live closure is missing. Chat turns and
messenger replies now take exactly that shape: the live path persists its
stash snapshot in host storage; a replay whose closure is gone enters the
unchanged recovery engine with that snapshot plus stream evidence, keyed by
stable run ids (chat_<nonce>, msgr_<nonce>).
Fixtures, e2es, the example, docs, and changesets move to the replay
model; the SIGKILL e2es now prove resume-without-duplication (gapless seq
sequences) rather than finalize-at-cursor. The RFCs record the amendment.
Tasks and Streams are unreleased, so no compatibility surface changes.
…n definition, typed aperture, decomposition
- step.interrupted: the interrupted step ({ name, attempt } | null) is
first-class on the step surface, captured once at claim. The engine, the
test harness, and the e2e handler previously reimplemented the same raw
journal query; all three now read the API, and it is the documented way a
replayed handler branches on interruption evidence.
- The chat-turn Task definition lives once in agents/chat
(createChatTurnTaskDefinition): AIChatAgent and Think wire their
protected internals through a narrow hooks contract instead of carrying
60 identical lines each. The stash fire-and-forget storage writes are
.catch-guarded (previously an unhandled-rejection risk on the token hot
path), and the snapshot-key vocabulary is centralized. Messenger
registration renamed to _registerMessengerReplyTaskDefinition.
- The Streams internal sync aperture is fully typed: the raw exec escape
hatch is gone, replaced by latestRowByTag/deleteMany/importStream/
importChunk, so no SQL crosses the capability boundary. Chat streams
carry their request id as the indexed tag (metadata keeps only the
ownership marker), and the legacy-table migration reads chat's own
tables through the host-supplied sql handle.
- tasks.ts decomposed under 1k lines: store.ts owns the tables (DDL, row
access, fenced writes, snapshot projection) and engine-port.ts builds
the step-engine port. Queue-mirror syncs moved inside the settle
helpers, so a state transition cannot forget its wake.
- resumable-stream.ts decomposed: replay wire-framing extracted to
replay-frames.ts, collapsing three near-identical send loops; the three
latest-stream query variants collapsed onto latestRowByTag.
- sseResponse cancel race fixed: a heartbeat tick racing a client
disconnect no longer throws from the interval, and finish() tolerates a
cancelled controller.
- Fixture seeds updated to the tag column; docs, changesets, and mirrors
updated (tasks.md documents step.interrupted).
The exhaustion e2es exposed a queue-starvation regression from the work-queue port: _chatRecoveryRetry/_chatRecoveryContinue are schedule callbacks — queue jobs — and the replatform made them await the recovered turn inline. A turn is legitimately unbounded (a hanging model stream is capped only by the step timeout), so one stuck recovery dispatch starved every other job on the object: keepAlive stopped, the interrupted turn's own replay-wake never fired, budgets never advanced, and onExhausted never sealed. Legacy dispatch was fire-and-forget runFiber, which never had this property. The callbacks are now split: the public schedule-facing method awaits the bounded pre-turn phase and detaches at the turn boundary via a handoff (Promise.race of the dispatch against a reached-the-turn signal), so a platform transient thrown before the turn still reaches the queue and defers the job (#1730). A platform-class failure after the handoff — when the job has already completed — re-defers itself by rescheduling the same callback (isPlatformFailure, now re-exported through agents/chat). All incident bookkeeping (OOM intercept, budget evaluation, stranded-child reconcile) lives unchanged in the protected *Detached body, which fixtures drive directly when they need settled-state assertions. Verified end to end: the exhaustion e2es seal all three budget kinds again (3/3), think 887/887 (+2 react) including the #1730 deferral and storage-reset pairs, ai-chat 737/737, agents 1957/1957.
… wiring The two design records had accumulated amendments faster than their bodies: rfc-fibers.md was 2,785 lines of pre-implementation proposal in 'Fibers' vocabulary describing recover callbacks, checkpoints, and the alarm-contribution model — all superseded — with the corrections stacked at the bottom. Both files are rewritten as records of the shipped design (~180 lines each): the problem, the shipped API and architecture, an honest 'how the design evolved' section (runtime create() → constructor map; alarm contribution → job queue; recover shipped-then-removed with the reasoning; the checkpoint→cursor contract collapse), alternatives considered, deferred work, and the verification stance. Filenames stay for link stability. docs/agents/tasks.md gets a real fix (the install example declared 'readonly fibers' but installed 'this.tasks'), section reordering so both replay discussions sit together, and current-limits wording aligned with the record. streams.md drops the last 'recovery finalizes' phrasing. The Agent-side Tasks wiring Matt flagged is refactored at the source: Tasks runs onError through the standard runInHostContext boundary itself (hosts pass a plain callback — the hand-rolled runInInvocation scope bag is gone), and the definition-resolver aperture exports its value type (TaskDefinitionResolver, the input-erased TaskCallbacks form), replacing the ReturnType<Parameters<...>> cast gymnastics with one documented cast.
Local 'oxfmt --check' accepts these markdown files while CI's identical check rejects them; write mode settles the canonical form.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
The Agent-wiring refactor removed the last use; oxlint in CI rejects the leftover import.
- sseResponse: a fresh EventSource connection (no Last-Event-ID header) now starts at chunk 0 — Number(null) parsed as 0, which skipped the first chunk and made the ?from= fallback unreachable. - Tasks.handle().cancel() is scoped to its definition, matching get(). - run() rejects a runId/idempotencyKey pair that names two different runs instead of silently joining one of them. - Task wake jobs are namespaced task:<runId> so caller-selected run ids stay inside Tasks' own job-id space. - Documented why fire-and-forget stash writes are safe (Durable Object storage applies same-key operations in issuance order) in the chat turn and messenger reply definitions.
Four named rules now govern the queue (design/lifecycle-work-queue.md): - Job ids are scoped to their owner: push replaces only the owner's own job, and a cross-owner id collision throws instead of silently replacing the other owner's job. - Newer pushes win over drive results: every dispatched job carries a durable in-flight marker, a same-id push or reschedule clears it, and applyOutcome only applies to still-marked jobs — a wake pushed mid-drive can no longer be lost. The memory-limit breaker retimes through an unguarded path because its backoff must land regardless. - Dispatch must be bounded: a dispatch outliving its job's hung timeout warns and emits job:slow_dispatch telemetry. - Platform failures abort the drive loop (existing behavior, now a documented rule), and cross-owner drive order is explicitly unspecified so lanes or fairness can arrive without a contract change.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…e gaps - The drive loop refetches each due job before claiming it: a job replaced by an earlier dispatch in the same alarm cycle dispatches with fresh data, or is skipped when no longer due, instead of being driven and settled from its stale snapshot. - Tasks.onJob is bounded: a queue-driven attempt holds the serial dispatch loop for at most a small budget, then detaches and keeps executing while the isolate lives. The claim backstop remains the durable wake, and a detached settle re-syncs the wake mirror, which supersedes the returned outcome. - The idempotency key is the deduplication authority in run acceptance: a fresh runId alongside a key that names an existing run joins that run (the repeated-delivery pattern), while a run matched by ID with a different stored key still refuses the join. - sseResponse checks for already-aborted signals before wiring abort listeners, so a pre-aborted request ends instead of tailing a live stream forever.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
A backdated push can auto-fire its alarm between two awaited arms (seen on the slower CI runner), letting the victim dispatch before the retimer even existed. Push both jobs far-future, backdate them synchronously in one breath, then rearm.
- A platform-class failure in a task attempt (superseded isolate, memory-limit reset, storage transient) no longer settles the run as failed: the attempt unwinds and rethrows, the claim backstop stays the durable wake, and the next invocation reclaims and replays. - The messenger reply definition durably persists its initial accepted snapshot before delivery begins, mirroring the chat turn definition; an isolate lost mid-answer can always recover through it. - Chat tag lookups filter to cfChat-owned rows: the stream table is shared and tags are non-unique, so the newest row by tag alone could be an unrelated application stream masking chat recovery evidence. The sync aperture's latestRowByTag becomes rowsByTag. - readBatches re-polls after onUpToDate fires instead of sleeping: the callback is application code and a synchronous append inside it fired its wake before any waiter registered — a lost wakeup. - The last aborted waiter removes its stream's empty wake set, so abandoned reads stop accumulating map entries.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…live stream deleteMany (the sweep) and deleteUnchecked can remove a streaming row; a reader parked in the live tail never woke to observe the deletion and stayed pending until an unrelated abort.
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
1 participant
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.
Durable execution for Lifecycle Objects: the Tasks capability (
agents/tasks, replayable durable work), the Streams capability (agents/streams, durable incremental output), and the replatform of chat's in-flight streaming (AIChatAgent, Think) onto them. Supersedes #2168 — this PR carries that work, rebased onto the Lifecycle work-queue rework (#2175) and the WebSockets capability extraction (#2169).Tasks (
agents/tasks)Durable, replayable execution owned by one capability instance:
Replay-from-top with journal memoization, first-deadline-authority sleeps, per-step retry policy with backoff, cooperative cancellation, idempotency keys, and generation fencing against concurrent claims. There is no separate recovery mode: an unclean interruption replays the handler, and replay safety comes from step idempotency keys (external writes deduplicate) and durable evidence read at the top of the work — a producer that starts at
stream.cursorresumes instead of redoing. Atask:attempt:interruptedevent carries the step a lost attempt left mid-execution. (An earlier revision shipped a{ run, recover }callback surface; it was removed after using it end-to-end showed replay-plus-evidence covers every consumer — the RFC records the reasoning.)Queue integration (per #2175): every non-terminal run's authoritative
next_atis mirrored as one Lifecycle queue job (id = run id, so a retime is a same-id push); wakes dispatch throughonJob, whose outcome is derived from the run row after execution — the single source of truth. No capability touches the physical alarm; interrupted runs' overdue mirror jobs re-fire on the post-restart alarm derivation, which is the whole recovery trigger.Agent integration:
Agentinstallsreadonly taskswith adeclare readonly taskDefinitionstyping surface, and Think/AIChatAgent run their chat turns and messenger replies as internal task definitions — the recovery engines unchanged, now entered when a replayed turn finds its live closure gone. The legacyrunFiberengine remains for facet turns and existing users.Streams (
agents/streams)Durable incremental output: an ordered chunk log per stream with a monotonic cursor.
open()is idempotent on the id; reads replay from any cursor then tail live appends, independent of producer liveness;status()reports state and cursor — the durable evidence a replayed Task handler resumes from. Tags are indexed, deliberately non-unique lookup keys ("latest stream of this operation").sseResponseserves the whole lifecycle in one call: each chunk's seq rides the SSEid:field so a reconnectingEventSourceresumes viaLast-Event-IDwith zero client code, withup-to-date/done/errorcontrol events. The capability has no time-based behavior — appends happen in the producer's invocation, readers wake by append or reconnect — so it pushes zero jobs and works on facets.Composition contract: a task step appends to a stream it does not own and starts its loop at
stream.cursor, so a replay after interruption is a resume — the stream itself is the interruption evidence. Neither capability imports the other.The chat/think replatform
ResumableStream— the store behind resumable chat streaming — is now a thin adapter over Streams: chat's in-flight output lives in the shared chunk log (packed ~10 wire chunks per stored segment for write economy), completion/error map onto settlement, retention keys off the stream row'supdated_at(sweeps never scan the chunk table), and legacycf_ai_chat_stream_*tables migrate wholesale — an in-flight stream survives the upgrade — then drop. The wire protocol, replay handshake, and recovery behavior are unchanged; both hosts expose the backing capability asreadonly streams.Storage-op accounting (benchmarked in-suite on real DO SQLite, 20 turns × 100 chunks): legacy pattern 240 rows written, replatformed adapter 440 (the append fence per segment — buying settled-write rejection, the cursor, and
updated_at), naive per-chunk 4040. Retention sweeps drop from 239 rows read to 40 per pass. The benchmark asserts the model so regressions fail CI.Structure
agents/tasksis decomposed:tasks.ts(state machine, ~900 lines),store.ts(tables, fenced writes, snapshot projection),engine-port.ts(the step-engine port),replay.ts(ReplayStep), plus types/errors/serialization/duration.agents/streamsisstreams.ts+sse.ts+ types/errors.step.interrupted({ name, attempt } | null) — handlers branch on it instead of querying engine tables; atask:attempt:interruptedevent mirrors it.agents/chat(createChatTurnTaskDefinition); AIChatAgent and Think wire their protected internals through a narrow hooks contract. Replay wire-framing is one module (replay-frames.ts).tag, and the legacy-table migration reads chat's own tables through the host'ssql.Verification (all real DOs, no fakes)
design/rfc-fibers.md,design/rfc-streams.md,design/alarm-coordination.md(updated for the queue model).Deliberately deferred
open()and sliding-TTL retention in Streams (both need design; retention is one queue job when built).runFiber/startFiberdeprecation — awaits migration evidence.