fix(chat): reconnect a dropped response stream instead of freezing mid-turn - #1924
Conversation
…d-turn
A long turn's SSE stream can end before the run does. Reproduced on prod
2026-08-02: the connection closed at ~123s with a clean [DONE] and no
finish chunk while the workflow ran on to completion. useChat saw a
stream that ended without a terminal chunk, stopped rendering, and never
marked the message complete — no error, no retry, composer looking idle.
The user got 6 of 13 iterations and had to refresh to see the rest.
The AI SDK already has the machinery: DefaultChatTransport.reconnectToStream
defaults to GET {api}/{chatId}/stream, which is exactly the route
recoupable/api#809 implements. We had neither the client wiring nor the
endpoint.
- resume: true on useChat — re-attach to an in-progress response on mount,
so returning to a chat mid-turn keeps rendering.
- useStreamRecovery — watches an in-flight turn for silence and calls
resumeStream(). Also re-checks on visibilitychange, since a backgrounded
tab is where drops are most likely and least likely to be noticed.
- shouldRecoverStalledStream — the pure decision, unit-tested.
- prepareReconnectToStreamRequest on the transport — the resume route is
authenticated like every other endpoint, so without this the reconnect
401s and a dropped stream stays dropped.
Silence-based rather than duration-based: a turn still streaming is
healthy however long it runs, and a turn gone quiet is suspect even if it
just started. A cooldown stops a permanently dead stream being retried
every tick.
Depends on recoupable/api#809 (the resume route) and docs#286 (contract).
Refs #1923
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
Bugbot is not enabled for your account, so this pull request was not reviewed. Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs. |
📝 WalkthroughWalkthroughThe chat client detects stalled active streams, retries recovery, tracks consumed SSE chunks, and reconnects with a Privy Bearer token from the next unseen chunk. ChangesStalled-stream recovery
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant useVercelChat
participant useStreamRecovery
participant useChatTransport
useVercelChat->>useStreamRecovery: provide stream status and activity
useStreamRecovery->>useStreamRecovery: detect stalled stream
useStreamRecovery->>useChatTransport: call resumeStream()
useChatTransport->>useChatTransport: add Bearer token and startIndex
useChatTransport-->>useStreamRecovery: resume from next unseen chunk
Possibly related issues
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 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}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@hooks/useChatTransport.ts`:
- Around line 64-72: Update prepareReconnectToStreamRequest in useChatTransport
so reconnect requests use the effective API chat ID, matching the workflowChatId
?? id selection used when sending messages, rather than relying on useChat.id
alone. Preserve the existing authorization header behavior and return the
reconnect URL/config with the resolved chat identifier.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 8ec90ab8-c676-4ad2-a90d-098da25d4bf1
⛔ Files ignored due to path filters (1)
lib/chat/__tests__/shouldRecoverStalledStream.test.tsis excluded by!**/*.test.*and included bylib/**
📒 Files selected for processing (4)
hooks/useChatTransport.tshooks/useStreamRecovery.tshooks/useVercelChat.tslib/chat/shouldRecoverStalledStream.ts
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1ec8d635c9
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| lastRecoveryAt, | ||
| isRecoveryInFlight, | ||
| }: StreamRecoveryInput): boolean { | ||
| if (!IN_FLIGHT_STATUSES.has(status)) return false; |
There was a problem hiding this comment.
In the clean [DONE] / no finish case this change is targeting, the AI SDK still completes the stream and moves status back to ready; once that happens this predicate immediately returns false, so the polling hook never calls resumeStream() after the premature close. This means the reported frozen-mid-turn scenario remains unrecovered unless there happened to be a 20s silence window before the socket closed; consider tracking the missing finish signal/onFinish state instead of treating every ready transition as terminal.
Useful? React with 👍 / 👎.
| prepareReconnectToStreamRequest: async () => { | ||
| const accessToken = await getAccessToken().catch(() => null); | ||
| const headers: Record<string, string> = {}; | ||
| if (accessToken) headers.Authorization = `Bearer ${accessToken}`; | ||
| return { headers }; |
There was a problem hiding this comment.
When a new chat is opened through NewChatBootstrap, useChat({ id }) keeps the client placeholder id while workflowChatId becomes the real recoup-api chat id used for the POST body and URL. resumeStream() reconnects with the useChat id, and because this preparer only returns headers, the default reconnect URL stays /api/chat/<placeholder>/stream; dropped first-turn streams from /chat will therefore look up the wrong chat and fail to resume. Override the reconnect api with the current chatIdRef.current here, or make the useChat id match the workflow chat id before recovery can run.
Useful? React with 👍 / 👎.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
6 issues found across 5 files
Confidence score: 2/5
hooks/useVercelChat.tshas two concrete stream-reliability risks: resume can race persisted history loading and overwrite newer resumed output with olderinitialMessages, leaving turns incomplete again—gate resume on history-load completion (or merge responses deterministically) to prevent regressions.hooks/useVercelChat.tsalso appears to reconnect with the wrong stream identity in new-chat bootstrap flows (resumeStreamusinguseChatplaceholderidwhile sends useworkflowChatId), so dropped streams may never resume for affected users—align the transport/resume ID to the same chat ID used for sends.lib/chat/shouldRecoverStalledStream.tscan miss stalled streams after a clean[DONE]without afinishchunk becausestatusreturns toreadyand falls outsideIN_FLIGHT_STATUSES, which can leave interrupted generations unrecovered—expand the predicate/status handling for this terminal-but-incomplete path.- There are lower-severity follow-ups that could cause drift over time:
hooks/useChatTransport.tsduplicates auth header/token logic for reconnect requests, andhooks/useVercelChat.tsnow uses resume semantics that conflict with abort behavior—centralize auth/header construction and update cancellation handling to match AI SDK resume constraints.
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="hooks/useStreamRecovery.ts">
<violation number="1" location="hooks/useStreamRecovery.ts:106">
P2: Custom agent: **Code Structure and Size Limits for Readability and Single Responsibility**
This new file is 106 lines, which exceeds the 100-line limit for readability and single responsibility. Consider splitting it—e.g., extract the re-export of `STREAM_STALL_MS` into its own file or decompose the recovery logic into smaller focused helpers—to bring it under the limit.</violation>
</file>
<file name="hooks/useChatTransport.ts">
<violation number="1" location="hooks/useChatTransport.ts:68">
P3: Reconnect auth now has a second token/header implementation beside `headers`, so future changes to token retrieval or header format can make normal and recovery requests authenticate differently. Reusing the existing header resolver keeps both transport paths aligned.</violation>
</file>
<file name="hooks/useVercelChat.ts">
<violation number="1" location="hooks/useVercelChat.ts:223">
P1: Resuming an active chat races the persisted-history fetch, which can replace resumed output with the older `initialMessages` response and leave the turn incomplete again. Gate resume until history load completes, or make the loader merge/ignore its result once streaming has started.</violation>
<violation number="2" location="hooks/useVercelChat.ts:223">
P3: Now that `resume: true` is enabled for useChat, the abort semantics of the SDK change: per the AI SDK resume-streams documentation, resuming is incompatible with aborting the stream, and the recommended approach is a dedicated stop endpoint that persists partial results and cancels the active resumable stream. This hook still relies on the SDK's default `stop()` for cancellation, so a user hitting stop on a running workflow may leave an orphaned active stream server-side that the recovery loop in useStreamRecovery keeps attempting to reconnect to once the stall timer fires. Worth verifying stop behavior in the resume flow (ideally wiring the existing stop handler to recoup-api's stop/resume contract) before relying on it for cancellation.</violation>
<violation number="3" location="hooks/useVercelChat.ts:241">
P1: Dropped streams in chats created from the new-chat bootstrap still cannot reconnect: `resumeStream` uses `useChat`'s placeholder `id`, while sends use `workflowChatId`. Keep the `useChat` stream ID aligned with `transportChatId` (or customize the reconnect route) so recovery requests address the persisted chat.</violation>
</file>
<file name="lib/chat/shouldRecoverStalledStream.ts">
<violation number="1" location="lib/chat/shouldRecoverStalledStream.ts:46">
P1: Once the AI SDK closes the stream cleanly with `[DONE]` but no `finish` chunk, `status` transitions back to `ready`. Since `IN_FLIGHT_STATUSES` only includes `streaming`/`submitted`, this predicate returns `false` immediately after that transition, so `useStreamRecovery` never calls `resumeStream()` for the prematurely-closed case this PR targets — recovery only happens if there's a 20s silence window before the status flips. Consider tracking the missing `finish`/`onFinish` signal explicitly rather than treating every `ready` transition as terminal.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| transport, | ||
| // Re-attach to an in-progress response on mount, so returning to a chat | ||
| // mid-turn keeps rendering instead of showing a frozen half-message. | ||
| resume: true, |
There was a problem hiding this comment.
P1: Resuming an active chat races the persisted-history fetch, which can replace resumed output with the older initialMessages response and leave the turn incomplete again. Gate resume until history load completes, or make the loader merge/ignore its result once streaming has started.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At hooks/useVercelChat.ts, line 223:
<comment>Resuming an active chat races the persisted-history fetch, which can replace resumed output with the older `initialMessages` response and leave the turn incomplete again. Gate resume until history load completes, or make the loader merge/ignore its result once streaming has started.</comment>
<file context>
@@ -213,10 +214,13 @@ export function useVercelChat({
transport,
+ // Re-attach to an in-progress response on mount, so returning to a chat
+ // mid-turn keeps rendering instead of showing a frozen half-message.
+ resume: true,
experimental_throttle: 100,
generateId: generateUUID,
</file context>
| useStreamRecovery({ | ||
| status, | ||
| activityMarker: messages, | ||
| resumeStream, |
There was a problem hiding this comment.
P1: Dropped streams in chats created from the new-chat bootstrap still cannot reconnect: resumeStream uses useChat's placeholder id, while sends use workflowChatId. Keep the useChat stream ID aligned with transportChatId (or customize the reconnect route) so recovery requests address the persisted chat.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At hooks/useVercelChat.ts, line 241:
<comment>Dropped streams in chats created from the new-chat bootstrap still cannot reconnect: `resumeStream` uses `useChat`'s placeholder `id`, while sends use `workflowChatId`. Keep the `useChat` stream ID aligned with `transportChatId` (or customize the reconnect route) so recovery requests address the persisted chat.</comment>
<file context>
@@ -229,6 +233,14 @@ export function useVercelChat({
+ useStreamRecovery({
+ status,
+ activityMarker: messages,
+ resumeStream,
+ });
+
</file context>
| lastRecoveryAt, | ||
| isRecoveryInFlight, | ||
| }: StreamRecoveryInput): boolean { | ||
| if (!IN_FLIGHT_STATUSES.has(status)) return false; |
There was a problem hiding this comment.
P1: Once the AI SDK closes the stream cleanly with [DONE] but no finish chunk, status transitions back to ready. Since IN_FLIGHT_STATUSES only includes streaming/submitted, this predicate returns false immediately after that transition, so useStreamRecovery never calls resumeStream() for the prematurely-closed case this PR targets — recovery only happens if there's a 20s silence window before the status flips. Consider tracking the missing finish/onFinish signal explicitly rather than treating every ready transition as terminal.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At lib/chat/shouldRecoverStalledStream.ts, line 46:
<comment>Once the AI SDK closes the stream cleanly with `[DONE]` but no `finish` chunk, `status` transitions back to `ready`. Since `IN_FLIGHT_STATUSES` only includes `streaming`/`submitted`, this predicate returns `false` immediately after that transition, so `useStreamRecovery` never calls `resumeStream()` for the prematurely-closed case this PR targets — recovery only happens if there's a 20s silence window before the status flips. Consider tracking the missing `finish`/`onFinish` signal explicitly rather than treating every `ready` transition as terminal.</comment>
<file context>
@@ -0,0 +1,53 @@
+ lastRecoveryAt,
+ isRecoveryInFlight,
+}: StreamRecoveryInput): boolean {
+ if (!IN_FLIGHT_STATUSES.has(status)) return false;
+ if (isRecoveryInFlight) return false;
+ if (lastChunkAt === null) return false;
</file context>
| }, []); | ||
| } | ||
|
|
||
| export { STREAM_STALL_MS }; |
There was a problem hiding this comment.
P2: Custom agent: Code Structure and Size Limits for Readability and Single Responsibility
This new file is 106 lines, which exceeds the 100-line limit for readability and single responsibility. Consider splitting it—e.g., extract the re-export of STREAM_STALL_MS into its own file or decompose the recovery logic into smaller focused helpers—to bring it under the limit.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At hooks/useStreamRecovery.ts, line 106:
<comment>This new file is 106 lines, which exceeds the 100-line limit for readability and single responsibility. Consider splitting it—e.g., extract the re-export of `STREAM_STALL_MS` into its own file or decompose the recovery logic into smaller focused helpers—to bring it under the limit.</comment>
<file context>
@@ -0,0 +1,106 @@
+ }, []);
+}
+
+export { STREAM_STALL_MS };
</file context>
| // route, which is authenticated like every other endpoint. Without | ||
| // this the reconnect 401s and a dropped stream stays dropped. | ||
| prepareReconnectToStreamRequest: async () => { | ||
| const accessToken = await getAccessToken().catch(() => null); |
There was a problem hiding this comment.
P3: Reconnect auth now has a second token/header implementation beside headers, so future changes to token retrieval or header format can make normal and recovery requests authenticate differently. Reusing the existing header resolver keeps both transport paths aligned.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At hooks/useChatTransport.ts, line 68:
<comment>Reconnect auth now has a second token/header implementation beside `headers`, so future changes to token retrieval or header format can make normal and recovery requests authenticate differently. Reusing the existing header resolver keeps both transport paths aligned.</comment>
<file context>
@@ -61,6 +61,15 @@ export function useChatTransport({
+ // route, which is authenticated like every other endpoint. Without
+ // this the reconnect 401s and a dropped stream stays dropped.
+ prepareReconnectToStreamRequest: async () => {
+ const accessToken = await getAccessToken().catch(() => null);
+ const headers: Record<string, string> = {};
+ if (accessToken) headers.Authorization = `Bearer ${accessToken}`;
</file context>
| transport, | ||
| // Re-attach to an in-progress response on mount, so returning to a chat | ||
| // mid-turn keeps rendering instead of showing a frozen half-message. | ||
| resume: true, |
There was a problem hiding this comment.
P3: Now that resume: true is enabled for useChat, the abort semantics of the SDK change: per the AI SDK resume-streams documentation, resuming is incompatible with aborting the stream, and the recommended approach is a dedicated stop endpoint that persists partial results and cancels the active resumable stream. This hook still relies on the SDK's default stop() for cancellation, so a user hitting stop on a running workflow may leave an orphaned active stream server-side that the recovery loop in useStreamRecovery keeps attempting to reconnect to once the stall timer fires. Worth verifying stop behavior in the resume flow (ideally wiring the existing stop handler to recoup-api's stop/resume contract) before relying on it for cancellation.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At hooks/useVercelChat.ts, line 223:
<comment>Now that `resume: true` is enabled for useChat, the abort semantics of the SDK change: per the AI SDK resume-streams documentation, resuming is incompatible with aborting the stream, and the recommended approach is a dedicated stop endpoint that persists partial results and cancels the active resumable stream. This hook still relies on the SDK's default `stop()` for cancellation, so a user hitting stop on a running workflow may leave an orphaned active stream server-side that the recovery loop in useStreamRecovery keeps attempting to reconnect to once the stall timer fires. Worth verifying stop behavior in the resume flow (ideally wiring the existing stop handler to recoup-api's stop/resume contract) before relying on it for cancellation.</comment>
<file context>
@@ -213,10 +214,13 @@ export function useVercelChat({
transport,
+ // Re-attach to an in-progress response on mount, so returning to a chat
+ // mid-turn keeps rendering instead of showing a frozen half-message.
+ resume: true,
experimental_throttle: 100,
generateId: generateUUID,
</file context>
…isibility Three refinements after comparing against upstream open-agents. 1. Consume x-workflow-stream-tail-index. recoupable/api#809 now reports where the read it served ends; a custom transport fetch captures it and prepareReconnectToStreamRequest sends startIndex = tail + 1. Reconnects are now gap-free instead of replaying the turn from chunk zero. 2. Thresholds tightened: stall 20s -> 10s, cooldown 15s -> 8s (upstream's STREAM_RECOVERY_MIN_INTERVAL_MS), poll 5s -> 3s. Safe precisely because of (1) — an unnecessary reconnect now costs a request rather than re-rendering content the client already has. NOT upstream's STREAM_RECOVERY_STALL_MS = 4_000. That constant feeds a scheduler their shouldScheduleStallRecovery unconditionally disables (`void options; return false`), so it is not a live stall threshold to copy. It also would not survive our workload: a single legitimate tool call streams nothing for up to ~200s (measured on prod), so a 4s window would fire dozens of pointless reconnects per turn. 3. Visibility probe, upstream's only live recovery trigger. A backgrounded tab can have its connection killed silently and no amount of waiting produces a chunk to time out on, so a visibility check skips the silence window. The cooldown still applies, so a focus-flapping tab cannot spam reconnects. Kept our stall-based trigger rather than adopting upstream's posture wholesale: their live triggers are `status === "error"` and a visibility probe when `status === "ready"`, and our failure mode produces neither — the stream ends with a clean [DONE] and no error, on a visible tab. 366 chat tests pass; tsc delta 0 vs main. Refs #1923 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Lint flagged the placeholder param in the useRef initializer. Typing the ref gives the same call signature without an unused binding. Refs #1923 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Refinements after comparing against upstream open-agentsThree changes in 1. Consume
|
| before | now | upstream | |
|---|---|---|---|
| stall window | 20 s | 10 s | n/a (see below) |
| cooldown | 15 s | 8 s | 8 s (STREAM_RECOVERY_MIN_INTERVAL_MS) |
| poll | 5 s | 3 s | n/a |
Safe because of (1): an unnecessary reconnect now costs a request rather than re-rendering content the client already has.
I did not adopt upstream's STREAM_RECOVERY_STALL_MS = 4_000, and it is worth saying why. That constant feeds a scheduler their shouldScheduleStallRecovery unconditionally disables:
export function shouldScheduleStallRecovery(options: {…}): boolean {
void options;
return false;
}So it is not a live stall threshold to copy — upstream has no stall-based recovery at all. It also would not survive our workload: a single legitimate tool call streams nothing for up to ~200 s (measured on the prod repro), so a 4 s window would fire dozens of pointless reconnects per turn.
3. Visibility probe
Added, and it is upstream's only live recovery trigger. A backgrounded tab can have its connection killed silently, and no amount of waiting produces a chunk to time out on — so a visibility check skips the silence window. The cooldown still applies, so a focus-flapping tab cannot spam reconnects.
Why I kept the stall trigger rather than adopting upstream's posture wholesale
Upstream's live triggers are status === "error" and a visibility probe when status === "ready". Our failure mode produces neither: the stream ends with a clean [DONE], so there is no error, and the prod repro had the user watching a visible tab for nine minutes. Adopting upstream's posture exactly would have left our own bug unfixed except on tab re-focus.
Status
- 366 chat tests passing (3 new cases covering the visibility path and its cooldown).
tsc --noEmit: 7 errors, identical to themainbaseline, zero in files this PR touches.eslint: the one error that was mine (unused placeholder param in auseRefinitializer) is fixed in3cb9682f. The remaining 5 inhooks/are pre-existing onmain— includinguseVercelChat.ts'authenticated' is assigned a value but never used, which is line 83 onmainand only shifted to 84 by this diff.
Preview verification against a real dropped stream is still owed.
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
hooks/useChatTransport.ts (1)
82-96: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winConstruct the full reconnect URL before adding
startIndex.In
ai@6.0.165,apiis the base transport URL. The SDK appends/${options.chatId}/streamonly when the callback does not returnapi. This callback always returnsapi, so Line 95 sendsGET {baseUrl}/api/chatinstead ofGET {baseUrl}/api/chat/{chatId}/stream. Recovery therefore cannot reach the resume route. (raw.githubusercontent.com)Build the stream URL from the effective current chat ID, then append
startIndex. This is the same reconnect-path concern raised in the previous review; the current implementation still leaves it unresolved.Proposed fix
- prepareReconnectToStreamRequest: async ({ api }) => { + prepareReconnectToStreamRequest: async ({ api }) => { ... - const url = tail === null ? api : `${api}?startIndex=${tail + 1}`; + const streamApi = + `${api}/${encodeURIComponent(chatIdRef.current)}/stream`; + const url = + tail === null + ? streamApi + : `${streamApi}?startIndex=${tail + 1}`;Verify the dependency behavior with:
#!/bin/bash set -euo pipefail curl -fsSL 'https://raw.githubusercontent.com/vercel/ai/ai%406.0.165/packages/ai/src/ui/http-chat-transport.ts' | sed -n '200,230p'🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@hooks/useChatTransport.ts` around lines 82 - 96, Update prepareReconnectToStreamRequest to construct the full stream endpoint from the effective current chat ID before adding the resume query parameter. Ensure the returned api targets the /{chatId}/stream route, then append startIndex using tailIndexRef.current while preserving the existing authentication headers and null-tail behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@hooks/useChatTransport.ts`:
- Around line 38-41: Reset tailIndexRef at every stream boundary so stale tail
indices cannot affect later reconnects: clear it before each new submission and
when chatId changes, using the existing useChatTransport submission and chatId
lifecycle symbols. Preserve header updates for the active stream, and add a
regression test covering a follow-up turn whose response has no
x-workflow-stream-tail-index header.
---
Duplicate comments:
In `@hooks/useChatTransport.ts`:
- Around line 82-96: Update prepareReconnectToStreamRequest to construct the
full stream endpoint from the effective current chat ID before adding the resume
query parameter. Ensure the returned api targets the /{chatId}/stream route,
then append startIndex using tailIndexRef.current while preserving the existing
authentication headers and null-tail behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 898c0206-d30d-47f9-9137-96cc24ee3356
⛔ Files ignored due to path filters (1)
lib/chat/__tests__/shouldRecoverStalledStream.test.tsis excluded by!**/*.test.*and included bylib/**
📒 Files selected for processing (3)
hooks/useChatTransport.tshooks/useStreamRecovery.tslib/chat/shouldRecoverStalledStream.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- hooks/useStreamRecovery.ts
| // Highest chunk index the server has reported serving us, from the | ||
| // `x-workflow-stream-tail-index` response header. Drives `startIndex` on | ||
| // reconnect so a resume is gap-free rather than a replay. | ||
| const tailIndexRef = useRef<number | null>(null); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | rg '(^|/)useChatTransport\.ts$|useStreamRecovery\.ts$|Chat|chat' | sed -n '1,120p'
echo "== outline useChatTransport =="
if [ -f hooks/useChatTransport.ts ]; then
ast-grep outline hooks/useChatTransport.ts --view expanded || true
wc -l hooks/useChatTransport.ts
fi
echo "== relevant useChatTransport =="
if [ -f hooks/useChatTransport.ts ]; then
sed -n '1,160p' hooks/useChatTransport.ts
fi
echo "== search tail/header/usages =="
rg -n "tailIndexRef|tail-index|x-workflow-stream-tail-index|useStreamRecovery|streamRecovery|chatId|reconnectUrl" hooks use* src || trueRepository: recoupable/chat
Length of output: 16203
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== useVercelChat outline =="
ast-grep outline hooks/useVercelChat.ts --view expanded || true
echo "== useVercelChat relevant sections =="
sed -n '1,120p' hooks/useVercelChat.ts
sed -n '210,260p' hooks/useVercelChat.ts
sed -n '330,415p' hooks/useVercelChat.ts
echo "== useStreamRecovery full =="
cat -n hooks/useStreamRecovery.ts
echo "== package ai version =="
if [ -f package.json ]; then
node -e 'const fs=require("fs"); const p=JSON.parse(fs.readFileSync("package.json","utf8")); console.log(JSON.stringify({dependencies:p.dependencies?.ai, devDependencies:p.devDependencies?.ai}, null, 2))'
fi
echo "== API route hints for tail header/startIndex =="
rg -n "x-workflow-stream-tail-index|x-workflow-stream-tail|startIndex|api/chat/.*/stream|/stream" . --glob '!**/node_modules/**' --glob '!**/.next/**' --glob '!**/dist/**' | sed -n '1,200p'Repository: recoupable/chat
Length of output: 15219
Reset tailIndexRef at each stream boundary.
useChatTransport creates one stable transport instance for the hook lifetime, so tailIndexRef can carry a tail from a previous chat or turn into a later GET /api/chat/{chatId}/stream reconnect. Clear it before each new submission and when chatId changes, or keep the tail scoped to the current stream/response identity. Also add a regression test for a follow-up turn that has no tail header.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@hooks/useChatTransport.ts` around lines 38 - 41, Reset tailIndexRef at every
stream boundary so stale tail indices cannot affect later reconnects: clear it
before each new submission and when chatId changes, using the existing
useChatTransport submission and chatId lifecycle symbols. Preserve header
updates for the active stream, and add a regression test covering a follow-up
turn whose response has no x-workflow-stream-tail-index header.
There was a problem hiding this comment.
3 issues found across 4 files (changes from recent commits).
Confidence score: 3/5
- In
lib/chat/shouldRecoverStalledStream.ts, the stall/cooldown timing can treat long but healthy tool calls as stalled and repeatedly trigger resume requests, creating unnecessary reconnect churn and potential duplicate recovery behavior for a single turn — tighten the recovery gate so resume only fires for truly dropped streams (e.g., require sustained no-progress across polls or a longer cooldown than poll interval). - In
lib/chat/shouldRecoverStalledStream.ts, the visibility-change path can reconnect while a turn is stillsubmitted(before its first stream chunk), so an early 204 can consume the recovery window and leave a real later disconnect unrecovered — align visibility-triggered reconnect checks with the no-chunk/stall safeguards used for normal streaming recovery. - In
hooks/useChatTransport.ts, sharedstartIndexstate can move backward when overlapping responses arrive out of order, so the next reconnect may resume from an older tail index and risk replaying or skipping stream data — makestartIndexmonotonic (ignore lower late indexes or compare-and-set with max).
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="lib/chat/shouldRecoverStalledStream.ts">
<violation number="1" location="lib/chat/shouldRecoverStalledStream.ts:22">
P2: Long legitimate tool calls will repeatedly open resume requests rather than only recovering a dropped connection: after the first 10 s stall, the 8 s cooldown expires before the next 3 s poll. This is specifically a known ~200 s no-output workload; retain a longer/backing-off retry interval or track that a recovery for the current quiet period already succeeded.</violation>
<violation number="2" location="lib/chat/shouldRecoverStalledStream.ts:78">
P2: A visibility event can reconnect a just-submitted turn before its initial stream exists. Because the visibility path skips the no-chunk and stall checks for both `streaming` and `submitted`, a premature 204 consumes the cooldown and can delay recovery of a POST that drops during startup; limiting the immediate visibility probe to an already-streaming turn (or retaining the startup silence window) avoids this race.</violation>
</file>
<file name="hooks/useChatTransport.ts">
<violation number="1" location="hooks/useChatTransport.ts:78">
P2: A reconnect can regress `startIndex` when overlapping transport responses report tail indexes out of order. Since this ref is shared by all responses and the next reconnect uses its current value, a late lower header can make the client replay chunks it already rendered; retaining the maximum observed tail index would preserve gap-free recovery.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
| * Minimum gap between reconnect attempts, so a dead stream isn't retried every | ||
| * tick. Matches upstream's `STREAM_RECOVERY_MIN_INTERVAL_MS`. | ||
| */ | ||
| export const STREAM_RECOVERY_COOLDOWN_MS = 8_000; |
There was a problem hiding this comment.
P2: Long legitimate tool calls will repeatedly open resume requests rather than only recovering a dropped connection: after the first 10 s stall, the 8 s cooldown expires before the next 3 s poll. This is specifically a known ~200 s no-output workload; retain a longer/backing-off retry interval or track that a recovery for the current quiet period already succeeded.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At lib/chat/shouldRecoverStalledStream.ts, line 22:
<comment>Long legitimate tool calls will repeatedly open resume requests rather than only recovering a dropped connection: after the first 10 s stall, the 8 s cooldown expires before the next 3 s poll. This is specifically a known ~200 s no-output workload; retain a longer/backing-off retry interval or track that a recovery for the current quiet period already succeeded.</comment>
<file context>
@@ -1,8 +1,25 @@
+ * Minimum gap between reconnect attempts, so a dead stream isn't retried every
+ * tick. Matches upstream's `STREAM_RECOVERY_MIN_INTERVAL_MS`.
+ */
+export const STREAM_RECOVERY_COOLDOWN_MS = 8_000;
/** The `useChat` statuses that mean a turn is still expected to produce output. */
</file context>
|
|
||
| // A visibility check skips the silence window but keeps the cooldown, so a | ||
| // focus-flapping tab can't spam reconnects. | ||
| if (isVisibilityCheck) return true; |
There was a problem hiding this comment.
P2: A visibility event can reconnect a just-submitted turn before its initial stream exists. Because the visibility path skips the no-chunk and stall checks for both streaming and submitted, a premature 204 consumes the cooldown and can delay recovery of a POST that drops during startup; limiting the immediate visibility probe to an already-streaming turn (or retaining the startup silence window) avoids this race.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At lib/chat/shouldRecoverStalledStream.ts, line 78:
<comment>A visibility event can reconnect a just-submitted turn before its initial stream exists. Because the visibility path skips the no-chunk and stall checks for both `streaming` and `submitted`, a premature 204 consumes the cooldown and can delay recovery of a POST that drops during startup; limiting the immediate visibility probe to an already-streaming turn (or retaining the startup silence window) avoids this race.</comment>
<file context>
@@ -42,12 +67,18 @@ export function shouldRecoverStalledStream({
+
+ // A visibility check skips the silence window but keeps the cooldown, so a
+ // focus-flapping tab can't spam reconnects.
+ if (isVisibilityCheck) return true;
+
if (lastChunkAt === null) return false;
</file context>
| if (isVisibilityCheck) return true; | |
| if (isVisibilityCheck && status === "streaming") return true; |
| const tail = response.headers.get("x-workflow-stream-tail-index"); | ||
| if (tail !== null) { | ||
| const parsed = Number(tail); | ||
| if (Number.isInteger(parsed) && parsed >= 0) tailIndexRef.current = parsed; |
There was a problem hiding this comment.
P2: A reconnect can regress startIndex when overlapping transport responses report tail indexes out of order. Since this ref is shared by all responses and the next reconnect uses its current value, a late lower header can make the client replay chunks it already rendered; retaining the maximum observed tail index would preserve gap-free recovery.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At hooks/useChatTransport.ts, line 78:
<comment>A reconnect can regress `startIndex` when overlapping transport responses report tail indexes out of order. Since this ref is shared by all responses and the next reconnect uses its current value, a late lower header can make the client replay chunks it already rendered; retaining the maximum observed tail index would preserve gap-free recovery.</comment>
<file context>
@@ -61,14 +65,34 @@ export function useChatTransport({
+ const tail = response.headers.get("x-workflow-stream-tail-index");
+ if (tail !== null) {
+ const parsed = Number(tail);
+ if (Number.isInteger(parsed) && parsed >= 0) tailIndexRef.current = parsed;
+ }
+ return response;
</file context>
| if (Number.isInteger(parsed) && parsed >= 0) tailIndexRef.current = parsed; | |
| if (Number.isInteger(parsed) && parsed >= 0) { | |
| tailIndexRef.current = Math.max(tailIndexRef.current ?? -1, parsed); | |
| } |
…ised tail Preview testing of recoupable/api#809 caught this. The route reports x-workflow-stream-tail-index at the moment the read is OPENED, not when it ends: a live read that returned 22 chunks advertised a tail of 9. Resuming at tail + 1 would therefore have replayed 12 chunks the client had already rendered — the exact duplication startIndex exists to avoid. That matches the SDK contract on closer reading: the header is a base for computing absolute positions, and "subsequent retries always resume from the last received chunk". So count the chunks instead. The transport fetch tees the response body, counts SSE frames (excluding the [DONE] terminator), and tracks the absolute index as requestedStartIndex + framesSeen. Reconnect sends that + 1. Best-effort: a torn read just means the next reconnect resumes from the last index counted, which is still ahead of replaying from zero. Refs #1923 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Confidence score: 5/5
- Automated review surfaced no issues in the provided summaries.
- No files require special attention.
Requires human review: Auto-approval blocked by 10 unresolved issues from previous reviews.
Re-trigger cubic
Preview testing on the chat#1924 branch: every reconnect hit
`/api/chat?startIndex=376` and got a 405.
prepareReconnectToStreamRequest receives `api` as the BASE (`…/api/chat`),
not the reconnect URL — the SDK only falls back to `${api}/${id}/stream`
when the callback returns no `api` of its own. Returning one replaces the
whole URL, so appending `?startIndex=N` to the base produced a GET against
the POST-only chat endpoint.
Rebuilds the path from `api` + the `id` the callback is handed.
The rest of the chain was already working in that run: stall detection fired
twice, and the chunk counter produced startIndex 376 then 409 off real
deltas. Only the URL was wrong.
Refs #1923
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Confidence score: 5/5
- Automated review surfaced no issues in the provided summaries.
- No files require special attention.
Requires human review: Auto-approval blocked by 10 unresolved issues from previous reviews.
Re-trigger cubic
…nce id Second preview run: the URL shape was right but every reconnect 404'd — /api/chat/ca264e2f.../stream while the page was on chat d40b5147... The `id` prepareReconnectToStreamRequest receives is the useChat INSTANCE id. For a new chat that is still the client placeholder; the api-minted id arrives later and lives in chatIdRef, which is exactly why that ref exists (useChat captures the transport at mount and never swaps it). So the reconnect was addressing a chat that does not exist. Uses chatIdRef.current — the same value the request body already sends as `chatId`. Refs #1923 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Confidence score: 5/5
- Automated review surfaced no issues in the provided summaries.
- No files require special attention.
Requires human review: Auto-approval blocked by 10 unresolved issues from previous reviews.
Re-trigger cubic
Preview verification — the reconnect works end to endSigned in as the real account on preview
The main stream died at 123 s again — third independent reproduction, on a third account and deployment. The reconnect resumed at index 348 and carried the turn to completion. The turn finished with no duplication, which is the assertion that matters for a resume: Each word rendered exactly once. (
It took three runs, and each found a different bug
Worth stating: all 86 test files passed against every one of those broken URLs. The tests cover the decision to reconnect, not the URL the SDK ultimately builds. Nothing but a live preview run would have caught these. Consistent across all three runs: the stall detector fired reliably, and the chunk counter tracked real positions (376 → 409, 141 → 180 → 209, 348). Two findings worth acting on separately1. 2. The reconnect fired at 17:10:28, before the original stream died at 17:11:48. With a 10 s stall window and 45 s sleeps, legitimate tool-call silence trips the detector, so we briefly ran two concurrent reads of the same stream. Harmless here — resuming from 348 produced no duplication — but it is wasted work. If we want to avoid it, the lever is upstream's approach of probing whether the server still considers the stream live before reconnecting, rather than reconnecting blind. Flagging rather than changing it now, since the current behaviour is correct. |
| fetch: (async (input, init) => { | ||
| const response = await globalThis.fetch(input as RequestInfo, init); | ||
| if (!response.body) return response; | ||
|
|
||
| const url = typeof input === "string" ? input : (input as Request).url; | ||
| const requested = Number(new URL(url, baseUrl).searchParams.get("startIndex") ?? "0"); | ||
| let index = (Number.isInteger(requested) && requested >= 0 ? requested : 0) - 1; | ||
|
|
||
| const [toCaller, toCount] = response.body.tee(); | ||
| void (async () => { | ||
| const reader = toCount.getReader(); | ||
| const decoder = new TextDecoder(); | ||
| let buffered = ""; | ||
| try { | ||
| for (;;) { | ||
| const { done, value } = await reader.read(); | ||
| if (done) break; | ||
| buffered += decoder.decode(value, { stream: true }); | ||
| const lines = buffered.split("\n"); | ||
| buffered = lines.pop() ?? ""; | ||
| for (const line of lines) { | ||
| // `[DONE]` is the SSE terminator, not a stream chunk. | ||
| if (line.startsWith("data: ") && !line.startsWith("data: [DONE]")) { | ||
| index += 1; | ||
| lastChunkIndexRef.current = index; | ||
| } | ||
| } | ||
| } | ||
| } catch { | ||
| // Counting is best-effort; a torn read just means the next | ||
| // reconnect resumes from the last index we did count. | ||
| } finally { | ||
| reader.releaseLock(); | ||
| } | ||
| })(); | ||
|
|
||
| return new Response(toCaller, { | ||
| status: response.status, | ||
| statusText: response.statusText, | ||
| headers: response.headers, | ||
| }); | ||
| }) as typeof globalThis.fetch, | ||
| // Reconnect hits `GET {api}/{chatId}/stream` — recoup-api's resume | ||
| // route, which is authenticated like every other endpoint. Without | ||
| // this the reconnect 401s and a dropped stream stays dropped. | ||
| prepareReconnectToStreamRequest: async ({ api }) => { | ||
| const accessToken = await getAccessToken().catch(() => null); | ||
| const headers: Record<string, string> = {}; | ||
| if (accessToken) headers.Authorization = `Bearer ${accessToken}`; | ||
|
|
||
| // `api` here is the BASE (`…/api/chat`), not the reconnect URL — the | ||
| // SDK only falls back to `${api}/${id}/stream` when we return no | ||
| // `api` of our own. Returning one replaces the whole URL, so the | ||
| // path has to be rebuilt, not appended to: appending the query to | ||
| // the base produced `POST`-only `/api/chat?startIndex=N` and 405s. | ||
| // | ||
| // Built from `chatIdRef`, NOT the `id` the callback is handed: that | ||
| // is the `useChat` INSTANCE id, which for a new chat is still the | ||
| // client placeholder while the api-minted id lives in the ref. Using | ||
| // it reconnected to a chat that does not exist and 404'd. This is the | ||
| // same ref the request body already sends as `chatId`. | ||
| const last = lastChunkIndexRef.current; | ||
| const chatId = chatIdRef.current; | ||
| const url = `${api}/${chatId}/stream${last === null ? "" : `?startIndex=${last + 1}`}`; | ||
|
|
||
| return { headers, api: url }; | ||
| }, | ||
| }), |
There was a problem hiding this comment.
OCP
- actual: net new code added inline to useChatTransport
- required: new lib file for the net new code imported into useChatTransport
… read Addresses both review comments on chat#1924. OCP — the counting fetch and the reconnect-URL construction were net new logic inline in useChatTransport. Extracted to their own lib files, so the hook is wiring again and both units are directly testable: - lib/chat/createChunkCountingFetch.ts — wraps fetch, counts SSE frames off a tee of the body, reports the absolute position. - lib/chat/buildStreamReconnectUrl.ts — pure URL builder. Stale index — the position ref was never reset, so a reconnect could send a startIndex belonging to a previous turn or a different chat and skip chunks the client never saw. Valid, and narrower than it first looks: the counter re-seeds from each request's own startIndex, so it self-corrects on the first frame of any new read. The exposed window is between issuing a request and its first frame — which, with a 10s stall threshold and a slow sandbox start, a reconnect can land in. Closed at both ends: - createChunkCountingFetch reports null the moment it issues a read that has no startIndex, i.e. one starting from chunk zero, before awaiting the response. That covers a new turn and a new chat's first POST. - useChatTransport clears the ref when chatId changes, since the transport is memoised for the lifetime of the hook. 9 new unit tests cover the counting, the seeding, both reset paths, and pass-through. 88 chat test files pass; tsc delta 0 vs main. Refs #1923 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Both review comments addressed in
|
| file | LOC | role |
|---|---|---|
lib/chat/createChunkCountingFetch.ts |
89 | wraps fetch, counts SSE frames off a tee of the body, reports the absolute position |
lib/chat/buildStreamReconnectUrl.ts |
25 | pure URL builder |
hooks/useChatTransport.ts |
101 | wiring only |
Stale index — valid, and I've closed it at both ends
The concern is real. It is also narrower than it first reads, which is worth stating precisely rather than just agreeing: the counter re-seeds from each request's own startIndex, so it self-corrects on the first frame of any new read. A stale value can only be sent in the window between issuing a request and its first frame arriving.
That window is not theoretical. With a 10 s stall threshold and a slow sandbox start, a reconnect can land in it — and then startIndex points past chunks the client never saw, which is a silent gap rather than a visible error. Worse than a replay.
Closed at both ends:
createChunkCountingFetchreportsnullthe moment it issues a read with nostartIndex— i.e. one starting from chunk zero — before awaiting the response. Covers a new turn and a new chat's first POST. The test asserts the reset lands synchronously, since a reset that only arrived with the first frame would leave the window open.useChatTransportclears the ref whenchatIdchanges, because the transport is memoised for the lifetime of the hook and would otherwise carry a position across chats.
Note the comment referenced tailIndexRef; that ref was renamed to lastChunkIndexRef in cb8cea2e when the implementation moved from trusting the response header to counting the wire. Same object, same concern.
Coverage
9 new unit tests: frame counting, [DONE] exclusion, seeding from startIndex, both reset paths, non-reset on a genuine resume, body pass-through, and bodyless (204) responses.
88 chat test files passing; tsc --noEmit delta 0 vs main; eslint clean.
Re-running the live preview verification on this head before I'd call it done — the last three rounds each found something unit tests could not.
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@lib/chat/createChunkCountingFetch.ts`:
- Around line 33-89: Refactor createChunkCountingFetch so it remains under 50
lines and only configures the fetch wrapper. Extract the startIndex/resumesFrom
parsing into a focused helper and move response-body teeing, SSE counting, and
reader cleanup into a separate stream-instrumentation helper, preserving the
existing onPosition behavior and returned response semantics.
- Around line 54-81: Replace the eager `toCount` reader in the response-body tee
flow with a counting `TransformStream` applied to the caller’s consumed branch,
incrementing `index` and invoking `onPosition` only as chunks pass through that
transform. Preserve SSE parsing and `[DONE]` exclusion, and add a regression
test verifying reconnect position reflects only data consumed by the caller.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 5fd474e8-5baf-4ceb-adbd-ac52e4937c94
⛔ Files ignored due to path filters (2)
lib/chat/__tests__/buildStreamReconnectUrl.test.tsis excluded by!**/*.test.*and included bylib/**lib/chat/__tests__/createChunkCountingFetch.test.tsis excluded by!**/*.test.*and included bylib/**
📒 Files selected for processing (3)
hooks/useChatTransport.tslib/chat/buildStreamReconnectUrl.tslib/chat/createChunkCountingFetch.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- hooks/useChatTransport.ts
| export function createChunkCountingFetch({ | ||
| baseUrl, | ||
| onPosition, | ||
| fetchImpl, | ||
| }: ChunkCountingFetchOptions): typeof globalThis.fetch { | ||
| return (async (input: RequestInfo | URL, init?: RequestInit) => { | ||
| const doFetch = fetchImpl ?? globalThis.fetch; | ||
| const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url; | ||
|
|
||
| const startIndexParam = new URL(url, baseUrl).searchParams.get("startIndex"); | ||
| const requested = startIndexParam === null ? null : Number(startIndexParam); | ||
| const resumesFrom = | ||
| requested !== null && Number.isInteger(requested) && requested >= 0 ? requested : null; | ||
|
|
||
| // Reading from the beginning — any earlier position no longer applies. | ||
| if (resumesFrom === null) onPosition(null); | ||
|
|
||
| const response = await doFetch(input, init); | ||
| if (!response.body) return response; | ||
|
|
||
| let index = (resumesFrom ?? 0) - 1; | ||
| const [toCaller, toCount] = response.body.tee(); | ||
|
|
||
| void (async () => { | ||
| const reader = toCount.getReader(); | ||
| const decoder = new TextDecoder(); | ||
| let buffered = ""; | ||
| try { | ||
| for (;;) { | ||
| const { done, value } = await reader.read(); | ||
| if (done) break; | ||
| buffered += decoder.decode(value, { stream: true }); | ||
| const lines = buffered.split("\n"); | ||
| buffered = lines.pop() ?? ""; | ||
| for (const line of lines) { | ||
| // `[DONE]` terminates the SSE response; it is not a stream chunk. | ||
| if (line.startsWith("data: ") && !line.startsWith("data: [DONE]")) { | ||
| index += 1; | ||
| onPosition(index); | ||
| } | ||
| } | ||
| } | ||
| } catch { | ||
| // Counting is best-effort: a torn read just means the next reconnect | ||
| // resumes from the last index we did count, still ahead of replaying. | ||
| } finally { | ||
| reader.releaseLock(); | ||
| } | ||
| })(); | ||
|
|
||
| return new Response(toCaller, { | ||
| status: response.status, | ||
| statusText: response.statusText, | ||
| headers: response.headers, | ||
| }); | ||
| }) as typeof globalThis.fetch; | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Keep createChunkCountingFetch below 50 lines.
createChunkCountingFetch spans 57 lines. Extract request-position parsing and stream instrumentation into focused helpers. Keep this factory responsible only for wrapper setup.
As per path instructions, utility functions must “Keep functions under 50 lines”.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/chat/createChunkCountingFetch.ts` around lines 33 - 89, Refactor
createChunkCountingFetch so it remains under 50 lines and only configures the
fetch wrapper. Extract the startIndex/resumesFrom parsing into a focused helper
and move response-body teeing, SSE counting, and reader cleanup into a separate
stream-instrumentation helper, preserving the existing onPosition behavior and
returned response semantics.
Source: Path instructions
| const [toCaller, toCount] = response.body.tee(); | ||
|
|
||
| void (async () => { | ||
| const reader = toCount.getReader(); | ||
| const decoder = new TextDecoder(); | ||
| let buffered = ""; | ||
| try { | ||
| for (;;) { | ||
| const { done, value } = await reader.read(); | ||
| if (done) break; | ||
| buffered += decoder.decode(value, { stream: true }); | ||
| const lines = buffered.split("\n"); | ||
| buffered = lines.pop() ?? ""; | ||
| for (const line of lines) { | ||
| // `[DONE]` terminates the SSE response; it is not a stream chunk. | ||
| if (line.startsWith("data: ") && !line.startsWith("data: [DONE]")) { | ||
| index += 1; | ||
| onPosition(index); | ||
| } | ||
| } | ||
| } | ||
| } catch { | ||
| // Counting is best-effort: a torn read just means the next reconnect | ||
| // resumes from the last index we did count, still ahead of replaying. | ||
| } finally { | ||
| reader.releaseLock(); | ||
| } | ||
| })(); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
# Locate existing stream-position tests and assertions before adding the
# delayed-consumer regression case.
rg -n -C 4 'createChunkCountingFetch|onPosition|startIndex|lastChunkIndex' \
-g '*.{test,spec}.ts' .Repository: recoupable/chat
Length of output: 8358
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
printf 'Files:\n'
git ls-files | rg 'lib/chat/createChunkCountingFetch\.ts|hooks/useChatTransport\.ts|buildStreamReconnectUrl|chatReconnectUrl' || true
printf '\ncreateChunkCountingFetch outlines:\n'
ast-grep outline lib/chat/createChunkCountingFetch.ts || true
printf '\ncreateChunkCountingFetch lines:\n'
cat -n lib/chat/createChunkCountingFetch.ts | sed -n '1,130p'
printf '\ntransport relevant lines:\n'
cat -n hooks/useChatTransport.ts | sed -n '1,150p'
printf '\nsearch reconnect url usages:\n'
rg -n -C 4 'lastChunkIndexRef|buildStreamReconnectUrl|start.*reconnect|reconnectUrl|startIndex=' hooks lib app -g '*.{ts,tsx}' | sed -n '1,240p'Repository: recoupable/chat
Length of output: 17608
🌐 Web query:
Web Streams API ReadableStream tee backpressure consumer queue size limits source
💡 Result:
In the Web Streams API, the ReadableStream.tee() method creates two branches that consume data from a single underlying source [1][2]. A critical technical limitation of the current tee() implementation is that it does not provide full, robust backpressure to the slower of the two branches [1][3]. Instead, tee() signals backpressure based on the faster consumer [1][4]. When both branches have unread data, the original stream's internal controller queue will eventually fill up, triggering backpressure to the underlying source [1][3]. However, if one branch is consumed significantly faster than the other, the slower branch continues to buffer unread data internally [1]. This internal buffering has no enforced limit or automatic backpressure mechanism, meaning that if one branch is left unconsumed or is consumed very slowly, it will cause that data to persist in memory indefinitely until the memory is exhausted [1][4]. Because of this behavior, the built-in tee() is generally discouraged for use with very large streams or when there is a significant discrepancy in consumption speeds between the two branches [1][3]. While proposals have been made to modify tee() to support synchronization or backpressure tied to the slower branch [5][6][7], the standard behavior remains that unread data on a slow branch will buffer without limit [1][4]. Consequently, developers requiring strict backpressure for independent consumers often need to implement custom stream-branching logic [1][3].
Citations:
- 1: https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream/tee
- 2: https://streams.spec.whatwg.org/
- 3: https://docs.w3cub.com/dom/readablestream/tee
- 4: Add backpressure information to tee, clone mdn/content#16804
- 5: Proposal: ReadableStream tee() backpressure whatwg/streams#1235
- 6: Allow web devs to synchronize branches with tee()? whatwg/streams#1157
- 7: https://lists.w3.org/Archives/Public/public-webapps-github/2022Jun/0355.html
Count chunks on the caller’s return stream so reconnect position always matches consumed data.
toCount can drain ahead of toCaller, so onPosition may report chunks still queued in the caller branch. If a reconnect replaces the caller stream while those chunks remain unread, lastChunkIndexRef.current will resume past data the chat client did not process. Use a counting TransformStream in the caller branch and add a delayed-consumption regression test.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@lib/chat/createChunkCountingFetch.ts` around lines 54 - 81, Replace the eager
`toCount` reader in the response-body tee flow with a counting `TransformStream`
applied to the caller’s consumed branch, incrementing `index` and invoking
`onPosition` only as chunks pass through that transform. Preserve SSE parsing
and `[DONE]` exclusion, and add a regression test verifying reconnect position
reflects only data consumed by the caller.
1 participant
Client half of chat#1923. Depends on api#809 (the resume route) and docs#286 (the contract).
Why
Reproduced on prod 2026-08-02 by replaying the reported prompt through the real UI. The
/api/chatSSE response ended at ~123 s with a cleandata: [DONE]and nofinishchunk, while the workflow ran on to completion.No error, no retry, no visible state change — the composer looked idle, as if the turn had finished. Only a page refresh recovered it.
The SDK already had the machinery
DefaultChatTransport.reconnectToStreamdefaults toGET ${api}/${chatId}/stream. Our transport'sapiis${baseUrl}/api/chat, so the SDK has been prepared to reconnect to exactly the endpoint our docs describe — and which nothing ever implemented. We were missing both halves.What
hooks/useVercelChat.tsresume: true— re-attach to an in-progress response on mount, so returning to a chat mid-turn keeps rendering. WiresuseStreamRecovery.hooks/useStreamRecovery.tsresumeStream(). Also re-checks onvisibilitychange.lib/chat/shouldRecoverStalledStream.tshooks/useChatTransport.tsprepareReconnectToStreamRequestattaches the bearer token.Composed-hook shape rather than growing
useVercelChatinternals, matching how the other chat hooks extend it.Design calls
Silence-based, not duration-based. A turn that is still streaming is healthy however long it runs; a turn that has gone quiet is suspect even if it just started. Keying on elapsed total time would fight the legitimate long turns api#808 just made possible.
A cooldown, so a permanently dead stream isn't retried every tick. Reconnects are cheap but not free, and a 204 ends the loop naturally once the run finishes.
Auth on reconnect is not optional. The resume route authenticates like every other endpoint; without
prepareReconnectToStreamRequestthe reconnect 401s and a dropped stream stays dropped. Easy to miss because the failure looks identical to having no recovery at all.Not included: precise
startIndexapi#809 accepts
startIndexfor a gap-free resume, butuseChatdoes not expose a received-chunk count, so there is no honest value to send. The SDK's default reconnect replays the stream and reconciles by message id, which is correct if slightly wasteful. Passing a guessed index would risk skipping content — worse than replaying it. Wiring an exact index is a follow-up once we track it properly; the route already supports it.Tests
shouldRecoverStalledStream— 7 cases, RED before GREEN: recovers on a stalled streaming turn; does not while chunks arrive; not once the turn isready/error; recovers asubmittedturn that never started; holds off while an attempt is in flight; honours the cooldown; treats a never-seen chunk timestamp as no reason to act.tsc --noEmit: 7 errors, identical to themainbaseline (7), zero in any file this PR touches.One thing I could not verify locally
next buildin my worktree died withWorkerError: Call retries were exceeded— a worker crash, not a type or lint failure, which I attribute to the symlinkednode_modulesin the worktree rather than this diff. CI's build is the authority here; I'd rather flag that than claim a green build I didn't get. Preview verification against a real dropped stream is still owed and will be posted here.🤖 Generated with Claude Code
Summary by cubic
Automatically reconnects dropped chat response streams and resume from the last received chunk so turns keep rendering without duplication. Tightens stall detection and probes on tab visibility to prevent freezes like chat#1923.
Bug Fixes
useStreamRecoveryto detect silence and callresumeStream(); also checks onvisibilitychangeand polls every 3s with a 10s stall window and 8s cooldown.resume: trueinuseChatto reattach to in‑progress streams on mount.lib/chat/createChunkCountingFetchand sendingstartIndex=last+1;prepareReconnectToStreamRequestuseslib/chat/buildStreamReconnectUrlto rebuild${api}/${chatId}/stream, reads the server chat id from a ref, attaches bearer auth, and resets the position on fresh reads and chat switches to avoid stale indices. Unit tests cover both helpers and the decision logic.shouldRecoverStalledStream(unit‑tested) as the pure decision logic.Migration
GET /api/chat/{chatId}/streamfromapi#809(perdocs#286) to be available.Written for commit ea642b3. Summary will update on new commits.
Summary by CodeRabbit