feat(core): support refreshing access tokens for stream subscriptions - #4809
Closed
claude[bot] wants to merge 2 commits into
Closed
feat(core): support refreshing access tokens for stream subscriptions#4809claude[bot] wants to merge 2 commits into
claude[bot] wants to merge 2 commits into
Conversation
Realtime SSE subscriptions read their auth headers per HTTP attempt. When a refreshAccessToken callback is configured, a 401/403 now re-resolves the headers and retries the connect once instead of failing the stream.
…ting Pass the refresh resolver to the runShapeStream factory so subscribeToRun, subscribeToRunsWithTag and subscribeToBatch recover from an expired token. Only re-arm the refresh after a connection delivers a record, keep the auth error terminal when the refresher throws, and don't report a 401 the refresh recovered from.
🦋 Changeset detectedLatest commit: 525b842 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 |
3 tasks
Member
|
Closing in favor of a combined realtime-streams PR that folds this token-refresh work in alongside the session-read hook (#4808) and the start-from-latest subscribe option. |
ericallam
added a commit
that referenced
this pull request
Aug 28, 2026
…#4811) ## 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 `useSessionStream` hook for reading a Session's channels from React. ## `useRealtimeStream`: start-from-latest, bounded, resumable ```tsx const { parts, lastEventId } = useRealtimeStream<Frame>(runId, "frames", { from: "latest", // skip history, only new records after connect maxParts: 1, // keep just the most recent (bounded memory) lastEventId: saved, // resume from a persisted cursor (survives reload) onParts: (batch) => save(batch.at(-1)?.id), // per-batch event ids accessToken, }); ``` `from`, `lastEventId` (option and return), and the batching also apply to `streams.read()` and `fetchStream()`. ## `useSessionStream`: read a Session channel from React (new) A read-only hook for a Session's `out` (default) or `in` channel, with the same start / bound / resume options. `useSession` is reserved for two-way (read and write). ```tsx const { records, lastEventId } = useSessionStream<Frame>(sessionId, { io: "out", from: "latest", maxRecords: 5, onRecords: (batch) => {/* each throttled batch, with event ids */}, accessToken, }); ``` ## Access-token refresh Long-lived subscriptions can survive token expiry: pass `refreshAccessToken` and a 401/403 triggers one re-mint and reconnect. With no refresher, auth errors stay terminal exactly as before. ```tsx const { parts } = useRealtimeStream<Frame>(runId, "frames", { accessToken, // called on a 401/403 to mint a fresh public token from your backend refreshAccessToken: async () => { const res = await fetch("/api/realtime-token"); return (await res.json()).token; }, }); ``` 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 passing `from: "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, `lastEventId` resume across a reload, bounded memory, batched callbacks, and a real 401 to token-refresh to reconnect.
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.
Requested by Matt Aitken · Slack thread
✅ Checklist
Testing
Covered by unit tests in
packages/core, run with the package's vitest suite:packages/core/src/v3/apiClient/runStream.test.ts— 8 new cases againstSSEStreamSubscription: a 401 with noresolveHeadersstill fails the stream (unchanged behaviour); a 401 and a 403 each retry once with the refreshed headers; a refreshed token that is rejected again is terminal; a refresher that throws is terminal; a 401 the refresh recovered from is not reported toonError; a connection that is accepted but delivers no records does not get a second mint; and a connection that has delivered a record may refresh again.packages/core/src/v3/apiClient/refreshAccessToken.test.ts— 4 new cases on the dedupe helper: concurrent callers share one in-flight mint, different refreshers do not share, a new mint happens once the previous one settles, and a rejected mint rejects every concurrent caller without poisoning later calls.The "does not report a 401 that the refresh recovered from" case fails against the pre-fix code, where
onErrorwas invoked before the status was classified.Changelog
Before. An access token was fixed for the life of a streaming subscription. When it expired, the stream terminated with an auth error and there was no way to recover short of tearing down the subscription and creating a new one.
After. An optional
refreshAccessTokenon the client configuration (ApiClientConfiguration), and as an option onuseApiClient/TriggerAuthContext. Stream subscriptions now resolve their auth header per connection attempt, so an expired token triggers one re-mint and one retry. Behaviour is unchanged when no refresher is supplied: auth errors stay terminal exactly as before.How it works.
ApiClienthands stream subscriptions a per-attempt header resolver (resolveHeaders) instead of only a fixed header snapshot. It isundefinedunless arefreshAccessTokenwas configured, which is what keeps the no-refresher path byte-for-byte the old behaviour.refreshAccessTokenOncededupes concurrent refreshes, keyed on the refresher function itself, so multiple subscriptions or React hooks sharing one refresher mint a single token rather than one each.useApiClientkeeps the caller's refresher ref-stable so that key survives re-renders.The changeset is currently a
patchfor@trigger.dev/coreand@trigger.dev/react-hooks. This adds new public API surface, so whether it should be aminorbump instead is still under discussion — happy to switch it.Screenshots
n/a — no UI changes.
💯