chore(webapp,core,sdk): upgrade @s2-dev/streamstore to 0.25 and migrate S2 hosts - #4349
chore(webapp,core,sdk): upgrade @s2-dev/streamstore to 0.25 and migrate S2 hosts#4349ericallam wants to merge 25 commits into
Conversation
🦋 Changeset detectedLatest commit: 9186328 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
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
ericallam
changed the title
| async pull(controller) { | ||
| let result: ReadableStreamReadResult<PumpItem>; | ||
| try { | ||
| result = await internalReader.read(); | ||
| } catch (err) { | ||
| controller.error(err); | ||
| return; | ||
| } | ||
| if (result.done) { | ||
| self.options.onComplete?.(); | ||
| controller.close(); | ||
| return; | ||
| } | ||
| controller.enqueue(result.value.part); | ||
| }, | ||
| cancel(reason) { | ||
| self.cancelledByConsumer = true; | ||
| self.options.onComplete?.(); | ||
| internalReader.cancel(reason).catch(() => {}); | ||
| }, | ||
| }, | ||
| { highWaterMark: 0 } | ||
| ); |
There was a problem hiding this comment.
🟡 Stream-finished callback can fire twice when a reader is cancelled mid-read
The "stream finished" callback is invoked again (self.options.onComplete?.() at packages/core/src/v3/apiClient/runStream.ts:319) after it has already been invoked by the cancel path, so a consumer that cancels while waiting for the next record gets told twice that the stream ended.
Impact: Callers that do one-time work on completion (ending a trace span, resolving/cleanup logic) run it twice, producing spurious errors or duplicated side effects.
Mechanism: cancel resolves the in-flight internal read as done, which re-enters the pull completion branch
The returned stream is now pull-driven with highWaterMark: 0, so a consumer read() starts a pull that awaits internalReader.read(). If the consumer then calls reader.cancel() (a common pattern when racing read() against a timeout, e.g. apps/webapp/test/helpers/sessionStream.ts:254-270), the outer cancel(reason) handler fires onComplete (packages/core/src/v3/apiClient/runStream.ts:327) and cancels the internal reader. Cancelling the internal stream resolves the still-pending internalReader.read() with done: true, so the suspended pull continues into the result.done branch and calls onComplete a second time before controller.close() throws on the already-cancelled controller. packages/trigger-sdk/src/v3/streams.ts:385-387 ends an OTEL span in that callback, and it is also part of the public sessions.read({ onComplete }) surface (packages/trigger-sdk/src/v3/sessions.ts:419).
| async pull(controller) { | |
| let result: ReadableStreamReadResult<PumpItem>; | |
| try { | |
| result = await internalReader.read(); | |
| } catch (err) { | |
| controller.error(err); | |
| return; | |
| } | |
| if (result.done) { | |
| self.options.onComplete?.(); | |
| controller.close(); | |
| return; | |
| } | |
| controller.enqueue(result.value.part); | |
| }, | |
| cancel(reason) { | |
| self.cancelledByConsumer = true; | |
| self.options.onComplete?.(); | |
| internalReader.cancel(reason).catch(() => {}); | |
| }, | |
| }, | |
| { highWaterMark: 0 } | |
| ); | |
| async pull(controller) { | |
| let result: ReadableStreamReadResult<PumpItem>; | |
| try { | |
| result = await internalReader.read(); | |
| } catch (err) { | |
| controller.error(err); | |
| return; | |
| } | |
| if (result.done) { | |
| self.notifyComplete(); | |
| try { | |
| controller.close(); | |
| } catch { | |
| /* already closed / cancelled */ | |
| } | |
| return; | |
| } | |
| controller.enqueue(result.value.part); | |
| }, | |
| cancel(reason) { | |
| self.cancelledByConsumer = true; | |
| self.notifyComplete(); | |
| internalReader.cancel(reason).catch(() => {}); | |
| }, | |
| }, | |
| { highWaterMark: 0 } | |
| ); |
Was this helpful? React with 👍 or 👎 to provide feedback.
| "@opentelemetry/sdk-trace-node": "2.7.1", | ||
| "@opentelemetry/semantic-conventions": "1.41.1", | ||
| "@s2-dev/streamstore": "^0.22.10", | ||
| "@s2-dev/streamstore": "^0.25.0", |
There was a problem hiding this comment.
🟡 Published CLI package changed without a release note entry
The command-line tool's package was changed (streaming dependency bumped at packages/cli-v3/package.json:98) without adding the release-note entry the repository requires for any published package change, so this change ships with no user-visible note.
Impact: The CLI release notes will omit this change, and the CLI package will not get a version bump for it.
Repository rule
AGENTS.md ("Changesets and Server Changes") and CONTRIBUTING.md ("Adding changesets") state that any modification to a public package under packages/* requires a changeset. This PR adds changesets for @trigger.dev/core and @trigger.dev/sdk (.changeset/chat-session-caught-up-resume.md, .changeset/chat-agent-preload-oom-recovery.md) but none for trigger.dev (packages/cli-v3), whose @s2-dev/streamstore dependency was bumped from ^0.22.10 to ^0.25.0 (a change that also moves the default S2 hosts).
Prompt for agents
The PR bumps @s2-dev/streamstore in packages/cli-v3 (a published package, npm name `trigger.dev`) but the .changeset/ files only cover @trigger.dev/core and @trigger.dev/sdk. Per AGENTS.md and CONTRIBUTING.md, any change to a public package under packages/* needs a changeset. Add a patch changeset that includes "trigger.dev", with a user-facing one-line description of the streaming-backend dependency bump.
Was this helpful? React with 👍 or 👎 to provide feedback.
| async reconnect(abortSignal?: AbortSignal): Promise<ReadableStream<UIMessageChunk> | null> { | ||
| if (!this.state.started) return null; | ||
| return this.subscribeToSessionStream(abortSignal, { sendStopOnAbort: false }); | ||
| return this.subscribeToSessionStream(abortSignal, { | ||
| sendStopOnAbort: false, | ||
| peekSettled: true, | ||
| }); |
There was a problem hiding this comment.
🔍 Server-side reconnect now opts into the fast-close peek, which can close before an about-to-start turn writes anything
AgentChat.reconnect() is documented generically as "after a disconnect", but it now always sends X-Peek-Settled: 1. The server's peek (apps/webapp/app/services/realtime/s2realtimeStreams.server.ts:411-416) switches to wait=0 whenever the tail is a turn-complete (possibly behind a trim), so a reconnect issued after a message was appended but before the agent writes the turn's first record returns an immediately-closed stream with no records. docs/ai-chat/client-protocol.mdx:653 explicitly warns against sending the header on active-send paths. Mid-turn disconnects are unaffected (the tail is a data chunk), and this mirrors the browser transport, but unlike the browser transport AgentChat has no auto-resubscribe loop, so a caller using reconnect() right after a send would silently get an empty stream.
Was this helpful? React with 👍 or 👎 to provide feedback.
ericallam
changed the title
… server fast-close
d43512f to
9186328
Compare
1 participant
Summary
Upgrades
@s2-dev/streamstorefrom 0.22 to 0.25 and points the S2 client at its current hosts (a.s2.dev/{basin}.b.s2.dev, replacingaws.s2.dev). The host change is the operationally important piece: it tracks where S2 lives now, independent of any feature. It also folds in several chat.agent session-stream reliability fixes and full-stack e2e coverage that came out of this work.Included:
AgentChat.reconnect()now sendsX-Peek-Settled, so a server-to-server reconnect settles on the existing server fast-close instead of waiting out the long poll..outwritable on self-hosted setups that skip S2 access tokens.