feat(sdk,core): close resumed chat streams promptly when caught up - #4349
feat(sdk,core): close resumed chat streams promptly when caught up#4349ericallam wants to merge 24 commits into
Conversation
🦋 Changeset detectedLatest commit: cab3b03 The changes in this PR will be included in the next version bump. This PR includes changesets to release 27 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 |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughS2 Streamstore dependencies and REST endpoint defaults are updated to use the newer 🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
@trigger.dev/build
trigger.dev
@trigger.dev/core
@trigger.dev/python
@trigger.dev/react-hooks
@trigger.dev/redis-worker
@trigger.dev/rsc
@trigger.dev/schema-to-json
@trigger.dev/sdk
commit: |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
fdcd835 to
f202ccc
Compare
![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.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
Resuming a chat session stream (page reload or reconnect) held the SSE connection open for the whole long-poll window even after every buffered output record had already arrived. The client now detects when it has caught up to the latest output and closes the resumed stream right away, so reconnecting to an idle chat settles immediately. Detection reuses the stream's tail-carrying heartbeat: batch and ping frames now carry the tail, and CaughtUpTracker from @s2-dev/streamstore 0.25.0 turns "last delivered seq + 1 === tail" into a caught-up signal. When the tail is absent (older self-hosted stream backends) the client keeps its previous behavior, so nothing regresses. Also moves to the current S2 hosts that 0.25.0 defaults to.
In local development, Vite's dev-server CORS middleware reflected the request Origin on every Express response, on top of the app's own CORS handling. The two layers produced a duplicated Access-Control-Allow-Origin header, which browsers reject, breaking cross-origin API calls from a separate frontend in dev. Disabling Vite's dev CORS lets the app be the single source of CORS headers. Dev-only; production is unaffected.
caughtUp() is documented to reject when the stream ends before reaching the tail, but cancel(), non-retryable HTTP responses, and the user-abort branches closed or errored the stream without ending the tracker, so a caller awaiting caughtUp() could stay pending forever. End the tracker on every terminal path. Also removes a stray closing tag from the changeset and applies oxfmt formatting.
Two follow-up fixes to the caught-up close, from review of the streaming path: - A resumed stream that only receives a tail-bearing heartbeat (nothing new to deliver) now settles promptly. The client used to wait for the first visible record before it could observe caught-up, so a quiescent reconnect never triggered the fast close. - Caught-up is now reported only after the consumer has actually drained every record up to the tail, not when records are merely decoded. A resumed stream carrying a backlog that reaches the tail delivers all of it before the stream can close, with no dropped records. The decode path now hands records to a demand-driven reader that marks the tail boundary as the consumer pulls past it, and ends tracking at that same point.
The client-protocol reference now describes the tail carried on the heartbeat ping, the caught-up test a hand-rolled client can run against it, and the SSEStreamSubscription's own close-on-caught-up on the reconnect path.
Boots the real webapp plus Postgres, Redis, and s2-lite via testcontainers and drives the session `.out` wire protocol directly: a producer appends records straight to S2 while the client subscribes through the webapp SSE proxy using the real stream-subscription code. Covers basic delivery, multi-turn continuation, resume without duplication, trim and command-record filtering, a quiescent reconnect settling at the tail, and a backlog that reaches the tail delivering every record before the stream closes.
Adds a real-Chromium leg to the session-stream e2e: a page on a different origin than the webapp cross-origin fetches and streams the session `.out` through the proxy, exercising the browser CORS preflight and streaming read that a node-fetch client skips. Runs against the same testcontainer stack.
Download Chromium without `--with-deps` (the apt step exited 100 on the CI runner and aborted the whole e2e job before any test ran) and skip the leg if the browser still can't launch, so a browser hiccup can never take down the wire-protocol suite.
Adds wire-level legs drawn from the ai-chat smoke-test catalog: the `.in/append` client-channel round-trip, the server-side peek fast-close (`X-Session-Settled`) that pairs with the client caught-up close, a 413 on an oversized `.in/append` body still carrying CORS headers, and rejection of a subscribe with an invalid token. Seeds a real Session row so the append path resolves.
Runs the genuine chat.agent run loop in-process against the testcontainer stack (webapp + Postgres + Redis + s2-lite + MinIO): the agent's `.in`/`.out` go through the real webapp Session streams and its snapshots through the real object store, with the model injected as a deterministic MockLanguageModelV3. Turns are driven by appending to `.in` over HTTP and read back through the SSE proxy, so no real LLM or dev worker is involved. Covers a basic turn, multi-turn continuation on one run, and hydrateMessages. Adds two small test-only utilities to make this possible: a `StandardSessionStreamManager` export and a `sessionStreamManager` option on `runInMockTaskContext`, both from `@trigger.dev/core/v3/test`.
Extends the chat.agent e2e harness to cover the run-lifecycle scenarios that previously needed the run-engine and a supervisor: idle-timeout suspend and resume, chat.endRun() continuation, and chat.requestUpgrade(). session.in.wait() suspends a run on a waitpoint whose only task-visible effect is that runtime.waitUntil() eventually resolves with the next .in record. A TestRuntimeManager plus a SessionWaitpointBackend reproduce that in one process: the two waitpoint apiClient calls are stubbed, and the backend opens its own tail on the session channel and resolves the wait with the next record, so the same run() invocation continues in place (matching local-dev warm resume and the task-observable semantics of a deployed CRIU restore). endRun/requestUpgrade exit the run; a small session orchestrator then spawns the next run as a continuation, mirroring the server re-triggering on the next append. It deliberately does not model server-side waitpoint bookkeeping or process checkpoint/restore, none of which are visible to task code on the resume path.
…ation legs Builds on the waitpoint backend to add the run-lifecycle scenarios it unlocks: a HITL tool approval that crosses a suspend/resume boundary (the answer arrives after the run suspends on the idle waitpoint), a run that suspends and resumes across multiple turns, a requestUpgrade that defers the message so the continuation run processes it, and a three-hop endRun chain that restores history across each continuation.
Rounds out the run-lifecycle coverage: onChatSuspend/onChatResume fire around a real suspend/resume, an OutOfMemoryError on the first attempt fails the run the way a machine swap needs and the attempt-2 retry recovers the in-flight message, and a run with no next message times out on the waitpoint and exits. Adds waitpoint-timeout support to the in-process backend (honors the wire timeout, resolves ok:false) so the idle-timeout path is exercised.
A preloaded chat.agent run (started via transport.preload) that retried after an out-of-memory error dropped the message it was processing at crash time. Boot recovery restores the in-flight message into the dispatch queue and advances the .in cursor past it, but the preload branch fired onPreload and waited for a first message without draining that queue, so the recovered message was stranded behind the advanced cursor and the run sat waiting. The preload branch now dispatches a recovered message as the first turn before waiting. Includes a full-stack e2e that reproduces the drop (preloaded run, OOM on attempt 1, in-flight message on .in) and passes with the fix.
Cancelling the returned stream aborted the internal connection, but the retry path only bailed on the caller-provided abort signal, so a consumer cancel was treated as a transient drop and kept reconnecting (up to maxRetries), issuing SSE requests after the consumer was gone. Track consumer cancellation separately, wake any pending backoff, and short-circuit the retry. Adds a regression test. Also from review: drop the unused s2 networkEndpoint field, order the webapp test-server env so the worker-disable vars win over extraEnv as documented, abort the cross-origin browser read on a timer instead of a between-reads deadline, and correct the client-protocol docs (the caught-up check counts the highest raw seq_num including skipped command records; the chat transport closes on caught-up, SSEStreamSubscription only exposes the signal).
The e2e-webapp job timed out at 20 min and E9 (server-peek fast-close) failed. Three causes: - The e2e vitest config ran files in parallel, so several files each booted a full Docker stack (Postgres, Redis, s2-lite, MinIO) plus a webapp process at once and thrashed the runner. Run e2e files serially, matching the repo pnpm test --no-file-parallelism. - The wire test seeded a Session with no triggerConfig.basePayload, so .in/append threw a server ZodError. Seed an empty basePayload. - E9 raw peek fetch had no client timeout, so a peek that did not settle under load hung ~66s until the socket dropped. Bound the fetch with an abort deadline and retry the peek until it settles. Also raises the job timeout to 30 min for the larger suite.
The in-process chat.agent e2e harness runs every leg against the single globalThis API registry. When a leg closed while its run was still parked on a session waitpoint, that run stayed alive into the next leg and its later teardown unregistered the globals out from under the active run. In CI this surfaced as Noop run-metadata and "can only be used from inside task.run()" errors across most legs. close() now aborts the run and awaits full teardown instead of racing a 10s timeout; the abort disables the session-waitpoint backend so any in-flight or later wait() returns at once; and the backend keeps its pending entry until the wait resolves so disable() can abort it. Runs no longer overlap, so the shared globals stay consistent.
The e2e .out collector subscribed once and stopped at the first graceful SSE close. The proxy serves a bounded wait window and then closes; real clients reconnect with Last-Event-ID to keep draining. A pre-populated stream fits one window, so the wire legs passed, but a streaming agent turn spans several windows and a slower runner reliably lands the content past the first close, leaving the collector with empty output. collectSessionOut now re-subscribes from the last seq it saw (deduping) until its predicate holds or the deadline passes, matching how the session manager consumes the channel.
…pped Self-hosted deployments that run S2 with REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS and no configured access token got an empty token back from the session channel initialize call. The SDK's session-stream writer treats an empty access token as missing S2 credentials and never opens the writer, so a chat.agent turn's assistant output was silently dropped: .out stayed empty with no error surfaced. The initialize response now hands back a non-empty placeholder token when access tokens are skipped (S2 ignores it in that mode, but the client requires a token to be present). The session-agent e2e runs the skip-access-tokens path with no token so the regression stays covered.
The server-to-server AgentChat client's reconnect() now sends X-Peek-Settled, so resuming an .out that is already at a turn-complete tail closes promptly instead of holding the full SSE long-poll window. This matches the browser transport's reconnectToStream and gives server-to-server consumers the same fast tail-close. The active send-message path is unchanged (no peek, so it never races a newly-triggered turn's first chunk).
c02115a to
cab3b03
Compare
| if (options?.peekSettled) { | ||
| subscription | ||
| .caughtUp() | ||
| .then(() => { | ||
| if (!sawFirstChunk && !combinedSignal.aborted) { | ||
| internalAbort.abort(); | ||
| } | ||
| }) | ||
| .catch(() => {}); | ||
| } |
There was a problem hiding this comment.
🔴 Reloading a chat while the assistant is mid-answer can cut off the rest of the answer
A resumed chat stream is closed as soon as no further output is pending (internalAbort.abort() at packages/trigger-sdk/src/v3/chat.ts:1760) even when the assistant is only pausing mid-answer, so the remainder of that answer never reaches the page.
Impact: If a user reloads or reconnects during a pause in the assistant's reply (for example while a tool call is running), the reply stays frozen half-finished until they reload again or send a new message.
Why a mid-turn pause looks identical to a settled stream to the caught-up signal
reconnectToStream (packages/trigger-sdk/src/v3/chat.ts:1145-1173) passes peekSettled: true, which now also arms the client-side caught-up close. The server's X-Peek-Settled peek only fast-closes when the tail record is a turn-complete (apps/webapp/app/services/realtime/s2realtimeStreams.server.ts:411-416, 443-470); when the turn is still in flight it deliberately keeps the long-poll open so the next chunk is delivered on the same connection.
The new client signal makes no such distinction. CaughtUpTracker resolves from a bare tail-carrying ping with an empty backlog (see the new test "resolves caughtUp() from a ping tail with an empty backlog (open-at-tail)" in packages/core/src/v3/apiClient/runStream.test.ts:734-745). On a reconnect whose Last-Event-ID is already at the tail — exactly what happens when the agent is pausing between chunks, e.g. executing a tool — S2's heartbeat fires within one round-trip, no visible record is delivered, so sawFirstChunk is still false and the connection is aborted.
After the abort the loop takes the next.done path and closes the consumer stream without a turn-complete; state.isStreaming stays true, activeStreams is cleared in the finally, and nothing re-subscribes automatically, so the remaining chunks of the live turn are never rendered.
A correctness-preserving version of this optimisation needs the same "settled" condition the server uses (tail record is turn-complete, e.g. via the X-Session-Settled response header) in addition to being caught up, rather than treating "nothing pending right now" as "turn finished".
Prompt for agents
In packages/trigger-sdk/src/v3/chat.ts, the reconnect path (reconnectToStream -> subscribeToSessionStream with peekSettled: true) now aborts the SSE connection as soon as SSEStreamSubscription.caughtUp() resolves and no visible chunk has been seen. CaughtUpTracker resolves from a tail-bearing heartbeat ping with an empty backlog, which is indistinguishable from a turn that is merely paused (e.g. the agent is running a tool and has not written its next chunk yet). In that case the client closes the resumed stream, no turn-complete is observed, state.isStreaming remains true, and nothing re-subscribes, so the rest of the in-flight turn is never rendered. The server-side X-Peek-Settled peek is stricter: it only fast-closes when the tail record is a turn-complete control record (see apps/webapp/app/services/realtime/s2realtimeStreams.server.ts #peekIsSettled) and otherwise keeps the long-poll open. Consider gating the client-side caught-up close on evidence that the stream is actually settled — for example only arming it when the response carried X-Session-Settled: true, or requiring that the last record seen at/behind the tail be a turn-complete — so a mid-turn quiescent reconnect keeps the connection open as before.
Was this helpful? React with 👍 or 👎 to provide feedback.
| import type { CaughtUpBoundary } from "@s2-dev/streamstore"; | ||
| import { CaughtUpTracker } from "@s2-dev/streamstore"; |
There was a problem hiding this comment.
🔍 Root-package import of @s2-dev/streamstore lands in browser bundles
runStream.ts is part of the @trigger.dev/core/v3 surface consumed by the browser transports (packages/trigger-sdk/src/v3/chat.ts) and react-hooks. Importing CaughtUpTracker from the @s2-dev/streamstore root brings the whole S2 client entrypoint into that module graph; previously only server-side files (streamsWriterV2.ts, sessionStreamOneshot.ts) touched it. Worth checking whether the package exposes a narrower subpath (or whether tree-shaking actually drops the HTTP client) before this ships to browser bundles.
Was this helpful? React with 👍 or 👎 to provide feedback.
1 participant
Summary
Resuming a chat session stream (page reload, tab refocus, reconnect) is one of the most frequent things a live chat does, and it was doing more work than it needed to. On a resume the client either held the SSE connection open for the entire long-poll window or issued a separate probe request to decide whether the stream had settled, even when every buffered record had already been delivered. This makes the client detect when it has caught up to the latest output and close the resumed stream immediately.
Net effect on the resume path: one network round-trip instead of two, one connection instead of two, and an idle reconnect that settles in tens of milliseconds instead of waiting out the poll window.
Mechanism
S2's per-stream heartbeat already fires the moment a reader crosses the tail. As of
@s2-dev/streamstore0.25.0 both the batch frame and the heartbeatpingcarry the stream tail ({seq_num, timestamp}), so "last delivered seq + 1 === tail" is a reliable caught-up signal with no extra request.SSEStreamSubscriptionfeeds those raw wire signals intoCaughtUpTracker(a public export in 0.25.0, so no vendoring) and exposescaughtUp(). The SDK's resume path awaits that and closes the connection the moment the tracker reaches the tail, replacing the old readiness probe.One correctness note worth calling out: command records (trim, fence) consume a sequence number and count toward the tail, but are never delivered to consumers. The tracker therefore keys off raw wire counts, not post-filter counts, otherwise it would never register as caught up on a stream that ends with a trim. The tracker is also ended on every terminal path (cancel, non-retryable status, user abort, done, auth error, max retries) so a pending
caughtUp()can never leak.Impact
Before writing any of this I ran a standalone red/green benchmark of the current approach against the caught-up primitive, on real cloud S2, taking medians across storage classes. Absolute latencies are RTT-dominated (a local reader talking to cloud S2 has a ~100ms floor); in production the reader sits next to S2, so the durable win is the reduction in round-trips and connections, with latency following.
Express storage class (our default), median:
End to end in a real browser against cloud S2: after a completed turn, a reload settled to ready in about 50ms with no duplicated records, and a follow-up turn streamed cleanly.
Compatibility
Fully backward compatible and feature-detected. When the tail is absent (older self-hosted stream backends that do not emit it yet),
CaughtUpTrackernever reports caught up and the client falls back to the previous readiness behavior, so existing clients and self-hosters are unaffected. The change bumps@s2-dev/streamstoreto 0.25.0 across core, webapp, and cli, and moves to the S2 hosts that version defaults to.Dev CORS fix (bundled)
While testing this end to end, the dev server was returning a duplicated
Access-Control-Allow-Originheader: Vite's dev-server CORS middleware reflects the request Origin on every Express response, on top of the app's ownapiCors, and browsers reject the doubled value. Settingserver.cors: falseinapps/webapp/vite.config.tsmakes the app the single source of CORS headers. Dev-only; production builds do not run the Vite middleware, so they are unaffected.