Uh oh!
There was an error while loading. Please reload this page.
Poc 2 - #3
Merged
Merged
Conversation
4 tasks
tombeckenham added a commit
that referenced
this pull request
Jul 17, 2026
…ent 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>
AlemTuzlak added a commit
that referenced
this pull request
Jul 20, 2026
…uggable 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>
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 freeto 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.
No description provided.