feat(ai): materialize interrupted streams on dead-row reap#2077
Merged
Conversation
Turns a provably-dead ai_stream_sessions row into an honest, terminal `status: 'interrupted'` assistant message instead of a silent loss. - materializeInterruptedStream(row): builds the message payload from the row's parts snapshot (reusing buildAssistantPersistencePayload, the same primitive execute-end/onFinish already use), upserts chat_messages or messages (routed by parseGlobalChannelId) guarded by a setWhere ne(status, 'complete') on the conflict clause (the #2022 invariant, enforced atomically), then settles the session row terminal and broadcasts stream_complete({aborted: true}). Never throws; a failed step leaves the row eligible for the next pass to retry. - Wired into all three reap paths: stream-takeover.ts's reconcile set (now per-row instead of a bulk UPDATE), stream-abort-mark.ts's reconcileDeadStreamRows (re-selects full rows fresh), and a new lazy sweep in GET /api/ai/chat/active-streams (fire-and-forget, since that route already reads every streaming row on the channel). - Client: ConversationMessage.status threads to MessageRenderer, which renders an "Interrupted" badge + retry hint (including for a message with zero content — an honest empty marker, not silence). - 100% branch coverage on materialize-interrupted-stream.ts, gated via a new per-glob vitest threshold per the epic's own constraints. Filed a D-task at the epic level (gg2b2b8getkro3n1erfksj2v) for a narrower, non-blocking gap: a tab left open through the interruption doesn't badge it until reload (onStreamComplete doesn't thread `aborted` through synthesizeAssistantMessage) — the standard fetch/reload path already shows the badge correctly via the server's existing status field. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB
Contributor
📝 WalkthroughWalkthroughThe change materializes provably dead AI streams as interrupted assistant messages, reconciles them across active-stream, abort, and takeover paths, renders interruption and retry affordances, and prevents undo operations from modifying in-flight streaming rows. ChangesInterrupted stream lifecycle
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant ActiveStreamsRoute
participant StreamLiveness
participant Materializer
participant MessageStore
participant SessionStore
participant ChatRenderer
ActiveStreamsRoute->>StreamLiveness: evaluate authorized streaming rows
StreamLiveness->>Materializer: materialize provably dead rows
Materializer->>MessageStore: upsert interrupted assistant message
Materializer->>SessionStore: settle session as aborted
MessageStore-->>ChatRenderer: provide interrupted message status
ChatRenderer-->>ChatRenderer: render interruption and retry affordance
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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 |
![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: 06e95f351a
ℹ️ 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".
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (3)
apps/web/src/lib/ai/core/stream-abort-mark.ts (1)
290-317: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider materializing rows concurrently instead of sequentially.
The
status='streaming're-check before materializing is a solid safeguard. However, theforloop awaitsmaterializeInterruptedStream(row)one row at a time; since this function is documented to never throw,Promise.all(rows.map(...))would let independent rows materialize concurrently with no loss of fault isolation. This matters because this path runs as part oftakeOverConversationStreams's blocking send flow (viaawaitAbortSettled→reconcileDeadStreamRows).⚡ Proposed refactor
- for (const row of rows) { - await materializeInterruptedStream(row); - } + await Promise.all(rows.map((row) => materializeInterruptedStream(row)));🤖 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 `@apps/web/src/lib/ai/core/stream-abort-mark.ts` around lines 290 - 317, Update the row materialization loop in reconcileDeadStreamRows to invoke materializeInterruptedStream for all fetched rows concurrently via Promise.all, preserving the existing no-throw fault-isolation behavior and awaiting completion before returning.apps/web/src/lib/ai/core/stream-takeover.ts (2)
1-1: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winSequential materialization loops in both reconciliation paths. Both
takeOverConversationStreamsandreconcileDeadStreamRowsprocess independent rows withawait materializeInterruptedStream(...)one at a time inside aforloop, even though the function is documented to never throw. Since both loops sit on the takeover flow that can block a user's send, switching toPromise.alllets independent rows materialize concurrently without losing per-row fault isolation.
apps/web/src/lib/ai/core/stream-takeover.ts#L125-149: replace thefor (const messageId of reconcile) { ... await materializeInterruptedStream(...) }loop withawait Promise.all(reconcile.map(...)).apps/web/src/lib/ai/core/stream-abort-mark.ts#L290-317: replace thefor (const row of rows) { await materializeInterruptedStream(row); }loop withawait Promise.all(rows.map((row) => materializeInterruptedStream(row))).🤖 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 `@apps/web/src/lib/ai/core/stream-takeover.ts` at line 1, Update both reconciliation loops, takeOverConversationStreams and reconcileDeadStreamRows, to materialize independent rows concurrently with Promise.all over map callbacks. Preserve materializeInterruptedStream’s per-row fault isolation and existing behavior while removing sequential awaits from each for loop.
125-149: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winSame sequential-materialization pattern as
stream-abort-mark.ts.This loop awaits
materializeInterruptedStreamone row at a time on the send-blocking takeover path. Since the function is documented to never throw, running these concurrently would reduce latency on the (rare but real) crashed-stream takeover path.⚡ Proposed refactor
- const rowById = new Map(rows.map((r) => [r.messageId, r])); - for (const messageId of reconcile) { - const row = rowById.get(messageId); - if (!row) continue; - await materializeInterruptedStream({ - messageId, - channelId, - conversationId, - userId: row.userId, - parts: row.parts, - }); - } + const rowById = new Map(rows.map((r) => [r.messageId, r])); + await Promise.all(reconcile.map((messageId) => { + const row = rowById.get(messageId); + if (!row) return undefined; + return materializeInterruptedStream({ + messageId, + channelId, + conversationId, + userId: row.userId, + parts: row.parts, + }); + }));🤖 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 `@apps/web/src/lib/ai/core/stream-takeover.ts` around lines 125 - 149, Update the reconciliation loop in the takeover path to materialize all rows concurrently rather than awaiting each materializeInterruptedStream call sequentially. Preserve the existing row lookup and skip-missing-row behavior, and await completion of the full batch before continuing.
🤖 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 `@apps/web/src/lib/ai/core/materialize-interrupted-stream.ts`:
- Around line 58-64: Move the payload construction and tool-call/tool-result
JSON serialization in materializeInterruptedStream inside the existing try
block, including the now-dependent initialization as needed. Ensure errors from
buildAssistantPersistencePayload or JSON.stringify reach the function’s existing
error handling so the void invocation does not produce an unhandled rejection.
---
Nitpick comments:
In `@apps/web/src/lib/ai/core/stream-abort-mark.ts`:
- Around line 290-317: Update the row materialization loop in
reconcileDeadStreamRows to invoke materializeInterruptedStream for all fetched
rows concurrently via Promise.all, preserving the existing no-throw
fault-isolation behavior and awaiting completion before returning.
In `@apps/web/src/lib/ai/core/stream-takeover.ts`:
- Line 1: Update both reconciliation loops, takeOverConversationStreams and
reconcileDeadStreamRows, to materialize independent rows concurrently with
Promise.all over map callbacks. Preserve materializeInterruptedStream’s per-row
fault isolation and existing behavior while removing sequential awaits from each
for loop.
- Around line 125-149: Update the reconciliation loop in the takeover path to
materialize all rows concurrently rather than awaiting each
materializeInterruptedStream call sequentially. Preserve the existing row lookup
and skip-missing-row behavior, and await completion of the full batch before
continuing.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: a434bdf7-a8e1-4a87-aba4-2c614fc4522b
📒 Files selected for processing (13)
apps/web/src/app/api/ai/chat/active-streams/__tests__/route.test.tsapps/web/src/app/api/ai/chat/active-streams/route.tsapps/web/src/components/ai/shared/chat/MessageRenderer.tsxapps/web/src/components/ai/shared/chat/__tests__/MessageRenderer.interrupted.test.tsxapps/web/src/components/ai/shared/chat/message-types.tsapps/web/src/lib/ai/core/__tests__/materialize-interrupted-stream.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-mark.test.tsapps/web/src/lib/ai/core/__tests__/stream-takeover.test.tsapps/web/src/lib/ai/core/materialize-interrupted-stream.tsapps/web/src/lib/ai/core/stream-abort-mark.tsapps/web/src/lib/ai/core/stream-liveness.tsapps/web/src/lib/ai/core/stream-takeover.tsapps/web/vitest.config.ts
…ted-badge coupling Multi-angle review of PR #2077 surfaced several real issues, fixed here: - CRITICAL: stream-takeover.ts's local reconcile set conflated "provably dead" with "we just aborted it ourselves" (still alive moments ago, about to run its own onFinish). Materializing the latter raced that natural write with an older debounced checkpoint and could silently overwrite fresher content with staler content. Now only provably-dead rows are materialized; just-aborted-but-alive rows get the original cheap session-row wipe and their own generation's onFinish persists the real content, matching leaf 3.1's "only isProvablyDead rows" spec. - materialize-interrupted-stream.ts: moved payload-building inside the try/catch (the module's own "never throws" contract wasn't actually enforced); switched to the same extractStructuredContentFromParts pipeline saveMessageToDatabase/saveGlobalAssistantMessageToDatabase use, so file/data parts and ordering survive materialization instead of degrading to flat text; re-synced conversationId on conflict-update, matching the normal terminal-write path; logs broadcast failures instead of silently swallowing them. - active-streams route: the lazy sweep now runs over the subscription-authorized row set, not the raw per-channel query — a user with mere page-view access could otherwise trigger a reap side effect for another member's private conversation. - stream-abort-mark.ts / stream-takeover.ts: per-row reconciliation now runs concurrently (Promise.all) instead of sequentially — each row is independent and materializeInterruptedStream never throws, so there is no correctness reason to serialize a mass-crash-recovery batch. - MessageRenderer.tsx / CompactMessageRenderer.tsx: the interrupted badge was nested inside the text-block render, so a message interrupted mid-tool-call (no trailing text) never showed it. Moved to the message level, independent of which part groups exist. CompactMessageRenderer — the actual sidebar/global-assistant surface — never got the badge at all; added parity coverage. - ai-undo-service.test.ts: added the 3 tests D.2 (epic board, tza273pegokkzpd7ut5qvxky) flagged as claimed-tested-but-uncovered — previewAiUndo's count query and both of executeAiUndo's soft-delete updates now assert the ne(status,'streaming') guard is actually present. Filed two epic-level D-tasks for real-but-unscheduled gaps found along the way (not blocking): mention-notifications not firing for a materialized interrupted reply, and a live-tab socket-immediacy gap for the badge. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB
Owner
Author
|
Acknowledging CodeRabbit's 3 nitpick suggestions (sequential-vs-concurrent materialization in |
Owner
Author
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 4ff8b9c723
ℹ️ 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".
A follow-up review of the previous fix commit (which moved the
Interrupted badge out of TextBlock to a message-level render, so it
shows regardless of which part-group is last) found it dropped the
actual retry BUTTON for exactly the messages that motivated that move:
an interrupted message with empty or tool-only content never renders a
TextBlock, and the retry button previously lived only in TextBlock's
own footer. The message-level block had a text hint ("retry to
continue") but no actionable control.
Added a real retry button to the message-level interrupted block in
both MessageRenderer.tsx and CompactMessageRenderer.tsx, gated on
`!hasNonEmptyTextBlock` so a message that DID stream real text before
dying — which already gets a retry button from TextBlock's footer —
doesn't show two.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (1)
apps/web/src/components/ai/shared/chat/MessageRenderer.tsx (1)
447-455: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winProvide an explicit accessible name for the icon-only retry button.
While the
titleattribute gives a visual tooltip and can sometimes act as a fallback, explicitly setting anaria-labelensures all screen readers can consistently identify the control.
apps/web/src/components/ai/shared/chat/MessageRenderer.tsx#L447-L455: Addaria-label="Retry this message"to the Button props.apps/web/src/components/ai/shared/chat/CompactMessageRenderer.tsx#L445-L453: Addaria-label="Retry this message"to the Button props here as well.🤖 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 `@apps/web/src/components/ai/shared/chat/MessageRenderer.tsx` around lines 447 - 455, Add an explicit aria-label of “Retry this message” to the icon-only retry Button in MessageRenderer.tsx at lines 447-455 and CompactMessageRenderer.tsx at lines 445-453, while preserving the existing title and retry 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.
Nitpick comments:
In `@apps/web/src/components/ai/shared/chat/MessageRenderer.tsx`:
- Around line 447-455: Add an explicit aria-label of “Retry this message” to the
icon-only retry Button in MessageRenderer.tsx at lines 447-455 and
CompactMessageRenderer.tsx at lines 445-453, while preserving the existing title
and retry behavior.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 631821f3-2f9a-4cca-b97b-7ec984a83f61
📒 Files selected for processing (12)
apps/web/src/app/api/ai/chat/active-streams/__tests__/route.test.tsapps/web/src/app/api/ai/chat/active-streams/route.tsapps/web/src/components/ai/shared/chat/CompactMessageRenderer.tsxapps/web/src/components/ai/shared/chat/MessageRenderer.tsxapps/web/src/components/ai/shared/chat/__tests__/CompactMessageRenderer.interrupted.test.tsxapps/web/src/components/ai/shared/chat/__tests__/MessageRenderer.interrupted.test.tsxapps/web/src/lib/ai/core/__tests__/materialize-interrupted-stream.test.tsapps/web/src/lib/ai/core/__tests__/stream-takeover.test.tsapps/web/src/lib/ai/core/materialize-interrupted-stream.tsapps/web/src/lib/ai/core/stream-abort-mark.tsapps/web/src/lib/ai/core/stream-takeover.tsapps/web/src/services/api/__tests__/ai-undo-service.test.ts
🚧 Files skipped from review as they are similar to previous changes (7)
- apps/web/src/app/api/ai/chat/active-streams/route.ts
- apps/web/src/app/api/ai/chat/active-streams/tests/route.test.ts
- apps/web/src/lib/ai/core/stream-abort-mark.ts
- apps/web/src/lib/ai/core/stream-takeover.ts
- apps/web/src/lib/ai/core/tests/materialize-interrupted-stream.test.ts
- apps/web/src/lib/ai/core/materialize-interrupted-stream.ts
- apps/web/src/lib/ai/core/tests/stream-takeover.test.ts
…pplied onEdit/onDelete CI (ci / Unit Tests) failed both interrupted-affordance test files on the same test: "given an interrupted message WITH real text content and a retry handler, shows exactly ONE retry button" expected 1 button, found 0. Root cause: TextBlock's/CompactTextBlock's footer (MessageActionButtons, which is where its own retry button lives) only renders when BOTH onEdit AND onDelete are supplied — not onRetry alone. The test passed only onRetry, so TextBlock's footer never rendered at all, and the message-level button correctly stayed hidden (hasNonEmptyTextBlock=true), leaving zero buttons — a bug in the test's own setup, not in the retry-button fix from commit 9470bd2. Added onEdit/onDelete mocks to both tests so TextBlock's footer renders as it would on the real chat surfaces (which always wire edit/delete/retry together), correctly exercising the "don't duplicate the button" guard. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB
Two more real findings from a Codex re-review (requested after the previous convergence pass), both in materialize-interrupted-stream.ts: 1. The setWhere guard only excluded 'complete', not 'interrupted'. A row whose owning generation cleanly finished (onFinish already persisted the FULL correct content as status='interrupted') but whose ai_stream_sessions terminal write then failed separately (fire-and- forget) stays eligible for a later dead-row sweep. `!= 'complete'` would let that sweep overwrite the already-correct interrupted message with an older debounced checkpoint. Tightened to `== 'streaming'` — a true compare-and-swap: a row leaves 'streaming' exactly once, so this can never re-touch an already-terminal row in either direction. 2. The defensive insert-if-missing branch (only reached if PR 2's placeholder insert never happened) stamped `createdAt: now` (reap time) instead of the stream's actual start. A recovered reply could then sort after a user's later follow-up message. Added `startedAt` to `MaterializableStreamRow` and threaded it through all three call sites (stream-takeover.ts, stream-abort-mark.ts's reconcileDeadStreamRows, the active-streams lazy sweep — all three already had it in hand from their existing SELECTs, bar one new column added to reconcileDeadStreamRows' query). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB
CodeRabbit nitpick: the icon-only retry button (added for empty/tool-only interrupted messages) relied on `title` alone, which isn't a reliable accessible name across screen readers. Added an explicit aria-label matching the existing title text. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB
Owner
Author
|
Acknowledging CodeRabbit's accessibility nitpick (icon-only retry button relying on |
Owner
Author
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: fb8f3349c9
ℹ️ 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".
Codex review finding: takeOverConversationStreams's Promise.all over materializeInterruptedStream had no success signal, so a per-row DB failure was invisible to the caller — every id in `reconcile` was counted as `reconciled` regardless of whether the row actually settled. The old bulk-UPDATE design returned `reconciled: []` on any failure; this PR's per-row restructure silently dropped that accuracy. The same gap existed in the cross-instance path (remotelyReconciled was read straight from decideAbortOutcome's eligibility judgment, never from what reconcileDeadStreamRows actually did) and in reconcileDeadStreamRows itself. - materializeInterruptedStream now returns Promise<boolean> — true only if both the message write AND the session-row settle succeeded. - reconcileDeadStreamRows now returns Promise<string[]> — the messageIds it actually materialized, a subset of what it was asked to reconcile. - stream-takeover.ts's local reconcile path now filters provablyDead by actual materialize success, and the justAbortedStillAlive bulk wipe only contributes to `locallyReconciled` when it doesn't throw. - The cross-instance path now sets remotelyReconciled from reconcileDeadStreamRows's return value instead of the eligibility set. Updated all affected tests (several `.resolves.toBeUndefined()` assertions were factually wrong the moment the return type changed to boolean/array) and added regression tests for the accurate-reporting behavior itself (partial-failure batches, read failures). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB
Contributor
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
apps/web/src/lib/ai/core/materialize-interrupted-stream.ts (1)
174-205: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winapps/web/src/lib/ai/core/materialize-interrupted-stream.ts:174-205 — Treat a zero-row settle as unsuccessful
db.update(...).where(...)can return 0 when another path already terminalized the row. CheckrowCountand only returntruewhen this pass actually settles the session; add a regression test for the zero-row case.🤖 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 `@apps/web/src/lib/ai/core/materialize-interrupted-stream.ts` around lines 174 - 205, Update the session-settling update in materializeInterruptedStream to inspect its affected-row count and set settled to false when the conditional update changes zero rows, including when another path already terminalized the session. Preserve the existing exception handling and return settled only as true when this invocation actually updates the row, and add a regression test covering the zero-row case.
🤖 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.
Outside diff comments:
In `@apps/web/src/lib/ai/core/materialize-interrupted-stream.ts`:
- Around line 174-205: Update the session-settling update in
materializeInterruptedStream to inspect its affected-row count and set settled
to false when the conditional update changes zero rows, including when another
path already terminalized the session. Preserve the existing exception handling
and return settled only as true when this invocation actually updates the row,
and add a regression test covering the zero-row case.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: d5aa5a9e-984d-4360-bd8f-804ac4497c18
📒 Files selected for processing (7)
apps/web/src/lib/ai/core/__tests__/abort-stream-anywhere-pending.test.tsapps/web/src/lib/ai/core/__tests__/materialize-interrupted-stream.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-mark.test.tsapps/web/src/lib/ai/core/__tests__/stream-takeover.test.tsapps/web/src/lib/ai/core/materialize-interrupted-stream.tsapps/web/src/lib/ai/core/stream-abort-mark.tsapps/web/src/lib/ai/core/stream-takeover.ts
🚧 Files skipped from review as they are similar to previous changes (5)
- apps/web/src/lib/ai/core/stream-abort-mark.ts
- apps/web/src/lib/ai/core/tests/stream-abort-mark.test.ts
- apps/web/src/lib/ai/core/stream-takeover.ts
- apps/web/src/lib/ai/core/tests/materialize-interrupted-stream.test.ts
- apps/web/src/lib/ai/core/tests/stream-takeover.test.ts
…ccess CodeRabbit finding: a conditional UPDATE (WHERE messageId = X AND status = 'streaming') does not throw when it matches zero rows — it succeeds and changes nothing. That happens when a concurrent reap (another instance's takeover, or a second sweep racing this one) already settled the exact same row first. The previous "settled = true unless it throws" logic would then credit this call with settling a row it didn't actually touch. Now checks `(result.rowCount ?? 0) > 0`, matching the codebase's own established pattern for the same class of conditional update (compaction-repository.ts's version-guarded CAS). Added regression tests for both the zero-rowCount case and the defensive `?? 0` fallback (a driver/mock that omits rowCount entirely). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB
Owner
Author
|
Acknowledging CodeRabbit's outside-diff-range finding on |
Owner
Author
|
@codex address that feedback |
Summary
Testing
No commit was created and no PR was opened because there were no code changes to commit. |
Owner
Author
|
Thank you for the independent verification — confirms all 5 outstanding review findings (eligibility guard, insert timestamp, session-settle rowCount check, reconciliation-accuracy reporting, and retry-button accessibility) are correctly addressed on the current branch (commit 2f11e1f) with no further code changes needed. Leaving the 5 open threads unresolved per this session's convention (not self-resolving fixes made during this convergence pass) — ready for human sign-off. |
Owner
Author
|
All 6 review threads are now resolved. The 5 Codex-filed threads (structured-content pipeline, badge/retry on non-text-final groups, compare-and-swap guard, insert timestamp, reconciliation-accuracy reporting) were fixed and independently re-verified by Codex (09:46 UTC) hours before this convergence pass concluded; per the auto-resolve-safety rule they are resolved now because the fixes predate the current pass, not because they were fixed during it. CodeRabbit's thread was already self-resolved by CodeRabbit on re-scan. Three consecutive full re-scans (checks, mergeability, threads, comments) after resolution show: all CI green, mergeStateStatus CLEAN/MERGEABLE, 0/6 unresolved threads, no new comments. Ready for merge. |
2witstudios
added a commit
that referenced
this pull request
Jul 15, 2026
…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
6 tasks
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 3 of the "Server Stream Durability & Rejoin" epic — dependency (status column, PR #2076) merged, branch rebased onto current master.
materializeInterruptedStream(row)(apps/web/src/lib/ai/core/materialize-interrupted-stream.ts): from a provably-deadai_stream_sessionsrow, builds the message payload from its parts snapshot (reusingbuildAssistantPersistencePayload+extractStructuredContentFromParts, the same pipeline execute-end/onFinish use — preserves file/data parts and ordering, not just flat text), upsertschat_messages/messagestostatus: 'interrupted'guarded by a compare-and-swapsetWhere: eq(status, 'streaming')on the conflict clause (a row leaves'streaming'exactly once, so this can never re-touch an already-terminal row whether it landed at'complete'or'interrupted'by the normal path), re-syncsconversationIdon conflict, uses the stream's realstartedAt(not reap time) for the defensive insert-if-missing branch, then settles the session row terminal (checking actualrowCount, not just no-throw — a conditional UPDATE that matches zero rows because a concurrent reap already settled it does not throw) and broadcastsstream_complete({aborted: true}). Returns whether the row was genuinely driven terminal, never throws.stream-takeover.ts's reconcile set,stream-abort-mark.ts'sreconcileDeadStreamRows, and a new lazy sweep inGET /api/ai/chat/active-streams. Critical fix during review: the takeover's local reconcile set used to conflate "provably dead" (heartbeat stale — no generation will ever write this row again) with "we just aborted it ourselves this request" (still alive moments ago, about to run its ownonFinish, which persists the full, correct content). Materializing the latter raced that natural write with an older debounced checkpoint and could silently overwrite fresher content with staler content. Now only provably-dead rows are materialized; just-aborted-but-alive rows get the original cheap session-row wipe and their own generation'sonFinishpersists the real content. Every reconcile/materialize call site now reports actual success (not attempted eligibility) up throughreconciled/remotelyReconciled, and per-row reconciliation runs concurrently (Promise.all). The active-streams lazy sweep runs over the subscription-authorized row set, so a user with mere page-view access can't trigger a reap side effect for another member's private conversation.ConversationMessage.statusthreads toMessageRendererandCompactMessageRenderer(the actual sidebar/global-assistant surface), which render an "Interrupted" badge + a real retry button at the message level, independent of which part groups exist — including for a message with zero content, or one interrupted mid-tool-call with no trailing text — de-duplicated againstTextBlock's own retry button for messages with real text content.previewAiUndo's affected-message count query and both ofexecuteAiUndo's soft-delete updates now assert thene(status,'streaming')guard is actually present, not just claimed.Filed two epic-level
D —tasks for real-but-unscheduled gaps found during review (not blocking): mention-notifications not firing for a materialized interrupted reply, and a live-tab socket-immediacy gap for the badge.This PR went through an unusually thorough convergence pass: an internal 8-angle review followed by four rounds of external bot review (CodeRabbit + Codex) that independently confirmed and added to the findings — content-fidelity loss, badge/retry-button coupling to text-block rendering, an authorization gap in the lazy sweep, the compare-and-swap guard, the insert-timestamp ordering, per-row reconciliation-accuracy reporting, and a rowCount check for the session-settle write. All fixed, tested, and CI-green.
Staging kill-mid-stream drill procedure documented on the PR board page (
epuyvd3qh3u9rj04nzc2rs4m) per leaf 3.5 — not run from this environment.Test plan
bun run typecheck(Turbo build) — greenbun vitest runacrossai/core,ai/streams, chat components, chat routes, andai-undo-service— 108 files / 1338 tests passing (local, non-render)materialize-interrupted-stream.test.ts(24 tests) at 100% branch/line/stmt/func coveragestream-takeover.test.ts/stream-abort-mark.test.tsupdated for the corrected eligibility split, concurrent per-row processing, startedAt threading, and accurate reconciliation reportingactive-streamslazy-sweep tests (dead row reaped, live row untouched, ambiguous-cap row untouched, unauthorized-conversation row NOT reaped)MessageRenderer.interrupted.test.tsx+CompactMessageRenderer.interrupted.test.tsx(badge + retry button render independent of part-group shape, no duplicate button for text-bearing messages)ai-undo-service.test.ts— 3 new D.2 tests, 23/23 passing🤖 Generated with Claude Code
https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB