Loading workspace insights... Statistics interval
7 days30 daysLatest CI Pipeline Executions
10981ad8 feat: resumable streams — reconnect to in-flight SSE responses via pluggable delivery durability (#955)
* feat: resumable SSE streams via pluggable delivery durability
Add a transport-level StreamDurability seam to toServerSentEventsResponse:
chunks are appended to an ordered log before delivery and each SSE event is
tagged with an opaque adapter-owned id: offset. Reconnects (Last-Event-ID)
and joins (?offset=-1&runId) replay from the log without re-running the
provider. Ships memoryStream (in-core, dev/test) and the new
@tanstack/ai-durable-stream package (Durable Streams protocol adapter).
Client: fetchServerSentEvents now auto-resumes id-tagged streams, de-dupes
replayed prefixes, exposes joinRun(runId), and throws
DurableStreamIncompleteError when a durable run ends with no terminal event
and no forward progress.
Split out of #785 so state persistence and delivery durability land
independently.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01A6Arc9bWdLq1aRRnDRCLMt
* docs(resumable-streams): Cloudflare Durable Streams backend via service binding
Document running against the Durable Streams Cloudflare Workers + DO
backend: same protocol, no new adapter — inject the service binding's
fetch via the adapter's injectable fetch option, or point server at the
deployed Worker URL. Note the DO alarm satisfies the lease/reaper needed
for producer-death terminalization.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01A6Arc9bWdLq1aRRnDRCLMt
* docs(durable-stream): Cloudflare service-binding example; make server optional when fetch is provided
The durableStream adapter is a protocol client, so a Cloudflare Workers +
Durable Objects backend that speaks the same protocol needs no new adapter —
just the injected `fetch` seam. Over a service binding the host is irrelevant
(dispatch routes to the bound Worker by path), so `server` is now optional
whenever `fetch` is supplied and defaults to a reserved `.internal` base.
Passing neither `server` nor `fetch` throws loudly.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* fix(resumable-streams): bound eviction, reconnection, and surface silent failures
Addresses review findings on the resumable-streams PR:
#1 memoryStream never evicted its process-global log Map (unbounded growth).
Completed logs are now swept after a grace window with a hard LRU cap;
active runs are never evicted. Adds MemoryStreamOptions.
#3 memoryStream join/resume of an unknown or evicted run parked forever. A
concrete resume of a missing log now throws; a from-start join bounds the
wait for the first chunk (firstChunkDeadlineMs) instead of hanging.
#2 The client resumable-SSE reconnect loop and the durableStream read loop
were unbounded and backoff-free. The client now throttles between attempts
and caps the total (StreamReconnectLimitError); durableStream caps
consecutive body-read-failure retries. Normal long-poll advancement is
never throttled. Adds reconnect options to both.
#4 Durability terminal-append / close failures are rethrown to the live
consumer but invisible to a replaying joiner. toServerSentEventsResponse
now accepts `debug` to record the real cause server-side via the library's
logger.
Also: durableStream `server` is optional when `fetch` is provided (service
bindings). Docs + changeset updated; unit tests added for each path (timing-
and eviction-based behavior is covered by unit tests rather than the aimock
e2e harness, which can't exercise it deterministically).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* ci: apply automated fixes
* docs(example): add resumable-streams demo to ts-react-chat
New /resumable route pair: api.resumable.ts (memoryStream-backed POST that
appends+tags each SSE event, plus a GET joinRun replay endpoint) and
resumable.tsx (start a run, then join it by run ID — in a second tab or after a
reload — replaying from the durability log without re-running the model). Nav
link added to Header. Kept the shared api.tanchat route untouched.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* docs(resumable-streams): note other durability backends (Electric, etc.)
Clarify that durableStream works with any Durable Streams protocol server, and
that other systems (a Postgres-backed log via Electric, Redis streams, a queue)
can back durability by implementing the four-method StreamDurability interface.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* feat(ai, ai-client): resumable NDJSON + XHR delivery durability
Extend resumable streams (delivery durability) beyond SSE to NDJSON and
the XHR transports.
Server (@tanstack/ai):
- toHttpStream gains an optional getId; when present each NDJSON line is
emitted as an { id, chunk } envelope (NDJSON has no native event id).
Untagged streams stay bare lines, byte-identical to before.
- toHttpResponse gains durability/batch/debug, reusing the same
durableStreamSource as toServerSentEventsResponse.
Client (@tanstack/ai-client):
- Generalize the SSE-only reconnect loop into transport-agnostic
resumableStream(openEventSource, signal, reconnect). Shared line parsers
(linesToSSEEvents/linesToNdjsonEvents) feed fetch (fetchEventSource) and
XHR (xhrEventSource) thunks.
- fetchHttpStream, xhrServerSentEvents, xhrHttpStream are now resumable and
expose joinRun. XHR onerror surfaces StreamReadError so a durable XHR run
can reconnect; StreamReadError message is now transport-neutral.
Tests: NDJSON server durability suite, NDJSON+XHR resumable-transport client
suite, NDJSON arm on the delivery-durability e2e harness + spec.
Docs/skill/changeset updated for NDJSON + XHR.
* fix(ai, ai-client): CR round 1 — durability logging, memory-log leak, CRLF, [DONE] id parity
Round 1 review fixes (7-agent CR):
- HIGH: toServerSentEventsResponse / toHttpResponse constructed the durability
logger only when debug was passed, so terminal-append/close failures were
silently swallowed by default (fully lost on the client-disconnect path).
Now instantiate resolveDebugOption(debug) unconditionally, matching every
other activity (errors category is on by default).
- HIGH: memoryStream.read() called getOrCreateLog before the unknown-run
check, leaving a permanent empty log per unknown/evicted resume — unbounded
growth defeating the eviction logic. Peek with memoryLogs.get; a concrete
offset for an absent run throws without inserting; a from-start join creates
the log for the produce race but deletes it on the first-chunk deadline.
- readStreamLines (fetch) now strips a trailing CR, matching readXhrLines, so
CRLF SSE servers do not miss the [DONE] sentinel.
- Fetch SSE [DONE] synthesis now threads the run's ids (parity with the XHR
xhrSSEParser), so a [DONE]-terminating server that omits ids still yields a
correlated terminal.
- Clarified the ReconnectOptions.maxAttempts comment (counts total lifetime
reconnects, not consecutive no-progress ones).
- Softened the changeset claim: completion terminalizes when the source emits
its own terminal event.
- SKILL.md anti-pattern examples: gpt-4o -> gpt-5.5.
- Tests: fetch NDJSON reconnect test uses reconnect delayMs:0; parseNdjsonEvents
helper mirrors the production !('type' in value) envelope guard.
Call sites cleared:
- responseToSSEEvents: added optional 3rd param fallbackIds; all existing
callers (responseToSSEChunks, fetch connect/joinRun) pass <=3 args, backward
compatible.
- readStreamLines / memoryStream.read: signatures unchanged; behavior-preserving
except the removed phantom insertion (observable throw/reject paths unchanged,
already covered by stream-durability.test.ts).
* fix(ai-client): CR round 2 — reconnect resilience + joinRun id parity
Round 2 confirmation-round fixes:
- resumableStream: a transport drop (StreamTruncatedError/StreamReadError) now
retries whenever an offset is held, even if THAT attempt made no new progress
(a caught-up run whose parked long-poll socket drops, or a proxy that drops
right after replaying the de-duped overlap). The total-attempts ceiling still
bounds a genuine flapper; the per-attempt progress requirement only converted
recoverable drops into hard failures on flaky networks. The clean-end path
stays strict and now documents the invariant it relies on (a durable transport
must not surface an empty long-poll window as a clean end; both shipped
backends honor it).
- xhrServerSentEvents.joinRun now threads { runId } into the [DONE] fallback,
matching fetchServerSentEvents.joinRun (correlation parity).
- e2e parseNdjson + toHttpResponse @param prose aligned (envelope guard;
batch is nested under durability, debug documented).
- docs: durable sources must emit their own terminal; memoryStream is for
replaying completed runs (live mid-stream resume needs a backend whose
producer outlives the delivery socket); qualified producer-death headline as
backend-driven.
Covering test: a reconnect that replays only the de-duped overlap then drops is
retried, not surfaced as an error (connection-adapters-resumable.test.ts).
Call sites cleared: resumableStream catch condition — only relaxed the retry
guard (dropped '&& progressed'), kept StreamReadError/StreamTruncatedError type
gate + lastEventId gate, so a first-attempt failure with no offset still
rethrows (asserted by existing 'does not retry HTTP setup failures' test).
* fix(ai, ai-client): CR round 3 — producer robustness, fetch retry, NDJSON headers, docs
Round 3 confirmation-round fixes (scope widened per request to cover the
pre-existing durability-producer bugs).
Producer (durableStreamSource):
- Flush buffered-but-unflushed chunks to the log before terminalizing on the
abort/disconnect path (previously up to batchSize-1 already-produced chunks
were dropped, so a joiner replayed a truncated prefix). Matches the error path.
- Prefer the real provider error over a generic AbortError when a run both
fails and is aborted, so a joiner sees the true cause.
- Do not rethrow a post-terminal close()/append failure to the live consumer
once a terminal was already forwarded — rethrowing appended a contradictory
RUN_ERROR after RUN_FINISHED on the wire. Late cleanup failures are recorded
server-side via logger.errors instead.
- validateOffset now also rejects offsets with surrounding whitespace (the SSE
client .trim()s the id, so such an offset would not round-trip on reconnect).
Client:
- normalizeConnectionAdapter.send: guard terminal synthesis in the catch so a
missing-id throw can't mask the original error.
- fetchEventSource: wrap a fetch() rejection (offline/DNS/refused) as
StreamReadError so a reconnect retries from the offset, matching XHR; a
first-attempt failure with no offset still surfaces.
- readStreamLines: final decoder.decode() flush so a cut mid-multibyte-char is
reported as truncation.
Server transport:
- toHttpResponse defaults Content-Type to application/x-ndjson + no-cache
(overridable), matching the SSE helper, so intermediaries don't buffer it.
Docs/skill/harness:
- JSDoc @examples use openaiText('gpt-5.5') (chat has no model field) and wrap
durability examples in a POST handler; model ids normalized to gpt-5.5 across
connection-adapters.md + SKILL.md; SKILL sources += resumable-streams; doc
reconnection wording scoped to the clean-end path; e2e X-Run-Id no longer
advertised on reconnect; harness + seen-set comments.
Tests: flush-on-abort, double-terminal-suppression, whitespace-offset rejection,
NDJSON Content-Type, and fetch-rejection retry (+ first-attempt surfacing).
Deferred (low, out of delta subject): fetch body not cancelled on early terminal
return (reverted — the reader-cancel broke mock-reader teardown in ~26
pre-existing tests; XHR-parity nit, durable backends close on terminal anyway).
* fix(ai-client, docs): CR round 4 — doc-accuracy, comment precision, coverage
Round 4 confirmation-round fixes (docs/comments/tests + one defensive guard;
no new production logic bugs were found this round).
Docs (correcting inaccuracies introduced in earlier CR rounds):
- Reconnection-bounding section now states the durable-vs-non-durable
distinction accurately: a transport error retries while an offset is held; a
durable clean end with no progress fails with DurableStreamIncompleteError;
only a non-durable clean end is a completed run. Documents why the asymmetry
is deliberate.
- connection-adapters.md no longer groups xhrServerSentEvents (SSE) under the
NDJSON/toHttpResponse sentence.
- Added a reconnect-safety warning: the client auto-reconnects by re-POSTing, so
non-idempotent POST-handler work must be guarded behind a resume check.
- config.json: dropped the redundant updatedAt on the newly-added overview page.
- changeset: reconnect option applies to all four HTTP adapters, not just
fetchServerSentEvents.
Code:
- readXhrLines.finish() now guards status===0 like enqueueDelta (avoids a bogus
'status: 0' error if loadend fires before load/error/abort).
- Comments: linesToSSEEvents one-id-per-data-event assumption; clarified the
fetch-rejection wrap note.
Tests:
- XHR onerror→reconnect (proves StreamReadError from onerror drives a retry with
Last-Event-ID) and NDJSON provider-throw terminal persistence — closing the
highest-value coverage gaps on the XHR/NDJSON surface.
- delayMs:0 on the reconnecting fetch-SSE tests (speed/consistency).
Deferred (pre-existing / by-design / out-of-delta-subject): reconnect clean-end
asymmetry (correct + documented), abortableIterable listener cleanup, fetch body
cancel on early exit (reverted — broke mock-reader teardown), pump finally-throw
surfacing, SSE persistent-id interop.
* fix(ai, ai-client, docs): CR round 5 — self-inflicted doc/comment staleness + one silent-swallow
Round 5 confirmation-round fixes. No genuine code-logic defects surfaced this
round; the items below are (a) one silent failure introduced by the R3
double-terminal guard and (b) doc/comment staleness introduced by earlier
rounds, plus a pre-existing doc-example hang bug.
Code:
- durableStreamSource: a producer error thrown AFTER a terminal was forwarded
was suppressed by the !terminalForwarded rethrow guard (correct — avoids a
contradictory second terminal) but never logged, so it vanished. Now logged
via logger.errors like the close/terminal-append failures. Covering test added.
Docs/comments (correcting staleness from earlier rounds):
- debug JSDoc (both response helpers) + overview.md prose no longer imply
server-side logging requires ; the errors category is on by default
(R1 change), and debug only routes/raises verbosity.
- toServerSentEventsStream JSDoc documents its getId param (parity w/ toHttpStream).
- overview.md GET join example guards a missing offset and its comment no longer
over-claims 'never iterates the provider' for a bodyless produce path.
- Removed review-artifact comments ('Finding 6', 'the R1 comment claims').
Docs (pre-existing example bug, flagged twice):
- WebSocket subscribe() example drains the queue before honoring (a
burst + close in one macrotask previously dropped queued chunks, incl. a
trailing RUN_FINISHED → client hang) and registers the abort listener once.
Deferred to a follow-up (pre-existing / off NDJSON-XHR subject / documented
design): SSE heartbeat/empty-data frame tolerance; abortableIterable orphan-
promise .catch; joinRun offset=-1 + Last-Event-ID precedence; reconnect
lifetime-ceiling on healthy socket-per-event runs; fetch body-cancel on early
terminal; assorted test-hygiene (shared FakeXhr, delayMs).
* feat(ai-client, docs): reconnect ceiling = consecutive-no-progress (default 5); custom-adapter guide
Addressing review feedback:
- Reconnect ceiling: maxAttempts now bounds CONSECUTIVE reconnects that deliver
no new events (default lowered 1000 -> 5); forward progress resets the counter.
A healthy long run (even a socket-per-event proxy) never approaches it; it
fires only when the run is genuinely stuck. This also resolves the CR finding
that the old total-lifetime ceiling could fail a healthy progressing run.
Ceiling test split into a no-progress-flapper (hits it) + a progress-resets
test (does not).
- stream-to-response: terminalForwarded lint fix (scoped no-unnecessary-condition
disable; the flag is only assigned inside the flush() closure that TS CFA
cannot observe).
Docs:
- New guide docs/resumable-streams/custom-adapter.md: implement the four-method
StreamDurability contract over your own store, the offset/park/terminalize
rules, wiring, and offset branding. Registered in config.json, cross-linked
from the overview.
- chat/connection-adapters: show the GET handler (joinRun) alongside POST.
- overview reconnection-bounding section updated to the new semantics + default.
NOTE: did NOT make memoryStream a silent default (explored per request, then
reverted on review) - durability stays opt-in to avoid shipping an in-process,
single-process-only, per-run-buffering backend to production by default.
advertiseRunId is a local var in the e2e harness route, not public API.
* fix(ai-client, ai-durable-stream): address CodeRabbit review feedback
- SSE id parsing: preserve the opaque offset verbatim (strip only a single
leading space per the SSE spec, no trim, which would mangle a valid offset),
and treat an empty id: as a resume-cursor reset (drop lastEventId + clear the
de-dupe set) rather than a durable empty offset.
- resolveReconnectOptions: reject non-finite / negative maxAttempts and delayMs
up front so a NaN/Infinity ceiling cannot cause unbounded reconnects.
- durable-stream read: throw on non-strictly-increasing record sequences within
a response instead of silently dropping later records.
- durable-stream: new operationTimeoutMs (default 30000) bounds create/append/
close via an AbortSignal so a stalled backend cannot hang delivery or
terminalization; long-poll reads are intentionally excluded.
- e2e delivery-durability spec: document the aimock-policy exemption.
Tests added: empty-id reset + invalid-reconnect-bounds (ai-client), non-monotonic
seq rejection + operation timeout (ai-durable-stream).
Not applied (verified against the code): peer-dep workspace:^ is consistent with
all sibling packages (changing to * would break sherif); memory retention is
already bounded (sweepMemoryLogs + TTL + cap); the throw-in-finally is
intentional aggregation and ESLint-suppressed (repo does not use Biome); batch
is already documented as nested; tests already follow the package tests/ dir
convention.
* ci: apply automated fixes
* docs(resumable-streams): simplify overview to the happy path, split advanced out
- overview.md: rewritten as the 3-step common case (pick an adapter, wrap the
response with POST+GET, client is automatic). Removed em dashes. No longer
makes it look harder than it is.
- advanced.md (new): moved the deep material here — durableStream options,
joinRun (attach-by-id), completion/stop/errors, memoryStream-in-production,
reconnection bounding, offset ownership, Cloudflare, process death, and
delivery-is-not-state.
- joinRun is now documented under Advanced (it is a manual, opt-in API; the
common reconnect-on-drop path needs no client code).
- Scrubbed em dashes from custom-adapter.md and the resumable sections I added
to connection-adapters.md; fixed the custom-adapter process-death link to
point at the advanced page.
- config.json: registered the Advanced page.
* docs(resumable-streams): drop dead chat() placeholder from GET replay handler
The resume path serves entirely from the durability log and never iterates
the source stream, so the chat() call and its replay: threadId were dead
code. Replace with an empty stream and guard that offset is present so a bare
GET does not fall through to the produce path.
* feat(ai): add resumeServerSentEventsResponse / resumeHttpResponse helpers
A resume GET is served entirely from the durability log and never iterates a
producer stream, so the response helpers previously forced callers to fabricate
an empty stream in every GET handler. These helpers take just the durability
adapter, do the replay, and return a 400 when the request carries no resume
offset. Dogfood them in the e2e harness and the example app, and simplify the
docs GET handler to a one-liner.
* ci: apply automated fixes
* docs(chat): use resumeServerSentEventsResponse in the connection-adapters GET example
The resumable-SSE server example still constructed a dead chat() with a
replay: threadId in its GET handler. Replace it with the resume helper.
* refactor(example): resumable demo uses useChat; docs show the reconnect side-effect guard
Rewrite the ts-react-chat resumable route from hand-rolled connection driving
(useState/useRef/drainInto + manual connect/joinRun) to a plain useChat page,
matching the overview doc: the durable route makes reconnect automatic with no
client code. Expand the overview reconnect gotcha into a runnable example that
guards one-time side effects behind durability.resumeFrom().
* ci: apply automated fixes
* fix(ai-client): don't truncate agentic runs at the first RUN_FINISHED
The resumable stream engine returned on the first RUN_FINISHED/RUN_ERROR.
An agent loop emits one RUN_STARTED/RUN_FINISHED pair per turn, so a
tool-calling run carries several terminals in a single response — returning
on the first dropped the tool result and the final answer. This engine drives
every stream (durable and non-durable alike), so it regressed existing
non-durable clients: all tool/agentic/custom-event/structured E2E tests hung
after the first turn while plain single-turn streams passed.
Drain the event source to its natural end (the server closes the response only
when the run is truly complete) and use the terminal flag post-loop to decide
done-vs-reconnect. Restores the pre-durability read-to-close behavior for
non-durable streams; durable behavior is unchanged (single-terminal responses
end right after the terminal, so every resumable unit test still holds).
* fix(ai-client): send durability run id as X-Run-Id header, not a query param
The resumable adapters appended `?runId=<id>` to every POST (useChat always
supplies a runId), rewriting the request URL for all existing clients — not
just durable ones. That broke callers/tests that match the bare endpoint URL
and violated the invariant that a non-durable request is byte-identical to a
plain fetch.
Send the client-chosen run id in an `X-Run-Id` request header instead. The
POST URL is now untouched, so existing clients are unaffected, while a
durability sink still keys its log by the client's run id (memoryStream's
readRunId reads the header first, then falls back to the `?runId` query the
GET join path still uses). Durability remains a transparent add-on.
* ci: apply automated fixes
---------
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Alem Tuzlak <t.zlak@hotmail.com> 2cb3af5c fix(ai, ai-memory): address memory-middleware review findings
Portable record ids
- middleware.ts mints ids via crypto.randomUUID(), which is NOT a bare global
on the declared Node 18 floor (Web Crypto became unflagged only in Node 19+)
and threw ReferenceError there — silently dropping the whole turn's memory in
non-strict mode. Add newRecordId() (real UUID when available, portable
Date.now()+Math.random() fallback via try/catch).
Close silent-failure gaps in the error plumbing
- Wrap the scope resolver + shouldRetrieve (onConfig), the scope resolver
(onAfterToolCall, onFinish), and shouldRemember (persistTurn) so a throwing
user callback emits memory:error/onError and honours strict instead of
escaping the hook and breaking chat in non-strict mode.
- Make emitError defensive (like safeEmit) so a throwing onError handler can't
break chat or mask the original failure.
Honest strict-mode docs
- Persistence runs via ctx.defer; the engine awaits deferred work with
Promise.allSettled and discards results, so a strict-mode rethrow on the
persist path does NOT abort the run. Correct the comments that claimed
otherwise; memory:error is the observable signal in both modes.
Redis: stop silently deleting malformed rows
- loadAllForScope routed malformed JSON through the expired-sweep, permanently
deleting possibly-recoverable rows behind a one-shot console.warn. Leave the
row in place and skip it; warn per-distinct-id (bounded) so ongoing
corruption keeps surfacing without spamming. Skill doc updated.
Docs
- Fix the memory overview's broken ../advanced/observability link (page moved
on main) -> ../getting-started/devtools.
Tests (+4)
- Cross-turn persist->retrieve round trip (the feature's headline behaviour).
- Throwing scope resolver / throwing shouldRemember don't break chat + emit
memory:error.
- Redis malformed row is skipped on read but NOT deleted.
Gates: @tanstack/ai 1186 tests, @tanstack/ai-memory 72 tests, test:types,
test:eslint (0 errors), knip, sherif, test:docs all green.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>