From 1e8039e0294592995e525ede7206d0ef596643f4 Mon Sep 17 00:00:00 2001 From: agent_coder Date: Thu, 2 Jul 2026 23:38:15 +0300 Subject: [PATCH] fix(#234 review r2): close the turn-2+ Stop-latch flash + first-turn deferred stop (F4/F5/F6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F4: the round-1 !localStreaming gate was insufficient — the PREVIOUS turn's terminal run stays cached under AI_CHAT_RUN_RQ_KEY(chatId) and cleared the latch early on turn 2+. handleServerStop now removeQueries that key so is null until the current turn's run is fetched fresh; the terminal effect's holds the latch until the current run is observed terminal. Safety net: if that refetch ERRORS while no longer streaming, release the latch so the view can't freeze on a transient failure. F5: first-turn Stop (before the start chunk adopts the chat id) latches a pending stop (stopPendingRef) fired by the onServerChatId adoption effect, so a detached run is authoritatively stopped instead of left running by a silent local-only abort. Known abort-ordering sub-window documented. F6: extract the latch-clear decision to a pure, unit-tested shouldClearStoppingLatch (run-polling.ts) — clears only when stopping, not the streamer, and the current run is terminal; tests are non-vacuous against the round-3/4 buggy behavior. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../ai-chat/components/ai-chat-window.tsx | 52 +++++++--- .../ai-chat/components/chat-thread.tsx | 44 ++++++++- .../ai-chat/utils/run-polling.test.ts | 99 +++++++++++++++++++ .../src/features/ai-chat/utils/run-polling.ts | 30 ++++++ 4 files changed, 210 insertions(+), 15 deletions(-) diff --git a/apps/client/src/features/ai-chat/components/ai-chat-window.tsx b/apps/client/src/features/ai-chat/components/ai-chat-window.tsx index 28b42d76..92c125ce 100644 --- a/apps/client/src/features/ai-chat/components/ai-chat-window.tsx +++ b/apps/client/src/features/ai-chat/components/ai-chat-window.tsx @@ -41,12 +41,16 @@ import { extractPageSlugId } from "@/lib"; import { AI_CHATS_RQ_KEY, AI_CHAT_MESSAGES_RQ_KEY, + AI_CHAT_RUN_RQ_KEY, useAiChatMessagesQuery, useAiChatRunQuery, useAiChatsQuery, useAiRolesQuery, } from "@/features/ai-chat/queries/ai-chat-query.ts"; -import { shouldObserveRun } from "@/features/ai-chat/utils/run-polling.ts"; +import { + shouldClearStoppingLatch, + shouldObserveRun, +} from "@/features/ai-chat/utils/run-polling.ts"; import { workspaceAtom } from "@/features/user/atoms/current-user-atom"; import ConversationList from "@/features/ai-chat/components/conversation-list.tsx"; import ChatThread from "@/features/ai-chat/components/chat-thread.tsx"; @@ -281,6 +285,17 @@ export default function AiChatWindow() { const handleServerStop = useCallback( (chatId: string): void => { setStoppingRun(true); + // #234 F4: drop the PREVIOUS turn's run from the cache so `run` becomes null + // until the CURRENT turn's run is fetched fresh. Without this, once the local + // stream aborts (localStreaming -> false) the run query re-enables and + // react-query SYNCHRONOUSLY returns the still-cached prior terminal run; the + // terminal effect would then clear the stopping latch against that STALE run + // before the current turn's (still-running, detached, growing) run is ever + // observed — re-opening the observer merge and flashing the growing output + // over the frozen row. With the cache cleared the terminal effect's + // `if (!run) return` holds the latch until the current run itself is observed + // terminal (see shouldClearStoppingLatch). + queryClient.removeQueries({ queryKey: AI_CHAT_RUN_RQ_KEY(chatId) }); void stopRun(chatId).catch(() => { setStoppingRun(false); notifications.show({ @@ -289,18 +304,30 @@ export default function AiChatWindow() { }); }); }, - [t], + [t, queryClient], ); // Poll the latest run of the open chat ONLY when we are a passive observer: // feature on, a chat is open, and we are NOT the local streamer (the streamer // already has the live SSE — polling/merging too would double-render). The // query's own status-keyed refetchInterval stops once the run is terminal. - const { data: runData } = useAiChatRunQuery( + const { data: runData, isError: runQueryFailed } = useAiChatRunQuery( activeChatId ?? undefined, autonomousRunsEnabled && !localStreaming, ); const run = runData?.run ?? null; + + // Safety net (#234 F4 review): after handleServerStop clears the run cache, + // `run` is null until the current turn's run is fetched fresh, and the terminal + // effect below holds the latch via `if (!run) return`. If that refetch instead + // ERRORS (a transient GET-run failure) while we are no longer the streamer, the + // run stays null, its status-keyed refetchInterval is off, and nothing would + // ever observe a terminal run — freezing the view with the observer merge + // suppressed. Release the latch on that error so the live view resumes rather + // than stays stuck (the local stopRun may already have succeeded independently). + useEffect(() => { + if (stoppingRun && !localStreaming && runQueryFailed) setStoppingRun(false); + }, [stoppingRun, localStreaming, runQueryFailed]); // The run's incrementally-persisted assistant message to merge into the thread, // but only while we are an observer (never when we are the streamer — guards // against a stale poll fighting the live stream). Includes a terminal run so the @@ -324,16 +351,15 @@ export default function AiChatWindow() { } // Terminal: a stop we requested has landed (or the run finished on its own), // so release the stopping latch — the observer merge can now show the final - // persisted (aborted/finished) output without any live re-stream. - // - // But ONLY release once this tab is no longer the streamer. While - // `localStreaming` is true the run query is disabled, so `run` may be the - // PREVIOUS turn's terminal run held in the react-query cache, not the run we - // just asked to stop. Releasing the latch against that stale run would re-open - // the re-stream flash for the current turn the instant we switch to observer - // role. Gating on `!localStreaming` means the latch only clears against a run - // we are actually observing (the current turn's). - if (stoppingRun && !localStreaming) setStoppingRun(false); + // persisted (aborted/finished) output without any live re-stream. The decision + // is the pure, unit-tested `shouldClearStoppingLatch` (run-polling.ts): release + // ONLY when we requested a stop, this tab is no longer the streamer, AND the + // CURRENT run is terminal. The #234 F4 cache removal in handleServerStop makes + // `run` null (this branch's `if (!run) return` above holds) until the current + // turn's run is fetched fresh, so the latch can never clear against a stale + // cached run. + if (shouldClearStoppingLatch({ stoppingRun, run, isLocalStreaming: localStreaming })) + setStoppingRun(false); if (finalizedRunIdRef.current === run.id) return; finalizedRunIdRef.current = run.id; queryClient.invalidateQueries({ diff --git a/apps/client/src/features/ai-chat/components/chat-thread.tsx b/apps/client/src/features/ai-chat/components/chat-thread.tsx index c0fbbad9..a6ec1908 100644 --- a/apps/client/src/features/ai-chat/components/chat-thread.tsx +++ b/apps/client/src/features/ai-chat/components/chat-thread.tsx @@ -244,6 +244,16 @@ export default function ChatThread({ const flushOnAbortRef = useRef(false); const interruptNextSendRef = useRef(false); + // #234 F5: the user pressed Stop while streaming a BRAND-NEW chat whose server + // chat id has not been adopted yet (the `start` chunk carrying it hadn't landed + // when Stop was pressed). A local SSE abort alone does NOT stop the DETACHED + // autonomous run — it keeps burning tokens and WRITING TO PAGES — so we cannot + // just no-op. We latch the stop as PENDING and fire the authoritative server + // stop the moment onServerChatId adopts the id (below). Read-and-cleared there; + // also defused on every new turn start so it can never fire against a later, + // unrelated turn's run. + const stopPendingRef = useRef(false); + // FIFO dequeue + send the next queued message (no-op when the queue is empty). // Returns whether a message was actually sent, so callers can tell an empty // dequeue (nothing to flush) from a real send. @@ -393,7 +403,14 @@ export default function ChatThread({ return; lastForwardedChatIdRef.current = serverChatId; onServerChatId(serverChatId); - }, [messages, onServerChatId]); + // #234 F5: if Stop was pressed before the id was known, the authoritative + // server stop was deferred to this adoption point — fire it now with the + // just-adopted id. One-shot (read-and-clear) so it can't fire twice. + if (stopPendingRef.current) { + stopPendingRef.current = false; + onServerStop?.(serverChatId); + } + }, [messages, onServerChatId, onServerStop]); // Live "turn was interrupted" marker for the CURRENT session. The red error // banner (driven by `error`) covers the error case; this covers an aborted @@ -469,8 +486,26 @@ export default function ChatThread({ // only the local abort happens (there is no server-side run handle to stop yet). const handleStop = useCallback(() => { stop(); - if (autonomousRunsEnabled && chatIdRef.current) { + if (!autonomousRunsEnabled) return; + if (chatIdRef.current) { onServerStop?.(chatIdRef.current); + } else { + // #234 F5: no chat id yet (brand-new chat in the first moment of its first + // turn, before the `start` chunk adopted the id). Latch the stop as pending; + // the onServerChatId adoption effect fires the deferred server stop as soon + // as the id appears, so the detached run is still authoritatively stopped + // instead of left running by a silent local-only abort. + // + // KNOWN LIMITATION (#234 F5 review): `stop()` above has already aborted the + // local SSE reader. In the rare sub-window where Stop is pressed while still + // `submitted` (request sent, not one chunk read yet), that abort can cancel + // the reader BEFORE the `start` chunk is applied to `messages`, so the + // adoption effect never runs and this pending stop never fires. The detached + // run then keeps going for that turn. This is not a regression (the pre-fix + // behavior sent no server stop at all); closing it fully would require + // deferring the local abort until adoption, which is riskier and out of scope + // for this fix. Documented so a future change can address the abort-ordering. + stopPendingRef.current = true; } }, [stop, autonomousRunsEnabled, onServerStop]); @@ -485,6 +520,11 @@ export default function ChatThread({ setStopNotice(null); flushOnAbortRef.current = false; interruptNextSendRef.current = false; + // #234 F5: a new turn is starting — drop any pending deferred-stop from a + // previous turn that never adopted an id, so it can never fire against this + // (or a later) unrelated turn's run. A deferred stop for the CURRENT turn is + // set AFTER this effect (on the Stop click), so this does not clobber it. + stopPendingRef.current = false; } }, [isStreaming]); diff --git a/apps/client/src/features/ai-chat/utils/run-polling.test.ts b/apps/client/src/features/ai-chat/utils/run-polling.test.ts index a1a9cb26..580f759f 100644 --- a/apps/client/src/features/ai-chat/utils/run-polling.test.ts +++ b/apps/client/src/features/ai-chat/utils/run-polling.test.ts @@ -6,6 +6,7 @@ import { isRunActive, runPollInterval, shouldObserveRun, + shouldClearStoppingLatch, mergeObservedMessage, } from "./run-polling.ts"; @@ -77,6 +78,104 @@ describe("shouldObserveRun (observer-vs-streamer decision)", () => { }); }); +describe("shouldClearStoppingLatch (#234 latch-release decision)", () => { + // The one case the latch SHOULD clear: we requested a stop, we are the passive + // observer (not streaming), and the CURRENT run is terminal. + it("clears only when stopping, observing, and the run is terminal", () => { + expect( + shouldClearStoppingLatch({ + stoppingRun: true, + run: makeRun("aborted"), + isLocalStreaming: false, + }), + ).toBe(true); + expect( + shouldClearStoppingLatch({ + stoppingRun: true, + run: makeRun("succeeded"), + isLocalStreaming: false, + }), + ).toBe(true); + expect( + shouldClearStoppingLatch({ + stoppingRun: true, + run: makeRun("failed"), + isLocalStreaming: false, + }), + ).toBe(true); + }); + + // Round-3 regression: clearing while THIS tab is still the local streamer would + // re-open the flash for the current turn the moment we switch to observer role. + // A predicate lacking the streaming gate would (wrongly) return true here. + it("does NOT clear while this tab is the local streamer", () => { + expect( + shouldClearStoppingLatch({ + stoppingRun: true, + run: makeRun("aborted"), + isLocalStreaming: true, + }), + ).toBe(false); + expect( + shouldClearStoppingLatch({ + stoppingRun: true, + run: makeRun("succeeded"), + isLocalStreaming: true, + }), + ).toBe(false); + }); + + // The detached run keeps growing after a local abort — while it is still + // active the latch MUST hold so the observer merge stays suppressed. + it("does NOT clear while the run is still active", () => { + expect( + shouldClearStoppingLatch({ + stoppingRun: true, + run: makeRun("running"), + isLocalStreaming: false, + }), + ).toBe(false); + expect( + shouldClearStoppingLatch({ + stoppingRun: true, + run: makeRun("pending"), + isLocalStreaming: false, + }), + ).toBe(false); + }); + + // #234 F4: on Stop the stale PREVIOUS-turn run is removed from the cache, so the + // observed `run` is null until the current turn's run is fetched fresh. A null + // run HOLDS the latch — it can never clear against the just-removed stale run, + // only against the current turn's own terminal run once observed. + it("does NOT clear against a removed/absent run (F4 stale-run guard)", () => { + expect( + shouldClearStoppingLatch({ + stoppingRun: true, + run: null, + isLocalStreaming: false, + }), + ).toBe(false); + expect( + shouldClearStoppingLatch({ + stoppingRun: true, + run: undefined, + isLocalStreaming: false, + }), + ).toBe(false); + }); + + it("does NOT clear when no stop was requested", () => { + expect( + shouldClearStoppingLatch({ + stoppingRun: false, + run: makeRun("aborted"), + isLocalStreaming: false, + }), + ).toBe(false); + }); +}); + describe("mergeObservedMessage", () => { it("replaces the message with the same id in place (per-step growth)", () => { const prev = [makeMsg("u1", "hi"), makeMsg("a1", "step 1")]; diff --git a/apps/client/src/features/ai-chat/utils/run-polling.ts b/apps/client/src/features/ai-chat/utils/run-polling.ts index c6e4c006..fa204f95 100644 --- a/apps/client/src/features/ai-chat/utils/run-polling.ts +++ b/apps/client/src/features/ai-chat/utils/run-polling.ts @@ -52,6 +52,36 @@ export function shouldObserveRun( return !!run && !localStreaming; } +/** + * Should the "stopping" latch — which suppresses the observer re-stream flash + * after the user pressed Stop — be RELEASED now? All three must hold: + * - `stoppingRun`: we actually requested a stop (otherwise nothing to release); + * - `!isLocalStreaming`: this tab is NOT the local streamer. While we are the + * streamer the run query is disabled, so the observed `run` is not the run we + * are following — releasing the latch then would re-open the flash for the + * current turn the instant we switch to observer role; + * - the observed `run` EXISTS and has reached a TERMINAL status. + * + * The null / still-active `run` case is the #234 F4 invariant. On Stop the stale + * PREVIOUS-turn run is removed from the query cache (`removeQueries`), so `run` + * is null until the CURRENT turn's run is re-fetched fresh; a null or active run + * therefore HOLDS the latch, so it can only ever clear against the current turn's + * OWN terminal run — never a stale cached one. (The cache removal itself is + * integration-level in AiChatWindow; this predicate encodes the decision given + * whatever run is currently observed, and a stale terminal run is + * indistinguishable from a current terminal run at the predicate level — hence + * the cache removal is what guarantees only the current run is ever passed here.) + */ +export function shouldClearStoppingLatch(args: { + stoppingRun: boolean; + run: IAiChatRun | null | undefined; + isLocalStreaming: boolean; +}): boolean { + const { stoppingRun, run, isLocalStreaming } = args; + if (!stoppingRun || isLocalStreaming) return false; + return !!run && !isRunActive(run); +} + /** * Merge an observed assistant message into the rendered list: replace the message * with the same id in place (the in-progress assistant row is already seeded from