feat(ai): server-owned AI streams — clients become pure subscribers (core)#2018
Conversation
…core)
A stream is now a SERVER-SIDE entity with its own identity, lifecycle and durable state.
A client is a subscriber. Any client that opens the conversation — the originator, a fresh
tab, another device — attaches to whatever is live.
This is the core of the work, split out from the Stop-button/ownership machine so it can be
reviewed on its own merits. See the PR body.
WHAT THIS FIXES
An AI stream used to be implicitly owned by the browser tab that started it: the tab
consuming the POST body dropped its own stream on the socket. But the server generation is
deliberately disconnect-immune. So when the originating tab reloaded — which on iPad happens
constantly, because iOS jetsams the WKWebView content process and Capacitor calls
webView.reload() — the stream was orphaned. The server kept generating and editing pages
while the reloaded tab showed no stream, no Stop button, and a live input. A second send
started a SECOND generation on the same conversation. Two agents, two bills.
AC1 A client can attach to its OWN live stream. The blanket own-session skip becomes
`!(isOwn && isConsuming)`, where isConsuming is a synchronous refcount marked before
the POST leaves. It is module state, so it is EMPTY after a reload — and that
emptiness IS the fix.
AC2 Attach works on reload, on reconnect and on cold open, with no gap and no
double-render. A failed cross-instance SSE join no longer suppresses the completion
event, so the finished reply is loaded from the DB instead of vanishing.
AC3 The `${pageId}-default` sentinel is gone. Gating on the isPersisted FACT (not the id
string) recovers every existing user's stranded history with NO migration; new chats
mint real cuids. The sentinel was also a real IDOR — a guessable id against a history
load with no user filter.
AC4 The resume path is alive when needed (gated on user editing, evaluated at fire time,
because iOS freezes JS while backgrounded and a render-time capture was always stale).
AC5 Per-conversation concurrency is TAKEOVER, never 409 — a 409 would self-lock the
conversation, since a crashed generation's terminal write dies with its process and
leaves a permanently-'streaming' row. Adds lastHeartbeatAt: a row whose heartbeat is
stale is dead, must never block a send, and must never be served as active.
AC6 A stream is never rendered into another conversation.
AC7 Disconnect-immunity is PRESERVED — and now enforced. request.signal must never reach
streamText: an aborted request means the client stopped LISTENING, not that the user
pressed Stop. It was a one-line change to break, it looks like an obvious cleanup, and
nothing failed if you did. It now has a tripwire.
ALSO CLOSED
A privacy leak: stream subscription was authorized on page access alone, which would have
handed a co-member's PRIVATE conversation, token by token, to anyone who could open the
page. Subscription now mirrors listConversations (`userId = me OR isShared`), fails closed,
and 404s rather than 403s.
A permanent lockout: legacy-conversation ownership was checked per-page — the one page an
attacker would not stand on — letting them squat a victim's conversation from outside it and
403 the victim out of their own history forever.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4
|
Warning Review limit reached
Next review available in: 38 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 (4)
📝 WalkthroughWalkthroughThe PR adds conversation-scoped authorization, heartbeat-based stream liveness, takeover and abort handling, client stream coordination, persisted conversation identity tracking, and global stop-slot ownership across AI chat and global assistant flows. ChangesConversation authorization and identity
Heartbeat liveness and takeover
Client stream coordination
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ 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: 7ea77760cd
ℹ️ 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".
| export function useChatTransport( | ||
| conversationId: string | null, | ||
| api: string | ||
| api: string, | ||
| channelId?: string, | ||
| ): DefaultChatTransport<UIMessage> | null { |
There was a problem hiding this comment.
Making channelId optional leaves the existing agent-mode callers (GlobalAssistantView and SidebarChatTab) invoking useChatTransport(..., '/api/ai/chat') without the agent page id. For those sends, createStreamTrackingFetch never marks selectedAgent.id as consuming, so useAgentChannelMultiplayer treats the originator's own chat:stream_start as attachable, joins its own SSE stream, and then appends a synthesized assistant message in addition to the one useChat already received. Please either require this argument or update all page-agent transport callers to pass the socket channel id.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
You are right, and this is a bug my own PR split introduced. Good catch — it would have shipped silently.
Splitting the ownership machine out of #2011 returned GlobalAssistantView and SidebarChatTab to their master state, where they call useChatTransport(convId, '/api/ai/chat') with no channelId. In the combined PR both passed selectedAgent?.id; the split dropped it.
The consequence is exactly as you traced: consumingChannels never marks the agent channel, so shouldAttachStream({isOwn: true, isConsuming: false}) returns true, the originator joins the multicast for its own stream, and useAgentChannelMultiplayer appends a synthesized assistant message on top of the one useChat already holds. Every agent reply renders twice — and nothing fails.
I took your stronger option rather than just fixing the two call sites, because "update all callers" is a fix that lasts until the next caller:
channelId: string | null, // REQUIRED. Not optional.A caller with genuinely no channel must now say null explicitly. Forgetting is a compile error. That matters here more than usual: this mark is the single signal separating "my own stream, already being consumed" from "a remote stream I should attach to", and its absence fails silently and visibly to the user.
The type change immediately surfaced a third caller passing string | undefined (GlobalChatContext) — which is the argument for making it required, made for me.
Fixed in the head commit; all three call sites now pass explicitly.
A test-suite audit — hunting the class where a fixture models an impossible state and gives false comfort — found six live instances. Two of them guarded this PR's own load-bearing claims and would have passed while the claim was violated. isShared REACHED NO ASSERTION ANYWHERE (the worst one) useChannelStreamSocket DROPS a co-member's stream when payload.isShared === false, so that flag is the SOLE signal making another member's shared conversation visible. Every assertion on it pinned only `false`; the one shared-conversation route test asserted merely that the lifecycle "was called" — not with what. Hardcode `isShared: false` in the route or the lifecycle and 100% of the suite stayed green while EVERY SHARED CONVERSATION WENT DARK FOR EVERY CO-MEMBER. Now asserted end to end. THE TAKEOVER WAS UNTESTED ON THE MAIN CHAT ROUTE (AC5's whole guarantee) stream-takeover was not even mocked in the chat-route suite, so the real function ran against a db mock with no .update(), threw, and was swallowed by its own catch. The call site could have been DELETED and nothing would have failed. Now pinned: called with the right conversation, called BEFORE the new lifecycle (the other order aborts the stream it just started), and never a 409 — a rejection would self-lock the conversation, because a crashed generation's terminal write dies with its process and leaves a permanently-'streaming' row. A MOCK LEAK ACROSS A WHOLE ROUTE FILE vi.clearAllMocks() clears calls, not implementations, and seven tests used mockResolvedValue rather than ...Once. The lifecycle-invocation test — which pins the full argument shape for "a new AI stream" — was silently running against an already-owned conversation. It passed either way, which is exactly why nobody noticed. HALF THE REVOCATION BACKSTOP WAS NEVER EXERCISED hasViewAccess() is `pageOk && canSubscribe()`, and every recheck test only varied page access. `return canSubscribe();` could be deleted and the suite stayed green — while a conversation UN-SHARED mid-stream kept streaming, token by token, to someone who may no longer read it. Page access and conversation access are revoked independently. TWO ASSERTIONS TOO WEAK TO FAIL expect.any(String) on the bootstrap conversationId (which decides WHICH surface's Stop button lights up) and on the takeover's conversationId/channelId — both known constants, so a route that SWAPPED them passed. Now exact. Every one verified by reintroducing the bug: deleting the takeover fails 2, hardcoding isShared fails 1, deleting the conversation half of hasViewAccess fails 1, swapping the takeover args fails 1. Before this commit, all of them were silent. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
apps/web/src/components/layout/middle-content/page-views/ai-page/AiChatView.tsx (1)
784-802: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winVerify the identity is still the one that started reconciliation.
The persistence guard does not catch an unpersisted-to-unpersisted switch. If deletion creates a fresh identity while this request is pending, Line 802 can restore the stale conversation and route the next send incorrectly.
Proposed stale-identity guard
if (!isPersistedRef.current) { + const identityAtSyncStart = currentConversationIdRef.current; + if (!identityAtSyncStart) return; const { parts, conversationId: streamConvId, startedAt } = stream; fetchWithAuth(`/api/ai/page-agents/${page.id}/conversations?pageSize=1`) .then(async (res) => { if (pageIdRef.current !== page.id) return; if (!res.ok) return; const data = (await res.json()) as ConversationListResponse; const persisted = data.conversations?.[0]; if (!persisted || persisted.id !== streamConvId) return; - if (isPersistedRef.current) return; + if ( + isPersistedRef.current || + currentConversationIdRef.current !== identityAtSyncStart + ) return; setIdentity(persisted.id);🤖 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/layout/middle-content/page-views/ai-page/AiChatView.tsx` around lines 784 - 802, Update the background reconciliation in the unpersisted branch of the stream handling flow to capture the initiating conversation identity before starting fetchWithAuth, then verify that identity still matches the current identity before calling setIdentity. Preserve the existing page and persisted-state guards, and prevent reconciliation from adopting a stale conversation after an unpersisted-to-unpersisted switch.apps/web/src/app/api/ai/chat/active-streams/route.ts (1)
79-105: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick winFilter stale heartbeats in SQL before selecting
parts.Crashed rows can remain
status='streaming'indefinitely, so every poll currently loads their potentially large JSONB snapshots before discarding them in JavaScript. Add alastHeartbeatAt >= cutoffpredicate; consider including heartbeat in the supporting index.🤖 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/app/api/ai/chat/active-streams/route.ts` around lines 79 - 105, Update the active-stream query around aiStreamSessions to compute a heartbeat cutoff and add a lastHeartbeatAt >= cutoff predicate alongside the channelId and status filters, so stale rows are excluded before selecting the potentially large parts snapshot. Preserve the existing isStreamRowLive and filterSubscribableStreams checks, and update the supporting index definition if one is present nearby to include lastHeartbeatAt.
🧹 Nitpick comments (5)
apps/web/src/lib/ai/streams/__tests__/shouldClaimGlobalStopSlot.test.ts (1)
1-1: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRename this test file to kebab-case.
Use
should-claim-global-stop-slot.test.ts.As per coding guidelines, “Use kebab-case for filenames (except React hooks, Zustand stores, and React components).” <coding_guidelines>
🤖 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/streams/__tests__/shouldClaimGlobalStopSlot.test.ts` at line 1, Rename the test file from shouldClaimGlobalStopSlot.test.ts to should-claim-global-stop-slot.test.ts, leaving its contents unchanged.Source: Coding guidelines
apps/web/src/lib/ai/streams/__tests__/shouldAttachStream.test.ts (1)
1-1: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRename this test file to kebab-case.
Use
should-attach-stream.test.ts.As per coding guidelines, “Use kebab-case for filenames (except React hooks, Zustand stores, and React components).” <coding_guidelines>
🤖 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/streams/__tests__/shouldAttachStream.test.ts` at line 1, Rename the test file shouldAttachStream.test.ts to should-attach-stream.test.ts, leaving its contents unchanged.Source: Coding guidelines
apps/web/src/lib/ai/streams/shouldClaimGlobalStopSlot.ts (1)
1-1: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRename this module to kebab-case.
Use
should-claim-global-stop-slot.tsand update its imports.As per coding guidelines, “Use kebab-case for filenames (except React hooks, Zustand stores, and React components).” <coding_guidelines>
🤖 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/streams/shouldClaimGlobalStopSlot.ts` at line 1, Rename the shouldClaimGlobalStopSlot module to should-claim-global-stop-slot.ts and update every import or reference to use the new kebab-case path, preserving the module’s behavior and exports.Source: Coding guidelines
apps/web/src/app/api/ai/chat/__tests__/stream-socket-events.test.ts (1)
566-574: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winThis broadcast-suppression test no longer exercises the broadcast code path.
With the new conversationId authorization gate added in this same PR (route.ts 444-460),
userId:'other-user', isShared:falsenow gets rejected with 403 before the request ever reaches theshouldBroadcast/broadcast logic this test is meant to verify — it's now redundant with the "conversationId authorization" 403 test for the exact same combination, and thenot.toHaveBeenCalled()assertion passes for the wrong reason (early return, not suppression logic).Either drop this case (it's covered by the authorization describe block) or change the fixture so the request actually reaches the broadcast code for the scenario under test (e.g. owned by the caller but not shared, or shared-but-different-page if that's meant to be distinguished).
🤖 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/app/api/ai/chat/__tests__/stream-socket-events.test.ts` around lines 566 - 574, Update the test “should suppress broadcast when private conversation is owned by someone else” so it reaches the broadcast decision logic instead of the conversationId authorization early return. Either remove this redundant case, or change its fixture to an authorized scenario such as a caller-owned private conversation, while preserving the assertion that mockBroadcastChatUserMessage is not called.apps/web/src/app/api/ai/chat/route.ts (1)
391-463: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider extracting the conversationId authorization block into a dedicated, unit-testable function.
This is correct as written (verified all four ownership/shared/page-match branches), but it's ~70 lines of security-critical logic inline in an already ~1900-line handler, and the same PR already extracts comparably tricky logic (
stream-liveness.ts,stream-takeover.ts) into pure, directly-testable modules. Doing the same here (e.g.validateConversationAccess({ conversationId, userId, pageId })) would let it be unit-tested in isolation the waydecideStreamTakeoveris, rather than only indirectly through full-route integration tests.🤖 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/app/api/ai/chat/route.ts` around lines 391 - 463, Extract the inline conversationId authorization logic from the chat route into a dedicated, unit-testable function such as validateConversationAccess, preserving the existing cuid, legacy-message-owner, ownership/shared-access, and page-match checks. Keep database failures propagating to the route’s 500 handling and have the helper return or expose the same 400/403 outcomes required by the current flow; replace the inline block with a call to the helper and add focused unit coverage for all authorization branches.
🤖 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/stream-takeover.ts`:
- Around line 90-131: Separate the reconciliation UPDATE around aiStreamSessions
from the outer takeover failure handling by giving it its own try/catch. If it
fails after the abort loop succeeds, preserve and return the accumulated aborted
IDs, report reconciled as empty or only successfully reconciled IDs, and emit a
distinct warning; keep SELECT-stage or other pre-abort failures using the
existing empty safe fallback.
In `@packages/db/src/schema/ai-streams.ts`:
- Around line 16-23: The lastHeartbeatAt-based liveness change must remain
compatible with pre-heartbeat writers during rollout. Update the active-stream
and takeover logic around lastHeartbeatAt so old workers are not immediately
treated as dead, using a staged compatibility path or requiring old workers to
drain before heartbeat-only authority is enabled; preserve
heartbeat-authoritative behavior once the rollout is complete.
---
Outside diff comments:
In `@apps/web/src/app/api/ai/chat/active-streams/route.ts`:
- Around line 79-105: Update the active-stream query around aiStreamSessions to
compute a heartbeat cutoff and add a lastHeartbeatAt >= cutoff predicate
alongside the channelId and status filters, so stale rows are excluded before
selecting the potentially large parts snapshot. Preserve the existing
isStreamRowLive and filterSubscribableStreams checks, and update the supporting
index definition if one is present nearby to include lastHeartbeatAt.
In
`@apps/web/src/components/layout/middle-content/page-views/ai-page/AiChatView.tsx`:
- Around line 784-802: Update the background reconciliation in the unpersisted
branch of the stream handling flow to capture the initiating conversation
identity before starting fetchWithAuth, then verify that identity still matches
the current identity before calling setIdentity. Preserve the existing page and
persisted-state guards, and prevent reconciliation from adopting a stale
conversation after an unpersisted-to-unpersisted switch.
---
Nitpick comments:
In `@apps/web/src/app/api/ai/chat/__tests__/stream-socket-events.test.ts`:
- Around line 566-574: Update the test “should suppress broadcast when private
conversation is owned by someone else” so it reaches the broadcast decision
logic instead of the conversationId authorization early return. Either remove
this redundant case, or change its fixture to an authorized scenario such as a
caller-owned private conversation, while preserving the assertion that
mockBroadcastChatUserMessage is not called.
In `@apps/web/src/app/api/ai/chat/route.ts`:
- Around line 391-463: Extract the inline conversationId authorization logic
from the chat route into a dedicated, unit-testable function such as
validateConversationAccess, preserving the existing cuid, legacy-message-owner,
ownership/shared-access, and page-match checks. Keep database failures
propagating to the route’s 500 handling and have the helper return or expose the
same 400/403 outcomes required by the current flow; replace the inline block
with a call to the helper and add focused unit coverage for all authorization
branches.
In `@apps/web/src/lib/ai/streams/__tests__/shouldAttachStream.test.ts`:
- Line 1: Rename the test file shouldAttachStream.test.ts to
should-attach-stream.test.ts, leaving its contents unchanged.
In `@apps/web/src/lib/ai/streams/__tests__/shouldClaimGlobalStopSlot.test.ts`:
- Line 1: Rename the test file from shouldClaimGlobalStopSlot.test.ts to
should-claim-global-stop-slot.test.ts, leaving its contents unchanged.
In `@apps/web/src/lib/ai/streams/shouldClaimGlobalStopSlot.ts`:
- Line 1: Rename the shouldClaimGlobalStopSlot module to
should-claim-global-stop-slot.ts and update every import or reference to use the
new kebab-case path, preserving the module’s behavior and exports.
🪄 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: 811a74f7-45d5-44eb-950c-b37442ff5958
📒 Files selected for processing (53)
apps/web/src/app/api/ai/chat/__tests__/stream-socket-events.test.tsapps/web/src/app/api/ai/chat/active-streams/__tests__/route.test.tsapps/web/src/app/api/ai/chat/active-streams/route.tsapps/web/src/app/api/ai/chat/route.tsapps/web/src/app/api/ai/chat/stream-join/[messageId]/__tests__/route.test.tsapps/web/src/app/api/ai/chat/stream-join/[messageId]/route.tsapps/web/src/app/api/ai/global/[id]/messages/__tests__/credit-gate.test.tsapps/web/src/app/api/ai/global/[id]/messages/__tests__/stream-socket-events.test.tsapps/web/src/app/api/ai/global/[id]/messages/route.tsapps/web/src/components/layout/middle-content/page-views/ai-page/AiChatView.tsxapps/web/src/components/layout/middle-content/page-views/ai-page/__tests__/AiChatView.test.tsxapps/web/src/contexts/GlobalChatContext.tsxapps/web/src/contexts/__tests__/GlobalChatContext.test.tsxapps/web/src/hooks/__tests__/useAgentChannelMultiplayer.test.tsapps/web/src/hooks/__tests__/useAppStateRecovery.test.tsapps/web/src/hooks/__tests__/useChannelStreamSocket.test.tsapps/web/src/hooks/useAppStateRecovery.tsapps/web/src/hooks/useChannelStreamSocket.tsapps/web/src/lib/ai/core/__tests__/disconnect-immunity.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-client.test.tsapps/web/src/lib/ai/core/__tests__/stream-lifecycle.test.tsapps/web/src/lib/ai/core/__tests__/stream-liveness.test.tsapps/web/src/lib/ai/core/__tests__/stream-multicast-registry.test.tsapps/web/src/lib/ai/core/__tests__/stream-subscription-authz.test.tsapps/web/src/lib/ai/core/__tests__/stream-takeover.test.tsapps/web/src/lib/ai/core/stream-abort-client.tsapps/web/src/lib/ai/core/stream-abort-registry.tsapps/web/src/lib/ai/core/stream-horizons.tsapps/web/src/lib/ai/core/stream-lifecycle.tsapps/web/src/lib/ai/core/stream-liveness.tsapps/web/src/lib/ai/core/stream-multicast-registry.tsapps/web/src/lib/ai/core/stream-subscription-authz.tsapps/web/src/lib/ai/core/stream-takeover.tsapps/web/src/lib/ai/shared/hooks/__tests__/useConversationIdentity.test.tsapps/web/src/lib/ai/shared/hooks/index.tsapps/web/src/lib/ai/shared/hooks/useChatTransport.tsapps/web/src/lib/ai/shared/hooks/useConversationIdentity.tsapps/web/src/lib/ai/streams/__tests__/consumingChannels.test.tsapps/web/src/lib/ai/streams/__tests__/isPlaceholderConversationId.test.tsapps/web/src/lib/ai/streams/__tests__/shouldAttachStream.test.tsapps/web/src/lib/ai/streams/__tests__/shouldClaimGlobalStopSlot.test.tsapps/web/src/lib/ai/streams/consumingChannels.tsapps/web/src/lib/ai/streams/isPlaceholderConversationId.tsapps/web/src/lib/ai/streams/resolveActiveAssistantMessageId.tsapps/web/src/lib/ai/streams/shouldAttachStream.tsapps/web/src/lib/ai/streams/shouldClaimGlobalStopSlot.tsapps/web/src/lib/repositories/__tests__/conversation-repository.test.tsapps/web/src/lib/repositories/conversation-repository.tsapps/web/src/lib/websocket/socket-utils.tspackages/db/drizzle/0203_adorable_amphibian.sqlpackages/db/drizzle/meta/0203_snapshot.jsonpackages/db/drizzle/meta/_journal.jsonpackages/db/src/schema/ai-streams.ts
💤 Files with no reviewable changes (2)
- apps/web/src/lib/ai/streams/isPlaceholderConversationId.ts
- apps/web/src/lib/ai/streams/tests/isPlaceholderConversationId.test.ts
Test-suite audit: six tests that could not failAfter the split I audited the suite for the class that has bitten this work twice — a fixture that models an impossible state, so the test passes whether the code is right or not. Six live instances, two of them guarding this PR's own load-bearing claims. The two that matter:
The takeover was untested on the main chat route — AC5's entire guarantee.
Also closed: a mock-implementation leak across a whole route file ( Verified, not assumedEach is confirmed by reintroducing the bug and watching the test fail:
Before this commit, every one of them was silent. I'll note one thing against myself: my first attempt to verify the swap used a regex that didn't match the route's actual formatting, so it "passed" vacuously — the exact trap I was auditing for. Caught it, fixed the check, re-ran. That is why every claim above is a reintroduce-and-observe result rather than a reasoned one.
|
… failure
Three reviewer findings on the core PR. The first is a bug my own split introduced.
1. THE SPLIT DROPPED channelId ON BOTH AGENT TRANSPORTS (Codex, P1 — correct)
Splitting the ownership machine out returned GlobalAssistantView and SidebarChatTab to their
master state, where they call useChatTransport WITHOUT a channelId. So the agent channel was
never marked as being consumed — and consumingChannels is the ONE signal that stops the
originating tab from also attaching to its own stream off the socket. The originator treated
its own chat:stream_start as an attachable remote stream, joined its own multicast, and
appended a synthesized assistant message on top of the one useChat already had.
Every agent reply would have rendered twice, and nothing would have failed.
Taking the reviewer's stronger option: channelId is now REQUIRED (`string | null`, never
optional). A caller with no channel must say so explicitly; forgetting is a compile error.
The type change immediately surfaced a third caller that was passing `string | undefined` —
which is precisely the point.
2. THE TAKEOVER LIED ON PARTIAL FAILURE (CodeRabbit — correct)
One try/catch wrapped the whole function. The aborts land FIRST — real in-process generations
are STOPPED — and only then does the reconcile UPDATE run. If that UPDATE threw, the catch
returned `{aborted: [], reconciled: []}`: "nothing happened", when streams had in fact been
stopped and their rows still read 'streaming'. Every reader then treats a dead generation as
live, and the caller is told the exact opposite of the truth.
The UPDATE now has its own catch and returns what was ACTUALLY aborted. The rows self-heal: a
stopped generation stops beating, so within STREAM_HEARTBEAT_STALE_MS the liveness predicate
calls them dead and the next takeover reconciles them — which is exactly the failure mode the
heartbeat exists to absorb.
3. HEARTBEAT-AUTHORITATIVE LIVENESS HAS A ROLLOUT WINDOW (CodeRabbit — real, documented, not
gated)
Old workers do not beat, so during a deploy their still-running streams age out and get driven
terminal. Documented in the schema rather than hidden behind a two-phase flag, because the
blast radius is smaller than the flag: no content is lost (the message persists through the
normal path; `parts` is only a crash-recovery snapshot, and the old worker's own terminal write
corrects status), and the worst case — a concurrent generation — is EXACTLY what master does
today, since master has no takeover at all. The window is one rolling-deploy drain.
Both fixes pinned by tests verified to fail without them: restoring the single catch fails the
partial-failure test; dropping channelId is now a compile error.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4
Deploy note: run the index build out-of-band firstAuditing my own migration, not because a reviewer asked, but because this is the class of thing that surfaces as an incident rather than a failing test.
ALTER TABLE "ai_stream_sessions" ADD COLUMN "last_heartbeat_at" timestamp DEFAULT now() NOT NULL;
CREATE INDEX IF NOT EXISTS "ai_stream_sessions_conversation_status_idx"
ON "ai_stream_sessions" USING btree ("conversation_id","status");The The That matters more than it looks, for a reason I only found by checking: this table is never pruned. The safe path, which the generated SQL already supportsThe statement is -- Before deploying. Non-blocking; safe to run against live traffic.
CREATE INDEX CONCURRENTLY IF NOT EXISTS "ai_stream_sessions_conversation_status_idx"
ON "ai_stream_sessions" USING btree ("conversation_id","status");Then run the migration normally — the If the table turns out to be small (check Separately worth a follow-up (pre-existing, not this PR)
|
I wrote a source-level tripwire for disconnect-immunity, verified it against
`AbortSignal.any([abortSignal, request.signal])`, and called it done.
Then I tried to defeat my own guard, and did — on the first try:
const { signal } = request;
A destructure. The single most natural way a developer would actually reach for the request's
abort signal. It sailed straight past a member-access-only regex: the test stayed GREEN while
disconnect-immunity — the load-bearing invariant of the entire architecture — was destroyed.
That is exactly the false comfort this file was written to prevent, reproduced inside the file
itself. A tripwire that only catches the spelling you happened to think of is not a tripwire;
it is a green light that measures nothing, which is worse than no light at all.
Now matches both the member access and the destructure, and verified against all four realistic
forms (`request.signal`, `req.signal`, `const { signal } = request`, `const { signal: alias } =
req`, and the AbortSignal.any composition) — each reintroduced independently, each caught.
Stated plainly in the file that this remains a HEURISTIC: it cannot catch every aliasing
(`const r = request; r.signal`), and it is not pretending to be a type system. It catches the
ways the mistake actually gets made, and fails loudly with an explanation. That is worth more
than a proof nobody writes.
Tests only. No behaviour change.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4
AC4 is surface-scoped — stating it so nobody assumes otherwiseHunting for other places an optional parameter could be silently omitted (the failure mode behind Codex's
const resumeEnabled = useCallback(
() => currentConversationIdRef.current !== null && !useEditingStore.getState().isAnyEditing(),
[],
);
I am deliberately NOT "fixing" that here, and the reason mattersMy first instinct was to drop Making them resume-safe means porting So, accurately:
Flagging it because "AC4 done" reads like all surfaces recover correctly, and they do not. |
An adversarial pass found a runaway I had looked at earlier and WRONGLY dismissed. streamId and messageId are BOTH minted server-side, and the client learns neither until the response headers land (X-Stream-Id). But a real agent send spends 0.5-3 seconds before that: auth, rate limit, DB reads, context assembly, connecting to the provider. Press Stop in that window — exactly when a user who has spotted a typo presses it — and the client had NOTHING to name. `abortActiveStream` was a guaranteed map miss, so it cancelled the local fetch and returned. The button flipped back to Send. The user believed it worked. Cancelling the fetch stops NOTHING. Streams are deliberately server-owned and survive a client disconnect — that is the entire architecture. So the generation kept running, kept calling write tools, and kept BILLING, while the UI said it had stopped. One click, every time, in a multi-second window. I had traced this window earlier and concluded "there is no id the client could name." That was wrong. There is one: the CONVERSATION id, which the client holds from t=0 — and the server already aborts by conversation (takeOverConversationStreams does exactly that). I dismissed a real runaway on a claim I had not actually checked. /api/ai/abort now accepts conversationId as a fallback (messageId and streamId still take precedence — they name the stream exactly, whereas conversationId names all of them). AUTHZ IS DELIBERATELY STRICTER THAN THE TAKEOVER'S takeOverConversationStreams aborts as the STREAM's owner (row.userId), because a second send on a SHARED conversation must be able to take over a co-member's generation. This is not that. This is an explicit user Stop, so it may only ever stop the CALLER'S OWN streams: the userId filter is in the query, and the registry re-checks ownership on every abort. Naming someone else's conversation cannot kill their agent mid-answer. Pinned by 11 tests. The authz one is verified by removing the caller-scoping — i.e. adopting the takeover's rule — and watching it fail. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4
A runaway I had looked at and wrongly dismissedAn adversarial pass found a bug I had personally traced earlier in this work and declared unfixable. It was not. Stop pressed before the response headers arrived did nothing at all.
And cancelling the fetch stops nothing — streams are deliberately server-owned and survive a client disconnect; that is the entire architecture. So the generation kept running, kept calling write tools, and kept billing, while the UI told the user it had stopped. One click, every time, in a multi-second window. What I got wrongI traced this window earlier and concluded "there is no id the client could name — that's inherent to disconnect-immunity." There is one: the conversationId, which the client holds from t=0 — and the server already aborts by conversation ( The fix
Authorization is deliberately stricter than the takeover's. Pinned by 11 tests. The authz one is verified by removing the caller-scoping — i.e. adopting the takeover's rule — and watching it fail. CI green on this exact commit ( |
A stale closure in these files is not a re-render bug. It is a RUNAWAY AGENT.
Streams are server-owned and deliberately survive a client disconnect, so the only thing that
stops a generation is an explicit abort that names it correctly. Every stale value here — a
captured conversationId, a captured messageId, a captured isStreaming — means Stop names the
wrong stream or nothing at all: the fetch is cancelled, the button flips back to Send, and the
server keeps generating, keeps running write tools, and KEEPS BILLING while the user believes
it stopped. That has now happened repeatedly in this work.
react-hooks/exhaustive-deps catches exactly this, and it was ALREADY ENABLED — as a warning,
via next/core-web-vitals. `bun run lint` exits 0 on warnings, so CI reported "14/14 successful"
while the rule pointed straight at a missing dep in SidebarChatTab's handleStop.
The signal existed and was wired to nothing. That is the same failure as the six tests that
could not fail, the CI that never ran on the stacked PR, and the migration whose lock risk no
test could see.
Promoted to an error for the ten files that own stream identity. It immediately found four
violations I had confidently measured as zero:
- AiChatView: `attachments.length` was a redundant dep (getFilesForSend is already memoized
on [attachments] and is already listed).
- GlobalChatContext: three useCallbacks omitted `dispatchIdentity`, and the context-value
useMemo omitted `setCurrentConversationId`. Both are stable useCallback identities, so
listing them is a runtime no-op — but the omission is exactly the habit that lets a
NON-stable value slip through unnoticed in this subsystem.
I had run `next lint --file` over these files, got zero, and reported it as free. It was not.
The real lint run disagreed. Measuring with the wrong tool is how you get a green light that
means nothing — which is the entire lesson of this PR.
If this rule fires, do not silence it. A dep you are tempted to omit here is a stop button you
are about to break.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4
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/app/api/ai/abort/route.ts (1)
91-97: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winInclude
conversationIdin the abort info log.When the abort is dispatched by
conversationId(the pre-headers fallback),streamIdandmessageIdare bothundefinedin this log entry, andconversationIdis omitted entirely — making conversation-level aborts untraceable in the info log. The audit log captures it viaresourceId, but the info log does not.📝 Proposed fix
loggers.api.info('AI stream abort requested', { streamId, messageId, + conversationId, userId, aborted: result.aborted, reason: result.reason, });🤖 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/app/api/ai/abort/route.ts` around lines 91 - 97, Update the AI abort info log in the abort route to include the available conversationId alongside streamId and messageId. Preserve the existing fields and values so conversation-level aborts remain traceable when streamId and messageId are undefined.
🤖 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/app/api/ai/abort/route.ts`:
- Around line 91-97: Update the AI abort info log in the abort route to include
the available conversationId alongside streamId and messageId. Preserve the
existing fields and values so conversation-level aborts remain traceable when
streamId and messageId are undefined.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: f71c9d2c-98a5-44de-ac0e-d896936d37c8
📒 Files selected for processing (20)
apps/web/eslint.config.mjsapps/web/src/app/api/ai/abort/__tests__/route.test.tsapps/web/src/app/api/ai/abort/route.tsapps/web/src/app/api/ai/chat/__tests__/stream-socket-events.test.tsapps/web/src/app/api/ai/chat/stream-join/[messageId]/__tests__/route.test.tsapps/web/src/app/api/ai/global/[id]/messages/__tests__/stream-socket-events.test.tsapps/web/src/components/layout/middle-content/page-views/ai-page/AiChatView.tsxapps/web/src/components/layout/middle-content/page-views/dashboard/GlobalAssistantView.tsxapps/web/src/components/layout/right-sidebar/ai-assistant/SidebarChatTab.tsxapps/web/src/contexts/GlobalChatContext.tsxapps/web/src/hooks/__tests__/useChannelStreamSocket.test.tsapps/web/src/lib/ai/core/__tests__/abort-conversation-streams.test.tsapps/web/src/lib/ai/core/__tests__/disconnect-immunity.test.tsapps/web/src/lib/ai/core/__tests__/stream-takeover.test.tsapps/web/src/lib/ai/core/abort-conversation-streams.tsapps/web/src/lib/ai/core/client.tsapps/web/src/lib/ai/core/stream-abort-client.tsapps/web/src/lib/ai/core/stream-takeover.tsapps/web/src/lib/ai/shared/hooks/useChatTransport.tspackages/db/src/schema/ai-streams.ts
🚧 Files skipped from review as they are similar to previous changes (9)
- apps/web/src/lib/ai/core/tests/disconnect-immunity.test.ts
- apps/web/src/app/api/ai/global/[id]/messages/tests/stream-socket-events.test.ts
- packages/db/src/schema/ai-streams.ts
- apps/web/src/app/api/ai/chat/stream-join/[messageId]/tests/route.test.ts
- apps/web/src/lib/ai/core/stream-takeover.ts
- apps/web/src/contexts/GlobalChatContext.tsx
- apps/web/src/app/api/ai/chat/tests/stream-socket-events.test.ts
- apps/web/src/hooks/tests/useChannelStreamSocket.test.ts
- apps/web/src/components/layout/middle-content/page-views/ai-page/AiChatView.tsx
… have
The call site said: "at most one may be generating on a conversation at a time."
That is false. takeOverConversationStreams is a check-then-act with NO serialization — the
SELECT inside it and the INSERT inside createStreamLifecycle are not atomic together. Two
near-simultaneous sends can both find zero in-flight rows and both proceed: two generations,
two sets of tool calls, two bills. Exactly what the guard exists to prevent.
This is the fourth comment in this work that asserted a property the code lacked:
- the heartbeat docblock called the parts checkpoint a MITIGATION for the immortal-ghost
problem, when the checkpoint WAS the hole;
- holdForStream's docstring said the messageId is "captured when the first chunk arrives",
which is the exact false premise that made Stop abort the previous turn's message;
- resolveActiveAssistantMessageId's docstring called the submitted window one "where no
assistant id exists yet" — true only on the very first turn;
- and this one.
Each was load-bearing. A comment that promises a guarantee the code does not have is worse
than no comment: the next person reads it, trusts it, and builds on a false premise. That is
not a documentation problem, it is the same failure as a test that cannot fail — a signal that
looks like it attests to something it never examined.
The race is NOT closed here, deliberately: it needs DB-level serialization (an advisory lock
spanning takeover+insert, or a partial unique index on (conversation_id) WHERE
status='streaming' — whose migration would fail outright on pre-existing duplicate rows, so it
needs a reconciliation step first). That is its own change with its own migration risk, and
master has no takeover at all — concurrent sends there ALWAYS double-generate — so this
narrows the window rather than opening one.
Now stated in both places a future engineer will actually look: the call site and the module.
Comments only. No behaviour change.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4
Sweeping the diff for absolute claims ("always", "never", "cannot", "guarantee", "at most
one") found the chat route was not the only place asserting a property the code lacks — the
global route said the same thing, word for word:
"A stream is a server-owned entity: at most one may be generating on a conversation at a
time."
It is not. takeOverConversationStreams is a check-then-act with no serialization, so two
near-simultaneous sends can both find nothing in flight and both proceed. I corrected the
chat route and did not check its twin — the same incompleteness that produced half the bugs
in this work.
The other absolute claims in the diff were VERIFIED rather than assumed:
- "a conversation may only ever be CREATED from a cuid" — holds: the create branch is gated
on isCuid, and a legacy non-cuid id passes only when its row already exists.
- "Never 409" — holds: there is no 409 on any stream path (the only matches are the comments
saying so).
That verification is the point. Four comments in this work promised guarantees the code did
not have, each load-bearing, and each was believed by the next person to read it — including
me. A claim in a comment is a signal like any other, and an unverified one is a signal that
attests to nothing.
Comments only. No behaviour change.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4
Extending the "signals that attest to nothing" audit from tests and comments to LOGS found one that lies at exactly the moment an operator most needs the truth. takeOverConversationStreams logged "took over in-flight stream(s) on this conversation" unconditionally — including when `aborted` and `reconciled` were BOTH empty. That case is not a no-op: it means the live row belongs to ANOTHER web instance (the abort registry is in-process), it is still generating, and this send is about to start a SECOND generation alongside it. Two agents, two sets of tool calls, two bills. The `unstoppable` warn immediately above says so correctly. The info line then contradicted it in the same breath. An operator scanning messages during an incident would read "took over" and move on. Now it only claims a takeover when one actually happened. Pinned by a test that asserts the warn fires AND that nothing claims a takeover; verified by restoring the unconditional log and watching it fail. This is the same defect as the six tests that could not fail, the CI that never ran, the tripwire blind to the obvious break, the lint rule at warning level, and the five comments promising guarantees the code did not have. A log is a signal like any other, and one that misreports attests to nothing. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4
…2011) * fix(ai): Stop-button / stream-ownership machine Stacked on the server-owned-streams core. This is the subsystem that produced 26+ defects across fourteen review rounds of the combined PR — two of them mine — and it is split out so it gets the focused review that record demands. Every bug in it has been the same bug: A PIECE OF STATE THAT NAMED THE SURFACE WHEN IT NEEDED TO NAME THE STREAM. A chatId, a panel, a tab, the live `messages` array — used where an (agentId, conversationId, messageId) triple was required. Streams deliberately survive a client disconnect. So a Stop that misses is not cosmetic: it is a runaway agent that keeps running write tools and keeps billing. holdForStream capture a stream's identity when it starts, hold it until it ends selectLiveAssistantIds resolve EACH chat's live assistant id from ITS OWN status/messages shouldClaim*StopSlot single-writer guards on the shared stop slots + the surfaces and the page-agents store, rekeyed by (agentId, conversationId) THE THREE THAT MATTER 1. Stop aborted the PREVIOUS turn's message, on every turn after the first. useChat sets status='submitted' BEFORE issuing the request and pushes the new assistant message only inside write(), which flips to 'streaming' in the same job — so during submitted, the array's last assistant message is the previous turn's reply. holdForStream latches on the first live render, which is a submitted render, so it captured and held a message that had finished minutes ago. Every messageId now resolves only at status === 'streaming'. 2. The dedup guard TRUNCATED the reply. "We have a message with this id" is not "we have its content": the server names the assistant message, so useChat's copy shares the stream's id — and useChat does not roll back on error, leaving a HALF-STREAMED message in the array on a network drop. That is exactly when recovery rejoins and the full reply arrives. It now replaces by id instead of skipping. 3. A surface may only free the activeStreams key it allocated. The sidebar registered `sidebar:<convId>` and cleared the bare `<convId>` — a key it never writes, and which belongs to the dashboard. WHY selectLiveAssistantIds EXISTS My first fix for (1) passed every test with the bug still present. The defect was in the WIRING, and a test of holdForStream is blind to which chat's id the caller hands it. So the wiring itself became the unit under test: it takes both chats and returns both ids, structurally unable to let one reach the other's slot. STILL WANTED, DELIBERATELY NOT DONE HERE Every key here is a string, correct by convention and by comment. The real fix is a branded type the transport alone can mint, so "named the surface instead of the stream" becomes a COMPILE ERROR rather than a bug found in round fifteen. A wide-blast-radius refactor of the highest-defect subsystem, at the tail of a long PR, with my recent error rate, is how bug #29 gets written. It should be a deliberate follow-up. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4 * fix(ai): three more runaways in the ownership machine An adversarial pass on the split-out ownership PR. All three let the server keep generating and keep billing while the UI said otherwise. 1. STOP AFTER A MID-STREAM CONVERSATION SWITCH ABORTED NOTHING The last surviving instance of the bug class: stream-owned state torn down on a SURFACE event. The conversation-change cleanup deletes the running stream's activeStreams entry, so the chatId fallback then named an entry the surface had just deleted. Send, click New (no guard), click Stop — the fetch aborts, the abort is a map miss, the generation keeps running. Fixed at the shared choke point: `useChatStop` and both surfaces' stop paths now pass the STREAM's conversation — held from when it started, never the live one — so the abort reaches the server by conversation even when the map entry is gone or was never written. 2. AFTER A REFRESH MID-AGENT-STREAM, THE DASHBOARD HAD NO STOP BUTTON AT ALL `useGlobalEffectiveStream` ORed the bootstrap-restored stream into `isStreaming` for GLOBAL mode only. So the multiplayer hook claimed the store slot, the SIDEBAR read it and showed a working Stop, and the DASHBOARD — the surface that started the stream — rendered Send. The user was looking at their own runaway generation with no way to stop it. The (agentId, conversationId)-keyed selectors were already there; the dashboard just never read them. 3. THE JOIN-FAILED RACE COULD LOSE THE REPLY `joinFailed` is populated in an ASYNC catch, and a join commonly 404s *because* the stream just ended — so `chat:stream_complete` can land before the rejection settles. In that window the store held only the seeded, debounced DB snapshot, which can be SHORTER than what useChat already had. The replace-by-id then swapped a longer visible reply for a truncated one AND suppressed the reload-from-DB, so the full persisted message was never fetched. Now tracked synchronously: a join that DELIVERED nothing is not authoritative, whatever the catch has managed to record. Both cases route to reload-from-DB. I had previously traced this exact path and declared it safe. I was wrong — I checked the joinFailed branch and never asked what happens before it is set. THREE FIXTURES MODELLED IMPOSSIBLE STATES Fixing (3) exposed tests that could not have caught it: a stream that completes with data in the store but whose join never delivered anything (delivery is HOW the data gets there), and a `chat:stream_complete` payload with no conversationId (the server always sends it; the type merely marks it optional). Corrected to model reality. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4 * fix(ai): handleStop read isActuallyStreaming without depending on it The callback reads `isActuallyStreaming` — the flag that stops it resolving the PREVIOUS turn's messageId during the submitted window — but the dep array omitted it. It only stayed fresh because `lastAssistantMessageId` happens to co-vary on the submitted -> streaming transition. That is an accident, not a guarantee: one refactor away from Stop capturing a stale value and aborting the wrong message while the real generation keeps billing. Worth noting HOW this survived: react-hooks/exhaustive-deps IS enabled (via next/core-web-vitals) and does flag it — as a WARNING. `bun run lint` exits 0 on warnings, so CI reported 14/14 successful while the rule was pointing straight at it. The signal existed and was not wired to anything, which is the same failure as the six tests that could not fail and the CI that never ran. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4 * fix(ai): the GLOBAL stop's pre-first-chunk abort still named nothing Auditing the abort call sites — the last place a "signal wired to nothing" could hide — found the global-mode stop function had never received the conversationId fix: await abortActiveStream({ chatId: convId }); // no conversationId That is the exact window the fix exists for. The chatId map is EMPTY before the response headers land (0.5-3s into a real send, while auth/rate-limit/context-assembly/provider-connect run), so this abort was a guaranteed no-op: the local fetch stopped, the button flipped back to Send, and the server kept generating, kept running write tools, and kept BILLING. I fixed the agent-mode stop and the sidebar and AiChatView, and missed this one. That is the THIRD time in this work I have fixed one instance of a defect and not checked its siblings — the same incompleteness as the false comment I corrected in the chat route but not the global route. So this time I swept exhaustively: `grep abortActiveStream({` across the app now shows ZERO call sites without a conversationId. Every chatId abort names its conversation. KNOWN GAP, DELIBERATELY NOT GUESSED AT Every caller discards the abort's return value (`void abortActiveStream(...)`), so `{aborted:false}` is itself a signal wired to nothing: the button flips back to Send whatever happened. I am not inventing UI for it, because `aborted:false` has two very different meanings and the right response differs: - "No in-flight stream on this conversation" — the stream already finished. A benign race; a toast here would fire constantly and train users to ignore it. - "In-flight stream(s) found but none could be aborted from this instance" — the generation is REAL, running on another web instance, and still billing. The user should almost certainly be told. Which of those surfaces, and how, is a product decision I cannot infer. Flagged rather than guessed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4 --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…nd, reload, or switch (#2061) * fix(ai): stop global-assistant load-on-select effects from clobbering live streams Sending a message from the global assistant (dashboard or right-sidebar surface) could visibly flash and make the in-progress reply disappear; reloading or switching conversations mid-stream never reattached to the live content. Root cause: four "load-on-select" effects across GlobalAssistantView.tsx and SidebarChatTab.tsx (global-mode and agent-mode variants in each) reapplied a conversation's persisted messages into the surface's own local useChat state whenever the reference changed (mount, reload, conversation/agent switch) with no guard against the surface actively streaming — unlike their sibling refresh-effects, which already had one. The unconditional overwrite clobbered the in-progress assistant bubble with a stale DB snapshot that predated the reply. Add the same streaming guard already used by the sibling effects to all four call sites. No new rendering path was needed: the existing remoteStreams/inflightRemoteStreams pipeline already renders a bootstrapped stream's live content independent of local message state, once the clobber is prevented. Also flagged, not addressed here (separate, larger, already-documented effort per PR #2011/#2018/#2022): cross-instance live-token rejoin doesn't work when a reload lands on a different web instance than the one running the generation, since the multicast stream registry is single-process. * fix(ai): advance load-on-select refs only when the guarded apply runs Codex review on this PR (3 threads, P1) found that the streaming guards added to the load-on-select effects advanced their "seen" ref BEFORE checking the guard. When the guard blocked (streaming), the ref was still marked seen — so once streaming ended, the effect would see "no change" on the same reference and permanently skip the deferred load, leaving the surface on stale or empty history instead of reattaching. Fix: move the ref assignment inside the guarded branch in all three flagged effects (GlobalAssistantView's agent load-signal effect, SidebarChatTab's global and agent load-on-select effects). While touching this, applied the identical fix to the pre-existing refreshSignal effect in SidebarChatTab (same defect shape, not introduced by this PR but same file/subject), and discovered + fixed a related gap: GlobalAssistantView's refreshSignal effect had no streaming guard at all, meaning it could clobber an in-progress reply the same way the original bug did. Added a stateful "ref-advances-only-on-apply" simulator to both test files to pin the actual multi-render sequence the bug depends on (guard blocks, then guard passes with the SAME reference must still apply) — the previous single-call boolean tests couldn't express this. * fix(ai): dedup GlobalAssistantView's global load-on-select on the message reference CodeRabbit review found a severe regression in the previous commit: this effect had no prevRef/dedup check, only a streaming guard. Since effectiveIsStreaming is itself a dependency, the effect re-fires on every streaming transition — including the streaming-to-not-streaming edge at the end of every ordinary send. GlobalChatContext's onStreamComplete deliberately does not refresh globalInitialMessages for an own fresh completion ("surface's useChat already has the message"), so without a dedup check this effect would reapply the stale PRE-SEND snapshot the instant a stream finished, wiping every just-completed reply in GlobalAssistantView's global mode. Added the same prevGlobalInitialMessagesRef dedup + ref-advances-only- on-apply discipline already used by every other load-on-select effect in this PR, and a regression test pinning the exact failure sequence: same snapshot reference, effectiveIsStreaming flips true -> false, must not re-apply. * fix(ai): scope streaming guards to the conversation being loaded, not the surface Found via proactive review (code-review skill, not a reviewer comment): the streaming guards added earlier in this PR (`!displayIsStreaming`/ `!effectiveIsStreaming`) blocked a load-on-select/refresh effect whenever this surface's useChat was streaming AT ALL — but useChat's id is a stable constant per surface, not per conversation, so switching to a different conversation for the same agent (or a different global conversation) does NOT abort the old conversation's in-flight send. The old guard would therefore stay blocked by an unrelated stream still running in a conversation the user already left, stranding the newly selected conversation on stale/previous messages until that unrelated stream happened to finish. Fixed by comparing the surface's own held stream-start conversation id (streamConvIdRef/globalStreamConvIdRef in GlobalAssistantView, heldStreamConvIdRef in SidebarChatTab — both already existed for the Stop/abort path) against the conversation actually being loaded, so the guard only blocks when the live stream truly belongs to that conversation. Six effects updated: SidebarChatTab's refreshSignal, global load-on-select, and agent load-on-select effects; GlobalAssistantView's refreshSignal, agent load-signal, and global load-on-select effects. Added regression tests pinning the conversation-scoping property directly (own stream in conversation A + switch to idle conversation B must not block B's load) in both test files. Also removed a test that inlined a standalone reimplementation of a bug no longer present anywhere in production code, so it could never actually regress (documentation dressed as an assertion, flagged by the same review pass).
1 participant
This is PR 1 of 2, split out of #2011. Same work, reviewable in two coherent pieces. #2011 is now retargeted onto this branch and carries the rest.
Why the split
#2011 went fourteen review rounds. The server-owned-streams core — this PR — has been stable for the last nine of them. Every new bug in those nine rounds came from the Stop-button/ownership machine, which is pre-existing, orthogonal to all seven acceptance criteria, and has now produced 26+ defects. Two of the most recent were introduced by me, while fixing something else.
So: merge the architecture on its own merits; give the ownership machine the focused review its defect density has earned. That's #2011, stacked on this.
The architecture
A stream is a server-side entity with its own identity, lifecycle and durable state. A client is a subscriber. Any client that opens the conversation — the originator, a fresh tab, another device — attaches to whatever is live.
Previously a stream was implicitly owned by the tab that started it: the tab consuming the POST body dropped its own stream on the socket. But the server generation is deliberately disconnect-immune. So when the originating tab reloaded — which on iPad happens constantly, because iOS jetsams the WKWebView content process and Capacitor calls
webView.reload()— the stream was orphaned. The server kept generating and editing pages while the reloaded tab showed no stream, no Stop button, and a live input. A second send started a second generation on the same conversation. Two agents, two bills.The iPad symptoms disappear as a consequence of the architecture, not as a patch.
Acceptance criteria
!(isOwn && isConsuming)— a synchronous refcount marked before the POST leaves (so it cannot lose the race againstbroadcastAiStreamStart). It is module state, so it is empty after a reload — and that emptiness is the fix.skipReplayCountdrops the duplicate prefix). A failed cross-instance SSE join no longer suppresses the completion event, so the finished reply loads from the DB instead of vanishing.${pageId}-defaultsentinel is gone, in two ordered steps. Gating on theisPersistedfact — not the id string — recovers every existing user's stranded history with no migration. Then new chats mint real cuids, with server-side validation.lastHeartbeatAt.Three things worth a reviewer's attention
AC5 is takeover, not 409, and that is load-bearing. A 409 would self-lock the conversation: the terminal status write is fire-and-forget and dies with its process, so a crashed generation leaves a permanently-
streamingrow. The user would be locked out of their own chat. Takeover aborts every in-flight row (liveness must never gate the abort — a stream inside a long tool call is quiet, not dead) and reconciles only provably finished ones.AC7 now has a tripwire, because comments do not fail CI. Wiring
request.signalintostreamTextis a one-line change, it looks like an obvious cleanup ("we're leaking a generation on disconnect!"), and nothing failed if you did it — the tab just quietly resumed murdering its own agent on every reload.disconnect-immunity.test.tsis a source-level assertion on both generation routes, verified by writing the exact "fix" a future dev would write. It also asserts the mirror:stream-joinmust readrequest.signal, or it leaks a subscriber per dead tab.Two security bugs fell out of this work, neither of which was the point:
listConversations(userId = me OR isShared), fails closed, and 404s rather than 403s.Validation
tscclean · lint 14/14 · 12,658 tests passing. Three pre-existing failures (activity-tools,admin-role-version,grouping) reproduce identically on a cleanmasterworktree — they need a Postgres role absent from this environment, plus one timezone-dependent test.Every fix here is pinned by a test I verified fails without it — I reverted each one to confirm the test actually catches the bug rather than assuming it would. That habit is not ceremony. Over the course of this work:
isShared: falsekept 100% of the suite green while every shared conversation went dark for every co-member, and one where the AC5 takeover call could have been deleted with nothing failing;const { signal } = request— a destructure — walked straight past a member-access regex).All three are fixed, and each fix is confirmed by reintroducing the bug and watching the test go red. If you review nothing else here, review
disconnect-immunity.test.tsand the takeover assertions instream-socket-events.test.ts— those are the guards the rest of the architecture leans on.0203adds a column and an index. TheALTERis safe (now()isSTABLE, so PG ≥11 stores a missing-value; no table rewrite). TheCREATE INDEXis the one to think about: it is notCONCURRENTLY, so it takes aSHARElock that blocks writes toai_stream_sessions— and that table is never pruned (one row per AI generation, forever, since #1159), so it is plausibly large. Blocking writes there blocks every new AI stream from starting.The generated SQL is
CREATE INDEX **IF NOT EXISTS**, so the safe path needs no code change — build it first, and the migration becomes a no-op:There is also a rollout window: old workers do not heartbeat, so during a deploy their in-flight streams age out and get driven terminal. Documented in the schema rather than gated behind a flag — no content is lost (the message persists via the normal path;
partsis only a crash-recovery snapshot), and the worst case is a concurrent generation, which is exactly whatmasterdoes today since master has no takeover at all. Deploy with a rolling strategy that lets old machines drain (Fly does this by default).Known residual, deliberately not fixed here
A TOCTOU on the takeover guard.
takeOverConversationStreamsdoes a plainSELECT, and the row that would block a peer is onlyINSERTed a few statements later — so two near-simultaneous sends can both see zero in-flight rows and both proceed.Closing it needs DB-level serialization: an advisory lock spanning takeover+insert, or a partial unique index on
(conversation_id) WHERE status='streaming'— whose migration would fail outright on any existing duplicate rows, so it needs a reconciliation step first. That is a risky change to bolt on at the end of a long PR, andmasterhas no takeover at all (concurrent sends there always double-generate), so this is already a strict improvement. It should be its own PR.🤖 Generated with Claude Code
https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4
Summary by CodeRabbit
New Features
Bug Fixes