fix(ai): make Stop work across web instances — cross-instance stream aborts#2022
Conversation
…aborts
AI streams are server-owned and disconnect-immune by design: `request.signal` is never
wired into `streamText`, which is what lets a generation survive an iPad reload. The
consequence is that cancelling the client fetch stops NOTHING — only an explicit
server-side abort that names the stream can stop it.
But the abort registry is an in-memory Map, and prod runs multiple web instances. A Stop
that load-balanced to an instance which did not own the stream found nothing and gave up:
the agent kept generating, kept calling write tools (editing the user's pages), and kept
billing, while the button flipped back to Send. `takeOverConversationStreams` had the same
hole, so a second send started a SECOND generation beside the first — two agents, two sets
of write tools, two bills. It logged 'in-flight stream(s) could not be aborted from this
instance' and proceeded. That warn path is now deleted.
APPROACH — a durable abort mark in Postgres, consumed by the owning instance.
The relay was evaluated and rejected: it is an emit-only fan-out to browsers (no web
instance subscribes to it), so receiving a broadcast would need a socket client in the Next
server, a new service identity, and a new room shape on a relay that is deliberately
generic — and being fire-and-forget, an owner mid-restart would miss the abort permanently.
Postgres is already the cross-instance channel for the heartbeat and the rate limiter.
- The receiving instance marks the row (`abort_requested_at`).
- The owning instance polls its OWN in-flight streams (~1s, lazily started, self-terminating,
zero queries when idle) and performs the abort locally, where the controller lives.
- The caller waits, briefly, and reports honestly what happened.
SECURITY — the WHERE clause IS the authorization. The only writer of the mark is an UPDATE
carrying the caller's user_id, so a user can only ever mark their own stream: there is no
message to forge and no payload to trust. The owner re-reads the owner id from its own row,
and refuses on any mismatch. A remote kill switch is not merely blocked, it is
unrepresentable. Proven by a test whose fake evaluates the predicate the code actually built.
Also fixes a latent bug this depends on: aborting the controller did NOT reliably drive the
row terminal (both routes note that `onAbort` only fires while a streamText is live, and
`onFinish` "may never fire when the mobile client backgrounds mid-stream" — exactly the
population of a cross-instance abort). The terminal write now rides the abort itself, so a
stopped stream can never be reported as "still running, still billing".
Client: `aborted:false` now carries a `code`. 'unconfirmed' (still running, still billing)
warns the user; 'not_found' (a benign race) stays silent. Stop calls the local stop FIRST so
the confirmation wait cannot hang the button.
Tests: pure decision core with zero mocks; every new test mutation-checked (break it, watch
it go red, restore). disconnect-immunity tripwire untouched and green.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
📝 WalkthroughWalkthroughAI stream stopping now persists stream identifiers and abort requests, coordinates aborts across instances, confirms or reconciles stream termination, and reports structured outcomes to clients. Chat lifecycle routes attach finishers so aborts complete the same persistence path. ChangesCross-instance AI stream abortion
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 docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 8ce11142ed
ℹ️ 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".
| // Step 2 — anything not stopped here is either on another instance, already finished, or was | ||
| // never ours. The mark's WHERE clause carries `userId`, so the third case updates zero rows and | ||
| // is reported as 'not_found'. See stream-abort-mark.ts: that predicate IS the authorization. | ||
| const marked = await markAbortRequested({ messageId, streamId, conversationId, userId }); |
There was a problem hiding this comment.
When the local registry already aborts a stream named by messageId or streamId, this still writes an abort mark and waits for the DB row to settle. abortStream() calls lifecycle.finish(true) only through a fire-and-forget terminal update, so the row can still read as status='streaming' here; if that write is slow or fails past the 2s settle timeout, the endpoint reports unconfirmed and shows the user a warning even though the in-process controller was already aborted. For precise local hits, return the confirmed aborted result instead of entering the cross-instance path (or make the finisher awaitable before waiting on the row).
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed in 825f3a0 — this was a real bug, and worse than the report suggests: it also contradicted this PR's own claim that local aborts stay instant.
Why it bites. Aborting the controller is what stops the generation; the row's status is only bookkeeping, written fire-and-forget by lifecycle.finish. So the settle poll was waiting for a fact our own abort had already guaranteed. If that bookkeeping write is slow or fails, the poll times out against a heartbeat that is still fresh (it beat seconds ago) → decideAbortOutcome classifies it stillLive → unconfirmed → we warn the user their agent is "still running and still billing" moments after we killed it in-process. That is exactly the false alarm this design exists to prevent, and it would have fired on the most common path there is — the stream this instance owns. It also spent a needless DB write + poll on every same-instance Stop.
Fix:
- A precise name (
messageId/streamId) that the local registry aborts now returns immediately — no mark, no poll. - The conversation path cannot short-circuit wholesale (it names a set, and a sibling stream may be owned by another instance), so it stops what it can and then drops those ids from the wait set. Otherwise the identical false alarm returns through the side door: the mark's
status='streaming'predicate can still catch a row we just aborted, because its terminal write has not landed yet.
I did not take the "make the finisher awaitable" alternative: lifecycle.finish is deliberately fire-and-forget (it also broadcasts and clears the heartbeat), and making Stop block on a DB write to report success would trade a false alarm for a slow button. Short-circuiting is strictly less machinery.
Tests (abort-stream-anywhere.test.ts), each mutation-checked — restore the bug and they go red:
never marks or waits on a precisely-named stream it aborted itself(assertsmarkAbortRequestedandawaitAbortSettledare not called)- same for
streamId does not wait on the conversation rows it aborted itself(asserts the wait set is exactly the ids we could not stop)confirms the abort when every stream on the conversation was stopped locally
typecheck exit 0 · lint clean · 832 targeted tests green.
…top machine #2011 (Stop-button / stream-ownership) landed on master after this branch was cut, and it rewrote the same three Stop paths. Its versions are strictly better than the base this branch forked from — held stream ids (heldStreamMsgIdRef / globalStreamMsgIdRef) so a mid-stream conversation switch still names the running generation, ownership-guarded stop slots (clearGlobalStopIfOurs / clearAgentStopIfOurs) so a surface cannot destroy a Stop button it does not own, and messageId-first aborts. So master's versions were taken WHOLESALE and this branch's two deltas re-applied on top: 1. Local stop runs FIRST, not in a `finally`. The server abort is no longer instant — when the generation lives on another web instance it marks the stream and WAITS to learn whether the owner stopped it. Awaiting that before the local stop would hang the Stop button for seconds with tokens still rendering. Running it up front guarantees it strictly harder than the `finally` did. 2. The abort outcome is reported (reportAbortOutcome), so a stream that could not be confirmed stopped — still generating, still calling write tools, still billing — reaches the user, instead of being silently discarded while the button says "Send". useChatStop keeps #2011's `conversationId` parameter and its no-chatId fallback. typecheck exit 0 · lint clean · 12,765 tests passing (the 16 failures in activity-tools, admin-role-version and grouping are pre-existing and fail identically on master). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
Codex review (P2, abort-stream-anywhere.ts): a local registry hit still fell through to the
cross-instance path — writing an abort mark and polling the DB row for a terminal status.
That is wrong twice over.
Aborting the controller is what STOPS the generation; the row's `status` is only bookkeeping,
and `lifecycle.finish` writes it fire-and-forget. So the poll waits for a fact our own abort
already guaranteed — and if that bookkeeping write is slow or fails, the poll times out against
a heartbeat that is still FRESH (it beat seconds ago), yielding 'unconfirmed'. The endpoint
would then warn the user that their agent is "still running and still billing" moments after we
killed it in-process. That is precisely the false alarm this design exists to prevent, and it
would have fired on the most common path there is: the stream this instance owns.
It also spent a DB write and a poll on every same-instance Stop — contradicting this PR's own
claim that local aborts stay instant.
- A precise name (messageId / streamId) that the local registry aborts now returns immediately.
- The conversation path cannot short-circuit wholesale (it names a SET, and a sibling stream may
live on another instance), so it stops what it can and then drops those ids from the wait —
the same false alarm would otherwise return through the side door.
Both fixes are mutation-checked: restore either bug and the new tests go red.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (4)
apps/web/src/lib/ai/core/stream-abort-watcher.ts (1)
69-118: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueSingleton timer + batched local-ownership check design looks solid.
The
globalThis-keyed handle correctly survives HMR/double module instantiation, the tick self-stops whenlistLocalStreams()is empty, and errors fromreadMarkedStreams/decideWatcherActions/clearAbortMarksare caught and logged without throwing out of the interval. One nit:listLocalStreams()itself sits outside thetryblock, so the "never throw out of an interval" invariant stated in the comment at Line 112 doesn't technically cover it — though this is a plainMapiteration that's effectively never going to throw in practice.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/web/src/lib/ai/core/stream-abort-watcher.ts` around lines 69 - 118, Move the listLocalStreams() call and its empty-result handling inside the try block in runAbortWatchTick so every interval-tick operation is covered by the existing catch and preserves the “never throw out of an interval” invariant. Keep the self-stop behavior unchanged when no local streams exist.apps/web/src/lib/ai/core/stream-lifecycle.ts (1)
91-98: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueWatcher startup and registry registration have inconsistent error handling.
streamMulticastRegistry.register(...)right below is wrapped in try/catch with a warn log, but the newly addedensureStreamAbortWatcher()call is bare. In practice this is not a live bug — any throw here would just propagate to the caller's own try/catch in the route handlers — but it's an inconsistency worth a quick look for defensive-coding parity with the surrounding code.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/web/src/lib/ai/core/stream-lifecycle.ts` around lines 91 - 98, Review the ensureStreamAbortWatcher() call in createStreamLifecycle and align its error handling with the nearby streamMulticastRegistry.register(...) operation by catching startup failures and issuing the same defensive warning treatment before continuing or propagating according to the existing lifecycle contract.apps/web/src/lib/ai/core/stream-abort-client.ts (1)
55-65: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winType
bodyasunknownand narrow, instead of relying on the implicitanyfromresponse.json().
await response.json()returnsany, sobody.codeandbody as AbortResultbypass strict typing. Narrow explicitly so the validation is type-checked.As per coding guidelines: "Never use
anytypes - always use proper TypeScript types".♻️ Proposed narrowing
const parseAbortResult = async (response: Response): Promise<AbortResult> => { - const body = await response.json(); - - if (!body || typeof body.code !== 'string') { + const body: unknown = await response.json(); + + if (typeof body !== 'object' || body === null || typeof (body as { code?: unknown }).code !== 'string') { // The endpoint answered with something we do not understand (an error envelope, a proxy page). // We cannot claim the stream stopped. return { aborted: false, code: 'unconfirmed', reason: 'Unrecognized abort response' }; } return body as AbortResult; };🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/web/src/lib/ai/core/stream-abort-client.ts` around lines 55 - 65, Update parseAbortResult so the response JSON is first treated as unknown, then explicitly narrow it to an object with a string code before accessing body.code or returning it as AbortResult. Preserve the existing unconfirmed result for invalid or unrecognized response bodies.Source: Coding guidelines
apps/web/src/components/layout/middle-content/page-views/ai-page/AiChatView.tsx (1)
702-709: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDead comparison:
streamTrackingId !== page.idis always false.
streamTrackingIdis declared as a local variable set directly frompage.id(see line 179:const streamTrackingId = page.id;), and both are read from the same render inside this callback's dependency array. SostreamTrackingId !== page.idcan never be true, and the secondabortActiveStreamcall in the array is unreachable — the comment above ("Both keys are tried, and they may resolve to the same stream") describes a scenario that cannot occur in this file. This is harmless today (the single real key is still aborted), but it's misleading and adds unnecessary complexity for a comparison that can never diverge.♻️ Proposed simplification
- const conversationId = currentConversationIdRef.current; - void Promise.all([ - streamTrackingId ? abortActiveStream({ chatId: streamTrackingId, conversationId }) : null, - streamTrackingId !== page.id ? abortActiveStream({ chatId: page.id, conversationId }) : null, - ]).then((outcomes) => reportAbortOutcomes(outcomes.filter((o) => o !== null))); + const conversationId = currentConversationIdRef.current; + void abortActiveStream({ chatId: streamTrackingId, conversationId }).then(reportAbortOutcome);🤖 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 702 - 709, Remove the unreachable second abort path from the callback using streamTrackingId and page.id, since streamTrackingId is always assigned from page.id. Simplify the Promise.all input and update the nearby comment so it only describes the single key being aborted, while preserving reportAbortOutcomes handling.
🤖 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-lifecycle.ts`:
- Around line 20-27: Make the streamId property required in the relevant
lifecycle type or object definition by removing its optional marker. Keep the
existing streamId documentation and current call-site behavior unchanged,
ensuring all stream creation and update paths continue supplying this value.
---
Nitpick comments:
In
`@apps/web/src/components/layout/middle-content/page-views/ai-page/AiChatView.tsx`:
- Around line 702-709: Remove the unreachable second abort path from the
callback using streamTrackingId and page.id, since streamTrackingId is always
assigned from page.id. Simplify the Promise.all input and update the nearby
comment so it only describes the single key being aborted, while preserving
reportAbortOutcomes handling.
In `@apps/web/src/lib/ai/core/stream-abort-client.ts`:
- Around line 55-65: Update parseAbortResult so the response JSON is first
treated as unknown, then explicitly narrow it to an object with a string code
before accessing body.code or returning it as AbortResult. Preserve the existing
unconfirmed result for invalid or unrecognized response bodies.
In `@apps/web/src/lib/ai/core/stream-abort-watcher.ts`:
- Around line 69-118: Move the listLocalStreams() call and its empty-result
handling inside the try block in runAbortWatchTick so every interval-tick
operation is covered by the existing catch and preserves the “never throw out of
an interval” invariant. Keep the self-stop behavior unchanged when no local
streams exist.
In `@apps/web/src/lib/ai/core/stream-lifecycle.ts`:
- Around line 91-98: Review the ensureStreamAbortWatcher() call in
createStreamLifecycle and align its error handling with the nearby
streamMulticastRegistry.register(...) operation by catching startup failures and
issuing the same defensive warning treatment before continuing or propagating
according to the existing lifecycle contract.
🪄 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: f5a50d55-e163-464c-a813-477d3cf0c73f
📒 Files selected for processing (35)
apps/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/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/lib/ai/core/__tests__/abort-conversation-streams.test.tsapps/web/src/lib/ai/core/__tests__/abort-mark-reset.test.tsapps/web/src/lib/ai/core/__tests__/abort-stream-anywhere.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-client.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-decisions.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-mark.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-registry.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-watcher.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/abort-stream-anywhere.tsapps/web/src/lib/ai/core/client.tsapps/web/src/lib/ai/core/stream-abort-client.tsapps/web/src/lib/ai/core/stream-abort-decisions.tsapps/web/src/lib/ai/core/stream-abort-mark.tsapps/web/src/lib/ai/core/stream-abort-registry.tsapps/web/src/lib/ai/core/stream-abort-watcher.tsapps/web/src/lib/ai/core/stream-lifecycle.tsapps/web/src/lib/ai/core/stream-takeover.tsapps/web/src/lib/ai/shared/hooks/useChatStop.tspackages/db/drizzle/0204_next_mandroid.sqlpackages/db/drizzle/meta/0204_snapshot.jsonpackages/db/drizzle/meta/_journal.jsonpackages/db/src/schema/ai-streams.ts
…s-instance Stop An adversarial review sweep (prompted by Codex finding a real one) turned up four more bugs of the same class — a Stop that lies, in one direction or the other. All four are fixed and mutation-checked. 1. FALSE ALARM ON THE MOST COMMON STOP THERE IS. `onFinish` unregisters the controller at its TOP but writes the terminal status at its BOTTOM — after persisting the message, settling the credit hold, and billing each tool call in turn. For that whole window the row still reads 'streaming' with a fresh heartbeat, so a Stop pressed as the last tokens render looked exactly like a stream owned by another instance: we marked a row nobody would ever consume, timed out against the live heartbeat, and warned the user their agent was "still running and still billing" — about a generation that had already finished. The instance KNOWS it finished; it was just throwing that away. `removeStream` now leaves a tombstone (`wasRecentlyFinishedHere`), and a Stop naming a stream that ended here is answered with the truth: nothing is running. Silent. The takeover consults it too, so a rapid follow-up send no longer waits on, and warns about, a stream that is merely finishing. 2. A SWALLOWED WRITE REPORTED AS THE BENIGN CODE. If the UPDATE that RECORDS the abort threw, the catch returned `[]` — indistinguishable from "nothing was in flight", which the client is designed to stay SILENT about. The DB blips, the Stop reaches nobody, and the agent generates and bills on while the user is told nothing at all. `markAbortRequested` now reports `failed`, and a failure to record surfaces as `unconfirmed`. Same for the takeover's mark, which previously logged nothing while starting a second generation beside a live one. 3. A LIVE >1h GENERATION COULD BE TERMINAL-WRITTEN AND REPORTED AS ABORTED. The heartbeat stops at STREAM_MAX_LIFETIME_MS by design; the generation does not. Past that cap a silent heartbeat is the EXPECTED state of a healthy stream — but it was read as death, which would have wiped the row's parts snapshot and hidden a still-generating stream from every subscriber, while telling the user it had stopped. Past the cap we now report `unconfirmed` and never touch the row. 4. THE SETTLE BUDGET HAD NO HEADROOM over the watcher tick (2s vs a 1s tick plus a parts persist plus the terminal write), so a single p99 DB spike manufactured the same false alarm for a stream that HAD been aborted. The budgets now derive from the tick, in stream-horizons.ts — the module that exists precisely because these numbers drifted apart once before. TESTS. The sweep also found that several predicates were mocked at every call site and asserted nowhere — green lights wired to nothing. Worst: deleting `isNotNull(abort_requested_at)` from `readMarkedStreams` would make the watcher abort EVERY generation on the instance within a second of it starting, with the entire suite still green. Now covered, along with `markAbortRequested`'s SET (it asserted only the WHERE), `reconcileDeadStreamRows`' destructive status guard, `clearAbortMarks`, and `markAbortRequestedAsOwner`. The source-level reset tripwire is replaced by behavioural tests, because it only inspected the onConflict branch — dropping `streamId` from the INSERT would have left every fresh row with stream_id=NULL, killing the headline feature, and the tripwire could not see it. Also fixed: a `streamId` matching no row (a stream from a pre-migration worker, whose stream_id is NULL) now falls back to the conversation instead of reporting the silent `not_found`. typecheck exit 0 · lint clean · 12,789 tests passing (the 16 failures in activity-tools, admin-role-version and grouping are pre-existing and fail identically on master). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
…on fallback
CodeRabbit (Major, stream-lifecycle.ts): `streamId` was optional. Both call sites happened to pass
it, but nothing enforced that — and an omission would not fail loudly: the column would simply be
NULL, and cross-instance Stop for that stream would degrade silently, with no signal at build time
or run time. It is now required, so the type catches what nothing else can.
Verifying that finding exposed a bigger one of my own: the streamId -> conversation fallback added
in the previous commit was only HALF wired. The server would fall back when a streamId resolved to
no row — but the client sent `{ streamId }` alone and never the conversation, so there was no
second name to fall back TO and the fallback could never fire. Precisely the rolling-deploy case it
was written for (a stream from a pre-migration worker has no stream_id) would still have been
reported as `not_found`, which the client stays silent about by design, while the generation ran on
and billed. The client now sends both names.
Both halves are mutation-checked: stop sending the conversation, or drop the server-side fallback,
and a test goes red for each.
Also covers the client's conversation fallback in the pre-headers window, which had no test at all
— deleting it would have restored a guaranteed no-op Stop with a green suite.
typecheck exit 0 · lint clean · 12,791 passing (16 pre-existing failures, identical on master).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
Adversarial self-review sweep — five more real bugs, all of the same classCodex finding a genuine bug (the local-hit short-circuit) told me the design had a systematic weak spot rather than a one-off, so I ran an adversarial correctness sweep plus a "which of these tests could not fail?" sweep over the whole diff. Both came back with real findings. Summary of what changed since the first review round: Correctness
Tests that could not failThe sweep found several predicates mocked at every call site and asserted nowhere. The worst: deleting The source-level reset tripwire was replaced by behavioural tests: it only inspected the Every fix above is mutation-checked: the bug is reintroduced, the test is watched go red, and the code restored. Where a test could pass with the bug present, it was rewritten (the cross-user authz test now evaluates the
|
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/web/src/lib/ai/core/stream-abort-registry.ts`:
- Around line 257-289: Update linkStream() to remove the recentlyFinished
tombstone for the reused messageId when establishing a new link. Preserve
existing stream registration behavior while ensuring wasRecentlyFinishedHere({
messageId }) cannot treat the newly linked live generation as recently
completed.
🪄 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: 5480d492-b934-43ff-a442-3c0d990c2355
📒 Files selected for processing (16)
apps/web/src/lib/ai/core/__tests__/abort-stream-anywhere.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-client.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-decisions.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-mark.test.tsapps/web/src/lib/ai/core/__tests__/stream-abort-registry.test.tsapps/web/src/lib/ai/core/__tests__/stream-lifecycle.test.tsapps/web/src/lib/ai/core/__tests__/stream-takeover.test.tsapps/web/src/lib/ai/core/abort-stream-anywhere.tsapps/web/src/lib/ai/core/stream-abort-client.tsapps/web/src/lib/ai/core/stream-abort-decisions.tsapps/web/src/lib/ai/core/stream-abort-mark.tsapps/web/src/lib/ai/core/stream-abort-registry.tsapps/web/src/lib/ai/core/stream-abort-watcher.tsapps/web/src/lib/ai/core/stream-horizons.tsapps/web/src/lib/ai/core/stream-lifecycle.tsapps/web/src/lib/ai/core/stream-takeover.ts
🚧 Files skipped from review as they are similar to previous changes (7)
- apps/web/src/lib/ai/core/tests/abort-stream-anywhere.test.ts
- apps/web/src/lib/ai/core/stream-abort-watcher.ts
- apps/web/src/lib/ai/core/tests/stream-takeover.test.ts
- apps/web/src/lib/ai/core/abort-stream-anywhere.ts
- apps/web/src/lib/ai/core/stream-abort-client.ts
- apps/web/src/lib/ai/core/stream-abort-decisions.ts
- apps/web/src/lib/ai/core/stream-takeover.ts
…silent again CodeRabbit (Major, stream-abort-registry.ts): `linkStream()` re-links a messageId without clearing `recentlyFinished`, so a stale tombstone from a PREVIOUS generation would answer for a new, live one — `wasRecentlyFinishedHere` returns true, `abortStreamAnywhere` reports `not_found`, and the client stays silent by design while the generation keeps running and keeps billing. That is the exact bug the tombstone was added to PREVENT, reachable through the tombstone itself. A stream that is STARTING is not a stream that finished, so registering now clears any tombstone on both of its names (linkStream covers both; createStreamAbortController also clears the streamId, since linkStream only runs when a messageId was given). Mutation-checked: drop either clear and a test goes red. typecheck exit 0 · lint clean · 12,793 passing (16 pre-existing failures, identical on master). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
…unning A third adversarial pass over the newest code found that the tombstone fix had re-opened the very bug it was written to close. Four fixes. 1. HIGH — THE TOMBSTONE SWALLOWED A LIVE STOP. The "did it finish here?" check ran BEFORE the mark, and therefore before the streamId->conversation fallback. But a client's streamId can be STALE: `activeStreams` held the PREVIOUS turn's id until the new response headers landed. So a Stop pressed in turn 2's TTFB window — the single most likely moment for it — named turn 1's finished stream, hit the tombstone, and returned `not_found`, which the UI stays SILENT about by design, while TURN 2 kept generating, kept calling write tools, and kept billing. The short-circuit was also redundant: the tombstone's real job is to keep a finishing stream out of the WAIT SET, which it already does further down. Removing it fixes the bug and deletes code. A precise name that misses now falls through to the conversation and stops the generation that is ACTUALLY running. 2. `abortStream` left no tombstone (only `removeStream` did), so a SECOND Stop naming a stream we had just aborted — a double-click, or any of the surfaces that abort by messageId — escalated, timed out against a heartbeat that was still beating, and warned "still running, still billing" about a stream we killed ourselves seconds earlier. 3. THE TAKEOVER COULD TERMINAL-WRITE A LIVE GENERATION. `decideStreamTakeover` had no cap awareness: past STREAM_MAX_LIFETIME_MS the heartbeat stops BY DESIGN, but it read that silence as death and would write `status='aborted', parts=[]` over a >1h stream that was still running — hiding it from every subscriber and destroying its crash-recovery snapshot. That is the one write the module's own docblock forbids. The cap guard now lives in stream-liveness.ts and both callers share it. A row we actually ABORTED is still reconciled, cap or no cap. 4. Root cause of the stale name: the client never cleared `activeStreams` on normal completion — only on conversation switch or unmount. It now forgets the streamId when the response body ends. The streamId names one generation and dies with it. Every fix is mutation-checked; each has a test that goes red when the bug is restored. typecheck exit 0 · lint clean · 12,799 passing (16 pre-existing failures, identical on master). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
Third adversarial pass — the tombstone had re-opened the bug it was written to closeI ran one more correctness sweep specifically over the code added in the previous round (tombstone, The tombstone swallowed a Stop for a generation that was still running. The "did it finish here?" check ran before the mark, and therefore before the That is exactly the bug this PR exists to eliminate, re-entered through the fix for it. Fixed in The short-circuit turned out to be redundant: the tombstone's real job is to keep a finishing stream out of the wait set, which it already did further down. Removing it fixes the bug and deletes code. A precise name that misses now falls through to the conversation and stops the generation that is actually running. Three more from the same pass:
All four are mutation-checked: restore the bug, watch the test go red, restore the code.
|
…on the abort mark A fourth adversarial pass — on the code the third pass produced — found two HIGH bugs. Both were mine, and one was a straight regression. 1. THE CAP GUARD KILLED THE ONLY REAPER. Asking "has this row outlived the heartbeat cap?" makes EVERY row older than the cap unreconcilable — including one whose process crashed at minute five and which nobody looked at for an hour. That row is definitively dead, but it would sit at 'streaming' forever: a permanent ghost, poisoning every later Stop on its conversation with a false "may still be running" (the batch verdict is pessimistic, so one ghost condemns the whole conversation) and every later send with a warn. I traded a rare harm — a live >1h generation being terminal-written — for a common one: any crash, plus an hour of the user not looking. The row itself tells us which it is, and I missed it. A stream still alive at the cap beat right UP TO it (lastHeartbeatAt ≈ startedAt + cap). A stream that crashed at minute five stopped beating there. So the question is not "is the row old?" but "where did the beat stop?" — `heartbeatStoppedAtCap`. A beat that reached the cap is ambiguous (silent by design; we cannot prove death, so we never touch it). A beat that stopped SHORT of the cap is proof the process died, at any age, and is reconciled. Both the takeover and the Stop path share the predicate. 2. THE MARK'S FIRST-MATCH PRECEDENCE SWALLOWED A LIVE STOP. A stale streamId is not the same as one that resolves to nothing: during the tail of onFinish — after the controller is unregistered but before the terminal write lands — the finished stream's row STILL READS 'streaming'. So the stale name MATCHED, precedence stopped there, and the conversation was never marked. The generation that was actually running (a later turn, possibly on another instance) was never asked to stop, and the caller was told `not_found` — which the UI stays silent about. A silent Stop over a live generation: the cardinal sin, in its third distinct shape. The mark is now a UNION over every name the caller gave, which also makes it AGREE with the local step (which already aborts every one of the caller's streams on the conversation). Two steps disagreeing about what a "name" means is why this bug kept coming back. 3. Also: `forgetStream` had no identity guard, so a stream that was ENDING could wipe the map slot of one that was still RUNNING (turn 2 claims the slot; turn 1's body then closes and deletes turn 2's streamId). It now only forgets its own. All three mutation-checked. Note the cap mutation also turns the PRE-EXISTING reaper tests red, which is exactly the regression signature. typecheck exit 0 · lint clean · 12,803 passing (16 pre-existing failures, identical on master). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
Fourth pass — two more HIGH bugs, one of them a regression I introducedEach sweep so far has found real bugs in the previous sweep's code, so I ran one more against the last commit rather than assume convergence. It found two HIGH issues. Both were mine. 1. My cap guard killed the only reaper for dead rowsThe previous commit stopped the takeover reconciling rows past That row is definitively dead, and it would now sit at
I had traded a rare harm (a live >1h stream terminal-written) for a common one (any crash, plus an hour of the user not looking). That's a bad trade, and it's worse than the bug it fixed. The fix, and the thing I'd missed: the row already tells you which it is. A stream still alive at the cap beat right up to it (
Both the takeover and the Stop path now share the predicate. The mutation that reverts it turns the pre-existing reaper tests red — exactly the regression signature. 2. The mark's first-match precedence swallowed a Stop over a live generationA stale That is the cardinal sin — a silent Stop over a live generation — in its third distinct shape. The mark is now a union over every name the caller gave, which also makes it agree with the local step (which already aborts every one of the caller's streams on the conversation). Two steps disagreeing about what a "name" means is precisely why this bug kept coming back. 3.
|
A fifth adversarial pass, on the code the fourth produced. Two findings, both mine. 1. THE IDENTITY GUARD WENT ON ONE OF THE TWO DOORS. The previous commit stopped a finishing stream from wiping a running one's name in `forgetStream` — and left the identical clobber unguarded 160 lines above, in `abortActiveStream`. A cross-instance Stop can take seconds (it waits for the owning instance to confirm), and the user can send turn 2 inside that window: its headers land and claim the slot, then the turn-1 abort resolves and deletes the name of a generation that is now RUNNING. Both doors into that slot now check that it is still theirs. 2. THE GHOST SURVIVED IN THE LAST TWO MINUTES BEFORE THE CAP. `heartbeatStoppedAtCap` cannot tell a healthy long generation from a process that crashed just before the cap — both stopped beating at roughly the same point. Refusing to judge protects the live one, but it left the crashed one at 'streaming' with NOTHING in the system able to reap it. That is not cosmetic: the abort path's batch verdict is pessimistic, so ONE such row makes every future Stop on its conversation report "may still be running" — about generations that stopped perfectly — and makes every future send pay the takeover's wait and log a warn. Forever. And the union mark (previous commit) sweeps such siblings in on every Stop, so it made the surviving ghost strictly more contagious. So there is now an outer horizon, well beyond the point where the rest of the system has already given up on the stream: the abort AND multicast registries both evict at STREAM_MAX_LIFETIME_MS, so a generation still running out there can no longer be stopped, joined, or watched by anyone. Its row claiming 'streaming' in perpetuity is a worse lie than declaring it over — and it is the only thing that converges. Both terminal writers now share ONE predicate, `isProvablyDead`, so they cannot drift apart on the one write that is unforgiving in both directions. It is pinned by mutation tests from BOTH sides: remove the backstop and the ghost test goes red; make it reap everything past the cap and the live-long-generation tests go red. Also: the union's catch now returns the marks that DID land rather than discarding them — those rows are marked in the DB and their owners will stop them, so reporting an empty set was a second lie on top of the failure. typecheck exit 0 · lint clean · 12,810 passing (16 pre-existing failures, identical on master). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
…ndow is still open A sixth adversarial pass found no new correctness bug (the first pass in six to come back clean), but it did catch this docblock making a promise the code does not keep — which is the specific failure mode this codebase keeps warning about. `abort-conversation-streams.ts` claimed "the conversationId is the one name the client holds from t=0. So Stop can now always say something true." The first half is true; the conclusion is not. conversationId is a name the CLIENT holds from t=0, but the thing it resolves against — the ai_stream_sessions row — is not written until createStreamLifecycle, at the very END of the route's preflight (after auth, permissions, message persistence, context assembly). For most of that 0.5-3s window there is no row and no registry entry: the SELECT matches nothing, the cross-instance mark matches nothing, and the caller is told `not_found` — which the UI stays SILENT about by design. The generation then starts a moment later and runs to completion, write tools and billing included. So what this actually closes is the tail between the row's INSERT and the headers landing. The head of the window — before the row exists — is still a silent Stop over a generation that goes on to run. Pre-existing, not introduced here, and NOT fixed here: closing it needs the abort request to outlive the absence of a row (a durable pending-abort intent, keyed by conversation, that createStreamLifecycle consults at INSERT time instead of unconditionally clearing abortRequestedAt). Own storage, own expiry, own migration — its own change. Written down in the docblock and flagged in the PR rather than papered over. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
Sixth pass — the streak ends: no new correctness bugFive consecutive review rounds each found a bug introduced by the previous round's fix, so I kept sweeping rather than assume convergence. The sixth pass, run against What convinced me it isn't just reviewer fatigue — it verified the load-bearing math exhaustively against real numbers (
The ambiguity margin is exactly 6 beat intervals wide, so a live long generation must miss 6+ consecutive beats to be misjudged — and 6 consecutive misses is the system's own definition of dead. The one thing it did find — and it is not from this PRA Stop pressed during the pre-INSERT preflight window is a silent no-op. What makes this worth surfacing rather than shrugging at: It is pre-existing — it predates this PR and every round of it — and closing it needs its own storage, expiry, and migration. Deliberately not absorbed here.
|
…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).
#2028) (#2062) * fix(ai): close pre-INSERT abort window + tighten abort mark predicates (#2028) Item 1 — Pre-INSERT pending-abort intent: - New `ai_pending_abort_intents` table + migration (0206) - `pending-abort-intents.ts`: recordPendingAbort, consumePendingAbort, clearPendingAbort - `abort-stream-anywhere.ts`: records a pending-abort intent when Stop finds nothing in flight - `createStreamLifecycle`: consumes pending-abort at INSERT time; if present, writes the row as 'aborted' directly and returns a pre-finished handle (preAborted: true) - Both chat routes (chat/route.ts, global/[id]/messages/route.ts): abort the controller when preAborted is true, so streamText never starts Item 4a — False `unconfirmed` when locally aborted: - `abort-stream-anywhere.ts`: if markAbortRequested fails but locallyAborted is non-empty, return ABORTED (streams are provably dead in-process) instead of UNCONFIRMED Item 4b — clearAbortMarks missing status predicate: - `stream-abort-mark.ts`: adds `status='streaming'` filter to clearAbortMarks UPDATE so a mark written against a non-streaming row is not erased Item 4c — Takeover reconcile missing abortRequestedAt null: - `stream-takeover.ts`: adds `abortRequestedAt: null` to the inline reconcile UPDATE, matching reconcileDeadStreamRows' shape Tests: 13 new tests across pending-abort-intents, abort-stream-anywhere-pending, and stream-lifecycle (pre-aborted path). All existing tests pass (113 total). Closes #2028 items 1, 4a, 4b, 4c. Items 2 and 3 are separate changes with their own migration risk. Refs: #2022, #2028 * fix(ai): close pre-INSERT abort window + tighten abort edge cases (#2028) Closes the remaining Stop gaps from #2028: 1. Pre-INSERT preflight window (item 1, critical) - New `ai_pending_abort_intents` table with migration 0206 - `abortStreamAnywhere` records a durable pending-abort intent when it finds nothing in flight (Stop pressed during 0.5-3s preflight) - `createStreamLifecycle` consumes the intent at INSERT time; if present, writes the row as 'aborted' directly and returns preAborted=true so the route aborts the controller before streamText - Both chat routes handle preAborted by aborting + removing stream 4a. False unconfirmed when locally aborted (item 4a) - `abortStreamAnywhere` now returns ABORTED when mark fails but locally-aborted streams are stopped, instead of UNCONFIRMED 4b. clearAbortMarks missing status predicate (item 4b) - Added `eq(status, 'streaming')` to prevent clearing marks on non-streaming rows 4c. Takeover reconcile missing abortRequestedAt null (item 4c) - Inline reconcile in stream-takeover.ts now sets abortRequestedAt: null to match reconcileDeadStreamRows behavior * fix: remove unused vars in test file (lint) * fix(ai): close consume/insert race, fix drizzle snapshot, sweep pending-abort intents Addresses PR #2062 review feedback (Codex + CodeRabbit): - stream-lifecycle.ts: recheck consumePendingAbort() immediately after the aiStreamSessions INSERT, not just before it. A Stop landing in the gap between the pre-INSERT check and the INSERT resolving previously missed the current generation AND left an orphaned intent to wrongly pre-abort the next send within the 30s TTL. - packages/db/drizzle/meta/0206_snapshot.json: the committed snapshot was misnamed (0206_pending_abort_intents_snapshot.json) and internally wrong — its prevId didn't chain from 0205 and it didn't even declare the new table. Regenerated via drizzle-kit against a scratch out-folder truncated at 0205, verified the resulting SQL matches the already-committed 0206_pending_abort_intents.sql, and replaced the file under the convention every other snapshot in this repo follows. - chat/route.ts, global/[id]/messages/route.ts: short-circuit inside execute() when lifecycle.preAborted, skipping command-plan writes, model-capability resolution, and the agent loop instead of relying on the already-aborted signal to short-circuit streamText's fetch. - pending-abort-intents.ts: add sweepExpiredPendingAbortIntents(), wired into the existing /api/cron/sweep-expired route alongside the other append-with-TTL table sweeps. Closes the gap where an intent from a conversation that's never resumed would sit past its TTL indefinitely. - abort-stream-anywhere-pending.test.ts: assert result.code for the "marked but finished here" branch, not just the recordPendingAbort side effect. 65 stream-lifecycle/pending-abort tests + 18 sweep-route tests pass; full ai/core + chat + global test directories (1062 tests) green; typecheck and lint clean. * refactor(ai): merge pending-abort recheck into a single post-INSERT check Self-review (8 independent finder agents) surfaced that the two sequential consumePendingAbort() calls added in 4407c4d (one before the aiStreamSessions INSERT, one after) were: - Duplicative: both branches wrote near-identical 'aborted' state and the same noop-handle literal, just via INSERT vs UPDATE. - Wasteful: every non-preaborted stream start (the overwhelming majority) paid for 2 DB round-trips to check an intent that almost never exists, on a path this whole feature calls TTFB-sensitive. Collapsed to ONE check, placed after the INSERT: since nothing else consumes a pending-abort intent in between, checking once there covers both the original pre-INSERT-window case and the narrower insert-race case the prior commit added, at 1 fewer DB round-trip per stream start and ~95 fewer lines. Also tightened the check's comment to disclose the still-open (but narrow, TTL-bounded, self-healing) commit-ordering race between this consume and a concurrent recordPendingAbort — matching this codebase's own KNOWN RACE disclosure convention rather than implying the window is fully closed. And de-duplicated the identical 5-line comment copied into both routes' execute() guards down to a 3-line pointer at StreamLifecycleHandle.preAborted's existing JSDoc. 48 stream-lifecycle tests (down from 52, net fewer due to the merge) + full ai/core + chat + global + cron test dirs (1058 tests) pass; typecheck and lint clean. --------- Co-authored-by: Sprite <noreply@sprites.dev>
Two more real findings from a Codex re-review (requested after the previous convergence pass), both in materialize-interrupted-stream.ts: 1. The setWhere guard only excluded 'complete', not 'interrupted'. A row whose owning generation cleanly finished (onFinish already persisted the FULL correct content as status='interrupted') but whose ai_stream_sessions terminal write then failed separately (fire-and- forget) stays eligible for a later dead-row sweep. `!= 'complete'` would let that sweep overwrite the already-correct interrupted message with an older debounced checkpoint. Tightened to `== 'streaming'` — a true compare-and-swap: a row leaves 'streaming' exactly once, so this can never re-touch an already-terminal row in either direction. 2. The defensive insert-if-missing branch (only reached if PR 2's placeholder insert never happened) stamped `createdAt: now` (reap time) instead of the stream's actual start. A recovered reply could then sort after a user's later follow-up message. Added `startedAt` to `MaterializableStreamRow` and threaded it through all three call sites (stream-takeover.ts, stream-abort-mark.ts's reconcileDeadStreamRows, the active-streams lazy sweep — all three already had it in hand from their existing SELECTs, bar one new column added to reconcileDeadStreamRows' query). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB
* feat(ai): materialize interrupted streams on dead-row reap Turns a provably-dead ai_stream_sessions row into an honest, terminal `status: 'interrupted'` assistant message instead of a silent loss. - materializeInterruptedStream(row): builds the message payload from the row's parts snapshot (reusing buildAssistantPersistencePayload, the same primitive execute-end/onFinish already use), upserts chat_messages or messages (routed by parseGlobalChannelId) guarded by a setWhere ne(status, 'complete') on the conflict clause (the #2022 invariant, enforced atomically), then settles the session row terminal and broadcasts stream_complete({aborted: true}). Never throws; a failed step leaves the row eligible for the next pass to retry. - Wired into all three reap paths: stream-takeover.ts's reconcile set (now per-row instead of a bulk UPDATE), stream-abort-mark.ts's reconcileDeadStreamRows (re-selects full rows fresh), and a new lazy sweep in GET /api/ai/chat/active-streams (fire-and-forget, since that route already reads every streaming row on the channel). - Client: ConversationMessage.status threads to MessageRenderer, which renders an "Interrupted" badge + retry hint (including for a message with zero content — an honest empty marker, not silence). - 100% branch coverage on materialize-interrupted-stream.ts, gated via a new per-glob vitest threshold per the epic's own constraints. Filed a D-task at the epic level (gg2b2b8getkro3n1erfksj2v) for a narrower, non-blocking gap: a tab left open through the interruption doesn't badge it until reload (onStreamComplete doesn't thread `aborted` through synthesizeAssistantMessage) — the standard fetch/reload path already shows the badge correctly via the server's existing status field. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB * fix(ai): materialize-eligibility race, content fidelity, and interrupted-badge coupling Multi-angle review of PR #2077 surfaced several real issues, fixed here: - CRITICAL: stream-takeover.ts's local reconcile set conflated "provably dead" with "we just aborted it ourselves" (still alive moments ago, about to run its own onFinish). Materializing the latter raced that natural write with an older debounced checkpoint and could silently overwrite fresher content with staler content. Now only provably-dead rows are materialized; just-aborted-but-alive rows get the original cheap session-row wipe and their own generation's onFinish persists the real content, matching leaf 3.1's "only isProvablyDead rows" spec. - materialize-interrupted-stream.ts: moved payload-building inside the try/catch (the module's own "never throws" contract wasn't actually enforced); switched to the same extractStructuredContentFromParts pipeline saveMessageToDatabase/saveGlobalAssistantMessageToDatabase use, so file/data parts and ordering survive materialization instead of degrading to flat text; re-synced conversationId on conflict-update, matching the normal terminal-write path; logs broadcast failures instead of silently swallowing them. - active-streams route: the lazy sweep now runs over the subscription-authorized row set, not the raw per-channel query — a user with mere page-view access could otherwise trigger a reap side effect for another member's private conversation. - stream-abort-mark.ts / stream-takeover.ts: per-row reconciliation now runs concurrently (Promise.all) instead of sequentially — each row is independent and materializeInterruptedStream never throws, so there is no correctness reason to serialize a mass-crash-recovery batch. - MessageRenderer.tsx / CompactMessageRenderer.tsx: the interrupted badge was nested inside the text-block render, so a message interrupted mid-tool-call (no trailing text) never showed it. Moved to the message level, independent of which part groups exist. CompactMessageRenderer — the actual sidebar/global-assistant surface — never got the badge at all; added parity coverage. - ai-undo-service.test.ts: added the 3 tests D.2 (epic board, tza273pegokkzpd7ut5qvxky) flagged as claimed-tested-but-uncovered — previewAiUndo's count query and both of executeAiUndo's soft-delete updates now assert the ne(status,'streaming') guard is actually present. Filed two epic-level D-tasks for real-but-unscheduled gaps found along the way (not blocking): mention-notifications not firing for a materialized interrupted reply, and a live-tab socket-immediacy gap for the badge. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB * fix(ai): restore retry button for empty/tool-only interrupted messages A follow-up review of the previous fix commit (which moved the Interrupted badge out of TextBlock to a message-level render, so it shows regardless of which part-group is last) found it dropped the actual retry BUTTON for exactly the messages that motivated that move: an interrupted message with empty or tool-only content never renders a TextBlock, and the retry button previously lived only in TextBlock's own footer. The message-level block had a text hint ("retry to continue") but no actionable control. Added a real retry button to the message-level interrupted block in both MessageRenderer.tsx and CompactMessageRenderer.tsx, gated on `!hasNonEmptyTextBlock` so a message that DID stream real text before dying — which already gets a retry button from TextBlock's footer — doesn't show two. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB * fix(test): CI failure — the "exactly ONE retry button" tests never supplied onEdit/onDelete CI (ci / Unit Tests) failed both interrupted-affordance test files on the same test: "given an interrupted message WITH real text content and a retry handler, shows exactly ONE retry button" expected 1 button, found 0. Root cause: TextBlock's/CompactTextBlock's footer (MessageActionButtons, which is where its own retry button lives) only renders when BOTH onEdit AND onDelete are supplied — not onRetry alone. The test passed only onRetry, so TextBlock's footer never rendered at all, and the message-level button correctly stayed hidden (hasNonEmptyTextBlock=true), leaving zero buttons — a bug in the test's own setup, not in the retry-button fix from commit 9470bd2. Added onEdit/onDelete mocks to both tests so TextBlock's footer renders as it would on the real chat surfaces (which always wire edit/delete/retry together), correctly exercising the "don't duplicate the button" guard. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB * fix(ai): tighten #2022 guard to compare-and-swap, fix insert timestamp Two more real findings from a Codex re-review (requested after the previous convergence pass), both in materialize-interrupted-stream.ts: 1. The setWhere guard only excluded 'complete', not 'interrupted'. A row whose owning generation cleanly finished (onFinish already persisted the FULL correct content as status='interrupted') but whose ai_stream_sessions terminal write then failed separately (fire-and- forget) stays eligible for a later dead-row sweep. `!= 'complete'` would let that sweep overwrite the already-correct interrupted message with an older debounced checkpoint. Tightened to `== 'streaming'` — a true compare-and-swap: a row leaves 'streaming' exactly once, so this can never re-touch an already-terminal row in either direction. 2. The defensive insert-if-missing branch (only reached if PR 2's placeholder insert never happened) stamped `createdAt: now` (reap time) instead of the stream's actual start. A recovered reply could then sort after a user's later follow-up message. Added `startedAt` to `MaterializableStreamRow` and threaded it through all three call sites (stream-takeover.ts, stream-abort-mark.ts's reconcileDeadStreamRows, the active-streams lazy sweep — all three already had it in hand from their existing SELECTs, bar one new column added to reconcileDeadStreamRows' query). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB * fix(a11y): explicit aria-label on the message-level retry button CodeRabbit nitpick: the icon-only retry button (added for empty/tool-only interrupted messages) relied on `title` alone, which isn't a reliable accessible name across screen readers. Added an explicit aria-label matching the existing title text. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB * fix(ai): report actual reconciliation success, not attempted eligibility Codex review finding: takeOverConversationStreams's Promise.all over materializeInterruptedStream had no success signal, so a per-row DB failure was invisible to the caller — every id in `reconcile` was counted as `reconciled` regardless of whether the row actually settled. The old bulk-UPDATE design returned `reconciled: []` on any failure; this PR's per-row restructure silently dropped that accuracy. The same gap existed in the cross-instance path (remotelyReconciled was read straight from decideAbortOutcome's eligibility judgment, never from what reconcileDeadStreamRows actually did) and in reconcileDeadStreamRows itself. - materializeInterruptedStream now returns Promise<boolean> — true only if both the message write AND the session-row settle succeeded. - reconcileDeadStreamRows now returns Promise<string[]> — the messageIds it actually materialized, a subset of what it was asked to reconcile. - stream-takeover.ts's local reconcile path now filters provablyDead by actual materialize success, and the justAbortedStillAlive bulk wipe only contributes to `locallyReconciled` when it doesn't throw. - The cross-instance path now sets remotelyReconciled from reconcileDeadStreamRows's return value instead of the eligibility set. Updated all affected tests (several `.resolves.toBeUndefined()` assertions were factually wrong the moment the return type changed to boolean/array) and added regression tests for the accurate-reporting behavior itself (partial-failure batches, read failures). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB * fix(ai): check rowCount, not just no-throw, for session-row settle success CodeRabbit finding: a conditional UPDATE (WHERE messageId = X AND status = 'streaming') does not throw when it matches zero rows — it succeeds and changes nothing. That happens when a concurrent reap (another instance's takeover, or a second sweep racing this one) already settled the exact same row first. The previous "settled = true unless it throws" logic would then credit this call with settling a row it didn't actually touch. Now checks `(result.rowCount ?? 0) > 0`, matching the codebase's own established pattern for the same class of conditional update (compaction-repository.ts's version-guarded CAS). Added regression tests for both the zero-rowCount case and the defensive `?? 0` fallback (a driver/mock that omits rowCount entirely). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013c3xvumpqg33Ss6AKAP7XB --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 participant
The bug
AI streams are server-owned and disconnect-immune by design:
request.signalis never wired intostreamText. That is the architecture — it is what lets a generation survive an iPad reload. The consequence is that cancelling the client fetch stops nothing. The only thing that stops a generation is an explicit server-side abort that names it.But
stream-abort-registry.tsis an in-memory Map — single-process, and prod runs multiple web instances behind a load balancer:{aborted:false}. The generation on A runs to completion. Even a perfectly-named abort found nothing.takeOverConversationStreamshad the same hole: a live row owned by another instance could not be aborted, so a second send started a second generation beside the first — two agents, two sets of write tools, two bills.A Stop that misses is not a UI glitch. The agent keeps generating, keeps calling write tools (editing the user's pages), and keeps billing, while the button flips back to Send and the user believes it stopped.
Definition of done: the
'in-flight stream(s) could not be aborted from this instance'warn path is deleted. It is.Not what #2011 fixed (that was which stream the client names). Complementary.
Why a Postgres mark and not a relay broadcast
The brief proposed broadcasting the abort over the signed Socket.IO relay. I evaluated it and rejected it:
apps/realtimeis an emit-only fan-out to browsers (web→relay is an HTTP POST + HMAC). No web-server-side socket client exists anywhere in the monorepo; the relay's connect middleware accepts only user session tokens and its rooms are user/page/drive-scoped behind an allowlist. Receiving a broadcast would need a socket client in the Next server, a connection/reconnect lifecycle, a new service-identity auth path, and a new room shape on the relay — new attack surface on the component the brief wants kept dumb.chat.stop()is "a disconnect signal, not a request to stop generation"; itscancelActiveWorkis an explicit hole — "you might cancel the job or write a cancellation flag that the job checks." Vercel's own cross-instance substrate for server-owned streams (resumable-stream) uses Redis pub/sub, has no abort primitive at all, and vercel/ai#6502 is still open. There is no blessed broadcast answer to copy.LISTEN/NOTIFYis at-most-once with no replay, anddb.ts:21swallows the exact disconnect that kills a listener — a silently-dead listener means Stop silently stops working.distributed-rate-limit.ts— "Works across multiple server instances (Postgres is the source of truth)." The heartbeat is already a Postgres cross-instance channel.Local aborts stay instant: a precise name (
messageId/streamId) that this instance's registry aborts returns immediately — no mark, no poll. Only a stream this instance could not stop is escalated and waited on. (The conversation path names a set, so it stops what it can and drops those ids from the wait.) The watcher is lazily started, self-terminating, and issues zero queries on an idle instance.Security — the highest-risk part of this change
A broadcast abort that any instance honoured without re-checking ownership would be a remote kill switch for other users' streams. There is no such thing here:
UPDATEwhoseWHEREcarries the caller'suser_id. A user naming another user's stream updates zero rows. There is no message to forge and no payload to trust — the forged-abort case is not blocked, it is unrepresentable.user_idfrom its own row and refuses to abort on any mismatch (and on astream_idepoch mismatch, so a stale Stop cannot kill the next generation).abortStream's existing IDOR guard still runs last.not_foundand "not yours" are deliberately indistinguishable — telling them apart would confirm the existence of another user's stream.The cross-user test does not stub its answer: a fake evaluates the
WHEREclause the code actually built, against a real row, the way Postgres would. Delete theuser_idpredicate and it goes red.A latent bug this depends on
Aborting the controller did not reliably drive the row terminal. Both routes say so in their own comments:
onAbort"only fires while a streamText is live", andonFinish"may never fire when the mobile client backgrounds mid-stream" — which is exactly the population of a cross-instance abort. A naive confirmation poll would therefore time out on already-dead streams and warn "still running, still billing" — a false alarm on the one message that must never be false.Fixed by
attachStreamFinisher: the terminal write now rides the abort itself.lifecycle.finishis idempotent, so the happy path is unchanged.And when the wait does time out, a stale heartbeat is read as "the owner is gone, nothing is running" (reconcile the row, no alarm) rather than as "still running" — composing with
stream-liveness.tsinstead of inventing a second notion of liveness.Client —
aborted:falsefinally means somethingEvery caller used to
voidthe result, so the button flipped back to Send regardless. Now the response carries acode:unconfirmed→ found, asked to stop, still generating. Still billing. →toast.warning.not_found→ the stream ended a beat before Stop. A benign race → silent (toasting here would fire constantly and train users to ignore the warning that matters).useChatStop/GlobalAssistantViewnow call the local stop first, then await the server purely for the toast decision — otherwise the confirmation wait would visibly hang the Stop button.Tests
stream-abort-decisions.ts) with ZERO mocks — real inputs, so the tests cannot pass with the bug present.user_idpredicate, the epoch guard, the owner-mismatch refusal, the finisher, theabortRequestedAt: nullreset, or the watcher's self-termination → each goes red; restore → green.onConflictDoUpdatestill resetsabortRequestedAt(a silent, catastrophic regression otherwise — a stale Stop would kill the next generation).disconnect-immunity.test.tsuntouched and green.request.signalis still absent from both generation routes.activity-tools,admin-role-version,grouping) fail identically on master — pre-existing, unrelated, verified by stashing.bun run typecheckexit 0 ·bun run lintclean · migration generated viabun run db:generate(additive + nullable; no hand-written SQL).Explicitly out of scope — tracked in #2028
Everything below is deliberately not in this PR, and is filed as #2028 so it isn't lost:
A Stop during the pre-INSERT preflight window is still a silent no-op (the last surviving cardinal-sin path — the row isn't written until the end of the route). Needs a durable pending-abort intent keyed by conversation. Pre-existing; predates this PR.
streamMulticastRegistrycross-instance join (stream-join404s cross-instance) — same root cause, different fix: it must move data, not control.The takeover TOCTOU race — needs DB-level serialization and its own migration risk.
Three smaller known nits (partial-truth
unconfirmed,clearAbortMarkspredicate, takeover reconcile parity).The pre-INSERT preflight window. A Stop pressed before
createStreamLifecyclewrites theai_stream_sessionsrow — i.e. during most of the 0.5-3s TTFB — still resolves to nothing and is reportednot_found, which the UI stays silent about, while the generation starts a moment later and runs.abort-conversation-streams.tsused to claim it had closed this ("Stop can now always say something true"); it had not, and that docblock is now corrected. Closing it for real needs the abort request to outlive the absence of a row — a durable pending-abort intent keyed by conversation, whichcreateStreamLifecycleconsults at INSERT time instead of unconditionally clearingabortRequestedAt. Own storage, own expiry, own migration. Pre-existing (it predates this PR), and deliberately not absorbed into it.streamMulticastRegistrycross-instance join (stream-join/[messageId]404s cross-instance). Same root cause, different fix — it must move data, not control. Separate PR.The takeover TOCTOU race documented in
stream-takeover.ts. Needs DB-level serialization and its own migration risk. Untouched.Rollout
Old workers do not run the watcher, so during a rolling deploy a cross-instance Stop against an old-worker stream returns
unconfirmed— correctly, because it is still running. Window is one drain; same bet thelastHeartbeatAtrollout note already takes.Review log
Three reviewers (Codex, CodeRabbit, and an adversarial self-review sweep) found seven real bugs, all of one class: a Stop that lies — either a false "still billing" alarm, or a silent no-op. Every fix is mutation-checked (reintroduce the bug, watch the test go red, restore).
onFinishunregisters the controller at its top but writes the terminal status at its bottom (after message persistence + per-tool billing). A Stop pressed as the last tokens render looked exactly like a stream owned elsewhere → escalated → false "still billing" alarm about a finished generation.removeStreamleaves a tombstone (wasRecentlyFinishedHere); that Stop is answered with the truth — nothing is running — silently. The takeover consults it too.not_found, which the client stays silent about. DB blips → the Stop reaches nobody → the agent bills on → the user is told nothing.markAbortRequestedreportsfailed; surfaces asunconfirmed. Same for the takeover's mark.partsand hiding a still-generating stream.unconfirmed, never touch the row.stream-horizons.ts.streamIdwas optional; an omission would silently leave the column NULL and degrade cross-instance Stop with no signal. Verifying it exposed that thestreamId → conversationfallback was half-wired (the client never sent the conversation, so it could never fire).streamId: stringrequired; client sends both names.Tests that could not fail. The sweep also found predicates mocked at every call site and asserted nowhere. Worst: deleting
isNotNull(abort_requested_at)fromreadMarkedStreamswould make the watcher abort every generation on the instance within a second of it starting — with the whole suite green. Now covered, along withmarkAbortRequested'sSET(only itsWHEREwas asserted),reconcileDeadStreamRows' destructive status guard,clearAbortMarks, andmarkAbortRequestedAsOwner. The source-level reset tripwire was replaced by behavioural tests, because it inspected only theonConflictbranch — droppingstreamIdfrom theINSERTwould have NULLed every fresh row and it could not see it.Merged with #2011
#2011 (Stop-button / stream-ownership machine) landed on master after this branch was cut and rewrote the same three Stop paths. Its versions are strictly better than the base this branch forked from — held stream ids (
heldStreamMsgIdRef/globalStreamMsgIdRef) so a mid-stream conversation switch still names the running generation, ownership-guarded stop slots so a surface cannot destroy a Stop button it does not own, and messageId-first aborts.Master's versions were therefore taken wholesale, and this branch's two deltas re-applied on top:
finally(the server abort is no longer instant, so awaiting it first would hang the button); anduseChatStopkeeps #2011'sconversationIdparameter and its no-chatId fallback.🤖 Generated with Claude Code
https://claude.ai/code/session_01QtLjkwak9tnY1gGE9fCbDU
Summary by CodeRabbit
New Features
Bug Fixes