feat(realtime): start-from-latest streams and a useSessionStream hook - #4811
Merged
ericallam merged 10 commits intoAug 28, 2026
Merged
Conversation
Realtime stream subscribers could only replay from the beginning, so a new or reconnecting subscriber always re-read the full history. Expose the backends' native start-from-tail on the live subscribe path. useRealtimeStream, streams.read, and fetchStream gain `from: "latest"` (seed the current tail, then live-tail), and useRealtimeStream gains `maxParts` to bound the accumulated parts array. The client sends the start position only on the first connect and resumes from the last record it saw on reconnect or remount, so nothing is replayed or missed. S2 maps this to tail_offset, Redis to the $ special id.
… refresh Folds two related changes into the realtime stream work. useSessionStream reads one channel of a session's realtime stream (out by default, or in), accumulating records with automatic resume from the last record seen. It is read-only; useSession is reserved for two-way (read and write) communication. Realtime stream subscriptions can refresh an expired access token and reconnect once, via an optional refreshAccessToken on the client configuration and the React hooks. Behavior is unchanged when no refresher is supplied: auth errors stay terminal.
useRealtimeStream now takes a lastEventId option and returns the lastEventId of the last part seen, so a caller can persist the cursor (for example across a page reload) and resume with no replay and no gap. A new onParts callback delivers each throttled batch of parts with their event ids.
…onStream useSessionStream can start at the current tail (from: "latest"), bound the retained records (maxRecords), and report each throttled batch of records with their event ids (onRecords, replacing the per-record onRecord). The session SSE route reads the start-position header and maps it to the S2 tail start, matching the run-stream path.
🦋 Changeset detectedLatest commit: c493f90 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 |
Contributor
|
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:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
Included review availability: Your plan provides up to 10 included reviews per hour; 8 remain after this review. 📜 Recent review details⏰ Context from checks skipped due to timeout. (51)
🧰 Additional context used📓 Path-based instructions (12)Always import from `@trigger.dev/sdk`. Never use `@trigger.dev/sdk/v3` or deprecated `client.defineJob`.📄 CodeRabbit inference engine (AGENTS.md) Files:
Never use `request.signal` to detect client disconnects. Use `getRequestAbortSignal()` from `app/services/httpAsyncStorage.server.ts`, which is wired to Express response close events.📄 CodeRabbit inference engine (apps/webapp/CLAUDE.md) Files:
For dashboard changes, visually verify the running Remix app with Chrome DevTools MCP, using snapshots, screenshots, interaction, and console-message checks as appropriate.📄 CodeRabbit inference engine (apps/webapp/CLAUDE.md) Files:
**Prefer static imports over dynamic imports.** Only use dynamic `import()` when:📄 CodeRabbit inference engine (AGENTS.md) Files:
Add crumbs as you write code — not just when debugging. Mark lines with📄 CodeRabbit inference engine (AGENTS.md) Files:
Always import from `@trigger.dev/sdk`. Never use `@trigger.dev/sdk/v3` (deprecated path alias)📄 CodeRabbit inference engine (packages/trigger-sdk/CLAUDE.md) Files:
Use zod for validation in packages/core and apps/webapp📄 CodeRabbit inference engine (.github/copilot-instructions.md) Files:
In the Trigger.dev SDK (packages/trigger-sdk), prefer isomorphic code like fetch and ReadableStream instead of Node.js-specific code📄 CodeRabbit inference engine (.github/copilot-instructions.md) Files:
Access environment variables through the `env` export of `env.server.ts` instead of directly accessing `process.env`📄 CodeRabbit inference engine (.cursor/rules/webapp.mdc) Files:
Use function declarations instead of default exports📄 CodeRabbit inference engine (.github/copilot-instructions.md) Files:
Use types over interfaces for TypeScript📄 CodeRabbit inference engine (.github/copilot-instructions.md) Files:
When creating or editing OTEL metrics (counters, histograms, gauges), ensure metric attributes have low cardinality by using only enums, booleans, bounded error codes, or bounded shard IDs📄 CodeRabbit inference engine (.cursor/rules/otel-metrics.mdc) Files:
🔇 Additional comments (3)
WalkthroughRealtime streams now support refreshed credentials after rejected connections and latest-position subscriptions. Server adapters interpret the latest-start header and select tail reads for Redis and S2 streams. Core APIs and the SDK forward stream-position options. React hooks persist cursors, limit accumulated parts or records, and deliver batched callbacks. A new Merge Risk: 🟡 Moderate · up to This PR adds latest-position streaming, cursor resume, bounded buffering, and token-refresh reconnects. At the current head, unresolved edge cases can replay or skip events, restore stale cursors, interfere with subscription cleanup, or exceed the requested memory bound, so the change is not merge-ready without fixes or explicit owner acceptance. 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Full details: Description checkExplanation The description is detailed and directly covers the main changes, usage examples, compatibility behavior, token refresh, and end-to-end testing. It does not include the template headings or checklist, but it provides equivalent summary, testing, and changelog information. ✨ Finishing Touches 💡 2📝 Generate docstrings 💡
🛠️ Fix failing CI checks 💡
🧪 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.
Document the from/maxParts/maxRecords/lastEventId/onParts/onRecords options on useRealtimeStream and useSessionStream, add the session-stream React hook page, from: "latest" on streams.read(), and refreshAccessToken on the realtime hooks.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
Address review feedback on the realtime stream hooks: - useRealtimeStream clears its resume cursor when the stream identity (id, runId, streamKey) changes, so a new stream no longer inherits the previous cursor. - useSessionStream resumes from its persisted cursor on remount, matching useRealtimeStream and its documented behavior. - Both hooks update the parts ref inside the throttle flush so back-to-back flushes build on the latest batch, not a stale one. - Clarify docs and the changeset: from "latest" starts at the current tail (the latest record, then live updates); older servers fall back to a full replay.
Two more review fixes:
- startIndex: 0 is a valid start position. useRealtimeStream and streams.read
now check startIndex !== undefined instead of treating 0 as falsy, so
{ startIndex: 0 } starts from the beginning rather than falling through to
from: "latest".
- Docs: scope the localStorage resume-cursor keys by stream identity so a
component that changes its resource does not load another stream's cursor.
@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: |
ericallam
marked this pull request as ready for review
August 28, 2026 10:09
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
![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.
startIndex: 0 must read from the beginning, not send a "-1" resume cursor. Treat 0 as "from the beginning" (undefined cursor) again; the !== undefined form produced lastEventId "-1", which Redis rejects as an invalid id and S2 maps past the first record.
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.
ericallam
changed the title
- Add clamp: true to the S2 tail_offset read so from: "latest" on an empty or
short stream saturates to the tail and long-polls instead of erroring.
- An explicit startIndex now suppresses from in useRealtimeStream and
streams.read, so { startIndex: 0 } reads from the beginning rather than
falling through to from: "latest".
nicktrn
approved these changes
Aug 28, 2026
Address two review findings on the realtime stream hooks: - Seed the resume-cursor ref only on stream-identity change, not on every persisted-cursor update, so a mid-stream flush can no longer overwrite a newer live cursor with an older persisted value (which could replay parts on restart). Applies to useRealtimeStream and useSessionStream. - Apply maxParts / maxRecords to the SWR-cached parts on mount and when the bound changes, so a remount with a bound of 1 does not briefly return the full cached history before the next batch.
ericallam
deleted the
feature/tri-13511-realtime-streams-start-from-latest-tail-subscribe-option
branch
August 28, 2026 12:16
Merged
ericallam
pushed a commit
that referenced
this pull request
Aug 28, 2026
## Summary 4 improvements, 1 bug fix. ## Improvements - Native build server deploys now show a single updating build log line by default; pass `--build-logs full` to stream every line (always used in CI and when output is not a terminal). ([#4817](#4817)) - Realtime stream subscriptions can now refresh an expired access token and reconnect, via a new optional `refreshAccessToken` option on the client configuration and the React hooks. ([#4811](#4811)) - Subscribe to a realtime stream from its latest record instead of replaying the whole history. Pass `from: "latest"` to `useRealtimeStream`, `streams.read()`, or `fetchStream` to start at the current tail (the latest record, then live updates) instead of replaying (a live "last value" view), and `maxParts` to keep the accumulated `parts` array bounded. A reconnect or remount resumes from the last record it saw, so no records are missed and none are replayed. `from: "latest"` needs a server that supports it; older servers safely fall back to a full replay. ([#4811](#4811)) `useRealtimeStream` also gains a `lastEventId` option and returns the `lastEventId` of the last part seen, so you can persist the cursor (for example across a page reload) and resume exactly where you left off. An `onParts` callback delivers each throttled batch of parts with their event ids. ```tsx const { parts, lastEventId } = useRealtimeStream<Frame>(runId, "frames", { from: "latest", // skip history, start at the current tail maxParts: 1, // keep only the most recent frame lastEventId: savedCursor, // resume from a persisted cursor onParts: (batch) => save(batch.at(-1)?.id), // track the cursor accessToken, }); ``` - Added a `useSessionStream` React hook for reading a session's output or input channel in realtime. It accumulates records with automatic resume from the last record you received, and supports `from: "latest"` (start at the current tail, only new records after you connect), `maxRecords` (keep a bounded number of records in memory), a `lastEventId` resume cursor, and an `onRecords` callback that delivers each throttled batch of records with their event ids. ([#4811](#4811)) ## Server changes These changes affect the self-hosted Docker image and Trigger.dev Cloud: - Task retries that wait in the queue no longer count against the queue's internal redelivery limit, so runs with many long-delay retries are not wrongly failed with TASK_RUN_DEQUEUED_MAX_RETRIES. ([#4810](#4810)) <details> <summary>Raw changeset output</summary> # Releases ## @trigger.dev/build@4.5.14 ### Patch Changes - Updated dependencies: - `@trigger.dev/core@4.5.14` ## trigger.dev@4.5.14 ### Patch Changes - Native build server deploys now show a single updating build log line by default; pass `--build-logs full` to stream every line (always used in CI and when output is not a terminal). ([#4817](#4817)) - Updated dependencies: - `@trigger.dev/core@4.5.14` - `@trigger.dev/build@4.5.14` - `@trigger.dev/schema-to-json@4.5.14` ## @trigger.dev/core@4.5.14 ### Patch Changes - Realtime stream subscriptions can now refresh an expired access token and reconnect, via a new optional `refreshAccessToken` option on the client configuration and the React hooks. ([#4811](#4811)) - Subscribe to a realtime stream from its latest record instead of replaying the whole history. Pass `from: "latest"` to `useRealtimeStream`, `streams.read()`, or `fetchStream` to start at the current tail (the latest record, then live updates) instead of replaying (a live "last value" view), and `maxParts` to keep the accumulated `parts` array bounded. A reconnect or remount resumes from the last record it saw, so no records are missed and none are replayed. `from: "latest"` needs a server that supports it; older servers safely fall back to a full replay. ([#4811](#4811)) `useRealtimeStream` also gains a `lastEventId` option and returns the `lastEventId` of the last part seen, so you can persist the cursor (for example across a page reload) and resume exactly where you left off. An `onParts` callback delivers each throttled batch of parts with their event ids. ```tsx const { parts, lastEventId } = useRealtimeStream<Frame>(runId, "frames", { from: "latest", // skip history, start at the current tail maxParts: 1, // keep only the most recent frame lastEventId: savedCursor, // resume from a persisted cursor onParts: (batch) => save(batch.at(-1)?.id), // track the cursor accessToken, }); ``` ## @trigger.dev/python@4.5.14 ### Patch Changes - Updated dependencies: - `@trigger.dev/core@4.5.14` - `@trigger.dev/sdk@4.5.14` - `@trigger.dev/build@4.5.14` ## @trigger.dev/react-hooks@4.5.14 ### Patch Changes - Realtime stream subscriptions can now refresh an expired access token and reconnect, via a new optional `refreshAccessToken` option on the client configuration and the React hooks. ([#4811](#4811)) - Subscribe to a realtime stream from its latest record instead of replaying the whole history. Pass `from: "latest"` to `useRealtimeStream`, `streams.read()`, or `fetchStream` to start at the current tail (the latest record, then live updates) instead of replaying (a live "last value" view), and `maxParts` to keep the accumulated `parts` array bounded. A reconnect or remount resumes from the last record it saw, so no records are missed and none are replayed. `from: "latest"` needs a server that supports it; older servers safely fall back to a full replay. ([#4811](#4811)) `useRealtimeStream` also gains a `lastEventId` option and returns the `lastEventId` of the last part seen, so you can persist the cursor (for example across a page reload) and resume exactly where you left off. An `onParts` callback delivers each throttled batch of parts with their event ids. ```tsx const { parts, lastEventId } = useRealtimeStream<Frame>(runId, "frames", { from: "latest", // skip history, start at the current tail maxParts: 1, // keep only the most recent frame lastEventId: savedCursor, // resume from a persisted cursor onParts: (batch) => save(batch.at(-1)?.id), // track the cursor accessToken, }); ``` - Added a `useSessionStream` React hook for reading a session's output or input channel in realtime. It accumulates records with automatic resume from the last record you received, and supports `from: "latest"` (start at the current tail, only new records after you connect), `maxRecords` (keep a bounded number of records in memory), a `lastEventId` resume cursor, and an `onRecords` callback that delivers each throttled batch of records with their event ids. ([#4811](#4811)) - Updated dependencies: - `@trigger.dev/core@4.5.14` ## @trigger.dev/redis-worker@4.5.14 ### Patch Changes - Updated dependencies: - `@trigger.dev/core@4.5.14` ## @trigger.dev/rsc@4.5.14 ### Patch Changes - Updated dependencies: - `@trigger.dev/core@4.5.14` ## @trigger.dev/schema-to-json@4.5.14 ### Patch Changes - Updated dependencies: - `@trigger.dev/core@4.5.14` ## @trigger.dev/sdk@4.5.14 ### Patch Changes - Subscribe to a realtime stream from its latest record instead of replaying the whole history. Pass `from: "latest"` to `useRealtimeStream`, `streams.read()`, or `fetchStream` to start at the current tail (the latest record, then live updates) instead of replaying (a live "last value" view), and `maxParts` to keep the accumulated `parts` array bounded. A reconnect or remount resumes from the last record it saw, so no records are missed and none are replayed. `from: "latest"` needs a server that supports it; older servers safely fall back to a full replay. ([#4811](#4811)) `useRealtimeStream` also gains a `lastEventId` option and returns the `lastEventId` of the last part seen, so you can persist the cursor (for example across a page reload) and resume exactly where you left off. An `onParts` callback delivers each throttled batch of parts with their event ids. ```tsx const { parts, lastEventId } = useRealtimeStream<Frame>(runId, "frames", { from: "latest", // skip history, start at the current tail maxParts: 1, // keep only the most recent frame lastEventId: savedCursor, // resume from a persisted cursor onParts: (batch) => save(batch.at(-1)?.id), // track the cursor accessToken, }); ``` - Updated dependencies: - `@trigger.dev/core@4.5.14` </details> Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
ericallam
added a commit
that referenced
this pull request
Aug 29, 2026
…4815) ## Summary Adds **named side channels** to a Session: durable, two-way realtime streams that outlive a single run and are shared across every run of the session. Today a Session has exactly one reserved `.in`/`.out` pair (the chat transcript). This lets a session hold any number of *named* channels alongside it, each its own `.in`/`.out` pair, so an agent can stream out-of-band data (a feed of frames, telemetry, a control channel) on a stream separate from the transcript while many clients read it live. The two properties a named channel adds over the reserved pair: 1. It is addressed by a name that outlives a run and is shared across runs, not welded to the chat turn loop. 2. Writing its `.in` does **not** wake or trigger a run. A run observes it by subscribing; an external client writes it without spawning anything. This is the generalization half of the Momentic ask (stream browser screenshots from a `chat.agent` to the frontend on a channel separate from the chat). It builds directly on the start-from-latest / `useSessionStream` subscribe seam from #4811. ## Usage Declare the channel's record types once and infer them on both sides: ```ts // channels.ts (shared, client imports it type-only) import { sessions } from "@trigger.dev/sdk"; export const screenshots = sessions.defineChannel<{ out: ScreenshotFrame; in: ViewportControl }>( "screenshots" ); ``` Open a channel from a session handle (`sessions.open(id)` returns one for a known session id). Writing its `.out` is durable, cross-run, and wakes nothing; a run observes its `.in` by tailing, without suspending: ```ts import { sessions } from "@trigger.dev/sdk"; import { screenshots } from "./channels"; const channel = sessions.open(sessionId).channel(screenshots); await channel.out.append(frame); // frame: ScreenshotFrame (typed from the definition) channel.in.on((control) => { /* ... */ }); // control: ViewportControl, tail, no suspend ``` Passing the definition types `.out.append` / `.in.on` on the producer side; a bare name string also works, with records typed `unknown`. An external client writes the `.in` without waking a run, and reads the `.out` from React: ```ts sessions.open(sessionId).channel("screenshots").in.send({ paused: true }); const { records } = useSessionStreamChannel<typeof screenshots>("screenshots", { sessionId, accessToken, io: "out", from: "latest", maxRecords: 1, }); ``` `session.channel(name)` returns the same `{ in, out }` handle shape as the reserved pair, so `append` / `pipe` / `writer` / `read` / `writeControl` / `trimTo` on `.out` and `send` / `on` / `once` / `peek` on `.in` all carry over. Passing a name other than the declared one is a type error; a bare-string call without the generic stays valid with `records` typed `unknown`. ### With `chat.agent` This is the motivating case: a `chat.agent` answers on the reserved transcript as usual, and streams screenshot frames on a side channel in parallel. `chat.channel(name)` opens a channel on the current run's own Session, so there's no id to thread: ```ts import { chat } from "@trigger.dev/sdk/ai"; import { streamText } from "ai"; import { screenshots } from "./channels"; export const browserAgent = chat.agent({ id: "browser-agent", run: async ({ messages, signal }) => { const frames = chat.channel(screenshots); // client pause/resume arrives here without waking a turn frames.in.on((control: ViewportControl) => applyViewport(control)); // frames stream on their own channel, not the chat transcript driveBrowser({ signal, onFrame: (frame) => frames.out.append(frame) }); // the assistant reply still goes to the reserved transcript return streamText({ model: openai("gpt-4o"), messages, abortSignal: signal }); }, }); ``` `chat.channel(name)` is a shortcut for `chat.session().channel(name)`; `chat.session()` returns the current run's full `SessionHandle` if you need it. The frontend renders the transcript with `useChat` as before, and the screenshots with `useSessionStreamChannel<typeof screenshots>("screenshots", { sessionId: chatId, io: "out", from: "latest", maxRecords: 1 })`: a live view of the newest frame that survives across turns (each turn is a new run), because the channel is keyed on the session, not the run. ### From MCP An MCP client can observe and write a session's channels with two tools, built on the same apiClient surface as the hook and the dashboard viewer: - `read_session_channel` reads records from a channel (or the reserved pair). It is a point-in-time drain with cursor pagination (`afterEventId` / `nextCursor`, `maxRecords`); pass `timeoutInSeconds` to wait for the next record when none exist yet. - `write_session_channel` appends one record to a channel's `.in` (an object or a raw string), so an agent can send control input without waking a run. `.out` is producer-only, so it is not writable here. ### On the session page The session detail page lists a session's channels (via an S2 prefix list in the loader) and shows each as a tab beside `Rendered` and `Raw`. Selecting a channel renders its records in the same table as the Raw transcript view, sourced from that channel's `out` and `in` streams. ## How it works **Addressing.** A channel is a stream name segment: `sessions/{id}/channels/{name}/{io}`. The reserved pair keeps its two-part `sessions/{id}/{io}` name for back-compat, and the `channels/` segment means a user channel named `in`/`out` can never collide with it. The channel dimension is threaded through the session stream manager (keyed on `(session, channel, io)`, reserved = absent), `subscribeToSessionStream`, the session apiClient methods, and the `realtime.v1.sessions.$session.channels.$channel.$io.{ts,append,records}` routes. The reserved-pair routes are untouched. The start-from-latest tail path from #4811 is channel-agnostic, so `from: "latest"` and `maxRecords` compose unchanged. **No-wake.** The reserved `.in` append route ensures a run and drains waitpoints so a chat turn advances. The channel `.in` append route deliberately does neither: the record lands durably and a run picks it up when it next subscribes, so writing a side channel can't spawn or resume a run. A named channel's `.in` is therefore subscribe-only from the run side (`.on` / `.once` / `.peek`); `.wait()` / `waitWithIdleTimeout()` throw with a message pointing at the observe methods. **Auth.** Channel scope folds into the existing resource id (`sessions:<key>:channels:<channel>`), so no RBAC grammar change. A channel route authorizes both the channel-folded id and the bare session id, which means a session-wide token grants every channel while a channel-scoped token grants only its own. The per-io rule is preserved per channel: writing `.out` requires secret-key auth so a browser can't forge frames; `.in` is writable with the session token. **Retention.** Channel streams are created on demand on first write and inherit the org's stream retention (bounded age plus delete-on-empty from the store's default config), the same as the reserved chat streams. There is no per-channel control-plane call on the write path. Custom per-channel retention is deferred until the stream store can set config inline on the on-demand create, which avoids a control-plane round trip. **Spans.** Channel writes carry `channel` and `io` attributes, an accessory chip, and the session icon. Clicking a channel span in the run's span inspector renders the channel's actual records with the same viewer the run realtime streams use, rather than the raw properties JSON. ## Verification - **Unit (core):** the stream manager isolates channels: two channels on the same `(session, io)` never cross buffers, and a named channel is isolated from the reserved pair. - **Full-stack e2e** against a real stack (webapp, stream store, Postgres, real runs): - a named `.out` record is readable back **after the triggering run has gone terminal** (durable, cross-run); - a channel `.in` append creates **no** run, while a reserved `.in` append **does** wake one (the differential is the red/green); - `from: "latest"` on a named channel delivers the live record and does **not** replay the backlog from the start; - the span inspector renders a channel span's records, and the MCP read/write tools round-trip records on a real session; - an invalid channel name is rejected. ## Notes - **Channel listing works on the self-hosted store too.** The stream store's list operation is available on s2-lite, so the session page's channel list is an OSS feature. It is a control-plane call made once per session-page load (best-effort; a failure just hides the tabs), not on the write path. - **The ~1 MiB per-record cap is unchanged.** Large payloads (e.g. raw screenshots) still need object-store pointers on the channel rather than inline bytes; that's independent of this change. - Docs ride this branch: the side channels guide, the `useSessionStreamChannel` reference, and the MCP tools list are all updated here. ## Screenshots <img width="3444" height="1870" alt="CleanShot 2026-08-28 at 21 46 27@2x" src="https://github.com/user-attachments/assets/c192aaee-b946-4824-87b7-ca057514d25e" />
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
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.
Summary
Realtime streams get a live "last value" mode: subscribe from the latest record instead of replaying the whole history, keep memory bounded, and resume across reloads. Plus a new
useSessionStreamhook for reading a Session's channels from React.useRealtimeStream: start-from-latest, bounded, resumablefrom,lastEventId(option and return), and the batching also apply tostreams.read()andfetchStream().useSessionStream: read a Session channel from React (new)A read-only hook for a Session's
out(default) orinchannel, with the same start / bound / resume options.useSessionis reserved for two-way (read and write).Access-token refresh
Long-lived subscriptions can survive token expiry: pass
refreshAccessTokenand a 401/403 triggers one re-mint and reconnect. With no refresher, auth errors stay terminal exactly as before.It is also available on
useApiClient/TriggerAuthContext, so every hook under a provider shares one refresher.Notes
Server support (S2
tail_offset/ Redis$, and the start-position header on the run and session SSE routes) ships here; a client passingfrom: "latest"against an older server degrades safely to a full replay. Resume, bounded memory, batched callbacks, and token refresh are client-only.Supersedes #4808 and #4809, folded in here. Verified end to end on an isolated stack:
from: "latest"on the run and session paths against real S2,lastEventIdresume across a reload, bounded memory, batched callbacks, and a real 401 to token-refresh to reconnect.