fix(ai): Stop-button / stream-ownership machine (stacked on #2018)#2011
fix(ai): Stop-button / stream-ownership machine (stacked on #2018)#20112witstudios wants to merge 1 commit into
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughChangesAI stream lifecycle and authorization
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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: 332c691e4e
ℹ️ 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".
| // In-memory and single-process: a no-op when the stream belongs to another | ||
| // instance (or already ended). The reconcile below is what actually | ||
| // guarantees the conversation is unblocked. | ||
| const result = abortStreamByMessageId({ messageId, userId }); |
There was a problem hiding this comment.
In a shared conversation, if user B sends while user A's generation is still running, this calls abortStreamByMessageId with B's user id. The abort registry authorizes aborts against the stream owner, so the call returns unauthorized, after which the row is still marked aborted and B's new stream starts while A's old generation continues editing/billing in the background. Select/pass the row's original userId (or use an internal abort path) so takeover actually stops live streams started by other participants.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Correct, and fixed in a70e1d1 — thank you, this was the sharpest finding on the PR.
You're right that the abort registry authorizes against the stream's owner (stream-abort-registry.ts:141-143), so passing the caller's id made the guard refuse every cross-user abort. On a shared conversation that meant user B's send left user A's generation running — still calling tools, still editing pages, still billing — while B's new generation started beside it. Exactly the double-generation this guard exists to prevent.
That IDOR guard exists for the client-facing POST /api/ai/abort endpoint. Takeover is not that: it's a trusted server-side path, and the caller's right to write to this conversation is already established upstream (page edit access, plus a new ownership/shared check on conversationId in the route). So takeOverConversationStreams now selects aiStreamSessions.userId and aborts as the owner:
const owner = ownerByMessageId.get(messageId);
if (!owner) continue;
const result = abortStreamByMessageId({ messageId, userId: owner });The userId parameter is gone from the function signature entirely, so there is no way to reintroduce the caller-id variant by accident. Pinned by a test that asserts the abort is issued with the owner's id and that the takeover actually succeeds: "given a live stream owned by ANOTHER user, should abort it as its OWNER so the takeover actually stops it" (stream-takeover.test.ts).
Related: the second half of what you describe — the row being marked aborted even when the abort was refused — is also fixed. decideStreamTakeover now only reconciles rows we can prove are finished (actually aborted, or provably dead). A live row we cannot stop is left alone and logged, because stamping status='aborted', parts=[] over a running stream hides it from every subscriber and destroys its only crash-recovery snapshot.
There was a problem hiding this comment.
Bookkeeping: this file has moved to #2018.
This PR was split. stream-takeover.ts is no longer part of its diff — it now lives in #2018 (the server-owned-streams core), which this PR is stacked on.
The concern you raised (abort streams with the row's own owner id, not the caller's) was fixed and is in #2018. To restate it so it isn't lost in the move: the abort is issued as the stream's owner (row.userId), not the caller's. Passing the caller's id made the registry's ownership guard refuse every cross-user abort — so on a shared conversation, user B's send would leave user A's generation running (still calling tools, still editing pages, still billing) while B's new one started beside it. Which is the entire thing the takeover exists to prevent.
Also since fixed in #2018, from a later review: the takeover used to lie on partial failure — if the aborts landed but the reconcile UPDATE threw, it reported {aborted: [], reconciled: []} ("nothing happened") when streams had in fact been stopped.
Leaving this thread open rather than resolving it, since it's yours to verify — but please verify it against #2018, not here.
| now: number; | ||
| staleAfterMs?: number; | ||
| }): StreamTakeoverDecision => ({ | ||
| abort: rows.filter((r) => isStreamRowLive(r, now, staleAfterMs)).map((r) => r.messageId), |
There was a problem hiding this comment.
When a live generation is quiet for more than two minutes (for example during a long tool call, or before enough chunks arrive to hit the 20-part checkpoint), isStreamRowLive returns false and this takeover decision omits it from abort. The subsequent reconcile marks the DB row aborted, but the in-process abort controller is never signaled, so a second send can run concurrently with the still-running first stream and the old stream can later keep writing/tool-editing. Takeover should still attempt to abort every in-flight row that may be owned by this process, using liveness only to decide what to expose as resumable.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Agreed and fixed — this was a genuine defect that defeated AC5, and my original code comment had the logic exactly backwards (it claimed misjudging a live row as stale was "harmless").
Two changes:
1. Liveness no longer gates the abort. As you say, aborting an unknown messageId is free (abortStreamByMessageId returns {aborted:false}; it does not throw), whereas skipping an abort for a row we misjudged leaves a real generation running. The asymmetry is total, so decideStreamTakeover now returns every in-flight row in abort, and liveness is used only to decide what may be driven terminal:
abort: rows.map((r) => r.messageId),
reconcile: rows
.filter((r) => aborted.has(r.messageId) || !isStreamRowLive(r, now, staleAfterMs))
.map((r) => r.messageId),That also fixes the reconcile half you flag: a row we could not abort but that is still beating is now left alone, rather than being stamped aborted while it keeps writing.
2. The heartbeat is now a real timer, not the parts checkpoint. Your parenthetical is the root cause: a stream in a long tool call pushes no parts at all, so a checkpoint-driven heartbeat could never have covered it — no value of PERSIST_EVERY_N_PARTS would. stream-lifecycle.ts now writes lastHeartbeatAt on a 20s interval (cleared in finish(), unref'd), independent of pushPart. Without this the same quiet stream also vanished from GET /active-streams mid-flight, so no client could attach to it — which is what walked the user into sending again in the first place.
Tests pinning both: "given a row that LOOKS stale, should still attempt to abort it (a liveness guess must never gate the abort)", "given a LIVE row we could not abort, should NOT reconcile it" (stream-liveness.test.ts), and "given a stream that pushes NO parts at all (a long tool call), should still beat" (stream-lifecycle.test.ts).
There was a problem hiding this comment.
Bookkeeping: this file has moved to #2018.
This PR was split. stream-liveness.ts is no longer part of its diff — it now lives in #2018 (the server-owned-streams core), which this PR is stacked on.
Your point (do not skip aborting a quiet stream) was fixed and is in #2018. Restating it so it survives the move, because it was the most important correction in the whole effort: liveness must never gate the abort. A stream sitting in a long tool call pushes no parts and looks idle — it is quiet, not dead. The original guard skipped aborting it, marked its row aborted, and started a second generation beside the first. Exactly the double-generation the guard was written to prevent.
The split is now: abort EVERY in-flight row, live-looking or not (a redundant abort is a cheap no-op; a skipped one leaves a real generation running), and reconcile only what was actually stopped plus what is provably dead.
Leaving this open for you to verify — but against #2018, not here.
Convergence summary — what changed since the initial reviewFour adversarial review passes over this diff surfaced defects serious enough that the original AC5 guard was not merely incomplete, it was wrong. Flagging them here because the PR now differs materially from what was first reviewed, and two of them are security issues. The AC5 guard was defeating itself
Security
One of my own fixes was a regression, and was revertedWhile restoring the sentinel's one deliberate property ("concurrent openers share a conversation"), I dropped the conversation filter whenever the conversation was unpersisted — which is the normal state of a fresh page, New Chat, and post-delete. That rendered every stream on the page channel, i.e. other members' private conversations, and gave a blank "New Chat" a Stop button wired to the old conversation. The property I was chasing was the leak. Reverted; scoping is unconditional. Verification
Every fix above is pinned by a test that fails without it. |
Summary
Testing
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
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/lib/ai/shared/hooks/useConversationIdentity.ts (1)
74-88: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winStale resolve can still overwrite
isPersisted.RESOLVEDis dropped once identity moves on, butsetIsPersisted(result.isPersisted ?? true)still runs unguarded. That can leaveisPersistedout of sync with the active conversation and break the load/refresh logic that depends on it inAiChatView. Gate this update the same way asRESOLVED, or moveisPersistedinto the reducer state.🤖 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/shared/hooks/useConversationIdentity.ts` around lines 74 - 88, The runResolve callback currently lets stale resolve results update isPersisted even when the conversation identity has changed. Guard setIsPersisted with the same active-resolution/identity check used before dispatching RESOLVED, or move isPersisted into the reducer so stale results are rejected consistently; keep AiChatView’s active-conversation load and refresh behavior synchronized.apps/web/src/hooks/useChannelStreamSocket.ts (1)
228-255: 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy liftDo not treat an SSE join failure as stream finalization.
The code correctly notes that a 404 can mean the stream is still running on another web instance, but it immediately calls
fireOwnFinalize. That clears the owning surface’s streaming/Stop state beforechat:stream_complete; bootstrap snapshots remain rendered as non-live, while unseeded streams disappear. Keep ownership and Stop state until terminal completion/abort, while separately recording that token consumption is unavailable.🤖 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/hooks/useChannelStreamSocket.ts` around lines 228 - 255, Remove the fireOwnFinalize call from the SSE join-rejection path in useChannelStreamSocket, so a failed join only records joinFailed and applies the existing snapshot/removal behavior without finalizing ownership or Stop state. Ensure finalization remains driven by chat:stream_complete or an explicit abort, while preserving the separate handling of seeded and unseeded streams.
🧹 Nitpick comments (3)
apps/web/src/app/api/ai/chat/active-streams/__tests__/route.test.ts (1)
50-51: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the duplicate
filterSubscribableStreamsmock entry.The
vi.mockfactory object definesfilterSubscribableStreamstwice with identical implementations. The second silently overwrites the first — harmless here since they're identical, but it's a copy-paste artifact that will confuse future readers.🧹 Proposed fix
vi.mock('`@/lib/ai/core/stream-subscription-authz`', () => ({ filterSubscribableStreams: (args: { rows: unknown[] }) => mockFilterSubscribableStreams(args), - filterSubscribableStreams: (args: { rows: unknown[] }) => mockFilterSubscribableStreams(args), }));🤖 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/__tests__/route.test.ts` around lines 50 - 51, Remove the duplicate filterSubscribableStreams property from the vi.mock factory in the active-streams route test, keeping one implementation that delegates to mockFilterSubscribableStreams.apps/web/src/app/api/ai/chat/stream-join/[messageId]/route.ts (1)
44-62: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate page-access check — extract a shared helper.
hasPageAccessand thepageOkcomputation insidehasViewAccesscompute the identicalchannelOwner !== null ? channelOwner === userId : await canUserViewPage(...)expression independently. Since this is security-relevant logic re-evaluated on every periodic recheck, keeping two copies in sync manually is a future-drift risk.♻️ Proposed refactor to share the check
+ const checkPageAccess = async (): Promise<boolean> => + channelOwner !== null ? channelOwner === userId : canUserViewPage(userId, meta.pageId); + - const hasPageAccess = channelOwner !== null - ? channelOwner === userId - : await canUserViewPage(userId, meta.pageId); + const hasPageAccess = await checkPageAccess(); ... const hasViewAccess = async (): Promise<boolean> => { - const pageOk = channelOwner !== null - ? channelOwner === userId - : await canUserViewPage(userId, meta.pageId); + const pageOk = await checkPageAccess(); if (!pageOk) return false; return canSubscribe(); };Also applies to: 81-90
🤖 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/stream-join/`[messageId]/route.ts around lines 44 - 62, Extract the duplicated page-access expression into a shared helper near hasViewAccess, accepting channelOwner, userId, and meta.pageId as needed. Replace both the hasPageAccess assignment and the pageOk computation inside hasViewAccess with calls to this helper, preserving the existing authorization outcomes and recheck behavior.apps/web/src/app/api/ai/chat/route.ts (1)
409-462: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winRedundant DB round-trips for new conversations — route now duplicates
createConversation's internal checks.The new validation block already resolves both "does a
conversationsrow exist" (viagetConversation) and "is there a conflicting message owner" (viahasConflictingMessageOwner) for a fresh cuidconversationId.conversationRepository.createConversation(called shortly after, at line 527) re-runs both checks internally, doubling the query count on the first-message-of-a-new-conversation path.Consider threading the already-known result through, e.g. an options flag so
createConversationcan skip its own existence/conflict checks when the caller has already validated them:♻️ Proposed sketch to avoid the duplicate queries
- await conversationRepository.createConversation(conversationId, userId!, chatId).catch(() => {}); + await conversationRepository.createConversation(conversationId, userId!, chatId, { + // Route-level validation above already confirmed no row exists and no + // conflicting owner, when requestConversationId was supplied. + skipConflictCheck: Boolean(requestConversationId) && !existingConversation, + }).catch(() => {});// conversation-repository.ts async createConversation( conversationId: string, userId: string, pageId: string, options?: { skipConflictCheck?: boolean }, ): Promise<void> { const [existing] = await db.select({ id: conversations.id }).from(conversations) .where(eq(conversations.id, conversationId)).limit(1); if (existing) return; if (!options?.skipConflictCheck) { const hasConflictingOwner = await hasConflictingMessageOwner(pageId, conversationId, userId); if (hasConflictingOwner) return; } ... }Also applies to: 527-527
🤖 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 409 - 462, Thread the route’s completed conversation validation into createConversation to avoid repeating database checks for a validated requestConversationId. Add an explicit option recognized by conversationRepository.createConversation to skip both existence and conflicting-message-owner queries only when the route already performed them, while retaining the current checks for all other callers. Pass that option at the createConversation call site after the validation block.
🤖 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/components/layout/middle-content/page-views/ai-page/AiChatView.tsx`:
- Around line 480-488: The useEffect sync guard can record a conversation as
synced before it exists in the refreshed list. Update the effect around
syncedConversationRef and refreshConversations so the ref is assigned only after
confirming conversations contains currentConversationId, or otherwise defer the
refresh until sending has started; ensure a later effect run can retry when the
row appears.
In `@apps/web/src/hooks/useChannelStreamSocket.ts`:
- Around line 50-52: Make own-stream finalization scoped to the
conversation/claim that bootstrapped the stream: update onOwnStreamFinalize and
the related finalize paths to carry and validate the conversation or claim
identity before clearing state. Ensure a rejected stream cannot finalize or
clear another stream’s Stop control, while preserving finalization for the
stream that actually owns the slot.
- Around line 395-402: The failed-join completion flow around joinFailed,
removeStream, and onStreamComplete must preserve global reload behavior. Pass
explicit join-failure metadata to the completion handler, or update the global
completion handling to reload when the conversation matches and the stream entry
is absent, ensuring persisted assistant content is fetched after removal.
In `@apps/web/src/lib/ai/core/__tests__/stream-lifecycle.test.ts`:
- Around line 284-289: Update the comment above the mockUpdateSet expectation in
the stream lifecycle test to remove the outdated claim that no separate
heartbeat timer exists, and accurately describe this assertion as verifying the
parts checkpoint write and its lastHeartbeatAt timestamp. Leave the independent
heartbeat timer tests and behavior unchanged.
---
Outside diff comments:
In `@apps/web/src/hooks/useChannelStreamSocket.ts`:
- Around line 228-255: Remove the fireOwnFinalize call from the SSE
join-rejection path in useChannelStreamSocket, so a failed join only records
joinFailed and applies the existing snapshot/removal behavior without finalizing
ownership or Stop state. Ensure finalization remains driven by
chat:stream_complete or an explicit abort, while preserving the separate
handling of seeded and unseeded streams.
In `@apps/web/src/lib/ai/shared/hooks/useConversationIdentity.ts`:
- Around line 74-88: The runResolve callback currently lets stale resolve
results update isPersisted even when the conversation identity has changed.
Guard setIsPersisted with the same active-resolution/identity check used before
dispatching RESOLVED, or move isPersisted into the reducer so stale results are
rejected consistently; keep AiChatView’s active-conversation load and refresh
behavior synchronized.
---
Nitpick comments:
In `@apps/web/src/app/api/ai/chat/active-streams/__tests__/route.test.ts`:
- Around line 50-51: Remove the duplicate filterSubscribableStreams property
from the vi.mock factory in the active-streams route test, keeping one
implementation that delegates to mockFilterSubscribableStreams.
In `@apps/web/src/app/api/ai/chat/route.ts`:
- Around line 409-462: Thread the route’s completed conversation validation into
createConversation to avoid repeating database checks for a validated
requestConversationId. Add an explicit option recognized by
conversationRepository.createConversation to skip both existence and
conflicting-message-owner queries only when the route already performed them,
while retaining the current checks for all other callers. Pass that option at
the createConversation call site after the validation block.
In `@apps/web/src/app/api/ai/chat/stream-join/`[messageId]/route.ts:
- Around line 44-62: Extract the duplicated page-access expression into a shared
helper near hasViewAccess, accepting channelOwner, userId, and meta.pageId as
needed. Replace both the hasPageAccess assignment and the pageOk computation
inside hasViewAccess with calls to this helper, preserving the existing
authorization outcomes and recheck behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 86c86eb5-1435-4e83-8b02-143888a742d0
📒 Files selected for processing (46)
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/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/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/useAgentChannelMultiplayer.tsapps/web/src/hooks/useAppStateRecovery.tsapps/web/src/hooks/useChannelStreamSocket.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-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-lifecycle.tsapps/web/src/lib/ai/core/stream-liveness.tsapps/web/src/lib/ai/core/stream-subscription-authz.tsapps/web/src/lib/ai/core/stream-takeover.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/consumingChannels.tsapps/web/src/lib/ai/streams/isPlaceholderConversationId.tsapps/web/src/lib/ai/streams/shouldAttachStream.tsapps/web/src/lib/repositories/conversation-repository.tsapps/web/src/lib/websocket/socket-utils.tspackages/db/drizzle/0200_strange_viper.sqlpackages/db/drizzle/meta/0200_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
|
@chatgpt-codex-connector — good catch on the stale docs, and you were right that they were the inverse of the code. Landed in Both comments you flagged had drifted during the fixes:
Two more staleness bugs in the same comments while I was there: the takeover guard is no longer Re: your test-run warning — |
Summary
Testing
|
|
@chatgpt-codex-connector — thanks, this landed on a stale SHA ( Adopted: the stale resolve clobbering
|
Summary
Testing
|
|
@chatgpt-codex-connector — confirmed, thanks for the verification pass. Nothing further needed on that one. For completeness, since you last looked the branch also picked up two things worth knowing:
|
Summary
Testing / Checks
|
|
| # | Bug | Consequence |
|---|---|---|
| 1 | Blanket own-session skip | Reloaded tab drops its own stream (the PR's headline bug) |
| 2 | Ownership guard conflated two claims | isStreaming stranded true — composer bricked until reload |
| 3 | Updater form passed to a zustand value setter | Stop button silently did nothing while the stream kept billing |
| 4 | Claim released only by an event that couldn't fire | Permanent strand after a socket-instance swap |
| 5 | Reconciliation trusted /active-streams as "what's alive" |
It means "started in the last 10 min" — released Stop on a live stream |
| 6 | Three lifetime horizons disagreed (10 min vs 60 min) | Live stream you can't join and can't stop |
| 7 | GlobalAssistantView never joined the claim protocol |
Wiped the slot from outside — the root cause rounds 2–6 kept accommodating |
All seven are fixed, and each is pinned by a test I verified fails without the fix (I reverted every fix to confirm the test catches the bug rather than assuming). I also retyped the slot so bug 3's class is now a compile error — reintroducing it fails tsc:
error TS2345: Argument of type '() => () => Promise<void>' is not assignable
to parameter of type '() => void | Promise<void>'.
But round 7 finding the root cause means rounds 2–6 were treating symptoms. That's a design smell, not a testing gap: the slot is a mutable singleton written by four surfaces with no ownership discipline. It pre-dates this PR — but this PR is the first thing to lean on it hard (a bootstrapped stream now claims Stop across a reload, which nothing previously did), which is why it started shedding bugs.
Three options — I'd like a human call
- Ship as-is. Strictly better than master; file a follow-up to redesign the slot.
- Redesign the slot in this PR. Single writer + explicit handoff. Bigger diff, more risk.
- Split. Merge the server-owned-streams core (takeover, subscription authz, sentinel removal, heartbeat — the actual ACs, all well-tested and independently valuable) and move the Stop-button/ownership work to its own PR.
My recommendation is (3). The core is what the PR was for; the stop-slot machine is a pre-existing landmine this PR merely walked onto, and it deserves review on its own terms rather than being the tail wagging this one.
Happy to execute any of the three — just say which. Meanwhile I'm continuing to converge everything else.
Update: bug 8 found, and it forced the design fixFollowing the decision request above — an eighth pass found a bug that is not a race and needs no refresh. It is reachable by clicking, and it settles the question of which option to take.
They shared one identity-less slot: a bare
This is the redesign, and it was the right callSeven rounds hardened the claim protocol; none asked what the slot means. So it now carries its owner: streamingAgentId: string | null
agentStopSlot: { agentId, stop } | nullEvery read goes through So: option 2, scoped tightly. Not a speculative rewrite — the minimum change that makes the bug class unrepresentable. The diff is contained (one store, two consumers, one predicate). Two things worth saying plainly:
Verification: |
There was a problem hiding this comment.
Actionable comments posted: 3
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/hooks/useChannelStreamSocket.ts (2)
278-291: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winDo not finalize an own stream solely because its SSE join failed.
A cross-instance join failure is explicitly treated as a still-live stream, but
fireOwnFinalize()immediately releases its Stop claim. Keep the claim untilchat:stream_completeor an authoritative active-stream snapshot excludes the message.Proposed fix
if (skipReplayCount === 0) { removeStream(messageId); } - fireOwnFinalize(messageId); + // A failed local join does not establish server-side completion. + // Release ownership on stream completion or snapshot reconciliation.🤖 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/hooks/useChannelStreamSocket.ts` around lines 278 - 291, Remove the fireOwnFinalize call from the SSE join-failure path in the stream handling logic, including the skipReplayCount branches. Preserve the Stop claim after a failed join, and release it only when chat:stream_complete arrives or an authoritative active-stream snapshot excludes the message.
309-372: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winPrevent superseded bootstraps from attaching stale streams.
The generation check only suppresses
onActiveStreamsSnapshot. An older response arriving after a newer snapshot can still add a stale stream, invokeonOwnStreamBootstrap, and resurrect a released Stop claim. Gate row processing on generation too, or track the latest successfully applied bootstrap generation.🤖 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/hooks/useChannelStreamSocket.ts` around lines 309 - 372, The runBootstrap row-processing loop must ignore responses from superseded bootstrap generations, not just suppress onActiveStreamsSnapshot. After the fetch/JSON awaits and before iterating data.streams, verify generation === bootstrapGeneration and return when stale, preventing addStream, onOwnStreamBootstrapRef, and startConsume from attaching outdated streams or restoring released claims.
🤖 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/contexts/GlobalChatContext.tsx`:
- Around line 279-313: Scope the bootstrap Stop-slot claim to both conversation
identity and message identity, adding a claimed conversation ID alongside
claimedStopMessageIdRef and validating both when releasing or reconciling. When
activeId resolves from null, release any claim belonging to a different
conversation and rejoin/reconcile the slot for the resolved conversation instead
of retaining the unscoped bootstrap. Update releaseStopSlotIfStillOurs and the
related claim/reconciliation logic around the additional range to use the
conversation ID consistently.
In `@apps/web/src/hooks/useAgentChannelMultiplayer.ts`:
- Around line 103-109: Implement a reclaim handoff for declined live streams in
the agent-slot ownership flow used by useAgentChannelMultiplayer: subscribe to
the active agent/conversation slot and retry claiming it when the current owner
releases it, but only while this surface’s own stream remains in flight. Ensure
the reclaimed owner restores the Stop control and cleanly unsubscribes or stops
retrying when the stream ends, the component unmounts, or the scope changes.
In `@apps/web/src/stores/page-agents/usePageAgentDashboardStore.ts`:
- Around line 62-94: The streaming state contract currently keys ownership only
by agent, so concurrent conversations for the same agent overwrite each other.
Update the store fields, writers, and selectors—including selectIsAgentStreaming
and selectAgentStop and the referenced stream lifecycle logic—to key entries by
both conversationId and agentId, ensuring stop callbacks and cleanup affect only
the matching conversation while preserving independent concurrent streams.
---
Outside diff comments:
In `@apps/web/src/hooks/useChannelStreamSocket.ts`:
- Around line 278-291: Remove the fireOwnFinalize call from the SSE join-failure
path in the stream handling logic, including the skipReplayCount branches.
Preserve the Stop claim after a failed join, and release it only when
chat:stream_complete arrives or an authoritative active-stream snapshot excludes
the message.
- Around line 309-372: The runBootstrap row-processing loop must ignore
responses from superseded bootstrap generations, not just suppress
onActiveStreamsSnapshot. After the fetch/JSON awaits and before iterating
data.streams, verify generation === bootstrapGeneration and return when stale,
preventing addStream, onOwnStreamBootstrapRef, and startConsume from attaching
outdated streams or restoring released claims.
🪄 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: c35bd039-82be-4b0c-8063-a15d508cf57f
📒 Files selected for processing (29)
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/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/contexts/__tests__/GlobalChatContext.test.tsxapps/web/src/hooks/__tests__/useAgentChannelMultiplayer.test.tsapps/web/src/hooks/__tests__/useChannelStreamSocket.test.tsapps/web/src/hooks/useAgentChannelMultiplayer.tsapps/web/src/hooks/useChannelStreamSocket.tsapps/web/src/lib/ai/core/__tests__/stream-lifecycle.test.tsapps/web/src/lib/ai/core/__tests__/stream-multicast-registry.test.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-takeover.tsapps/web/src/lib/ai/shared/hooks/__tests__/useConversationIdentity.test.tsapps/web/src/lib/ai/shared/hooks/useConversationIdentity.tsapps/web/src/lib/ai/streams/__tests__/shouldClaimAgentStopSlot.test.tsapps/web/src/lib/ai/streams/shouldClaimAgentStopSlot.tsapps/web/src/stores/page-agents/index.tsapps/web/src/stores/page-agents/usePageAgentDashboardStore.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 skipped from review due to trivial changes (3)
- packages/db/drizzle/0203_adorable_amphibian.sql
- packages/db/drizzle/meta/_journal.json
- packages/db/drizzle/meta/0203_snapshot.json
🚧 Files skipped from review as they are similar to previous changes (10)
- packages/db/src/schema/ai-streams.ts
- apps/web/src/app/api/ai/chat/active-streams/tests/route.test.ts
- apps/web/src/lib/ai/core/stream-takeover.ts
- apps/web/src/lib/ai/core/tests/stream-lifecycle.test.ts
- apps/web/src/hooks/tests/useChannelStreamSocket.test.ts
- apps/web/src/lib/ai/core/stream-lifecycle.ts
- apps/web/src/contexts/tests/GlobalChatContext.test.tsx
- apps/web/src/lib/ai/core/stream-liveness.ts
- apps/web/src/lib/ai/shared/hooks/useConversationIdentity.ts
- apps/web/src/components/layout/middle-content/page-views/ai-page/AiChatView.tsx
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/components/layout/right-sidebar/ai-assistant/SidebarChatTab.tsx (1)
1094-1130: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winAdd
isActuallyStreamingtohandleStop’s deps
handleStopreadsisActuallyStreamingbut only depends onisStreamingandlastAssistantMessageId. That can leave it using a stale'submitted'vs'streaming'decision and falling back to the wrong abort path.🤖 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/right-sidebar/ai-assistant/SidebarChatTab.tsx` around lines 1094 - 1130, Add isActuallyStreaming to the dependency array of handleStop so resolveActiveAssistantMessageId uses the current streaming state. Keep the existing stop and abort logic unchanged.
🤖 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/components/layout/right-sidebar/ai-assistant/SidebarChatTab.tsx`:
- Around line 1094-1130: Add isActuallyStreaming to the dependency array of
handleStop so resolveActiveAssistantMessageId uses the current streaming state.
Keep the existing stop and abort logic unchanged.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 5b5d9c5c-4d69-490c-93ef-c7e9f171f9b6
📒 Files selected for processing (14)
apps/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/contexts/__tests__/GlobalChatContext.test.tsxapps/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/stream-abort-client.tsapps/web/src/lib/ai/core/stream-lifecycle.tsapps/web/src/lib/ai/shared/hooks/useChatTransport.tsapps/web/src/lib/ai/streams/__tests__/holdForStream.test.tsapps/web/src/lib/ai/streams/__tests__/shouldClaimGlobalStopSlot.test.tsapps/web/src/lib/ai/streams/resolveActiveAssistantMessageId.tsapps/web/src/lib/ai/streams/shouldClaimGlobalStopSlot.ts
✅ Files skipped from review due to trivial changes (1)
- apps/web/src/lib/ai/streams/resolveActiveAssistantMessageId.ts
🚧 Files skipped from review as they are similar to previous changes (7)
- apps/web/src/lib/ai/streams/tests/holdForStream.test.ts
- apps/web/src/lib/ai/core/tests/stream-abort-client.test.ts
- apps/web/src/lib/ai/core/stream-lifecycle.ts
- apps/web/src/lib/ai/core/tests/stream-lifecycle.test.ts
- apps/web/src/contexts/GlobalChatContext.tsx
- apps/web/src/components/layout/middle-content/page-views/dashboard/GlobalAssistantView.tsx
- apps/web/src/components/layout/middle-content/page-views/ai-page/AiChatView.tsx
Round 14: seven more bugs. Two of them were mine. One I introduced last round.Two independent adversarial passes (one client, one server). Both found real defects. Fixed and pushed:
Every one is pinned by a test I verified fails without the fix. The one that should worry you#1 was mine, and it was a regression I introduced while fixing something else. Verified against the vendored SDK, not assumed: Stop then aborted an id the registry no longer knew. The local fetch stopped, the button looked like it worked, and the server kept generating, kept running write tools, and kept billing. On every turn after the first. Last round I hoisted the held id into the primary Stop button. That is what promoted this from a rare failure to a per-turn one. I made it worse, and only an independent adversarial pass caught it. Known residual, deliberately NOT fixed hereTOCTOU on the takeover guard. I am not fixing it in this PR. It needs DB-level serialization ( It should be its own PR, with its own review. Where this leaves usTwenty-six bugs, fourteen rounds, one subsystem. The rate is not falling, and this round it was me generating them. The server-owned-streams core — takeover, subscription authz (a real IDOR and a real privacy leak, both closed), sentinel removal recovering stranded history with no migration, the heartbeat — is stable and has been for many rounds. Every new bug for the last nine rounds has come from the Stop-button/ownership machine, which is pre-existing and orthogonal to all seven acceptance criteria. I have asked five times to split this PR. Asking again, plainly: merge the core; give the ownership machine its own PR. It is a subsystem with a defect density that has now defeated fourteen review rounds, two of them mine, and it deserves a focused review rather than a ride-along. I will keep converging if that is the call. But I am not going to tell you it is clean. I have no evidence for that, and the last time I asserted something I had not verified — telling CodeRabbit a thread was "dissolved" — I was wrong about that too. CI green (lint/typecheck, unit, CodeQL, security suite, secret scanning, dependency audit). |
2witstudios
changed the title
2witstudios
force-pushed
the
pu/server-owned-streams
branch
2 times, most recently
from
b19a7b3 to
f9049c7
Compare
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
2witstudios
force-pushed
the
pu/server-owned-streams
branch
from
f9049c7 to
53c9c79
Compare
1 participant
This is PR 2 of 2. It is stacked on #2018 — review and merge that first.
#2018 carries the server-owned-streams architecture (all seven ACs, the IDOR, the privacy leak, the AC7 tripwire). This PR carries the part that kept breaking.
Read this before reviewing
This PR is the reason the original #2011 went fourteen rounds. Twenty-six-plus defects, all in this subsystem, and the rate never fell. Two of the most recent were introduced by me, while fixing something else. It has been split out precisely so it gets the focused review that record demands, instead of riding along with an architecture that has been stable for nine rounds.
Please do not review this as a routine diff. Every bug here has been the same bug:
A chatId, a panel, a tab, the live
messagesarray — used where an(agentId, conversationId, messageId)triple was required. The symptoms were always some combination of: a Stop button lit for a stream the user isn't looking at; a Stop button that looks like it works and silently does nothing; a composer permanently disabled; one conversation's tokens rendered into another; and, in every case, a server that kept generating, kept running write tools, and kept billing.Streams deliberately survive a client disconnect (that is the whole architecture). So a Stop that misses is not a cosmetic bug. It is a runaway agent and a real invoice.
What is in here
holdForStream— capture a stream's identity when it starts, hold it until it ends. The surface moves independently; the stream's identity does not.selectLiveAssistantIds— resolves each chat's live assistant messageId from its own status and messages. This exists because the bug lived in the wiring, not the reducer: a test ofholdForStreamis blind to which chat's id the caller hands it, and passed happily while the caller handed it the wrong one.shouldClaimAgentStopSlot/shouldClaimGlobalStopSlot— single-writer guards on the shared stop slots.(agentId, conversationId).Three that are worth understanding, because they will come back otherwise
1. Stop aborted the PREVIOUS turn's message, on every turn after the first.
useChatsetsstatus: 'submitted'before issuing the request, and pushes the new assistant message only insidewrite(), which flips the status to'streaming'in the same job. So for the whole submitted window, the array's last assistant message is the previous turn's reply.holdForStreamlatches on the first live render — a submitted render — so it captured and held a message that had finished minutes ago. Stop then aborted an id the registry no longer knew: the fetch stopped, the button looked like it worked, and the real generation kept running.Every messageId here now resolves only at
status === 'streaming'.holdForStream's docstring used to state the false premise outright ("captured when the first chunk arrives") — that sentence is what the bug was made of, and it is corrected in place.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, so a mid-stream network drop leaves its half-streamed message in the array. That is exactly when recovery rejoins the multicast and the full reply arrives. Skipping threw the complete version away. It now replaces by id.
3. A surface may only free the
activeStreamskey it allocated. The sidebar registeredsidebar:<convId>and cleared the bare<convId>— a key it never writes, and which belongs to the dashboard. It leaked its own entry and deleted someone else's.The structural fix this subsystem still wants
Every one of these is a string used as a key, correct by convention and by comment. The real fix is to make the key 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.
I did not do that here. Doing a wide-blast-radius refactor of the highest-defect subsystem, at the tail of a long PR, with my own recent error rate, is how bug #29 gets written. It should be a deliberate follow-up.
Validation
tscclean · lint 14/14 · 12,688 tests passing (3 pre-existing failures, reproduced on cleanmaster).Every fix is pinned by a test I verified fails without it. That habit is not ceremony: earlier in this work a regression test of mine passed with the bug deliberately reintroduced (the helper reimplemented the production selector instead of calling it), and two bugs survived review by hiding behind fixtures that modelled impossible states. When a fix could not be pinned by a pure test, the wiring itself was extracted into a tested function — that is what
selectLiveAssistantIdsis.🤖 Generated with Claude Code
https://claude.ai/code/session_016ie3Bmn3sgiUJ7a6ErwEc4