feat(ai): user-scoped stream discovery, history rejoin, SSE hardening#2078
Merged
Conversation
Server Stream Durability & Rejoin epic, PR 5 (leaves 5.1-5.7).
- 5.1: GET /api/ai/chat/active-streams?scope=user — cross-channel discovery
of the caller's own in-flight streams (own rows, heartbeat-filtered, no
parts), for the history tab's streaming badge. New (user_id,status) index.
- 5.2: SidebarHistoryTab badges streaming conversations, seeded from the new
scope=user endpoint and kept fresh via the existing chat:stream_start/
chat:stream_complete socket events. GlobalChatContext.loadConversation now
requests includeStreaming=1 so a clicked streaming conversation's placeholder
row is visible and mergeServerAndPending can replace it with the live stream
once the existing channel-wide attach path (already running) discovers it.
- 5.3: stream-join SSE sends `: ping` comment frames every ~20s so a silent
multi-minute tool call doesn't get reaped by an idle-connection timeout.
- 5.4: when a stream-join 404s (the common cross-instance case — the
multicast registry is per-process), fall back to polling the channel's
parts snapshot at ~1s (matching the server's own checkpoint cadence) for a
near-live view instead of freezing until chat:stream_complete arrives.
- 5.6/5.7 (triaged D-task fixes from PR 4 and PR 1 self-review):
- withAdvisoryLock now resolves a structural `connection_error` outcome for
pool.connect()/try-lock failures instead of throwing, so callers can
distinguish lock-machinery failure from `fn` itself throwing without an
unenforced catch-and-assume convention. startGenerationExclusive branches
on the outcome directly; reconcileMachineStorageSerialized explicitly
rethrows to preserve its existing propagate-and-let-cron-retry behavior.
- rawPartsCount now reflects the raw-stream position the surviving (capped)
checkpoint content actually starts at, not the unconditional raw total —
capping content away from a >5MB checkpoint no longer tells a rejoining
client to skip past (and permanently lose) the raw frames that fed the
dropped content.
Board corrections: marked the two epic-level D tasks resolved by 5.6/5.7 as
completed; filed a new D task for a gap found in review — agent-mode history
rejoin (dashboard/page-sidebar) never opts into includeStreaming=1, unlike
global mode, so a badged agent conversation's click-through won't rejoin the
same way. Noted a pre-existing duplicate D task pair on the epic board.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
|
Warning Review limit reached
Next review available in: 44 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (26)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![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: 28c9078a92
ℹ️ About Codex in GitHub
Your team has set up Codex to 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 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
…artsCount Two real P2 bugs found by automated review on PR #2078: - stream-join-poll-fallback.ts: a poll tick that finds no matching row (stream finished, or the 404 was never a liveness gap to begin with — e.g. a private conversation, filtered by active-streams the same way the join itself was) kept ticking forever. startStreamJoinPollFallback now takes an onNotFound callback, fires it once, and stops the interval; useChannelStreamSocket wires it to remove the now-permanently-stale synthesized store entry instead of leaving it frozen. - checkpoint-serialize.ts: when the single surviving merged part alone already exceeds CHECKPOINT_MAX_SERIALIZED_BYTES (nothing was actually droppable), capPartsToByteBudget reported wasCapped=true, which stream-lifecycle.ts used as the signal to persist survivingFromRawIndex instead of the true raw total. But survivingFromRawIndex there is just where that one part FIRST started — it discards every raw frame that went on to EXTEND it (e.g. thousands of text-delta chunks merged into one giant part), so a rejoining client would re-replay nearly the entire already-seeded content on top of itself. Added a distinct `prefixDropped` field (true only when a prefix was actually trimmed) and keyed the persisted rawPartsCount fallback on that instead. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Fan-out multi-agent self-review (line-by-line, removed-behavior, cross-file, reuse, simplification, efficiency, altitude, conventions) surfaced 3 real issues beyond the two already fixed from automated PR review: - useChannelStreamSocket.ts: the poll fallback's onNotFound callback removed the store entry but never called fireComplete/onStreamComplete. Since broadcastAiStreamComplete is itself best-effort (fire-and-forget), a dropped socket event left nothing to trigger a reload-from-DB — the synthesized bubble could vanish permanently instead of being replaced by the persisted message. onNotFound now fires the same joinFailed:true signal chat:stream_complete's own reload path does. - useChannelStreamSocket.ts: a re-entrant chat:stream_start for a messageId whose poll fallback was already running restarted `pollSeq` at 0, which the store's per-messageId monotonic seq gate would silently drop against the old fallback's already-higher watermark — freezing the UI on stale content for several ticks. The counter is now keyed by messageId outside any single poll-fallback closure, so a restart continues it instead of resetting. - SidebarHistoryTab.tsx: the streaming-badge discovery fetch is a snapshot taken when dispatched but resolving later; replacing state wholesale on resolve silently dropped any chat:stream_start/complete that landed while it was in flight. Deltas are now recorded per fetch cycle and replayed on top of the snapshot (same shape as useChannelStreamSocket's own bootstrapGeneration guard for the identical race), extracted to a pure `applyPendingDeltas` alongside the existing pure transitions. Also: deduplicated the pollControllers abort-loop (independently flagged by two finder angles) into a single `abortAllPolls` helper used at both bulk-teardown sites instead of two copies that could drift. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
….ts conflict PR #2080 (guard placeholder INSERT + harden run-failure classification) merged into master and conflicted with this branch's leaf 5.6 change to start-generation-exclusive.ts. Resolved in favor of this branch's design: withAdvisoryLock's structural connection_error outcome (leaf 5.6) means startGenerationExclusive never catches withAdvisoryLock's rejection at all, only branches on its resolved outcome. That makes #2080's runSettled flag — built to disambiguate a caught rejection's cause before deciding whether re-invoking run unlocked was safe — provably redundant here: with no catch site, there is no reclassification step left for it to guard, for any rejection cause including the "post-run lock release throws" edge case #2080 added a test for. Verified directly: all 21 tests in start-generation-exclusive.test.ts, including that edge case, pass unchanged against the simpler shape. Updated the file's docstring and inline comments to explain the reconciliation for future readers of either PR's history. The rest of the merge (both chat routes' try/catch around the placeholder INSERT, CI config, coverage-ratchet additions) applied cleanly with no conflicts and is orthogonal to this branch's changes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WDuPiAGSLXE3vxC59is3Ro
…t conflict PR #2077 (materialize interrupted streams on dead-row reap) merged into master right after the previous #2080 merge landed, and touched the same GET /api/ai/chat/active-streams route this branch's leaf 5.1 added the scope=user discovery mode to. route.ts itself auto-merged cleanly: leaf 5.1's scope=user branch is untouched, and #2077's lazy-reap loop (materializeInterruptedStream on provably-dead rows) is correctly scoped to the pre-existing channelId branch only — the two modes don't interact. Only the test file conflicted, on the vi.hoisted() mock declaration: leaf 5.1's tests need mockWhere (asserting the scope=user WHERE clause), #2077's tests need mockMaterializeInterruptedStream (asserting the reap side effect). Both are exercised elsewhere in the now-merged file, so both are kept in one combined vi.hoisted() block. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WDuPiAGSLXE3vxC59is3Ro
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
1 participant
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
PR 5 of the Server Stream Durability & Rejoin epic — leaves 5.1-5.7. Depends on (already merged): #2076 (status column + includeStreaming), #2073 (time-based checkpoints + rawPartsCount).
GET /api/ai/chat/active-streams?scope=user— cross-channel discovery of the caller's own in-flight streams (own rows, heartbeat-filtered, noparts), backing the history tab's streaming badge. New(user_id, status)index (migration 0211).SidebarHistoryTabbadges streaming conversations, seeded fromscope=userand kept fresh via the existingchat:stream_start/chat:stream_completesocket events.GlobalChatContext.loadConversationnow requestsincludeStreaming=1so a clicked streaming conversation's placeholder row is visible;mergeServerAndPending(already merged in feat(ai): assistant message row at stream start + status column (streaming|complete|interrupted) #2076) replaces it with the live stream once the existing channel-wide attach path discovers it — no new attach machinery needed.stream-joinSSE sends: pingcomment frames every ~20s so a silent multi-minute tool call survives idle-connection reaping by proxies/load balancers.chat:stream_completefinally arrives. Stopgap until the AI SDK 7 Phase 3 durable transport.withAdvisoryLocknow resolves a structuralconnection_erroroutcome forpool.connect()/try-lock failures instead of throwing, so callers can distinguish lock-machinery failure fromfnitself throwing without an unenforced catch-and-assume convention.startGenerationExclusivebranches on the outcome directly;reconcileMachineStorageSerializedexplicitly rethrows to preserve its existing propagate-and-let-cron-retry behavior for its own caller.rawPartsCountnow reflects the raw-stream position the surviving (capped) checkpoint content actually starts at, not the unconditional raw total — capping content away from a >5MB checkpoint no longer tells a rejoining client to skip past (and permanently lose) the raw frames that fed the dropped content.Review fixes (post-open)
Two real P2 bugs found by automated review, both fixed and replied to inline:
startStreamJoinPollFallbacknow takes anonNotFoundcallback that stops the interval and fires once;useChannelStreamSocketwires it to drop the now-permanently-stale synthesized store entry.wasCappedwastruebutsurvivingFromRawIndexonly marked where that one part first started — discarding every raw frame that went on to extend it (e.g. thousands of text-delta chunks merged into one giant part). A rejoining client would have re-replayed nearly the entire already-seeded content on top of itself. Added a distinctprefixDroppedfield and keyed the persistedrawPartsCountfallback on that instead ofwasCapped.Proactive self-review after that (8 finder angles: line-by-line, removed-behavior, cross-file tracer, reuse, simplification, efficiency, altitude, CLAUDE.md conventions — run as parallel background agents) surfaced 3 more real issues, all fixed:
onNotFoundremoved the store entry but never calledonStreamComplete.broadcastAiStreamCompleteis itself best-effort (fire-and-forget); if that socket event is also dropped, this poll noticing the row disappeared was the ONLY remaining signal, and without firing the samejoinFailed:truereload-from-DB signal, the message could vanish with nothing replacing it — worse than the pre-poll-fallback behavior. Fixed to firefireComplete(messageId, conversationId, { joinFailed: true }).chat:stream_startfor a messageId already being polled aborts the old poll fallback and starts a new one; a fresh localpollSeqcounter restarting at 0 would have its early writes silently dropped by the store's per-messageId monotonic seq gate against the old fallback's already-higher watermark, freezing the UI for several ticks. Counter is now keyed by messageId outside any single poll-fallback closure.scope=userdiscovery fetch is a snapshot taken when dispatched but resolving later; a blind replace on resolve silently dropped anychat:stream_start/completethat landed while it was in flight. Deltas are now recorded per fetch cycle and replayed on top of the snapshot (same shape asuseChannelStreamSocket's ownbootstrapGenerationguard for the identical race), extracted to a pureapplyPendingDeltas.Also deduplicated the
pollControllersabort-loop (flagged independently by two finder angles) into oneabortAllPollshelper instead of two copies that could drift.Filed two more
D —tasks at the epic level for lower-severity, out-of-scope-for-now findings: the poll fallback fetches the whole channel's payload every ~1s instead of a messageId-filtered query (rd7gfajzpnvmgk7ijrv68j4n), andactive-streams/route.ts'schannelId/scope=userbranches duplicate their row-fetch+liveness-filter query pattern (ih85mvlv9xwxds1qaggpurto).Merged master twice to resolve real conflicts (post-open)
Two unrelated PRs landed on master in quick succession after this PR opened, both touching files this PR also changed:
start-generation-exclusive.ts: it added arunSettledflag guarding a try/catch aroundwithAdvisoryLockto stop a caught rejection from being misclassified as "lock never engaged" and re-invokingruna second time. Resolved in favor of this PR's leaf 5.6 design instead:withAdvisoryLocknow resolves a structuralconnection_erroroutcome rather than throwing for lock-machinery failures, sostartGenerationExclusivenever catches its rejection at all — it only branches on the resolved outcome. With no catch site, there is no reclassification step left forrunSettledto guard, for any rejection cause including the "post-run lock release throws" edge case fix(ai): guard placeholder INSERT + harden run-failure classification #2080 added a regression test for. Verified directly: all 21 tests instart-generation-exclusive.test.ts, including that edge case, pass unchanged against the simpler shape. Left a comment in the resolved file explaining the reconciliation for future readers of either PR's history. The rest of fix(ai): guard placeholder INSERT + harden run-failure classification #2080's changes (try/catch around the placeholder INSERT in both chat routes) applied cleanly, are orthogonal to this PR, and are still in effect.active-streams/__tests__/route.test.ts'svi.hoisted()mock block — leaf 5.1's tests needmockWhere, feat(ai): materialize interrupted streams on dead-row reap #2077's tests needmockMaterializeInterruptedStream; both are exercised elsewhere in the merged file, so both are kept.route.tsitself auto-merged cleanly: leaf 5.1'sscope=userbranch and feat(ai): materialize interrupted streams on dead-row reap #2077's lazy-reap loop don't interact (the reap loop is scoped to the pre-existingchannelIdbranch only).Full monorepo typecheck and both directly-conflicted test files re-verified green after each merge.
Board corrections
D —tasks resolved by 5.6/5.7 as completed (lh1zuj3i6nfi5igcl690d7yc,fyyyo59td8si62gwxcxyk19l).D —task at the epic level (co2as25wcpme4m4gxqu4zgcj): agent-mode history rejoin (dashboard/page-sidebar) never opts intoincludeStreaming=1, unlike global mode —SidebarHistoryTab's badge lights up correctly for agent conversations too (discovery is user-wide), but click-through won't rejoin the same way sinceusePageAgentDashboardStore/usePageAgentSidebarState'sloadConversationweren't in this leaf's stated scope (SidebarHistoryTab + GlobalChatContext). Real, unscheduled, not blocking.D —tasks already on the epic board ("Live-connected clients don't badge an interrupted stream until reload...",ixpwr76xepu2x9v4pxgksyhz+gg2b2b8getkro3n1erfksj2v) — worth the planner's dedup pass.Test plan
bun run typecheck— clean (full monorepo, all 16 turbo tasks); re-verified after review fixes.bun run test:unit(packages/lib + web) — 926/930 test files, 13671/13693 tests pass. The 3 failing files (admin-role-version.test.ts,grouping.test.ts,activity-tools.test.ts) are untouched by this PR and fail identically on a cleanmastercheckout in this environment (missing Postgres role"test"for DB-integration tests; one timezone-dependent grouping test) — confirmed viagit stash.packages/dbfull suite — clean except the same class of pre-existing DB-role-dependent integration tests (activity-logs-compliance.test.ts, needstest-with-db.sh, not plainvitest run).stream-joinroute's "given a silent multi-minute tool call, should send a ping on every tick" advances 3× the ping interval), but I do not have staging access in this session.Not merging — R review must PASS first.
https://claude.ai/code/session_01WDuPiAGSLXE3vxC59is3Ro