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 5c67e177..5a918cb9 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,17 +41,10 @@ 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 { - shouldClearLatchOnQueryError, - 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"; @@ -85,6 +78,12 @@ const MIN_HEIGHT = 400; // Margin kept between the window and the viewport edges while dragging. const EDGE_MARGIN = 8; +// #184 phase 1.5: hard cap on the degraded-poll fallback. The poll is armed when +// a resume attempt could not attach to the live run and disarmed by the thread on +// settle / local stream; this cap is the ONLY backstop against an endless tick +// (a stuck 'streaming' row before the boot-sweep, or a user-tail 204 with no run). +const DEGRADED_POLL_MAX_MS = 10 * 60_000; + /** Compact token formatter: 1.2M / 3.4k / 950. */ function formatTokens(n: number): string { if (n >= 1_000_000) return `${(n / 1_000_000).toFixed(1)}M`; @@ -242,150 +241,62 @@ export default function AiChatWindow() { [roles], ); + // #184 phase 1.5: degraded-poll fallback (replaces the F4/F5/F7 latches). When + // ChatThread could not attach to a still-running run it arms this via + // onResumeFallback(true); the thread disarms it on settle / local stream. The + // window only OWNS the timer (armedAtRef stamps when it was armed for the cap). + const [degradedPoll, setDegradedPoll] = useState(false); + const armedAtRef = useRef(0); + const onResumeFallback = useCallback((active: boolean): void => { + if (active) armedAtRef.current = Date.now(); + setDegradedPoll(active); + }, []); + // Reset the degraded poll whenever the open chat changes: it is scoped to the + // resume attempt of the previously-open chat (invariant 8). + useEffect(() => { + setDegradedPoll(false); + }, [activeChatId]); + const { data: messageRows, isLoading: messagesLoading } = - useAiChatMessagesQuery(activeChatId ?? undefined); + useAiChatMessagesQuery( + activeChatId ?? undefined, + // DELIBERATELY DUMB (invariant 8 / task 2.4): poll every 2.5s while armed + // and under the 10-min cap; otherwise off. NO error checks (TanStack v5 + // resets fetchFailureCount each fetch, so consecutive errors are not + // expressible — and the poll must survive a server restart) and NO tail + // checks (the settled/local-stream semantics live in ChatThread, which + // disarms via onResumeFallback(false)). The time cap is the only backstop. + () => + degradedPoll === true && + Date.now() - armedAtRef.current < DEGRADED_POLL_MAX_MS + ? 2500 + : false, + ); // #184 reconnect-and-live-follow. Whether detached agent runs are enabled for - // this workspace. The reconnect endpoint itself is NOT flag-gated server-side - // (it is only owner-gated and returns `{ run: null }` when the chat has no - // run); but when the feature is off no runs are ever created, so polling it - // would always come back empty — we gate it off here to avoid pointless polls. + // this workspace. When the feature is off no runs are ever created, so the + // resume attempt would only ever 204; gating ChatThread's resume on it avoids a + // pointless attach round-trip. const workspace = useAtomValue(workspaceAtom); const autonomousRunsEnabled = workspace?.settings?.ai?.autonomousRuns === true; - // Whether THIS tab is the one actively streaming the open chat's run locally - // (it started the run here and holds the SSE). Reported up from ChatThread. We - // are the STREAMER while true and a passive OBSERVER while false — the basis of - // the observer-vs-streamer detection. Reset to false by the fresh ChatThread's - // mount effect on every chat switch. - const [localStreaming, setLocalStreaming] = useState(false); - const onStreamingChange = useCallback((streaming: boolean) => { - setLocalStreaming(streaming); - }, []); - - // #184 Stop wiring. While a detached run is being stopped we SUPPRESS the - // observer merge so the stopping run's still-persisting output does not - // re-stream back into view between the moment the user pressed Stop and the run - // actually settling as 'aborted' server-side. Polling itself keeps running (so - // the terminal transition is still detected) — only the visual merge is gated. - // Cleared when the run is observed terminal (below) or the chat is switched. - const [stoppingRun, setStoppingRun] = useState(false); - // Reset the stopping latch whenever the open chat changes: it is scoped to the - // run of the previously-open chat. - useEffect(() => { - setStoppingRun(false); - }, [activeChatId]); - // Authoritative stop of the open chat's detached run (the Stop button in - // autonomous mode). Latch "stopping" first (suppresses the re-stream flash), - // then request the server stop — the ONLY thing that ends a detached run; a mere - // local SSE abort is a client disconnect the server ignores. On failure we - // release the latch so the observer resumes (better to show the live run than to - // freeze the view) and surface the error. + // autonomous mode). Request the server stop — the ONLY thing that ends a + // detached run; a mere local SSE abort is a client disconnect the server + // ignores. On failure surface the error. 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({ message: t("Failed to stop the run"), color: "red", }); }); }, - [t, queryClient], + [t], ); - // 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, 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 PERMANENTLY (the GET-run keeps failing) 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). - // - // #234 F7: this must NOT fire on a TRANSIENT error while `run` is still an - // ACTIVE held run. In TanStack Query v5 (retry:false) the query's `data` is - // RETAINED on error, so `runQueryFailed` can be true while `run` is still - // pending/running — releasing then would re-open the observer merge and flash - // the growing detached run over the frozen row (the very flash F4 prevents). The - // decision is the pure, unit-tested `shouldClearLatchOnQueryError`, which gates - // on the run NOT being active: it cures only the genuine permanent-null-freeze - // (`run === null`) and never releases against an active run. - useEffect(() => { - if ( - shouldClearLatchOnQueryError({ - stoppingRun, - isLocalStreaming: localStreaming, - runQueryFailed, - run, - }) - ) - setStoppingRun(false); - }, [stoppingRun, localStreaming, runQueryFailed, run]); - // 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 - // final persisted output is shown on reopen. - const observedRow = - shouldObserveRun(run, localStreaming) && !stoppingRun - ? (runData?.message ?? null) - : null; - - // When the observed run reaches a terminal status, do a final messages refetch - // so the persisted final state (token/context badge, export source) is shown, - // then the query's refetchInterval has already stopped polling. Deduped per run - // id so it fires exactly once per run, not on every subsequent poll-less render. - const finalizedRunIdRef = useRef(null); - useEffect(() => { - if (!run || !activeChatId) return; - if (run.status === "pending" || run.status === "running") { - // Active again (a new run) — re-arm so its terminal transition fires once. - finalizedRunIdRef.current = null; - return; - } - // 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. 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({ - queryKey: AI_CHAT_MESSAGES_RQ_KEY(activeChatId), - }); - }, [run, activeChatId, queryClient, stoppingRun, localStreaming]); - // The page the user is currently viewing. AiChatWindow lives in a pathless // parent layout route, so useParams() can't see :pageSlug. Match the full // pathname against the authenticated page route instead so "the current page" @@ -1034,16 +945,13 @@ export default function AiChatWindow() { assistantName={currentRole?.name} onTurnFinished={onTurnFinished} onServerChatId={onServerChatId} - // #184: live-follow a still-running run when we reopened the chat as - // a passive observer; null when there is nothing to observe or this - // tab is the streamer. onStreamingChange lets the window stop polling - // while we are the streamer. - observedRow={observedRow} - onStreamingChange={onStreamingChange} + // #184 phase 1.5: arm/disarm the degraded-poll fallback when a + // resume attempt could not attach to the live run; the thread + // disarms it on settle / local stream. + onResumeFallback={onResumeFallback} // #184: in autonomous mode the Stop button must hit the authoritative // server stop (a local SSE abort is a client disconnect the server - // ignores). onServerStop also arms the "stopping" latch above so the - // stopped run's output does not re-stream via the observer merge. + // ignores). autonomousRunsEnabled={autonomousRunsEnabled} onServerStop={handleServerStop} /> diff --git a/apps/client/src/features/ai-chat/components/chat-thread.test.tsx b/apps/client/src/features/ai-chat/components/chat-thread.test.tsx index 0e9c184e..7d1027c1 100644 --- a/apps/client/src/features/ai-chat/components/chat-thread.test.tsx +++ b/apps/client/src/features/ai-chat/components/chat-thread.test.tsx @@ -1,6 +1,13 @@ -import { describe, it, expect, beforeEach, vi } from "vitest"; -import { render, screen, fireEvent, act, cleanup } from "@testing-library/react"; +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import { + render, + screen, + fireEvent, + act, + cleanup, +} from "@testing-library/react"; import { MantineProvider } from "@mantine/core"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; // Shared, hoisted mock state so the @ai-sdk/react and "ai" module mocks (hoisted // above the imports) can expose the captured useChat callbacks / transport and @@ -12,50 +19,61 @@ const h = vi.hoisted(() => ({ sendMessage: vi.fn(), stop: vi.fn(), setMessages: vi.fn(), + resumeStream: vi.fn(), + // The messages array useChat was seeded with (to assert strip/seed behavior). + seededMessages: null as null | unknown[], transport: null as null | { - prepareSendMessagesRequest: (arg: { + prepareSendMessagesRequest?: (arg: { messages: unknown[]; body: Record; }) => { body: Record }; + prepareReconnectToStreamRequest?: () => { api?: string }; + fetch?: (input: unknown, init?: { method?: string }) => Promise; }, }, })); -// Mock useChat: capture onFinish, return the spies and the controllable status. +// Mock useChat: capture onFinish + seeded messages, return the spies and the +// controllable status. vi.mock("@ai-sdk/react", () => ({ - useChat: (opts: { onFinish?: (arg: Record) => void }) => { + useChat: (opts: { + messages?: unknown[]; + onFinish?: (arg: Record) => void; + }) => { h.state.onFinish = opts.onFinish ?? null; + h.state.seededMessages = opts.messages ?? null; return { messages: [], sendMessage: h.state.sendMessage, status: h.state.status, stop: h.state.stop, error: null, - // #184: ChatThread reads setMessages to merge a polled observer run. setMessages: h.state.setMessages, + resumeStream: h.state.resumeStream, }; }, })); // Mock "ai": deterministic ids + a transport that records its options so the test -// can invoke prepareSendMessagesRequest and assert the `interrupted` flag. +// can invoke prepareSendMessagesRequest / prepareReconnectToStreamRequest / fetch. vi.mock("ai", () => { let counter = 0; return { generateId: () => `gid-${counter++}`, DefaultChatTransport: class { - constructor(opts: { - prepareSendMessagesRequest: (arg: { - messages: unknown[]; - body: Record; - }) => { body: Record }; - }) { - h.state.transport = opts; + constructor(opts: Record) { + h.state.transport = opts as never; } }, }; }); +// Keep the ai-chat-query import light: ChatThread only needs the messages RQ key, +// so stub the module to avoid pulling axios / i18n transitively. +vi.mock("@/features/ai-chat/queries/ai-chat-query.ts", () => ({ + AI_CHAT_MESSAGES_RQ_KEY: (chatId: string) => ["ai-chat-messages", chatId], +})); + // Stub the heavy children: MessageList (markdown/render) and ChatInput (the // composer). The ChatInput stub exposes a button that queues a message, the only // interaction this test needs to populate the queue while "streaming". @@ -63,49 +81,90 @@ vi.mock("@/features/ai-chat/components/message-list.tsx", () => ({ default: () =>
, })); vi.mock("@/features/ai-chat/components/chat-input.tsx", () => ({ - default: ({ onQueue }: { onQueue: (text: string) => void }) => ( - + default: ({ + onQueue, + onStop, + }: { + onQueue: (text: string) => void; + onStop: () => void; + }) => ( + <> + + + ), })); import ChatThread from "./chat-thread"; +import type { IAiChatMessageRow } from "@/features/ai-chat/types/ai-chat.types.ts"; -function renderThread() { +function row( + id: string, + role: string, + status?: string, + text = "", +): IAiChatMessageRow { + return { id, role, content: text, status, createdAt: "2026-01-01T00:00:00Z" }; +} + +function renderThread(props?: { + chatId?: string | null; + initialRows?: IAiChatMessageRow[]; + autonomousRunsEnabled?: boolean; +}) { const onTurnFinished = vi.fn(); - render( - - - , + const onResumeFallback = vi.fn(); + const onServerStop = vi.fn(); + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const invalidateSpy = vi.spyOn(queryClient, "invalidateQueries"); + const { unmount } = render( + + + + + , ); - return { onTurnFinished }; + return { onTurnFinished, onResumeFallback, onServerStop, invalidateSpy, unmount }; +} + +function resetState() { + h.state.status = "streaming"; + h.state.onFinish = null; + h.state.seededMessages = null; + h.state.transport = null; + h.state.sendMessage.mockClear(); + h.state.stop.mockClear(); + h.state.setMessages.mockClear(); + h.state.resumeStream.mockClear(); } describe("ChatThread — send now (#198)", () => { - beforeEach(() => { - h.state.status = "streaming"; - h.state.onFinish = null; - h.state.sendMessage.mockClear(); - h.state.stop.mockClear(); - h.state.transport = null; - }); + beforeEach(resetState); it("aborts the current turn and resends the queued message on the abort", () => { renderThread(); - // Queue a message while the turn is streaming. fireEvent.click(screen.getByTestId("queue-btn")); const sendNowBtn = screen.getByLabelText("Send now"); expect(sendNowBtn).toBeTruthy(); - // "Send now" interrupts the current turn (stop), but does NOT send yet — - // the resend happens once the abort lands in onFinish. fireEvent.click(sendNowBtn); expect(h.state.stop).toHaveBeenCalledTimes(1); expect(h.state.sendMessage).not.toHaveBeenCalled(); - // The abort we triggered reaches onFinish: the promoted head is flushed. act(() => { h.state.onFinish?.({ message: { id: "a", role: "assistant", parts: [] }, @@ -122,10 +181,8 @@ describe("ChatThread — send now (#198)", () => { fireEvent.click(screen.getByTestId("queue-btn")); fireEvent.click(screen.getByLabelText("Send now")); - const prep = h.state.transport!.prepareSendMessagesRequest; - // The send right after "send now" carries interrupted: true... + const prep = h.state.transport!.prepareSendMessagesRequest!; expect(prep({ messages: [], body: {} }).body.interrupted).toBe(true); - // ...and only that one (the flag is read-and-cleared). expect(prep({ messages: [], body: {} }).body.interrupted).toBe(false); }); @@ -136,42 +193,24 @@ describe("ChatThread — send now (#198)", () => { fireEvent.click(screen.getByTestId("queue-btn")); fireEvent.click(screen.getByLabelText("Send now")); - // No turn to interrupt: sent straight away, no abort, not flagged. expect(h.state.stop).not.toHaveBeenCalled(); expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); - const prep = h.state.transport!.prepareSendMessagesRequest; + const prep = h.state.transport!.prepareSendMessagesRequest!; expect(prep({ messages: [], body: {} }).body.interrupted).toBe(false); }); }); -// The turn-end decision lives in the `onFinish` handler: given the terminal -// outcome of a turn (`isAbort` / `isDisconnect` / `isError`, or none = clean), -// it decides whether to CONTINUE (flush the next queued message) or END (leave -// the queue intact for the user), and which stop notice — if any — to show. -// `sendNow` is exercised above; these tests pin down the plain outcomes. describe("ChatThread — turn-end decision (onFinish)", () => { - beforeEach(() => { - h.state.status = "streaming"; - h.state.onFinish = null; - h.state.sendMessage.mockClear(); - h.state.stop.mockClear(); - h.state.transport = null; - }); + beforeEach(resetState); - // Drive a fresh onFinish with the given terminal flags after queueing a - // message, and report both what the parent was told and whether the queue was - // flushed (a resend to the sendMessage spy). function finishWith(flags: { isAbort?: boolean; isDisconnect?: boolean; isError?: boolean; }) { - // Tear down any prior render so the loop-driven "every outcome" case does - // not leave duplicate queue buttons in the DOM. cleanup(); h.state.sendMessage.mockClear(); const { onTurnFinished } = renderThread(); - // Populate the queue while the turn is streaming. fireEvent.click(screen.getByTestId("queue-btn")); act(() => { h.state.onFinish?.({ @@ -187,16 +226,12 @@ describe("ChatThread — turn-end decision (onFinish)", () => { it("CONTINUES — flushes the next queued message on a clean finish", () => { finishWith({}); - // Clean finish (no terminal flag): the queued message is auto-sent. expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); - // A clean finish shows no stop notice. expect(screen.queryByText("Response stopped.")).toBeNull(); }); it("ENDS — keeps the queue intact on a user abort and shows the stopped notice", () => { finishWith({ isAbort: true }); - // A plain Stop (not the sendNow interrupt path) must NOT auto-resend: the - // queue is preserved for the user to decide. expect(h.state.sendMessage).not.toHaveBeenCalled(); expect(screen.getByText("Response stopped.")).toBeTruthy(); }); @@ -211,15 +246,11 @@ describe("ChatThread — turn-end decision (onFinish)", () => { it("ENDS — keeps the queue intact on a stream error (no auto-retry, no stopped notice)", () => { finishWith({ isError: true }); - // Blindly retrying after a failure would be wrong; the queue is left alone. expect(h.state.sendMessage).not.toHaveBeenCalled(); - // isError clears the neutral notice (the error banner covers this case). expect(screen.queryByText("Response stopped.")).toBeNull(); }); it("notifies the parent on EVERY terminal outcome", () => { - // The chat-list refresh / new-chat id adoption must run on success and on - // every failure path alike. for (const flags of [ {}, { isAbort: true }, @@ -232,55 +263,411 @@ describe("ChatThread — turn-end decision (onFinish)", () => { }); }); -// #184 passive-observer merge: when reconnecting to a still-running run, the -// parent feeds the polled run message via `observedRow`; ChatThread merges it via -// setMessages — but ONLY when this tab is NOT itself streaming (the streamer's -// SSE owns the view, so a stale observedRow must never overwrite it). -describe("ChatThread — observer run merge (#184)", () => { - beforeEach(() => { - h.state.onFinish = null; - h.state.setMessages.mockReset(); +// #184 phase 1.5: the resumable-SSE client. A reopened tab resumes the live run +// via the SDK's reconnect transport (attach: replay + tail) instead of polling. +describe("ChatThread — resume (attach) machinery (#184)", () => { + const streamingTail = () => [ + row("u1", "user", undefined, "hi"), + row("a1", "assistant", "streaming", "partial"), + ]; + const settledTail = () => [ + row("u1", "user", undefined, "hi"), + row("a1", "assistant", "succeeded", "done"), + ]; + const userTail = () => [row("u1", "user", undefined, "hi")]; + + const visibleMsg = { + id: "a1", + role: "assistant", + parts: [{ type: "text", text: "streamed answer" }], + }; + const emptyMsg = { id: "a1", role: "assistant", parts: [] }; + + beforeEach(resetState); + // NOTE: do NOT vi.unstubAllGlobals() here — vitest.setup.ts installs + // matchMedia/localStorage via vi.stubGlobal and unstubbing wipes them for the + // rest of the file. Fetch is re-stubbed per test that needs it. + afterEach(cleanup); + + it("resumes on mount only when the flag is on, chatId is set, and the tail is not a settled assistant", () => { + // streaming tail -> resume + renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() }); + expect(h.state.resumeStream).toHaveBeenCalledTimes(1); + + // user tail -> resume (the assistant row may not be seeded yet) + cleanup(); + h.state.resumeStream.mockClear(); + renderThread({ autonomousRunsEnabled: true, initialRows: userTail() }); + expect(h.state.resumeStream).toHaveBeenCalledTimes(1); + + // settled assistant tail -> NO resume + cleanup(); + h.state.resumeStream.mockClear(); + renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); + expect(h.state.resumeStream).not.toHaveBeenCalled(); + + // flag off -> NO resume + cleanup(); + h.state.resumeStream.mockClear(); + renderThread({ autonomousRunsEnabled: false, initialRows: streamingTail() }); + expect(h.state.resumeStream).not.toHaveBeenCalled(); + + // no chatId -> NO resume + cleanup(); + h.state.resumeStream.mockClear(); + renderThread({ + autonomousRunsEnabled: true, + chatId: null, + initialRows: streamingTail(), + }); + expect(h.state.resumeStream).not.toHaveBeenCalled(); }); - const observedRow = { - id: "a-run", - role: "assistant", - content: "step 1\nstep 2", - metadata: { - parts: [{ type: "text", text: "step 1\nstep 2" }], - }, - createdAt: "2026-01-01T00:00:00Z", - } as const; + it("strips the streaming tail from the seed, but keeps a user tail whole", () => { + renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() }); + // 2 rows in, streaming tail stripped -> 1 seeded message. + expect(h.state.seededMessages).toHaveLength(1); - function renderObserver(status: string) { - h.state.status = status; - render( + cleanup(); + renderThread({ autonomousRunsEnabled: true, initialRows: userTail() }); + // user tail is not stripped. + expect(h.state.seededMessages).toHaveLength(1); + }); + + it("builds the attach URL with expect=live&anchor only when the streaming tail was stripped", () => { + renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() }); + expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe( + "/api/ai-chat/runs/c1/stream?expect=live&anchor=a1", + ); + + cleanup(); + renderThread({ autonomousRunsEnabled: true, initialRows: userTail() }); + expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe( + "/api/ai-chat/runs/c1/stream", + ); + }); + + async function fetch204() { + vi.stubGlobal( + "fetch", + vi.fn().mockResolvedValue({ status: 204, ok: false }), + ); + await act(async () => { + await h.state.transport!.fetch!("http://x", { method: "GET" }); + }); + } + + it("204 on a user tail: no crash, no restore, reconcile+invalidate, onResumeFallback(true)", async () => { + const { onResumeFallback, invalidateSpy } = renderThread({ + autonomousRunsEnabled: true, + initialRows: userTail(), + }); + await fetch204(); + // No stripped row -> no restore merge. + expect(h.state.setMessages).not.toHaveBeenCalled(); + expect(invalidateSpy).toHaveBeenCalledWith({ + queryKey: ["ai-chat-messages", "c1"], + }); + expect(onResumeFallback).toHaveBeenCalledWith(true); + }); + + it("204 on a streaming tail: restore + invalidate + onResumeFallback(true)", async () => { + const { onResumeFallback, invalidateSpy } = renderThread({ + autonomousRunsEnabled: true, + initialRows: streamingTail(), + }); + await fetch204(); + // Stripped row is restored to the store. + expect(h.state.setMessages).toHaveBeenCalledTimes(1); + expect(invalidateSpy).toHaveBeenCalledWith({ + queryKey: ["ai-chat-messages", "c1"], + }); + expect(onResumeFallback).toHaveBeenCalledWith(true); + }); + + it("F7 restart-survival: a 500 attach failure restores the stripped row AND arms the poll (not lost)", async () => { + const { onResumeFallback, invalidateSpy } = renderThread({ + autonomousRunsEnabled: true, + initialRows: streamingTail(), + }); + vi.stubGlobal( + "fetch", + vi.fn().mockResolvedValue({ status: 500, ok: false }), + ); + await act(async () => { + await h.state.transport!.fetch!("http://x", { method: "GET" }); + }); + expect(h.state.setMessages).toHaveBeenCalledTimes(1); // stripped row restored + expect(invalidateSpy).toHaveBeenCalledWith({ + queryKey: ["ai-chat-messages", "c1"], + }); + expect(onResumeFallback).toHaveBeenCalledWith(true); // degraded poll armed + }); + + it("F7 restart-survival: a network throw restores the stripped row AND arms the poll", async () => { + const { onResumeFallback, invalidateSpy } = renderThread({ + autonomousRunsEnabled: true, + initialRows: streamingTail(), + }); + vi.stubGlobal( + "fetch", + vi.fn().mockRejectedValue(new Error("network down")), + ); + await act(async () => { + await h.state + .transport!.fetch!("http://x", { method: "GET" }) + .catch(() => undefined); // the wrapper rethrows; swallow here + }); + expect(h.state.setMessages).toHaveBeenCalledTimes(1); + expect(invalidateSpy).toHaveBeenCalledWith({ + queryKey: ["ai-chat-messages", "c1"], + }); + expect(onResumeFallback).toHaveBeenCalledWith(true); + }); + + it("unmount during a pending attach aborts the controller and gates late callbacks", async () => { + const { onResumeFallback, invalidateSpy, unmount } = renderThread({ + autonomousRunsEnabled: true, + initialRows: streamingTail(), + }); + let abortSeen = false; + let resolveFetch!: (v: unknown) => void; + vi.stubGlobal( + "fetch", + vi.fn().mockImplementation((_input: unknown, init: RequestInit) => { + init.signal?.addEventListener("abort", () => { + abortSeen = true; + }); + return new Promise((res) => { + resolveFetch = res; + }); + }), + ); + // Kick a reconnect GET (stays pending). + let pending!: Promise; + act(() => { + pending = h.state.transport!.fetch!("http://x", { method: "GET" }); + }); + // Unmount: the cleanup aborts the in-flight attach. + unmount(); + expect(abortSeen).toBe(true); + // A late 204 landing after unmount must NOT arm a poll / invalidate the (now + // different) chat. + onResumeFallback.mockClear(); + invalidateSpy.mockClear(); + await act(async () => { + resolveFetch({ status: 204, ok: false }); + await pending; + }); + expect(onResumeFallback).not.toHaveBeenCalledWith(true); + expect(invalidateSpy).not.toHaveBeenCalled(); + }); + + it("a resume fetch error clears resumedTurn so the next local turn flushes the queue", async () => { + renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() }); + h.state.status = "ready"; + vi.stubGlobal( + "fetch", + vi.fn().mockResolvedValue({ status: 500, ok: false }), + ); + await act(async () => { + await h.state.transport!.fetch!("http://x", { method: "GET" }); + }); + // Queue then clean-finish: suppression was cleared, so the queue flushes. + fireEvent.click(screen.getByTestId("queue-btn")); + act(() => { + h.state.onFinish?.({ + message: visibleMsg, + isAbort: false, + isDisconnect: false, + isError: false, + }); + }); + expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); + }); + + it("a resumed turn's onFinish does NOT flush the queue", () => { + renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() }); + fireEvent.click(screen.getByTestId("queue-btn")); + act(() => { + h.state.onFinish?.({ + message: visibleMsg, + isAbort: false, + isDisconnect: false, + isError: false, + }); + }); + expect(h.state.sendMessage).not.toHaveBeenCalled(); + }); + + it("a healthy resumed finish (visible content) arms nothing and keeps the store", () => { + h.state.status = "ready"; + const { onResumeFallback } = renderThread({ + autonomousRunsEnabled: true, + initialRows: streamingTail(), + }); + h.state.setMessages.mockClear(); + onResumeFallback.mockClear(); + act(() => { + h.state.onFinish?.({ + message: visibleMsg, + isAbort: false, + isDisconnect: false, + isError: false, + }); + }); + // No restore (would clobber the fuller streamed message), no poll arm. + expect(h.state.setMessages).not.toHaveBeenCalled(); + expect(onResumeFallback).not.toHaveBeenCalledWith(true); + }); + + it("isDisconnect WITH visible content arms the poll but does NOT restore", () => { + h.state.status = "ready"; + const { onResumeFallback, invalidateSpy } = renderThread({ + autonomousRunsEnabled: true, + initialRows: streamingTail(), + }); + h.state.setMessages.mockClear(); + onResumeFallback.mockClear(); + invalidateSpy.mockClear(); + act(() => { + h.state.onFinish?.({ + message: visibleMsg, + isAbort: false, + isDisconnect: true, + isError: false, + }); + }); + expect(onResumeFallback).toHaveBeenCalledWith(true); + expect(invalidateSpy).toHaveBeenCalledWith({ + queryKey: ["ai-chat-messages", "c1"], + }); + // Restore forbidden: the on-screen partial must not roll back. + expect(h.state.setMessages).not.toHaveBeenCalled(); + }); + + it("an empty resumed message (starved replay) restores the stripped row AND arms the poll", () => { + h.state.status = "ready"; + const { onResumeFallback } = renderThread({ + autonomousRunsEnabled: true, + initialRows: streamingTail(), + }); + h.state.setMessages.mockClear(); + onResumeFallback.mockClear(); + act(() => { + h.state.onFinish?.({ + message: emptyMsg, + isAbort: false, + isDisconnect: false, + isError: false, + }); + }); + expect(h.state.setMessages).toHaveBeenCalledTimes(1); // restore + expect(onResumeFallback).toHaveBeenCalledWith(true); // arm + }); + + it("degraded-merge: merges the tail per initialRows update, and settles disarm the poll", async () => { + h.state.status = "ready"; + const { rerender, onResumeFallback } = renderResumable(streamingTail()); + // Arm reconcile via a 204. + vi.stubGlobal( + "fetch", + vi.fn().mockResolvedValue({ status: 204, ok: false }), + ); + await act(async () => { + await h.state.transport!.fetch!("http://x", { method: "GET" }); + }); + h.state.setMessages.mockClear(); + onResumeFallback.mockClear(); + + // A streaming-tail update: merge, poll stays armed. + rerender([ + row("u1", "user", undefined, "hi"), + row("a1", "assistant", "streaming", "step 1\nstep 2"), + ]); + expect(h.state.setMessages).toHaveBeenCalledTimes(1); + expect(onResumeFallback).not.toHaveBeenCalledWith(false); + + // A settled-tail update: merge + disarm. + h.state.setMessages.mockClear(); + rerender([ + row("u1", "user", undefined, "hi"), + row("a1", "assistant", "succeeded", "final"), + ]); + expect(h.state.setMessages).toHaveBeenCalledTimes(1); + expect(onResumeFallback).toHaveBeenCalledWith(false); + }); + + it("a local stream disarms both the merge and the poll", () => { + h.state.status = "streaming"; + const { rerender, onResumeFallback } = renderResumable(streamingTail()); + onResumeFallback.mockClear(); + // A re-render while streaming: the reconciliation effect disarms. + rerender(streamingTail()); + expect(onResumeFallback).toHaveBeenCalledWith(false); + }); + + it("Send now is hidden on a resumed turn but visible on a local stream", () => { + // Resumed turn: hidden. + renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() }); + fireEvent.click(screen.getByTestId("queue-btn")); + expect(screen.queryByLabelText("Send now")).toBeNull(); + + // Local streaming turn (no resume): visible. + cleanup(); + resetState(); + renderThread({ initialRows: [] }); + fireEvent.click(screen.getByTestId("queue-btn")); + expect(screen.getByLabelText("Send now")).toBeTruthy(); + }); + + it("handleStop aborts the attach controller and calls onServerStop", async () => { + const { onServerStop } = renderThread({ + autonomousRunsEnabled: true, + initialRows: streamingTail(), + }); + // Establish an attach controller via a (pending) reconnect GET. + let abortSeen = false; + vi.stubGlobal( + "fetch", + vi.fn().mockImplementation((_input: unknown, init: RequestInit) => { + init.signal?.addEventListener("abort", () => { + abortSeen = true; + }); + return new Promise(() => undefined); // never resolves + }), + ); + act(() => { + void h.state.transport!.fetch!("http://x", { method: "GET" }); + }); + fireEvent.click(screen.getByLabelText("Stop")); + expect(abortSeen).toBe(true); + expect(onServerStop).toHaveBeenCalledWith("c1"); + }); +}); + +// Helper: render a resumable thread and expose a rerender that only swaps +// initialRows (the degraded-merge effect depends on it). +function renderResumable(initialRows: IAiChatMessageRow[]) { + const onResumeFallback = vi.fn(); + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const Wrapper = ({ rows }: { rows: IAiChatMessageRow[] }) => ( + - , - ); - } - - it("merges the polled run message when this tab is a passive observer", () => { - renderObserver("ready"); - expect(h.state.setMessages).toHaveBeenCalledTimes(1); - // The updater replaces/append the observed assistant row by id. - const updater = h.state.setMessages.mock.calls[0][0] as ( - prev: { id: string; parts: { text: string }[] }[], - ) => { id: string; parts: { text: string }[] }[]; - const merged = updater([{ id: "u1", parts: [{ text: "hi" }] }]); - expect(merged).toHaveLength(2); - expect(merged[1].id).toBe("a-run"); - expect(merged[1].parts[0].text).toBe("step 1\nstep 2"); - }); - - it("does NOT merge while THIS tab is the streamer (no double-render)", () => { - renderObserver("streaming"); - expect(h.state.setMessages).not.toHaveBeenCalled(); - }); -}); + + + ); + const view = render(); + const rerender = (rows: IAiChatMessageRow[]) => + act(() => view.rerender()); + return { rerender, onResumeFallback }; +} 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 a6ec1908..acf4326e 100644 --- a/apps/client/src/features/ai-chat/components/chat-thread.tsx +++ b/apps/client/src/features/ai-chat/components/chat-thread.tsx @@ -1,4 +1,5 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import { useQueryClient } from "@tanstack/react-query"; import { generateId } from "ai"; import { ActionIcon, Box, Group, Stack, Text, Tooltip } from "@mantine/core"; import { @@ -24,7 +25,14 @@ import { } from "@/features/ai-chat/utils/role-launch.ts"; import { describeChatError } from "@/features/ai-chat/utils/error-message.ts"; import { extractServerChatId } from "@/features/ai-chat/utils/adopt-chat-id.ts"; -import { mergeObservedMessage } from "@/features/ai-chat/utils/run-polling.ts"; +import { assistantMessageHasVisibleContent } from "@/features/ai-chat/utils/message-content.ts"; +import { + isStreamingTail, + isSettledAssistantTail, + seedRows, + mergeById, +} from "@/features/ai-chat/utils/resume-helpers.ts"; +import { AI_CHAT_MESSAGES_RQ_KEY } from "@/features/ai-chat/queries/ai-chat-query.ts"; import { dequeue, enqueueMessage, @@ -87,19 +95,13 @@ interface ChatThreadProps { * Copy/export button available mid-stream). Distinct from onTurnFinished, * which fires only at the terminal outcome. */ onServerChatId?: (serverChatId?: string) => void; - /** #184 reconnect-and-live-follow. When THIS tab reopened a chat whose agent - * run is still going (it is a PASSIVE OBSERVER — it did not start the run here), - * the parent polls the reconnect endpoint and feeds the run's incrementally- - * persisted assistant message here; we merge it into the live list so new - * steps/tool-calls appear as they are persisted. Null when there is nothing to - * observe (no run, feature off, or this tab IS the streamer). The merge is - * ADDITIONALLY guarded by our own `isStreaming`, so a stale value can never - * fight the local stream when we are the streamer. */ - observedRow?: IAiChatMessageRow | null; - /** Report this tab's live streaming status up to the parent, so it can stop - * polling the run while WE are the active streamer (the SSE owns the view) and - * resume once we go idle. Called from an effect on every transition. */ - onStreamingChange?: (streaming: boolean) => void; + /** #184 phase 1.5: arm/disarm the parent's degraded-poll fallback for THIS + * chat's window. Called `true` when a resume attempt could not attach to the + * live run (attach 204 / starved-or-torn resumed finish), so the window starts + * a dumb timed poll of the message history to follow the detached run to settle; + * called `false` the moment a local stream starts or the terminal settled row is + * merged (invariant 8). The window owns the timer + its 10-min cap. */ + onResumeFallback?: (active: boolean) => void; /** #184: whether detached/autonomous agent runs are enabled for this workspace. * When true the Stop button must additionally hit the AUTHORITATIVE server stop * (via onServerStop) — aborting only the local SSE is just a client disconnect, @@ -155,15 +157,58 @@ export default function ChatThread({ assistantName, onTurnFinished, onServerChatId, - observedRow, - onStreamingChange, + onResumeFallback, autonomousRunsEnabled, onServerStop, }: ChatThreadProps) { const { t } = useTranslation(); + const queryClient = useQueryClient(); + + // resume machinery refs (#184 phase 1.5) + const attachAbortRef = useRef(null); + const reconcileTailRef = useRef(false); + const noStreamHandledRef = useRef(false); + const onNoActiveStreamRef = useRef<(() => void) | null>(null); + // Live mount flag. The attach GET and the resumed `onFinish` are async and can + // land AFTER this thread unmounts (the parent remounts per chat via `key`); with + // chatIdRef then pointing at the NEW chat, an ungated late callback would arm a + // spurious poll + foreign invalidation on the newly-opened chat. Every parent- + // facing resume side-effect is gated on this. + const mountedRef = useRef(true); + const [resumedTurn, setResumedTurn] = useState(false); + const resumedTurnRef = useRef(false); + // Identity-stable pair setter (bare useState setter + ref write): it is closed + // over by the transport useMemo([]), so it MUST NOT capture state. + const setResumedTurnPair = useCallback((v: boolean) => { + resumedTurnRef.current = v; + setResumedTurn(v); + }, []); + + // Mount-time resume gating (in refs — computed once for this mount; the parent + // remounts per chat via `key`). + // + // Attempt resume for any non-settled tail: a streaming tail (strip + expect + // live replay) or a user tail (the run may exist but its assistant row is not + // seeded yet — attach to the pre-opened registry entry and wait for frames). + // A settled assistant tail must NEVER resume: replaying a finished run into a + // store that already contains its message duplicates parts (SDK text-start + // always pushes a new part). + const stripRef = useRef(chatId !== null && isStreamingTail(initialRows ?? [])); + const attemptResumeRef = useRef( + autonomousRunsEnabled === true && + chatId !== null && + !isSettledAssistantTail(initialRows ?? []), + ); + const strippedRowRef = useRef( + stripRef.current ? (initialRows ?? [])[initialRows!.length - 1] : null, + ); const initialMessages = useMemo( - () => (initialRows ?? []).map(rowToUiMessage), + () => + seedRows( + initialRows ?? [], + attemptResumeRef.current && stripRef.current, + ).map(rowToUiMessage), [initialRows], ); @@ -261,9 +306,12 @@ export default function ChatThread({ const { head, rest } = dequeue(queuedRef.current); if (!head) return false; setQueue(rest); + // Local send: clear any resume-suppression flag so this genuine local turn's + // onFinish flushes normally (invariant 8). + setResumedTurnPair(false); sendMessageRef.current?.({ text: head.text }); return true; - }, [setQueue]); + }, [setQueue, setResumedTurnPair]); const enqueue = useCallback( (text: string) => { @@ -283,6 +331,47 @@ export default function ChatThread({ new DefaultChatTransport({ api: "/api/ai-chat/stream", credentials: "include", + prepareReconnectToStreamRequest: () => ({ + // SDK default URL uses the useChat STORE id — always build from the real chat id. + // ?expect=live&anchor= ONLY when we stripped a streaming tail: expect=live + // is the only case where a finished-retained replay is safe (the row is stripped, + // replay rebuilds it), and the anchor pins the replay to OUR run — a mismatching + // (newer) run must 204 into the restore+poll path instead of replaying a foreign + // transcript into this store. + api: `/api/ai-chat/runs/${chatIdRef.current}/stream${ + stripRef.current + ? `?expect=live&anchor=${strippedRowRef.current!.id}` + : "" + }`, + }), + fetch: async (input: RequestInfo | URL, init: RequestInit = {}) => { + if ((init.method ?? "GET") !== "GET") return fetch(input, init); // send path untouched + // Reconnect GET: the SDK passes no AbortSignal, so wire our own controller + // for observer Stop / unmount abort. + const controller = new AbortController(); + attachAbortRef.current = controller; + try { + const response = await fetch(input, { + ...init, + signal: controller.signal, + }); + // No onFinish will come for a 204 (silent no-op) OR any non-2xx + // (5xx/502 — a server restart mid-attach). Both run the same + // no-active-stream recovery: restore the stripped row, invalidate, and + // arm the degraded poll (idempotent via noStreamHandledRef; its part-d + // also clears the resumedTurn flag). This is the restart-survival path + // the removed F7 latch used to guard — a transient attach failure must + // NOT drop the in-progress row or stop tracking the durable run. + if (response.status === 204 || !response.ok) + onNoActiveStreamRef.current?.(); + return response; + } catch (err) { + // Network throw: same no-onFinish recovery, then rethrow so the SDK + // still surfaces the error to its own machinery. + onNoActiveStreamRef.current?.(); + throw err; + } + }, // Inject the chat id and the currently-open page alongside the useChat // messages so the server can resolve an existing chat (or create one // when null) and tell the agent which page "this page" refers to. Both @@ -312,7 +401,15 @@ export default function ChatThread({ [], ); - const { messages, sendMessage, status, stop, error, setMessages } = useChat({ + const { + messages, + sendMessage, + status, + stop, + error, + setMessages, + resumeStream, + } = useChat({ // Stable per-mount key. Existing chats use their real id; new chats use a // generated client id (never `undefined`) so the store is NOT re-created on // every render mid-stream (see `chatStoreId` above). @@ -330,6 +427,38 @@ export default function ChatThread({ // would be wrong, so on Stop/disconnect/error the queue is left intact for // the user to decide. onFinish: ({ message, isAbort, isDisconnect, isError }) => { + // (1) Capture whether THIS finish belongs to a resumed (attach) turn and + // immediately clear the flag so it can never suppress a LATER local turn. + const wasResumed = resumedTurnRef.current; + setResumedTurnPair(false); + // (2) Recovery after a starved/torn resumed finish (invariant 9). The arm + // and the stripped-row restore are gated DIFFERENTLY. Skip entirely once + // unmounted (an abort-triggered onFinish landing after a chat switch must + // not arm a poll / invalidate on the new chat). + if (wasResumed && mountedRef.current) { + const hasVisibleContent = assistantMessageHasVisibleContent(message); + // ARM the reconcile + degraded poll when the resumed message carries no + // visible content (starved replay) OR the connection dropped mid-run — in + // both cases the poll must drive the row to its real terminal state. + if (isDisconnect || !hasVisibleContent) { + reconcileTailRef.current = true; + queryClient.invalidateQueries({ + queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current), + }); + onResumeFallback?.(true); + } + // RESTORE the stripped streaming row ONLY when the resumed message has no + // visible content. On isDisconnect WITH visible content restore is + // FORBIDDEN: the live stream may have advanced far past the mount-time + // snapshot, so restoring would clobber on-screen content (invariant 9) — + // the arm above suffices, the poll reaches the true terminal. + if (!hasVisibleContent && strippedRowRef.current) { + setMessages((prev) => + mergeById(prev, rowToUiMessage(strippedRowRef.current!)), + ); + } + } + // (3) Standard branches. // Forward the authoritative server chatId (streamed on the assistant // message metadata) so the parent adopts the REAL created chat id for a new // chat — see adopt-chat-id.ts for the full #137 design. `threadKey` lets the @@ -342,6 +471,10 @@ export default function ChatThread({ else if (isAbort) setStopNotice("manual"); else if (isDisconnect) setStopNotice("disconnect"); else setStopNotice(null); + // A resumed turn NEVER flushes the queue (invariant 7): skip BOTH the + // flush-on-abort branch and the plain flush. The local streamer is the only + // tab that owns the queue. + if (wasResumed) return; // "Send now": WE triggered this abort to interrupt the current turn and // immediately send the promoted head. Flush it even though the turn was // aborted (the normal abort path below keeps the queue intact). The @@ -423,26 +556,98 @@ export default function ChatThread({ const isStreaming = status === "submitted" || status === "streaming"; - // #184: report our live streaming status up so the parent stops polling the run - // while WE are the streamer (the SSE owns the view) and resumes once we go idle. - // Effect (not render) so it never updates parent state during our own render; - // fires on mount with `false`, which also re-syncs the parent after a chat - // switch remounts this thread (a fresh mount is idle until the user sends). - useEffect(() => { - onStreamingChange?.(isStreaming); - }, [isStreaming, onStreamingChange]); + // 204-handler (`onNoActiveStream`): the attach returned 204 — nothing live to + // resume (overflow / begin-failure / after retention / anchor-mismatch). One- + // shot via noStreamHandledRef (we do NOT null onNoActiveStreamRef). Exactly four + // parts. Kept in a ref (read by the transport's fetch closure) and refreshed + // each render below. + const onNoActiveStream = useCallback(() => { + // A late attach outcome after unmount must not arm a poll / invalidate on the + // now-different chat this thread's refs were reused for. + if (!mountedRef.current) return; + if (noStreamHandledRef.current) return; + noStreamHandledRef.current = true; + // (a) Restore the stripped streaming row to the store — ONLY when we actually + // stripped one (a user-tail 204 does NOT reach here with a stripped row, so do + // not dereference null). + if (strippedRowRef.current) { + setMessages((prev) => + mergeById(prev, rowToUiMessage(strippedRowRef.current!)), + ); + } + // (b) Reconcile the tail from the message history + invalidate it so the + // degraded poll starts from a fresh fetch. + reconcileTailRef.current = true; + queryClient.invalidateQueries({ + queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current), + }); + // (c) Arm the degraded poll (a dumb timer with a 10-min cap in the window); + // the thread disarms it via onResumeFallback(false) on settle / local stream. + onResumeFallback?.(true); + // (d) 204 means onFinish will NOT fire — clear the suppression flag so it + // cannot swallow the NEXT local turn's queue flush. + setResumedTurnPair(false); + }, [setMessages, queryClient, onResumeFallback, setResumedTurnPair]); + onNoActiveStreamRef.current = onNoActiveStream; - // #184 passive-observer merge: when the parent feeds a polled run message (we - // reopened a chat whose run is still going and did NOT start it here), merge it - // into the live list so new steps/tool-calls appear as they are persisted. Hard- - // gated by `!isStreaming`: if THIS tab is actually the streamer, the local SSE - // owns the view and a stale observedRow must never overwrite it. `observedRow` - // is a stable per-poll object, so this runs once per poll, not per render. + // Mount effect: kick off the resume attempt for a non-settled tail. Marking the + // turn as resumed BEFORE resumeStream so onFinish (invariant 7/8) sees it. useEffect(() => { - if (isStreaming || !observedRow) return; - const observed = rowToUiMessage(observedRow); - setMessages((prev) => mergeObservedMessage(prev, observed)); - }, [observedRow, isStreaming, setMessages]); + // Re-arm on (re)mount — StrictMode dev-mounts twice, and the cleanup below + // flips this false between the two. + mountedRef.current = true; + if (attemptResumeRef.current) { + setResumedTurnPair(true); + void resumeStream(); + } + // Unmount: mark unmounted (gates late attach/onFinish side-effects) and abort + // the in-flight attach GET so its callbacks don't fire against the next chat. + return () => { + mountedRef.current = false; + attachAbortRef.current?.abort(); + }; + // Mount-only by design; the parent remounts per chat via `key`. + // eslint-disable-next-line react-hooks/exhaustive-deps + }, []); + + // Reconciliation + degraded-merge (invariant 8). Deps are EXACTLY + // [initialRows, isStreaming, setMessages]. + useEffect(() => { + // A local stream owns the view: disarm BOTH the merge and the window poll. + if (isStreaming) { + reconcileTailRef.current = false; + onResumeFallback?.(false); + return; + } + if (!reconcileTailRef.current) return; + const rows = initialRows ?? []; + const tail = rows[rows.length - 1]; + if (!tail || tail.role !== "assistant") return; + // Merge the polled assistant tail on EVERY initialRows update — while the + // degraded poll is active this IS the live per-step progress. + setMessages((prev) => mergeById(prev, rowToUiMessage(tail))); + // Anchor-mismatch coherence: when we restored a stripped streaming row A but a + // DIFFERENT run's row B is now the tail (A finished, B replaced the registry + // entry, so the attach 204'd), A would otherwise linger forever as an orphan + // jumping-dots row over the real run. Settle it from fresh history (where A is + // now persisted) so no phantom row survives. No-op in the common case where A + // IS the tail (id match). + const stripped = strippedRowRef.current; + if (stripped && stripped.id !== tail.id) { + const historical = rows.find((r) => r.id === stripped.id); + if (historical) + setMessages((prev) => mergeById(prev, rowToUiMessage(historical))); + } + // Settled: the terminal merge is done — disarm the flag AND the window poll + // explicitly (the window only has a time cap, it will not disarm itself). + if (tail.status !== "streaming") { + reconcileTailRef.current = false; + onResumeFallback?.(false); + } + // onResumeFallback intentionally omitted (parent-stable callback); deps are + // fixed by the resume design. + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [initialRows, isStreaming, setMessages]); // "Send now" on a queued message: interrupt the current turn and immediately // send THIS message, keeping the agent's partial output. Other queued messages @@ -469,10 +674,12 @@ export default function ChatThread({ const msg = queuedRef.current.find((m) => m.id === id); if (!msg) return; setQueue(removeQueuedById(queuedRef.current, id)); + // Local send: clear any resume-suppression flag (invariant 8). + setResumedTurnPair(false); sendMessageRef.current?.({ text: msg.text }); } }, - [setQueue, stop], + [setQueue, stop, setResumedTurnPair], ); // Stop the current turn. ALWAYS abort the local SSE (`stop()`) so the composer @@ -485,6 +692,9 @@ export default function ChatThread({ // is not known yet — a brand-new chat in the first moment of its first turn — // only the local abort happens (there is no server-side run handle to stop yet). const handleStop = useCallback(() => { + // Abort the resume/attach GET first: the SDK does not pass it a signal, so an + // observer's Stop would otherwise leave the attach fetch running. + attachAbortRef.current?.abort(); stop(); if (!autonomousRunsEnabled) return; if (chatIdRef.current) { @@ -617,17 +827,23 @@ export default function ChatThread({ {m.text} - - sendNow(m.id)} - aria-label={t("Send now")} - > - - - + {/* "Send now" (interrupt) is hidden on a RESUMED turn: a local + stop() does not abort the resumed attach fetch, so the click + would be swallowed while flushOnAbortRef would fire minutes + later on the natural finish. Only the remove affordance stays. */} + {!resumedTurn && ( + + sendNow(m.id)} + aria-label={t("Send now")} + > + + + + )} )} sendMessage({ text })} + onSend={(text) => { + // Local send: clear any resume-suppression flag (invariant 8). + setResumedTurnPair(false); + sendMessage({ text }); + }} onQueue={enqueue} onStop={handleStop} isStreaming={isStreaming} diff --git a/apps/client/src/features/ai-chat/queries/ai-chat-messages-query.test.tsx b/apps/client/src/features/ai-chat/queries/ai-chat-messages-query.test.tsx new file mode 100644 index 00000000..2c85ac66 --- /dev/null +++ b/apps/client/src/features/ai-chat/queries/ai-chat-messages-query.test.tsx @@ -0,0 +1,90 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import React from "react"; +import { renderHook, waitFor } from "@testing-library/react"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; + +// react-i18next / notifications are pulled in transitively by ai-chat-query.ts +// (the mutation hooks use them); stub so the module imports cleanly. +vi.mock("react-i18next", () => ({ + useTranslation: () => ({ t: (key: string) => key }), +})); +vi.mock("@mantine/notifications", () => ({ + notifications: { show: vi.fn() }, +})); + +// Mock the service module; only getAiChatMessages is exercised, but the other +// named exports must exist so ai-chat-query.ts imports resolve. +vi.mock("@/features/ai-chat/services/ai-chat-service.ts", () => ({ + getAiChatMessages: vi.fn(), + getAiChats: vi.fn(), + getAiRoleCatalog: vi.fn(), + getAiRoleCatalogBundle: vi.fn(), + getAiRoles: vi.fn(), + importAiRolesFromCatalog: vi.fn(), + createAiRole: vi.fn(), + deleteAiChat: vi.fn(), + deleteAiRole: vi.fn(), + renameAiChat: vi.fn(), + updateAiRole: vi.fn(), + updateAiRoleFromCatalog: vi.fn(), +})); + +import { getAiChatMessages } from "@/features/ai-chat/services/ai-chat-service.ts"; +import { useAiChatMessagesQuery } from "@/features/ai-chat/queries/ai-chat-query.ts"; + +const emptyPage = { items: [], meta: { hasNextPage: false, nextCursor: null } }; + +function createWrapper() { + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + return function Wrapper({ children }: { children: React.ReactNode }) { + return ( + {children} + ); + }; +} + +// The degraded-poll fallback (#184 phase 1.5) is threaded into this query as a +// `refetchInterval`; AiChatWindow supplies the deliberately-dumb callback. These +// pin the plumbing the window depends on: the interval polls the message history, +// and — critically — fetch ERRORS do NOT stop the tick (TanStack v5 resets the +// failure count each fetch, so the poll must survive a server restart). +describe("useAiChatMessagesQuery — degraded refetchInterval", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("re-polls at the interval while the callback returns a duration", async () => { + vi.mocked(getAiChatMessages).mockResolvedValue(emptyPage as never); + renderHook(() => useAiChatMessagesQuery("c1", () => 30), { + wrapper: createWrapper(), + }); + await waitFor(() => + expect(vi.mocked(getAiChatMessages).mock.calls.length).toBeGreaterThan(2), + ); + }); + + it("does NOT re-poll when the callback returns false", async () => { + vi.mocked(getAiChatMessages).mockResolvedValue(emptyPage as never); + renderHook(() => useAiChatMessagesQuery("c1", () => false), { + wrapper: createWrapper(), + }); + await waitFor(() => + expect(vi.mocked(getAiChatMessages)).toHaveBeenCalledTimes(1), + ); + // Give any errant interval a chance to fire, then assert it did not. + await new Promise((r) => setTimeout(r, 60)); + expect(vi.mocked(getAiChatMessages)).toHaveBeenCalledTimes(1); + }); + + it("keeps ticking through fetch errors (errors do not gate the poll)", async () => { + vi.mocked(getAiChatMessages).mockRejectedValue(new Error("server down")); + renderHook(() => useAiChatMessagesQuery("c1", () => 30), { + wrapper: createWrapper(), + }); + await waitFor(() => + expect(vi.mocked(getAiChatMessages).mock.calls.length).toBeGreaterThan(2), + ); + }); +}); diff --git a/apps/client/src/features/ai-chat/queries/ai-chat-query.ts b/apps/client/src/features/ai-chat/queries/ai-chat-query.ts index 9a6851cd..2ffe715f 100644 --- a/apps/client/src/features/ai-chat/queries/ai-chat-query.ts +++ b/apps/client/src/features/ai-chat/queries/ai-chat-query.ts @@ -13,7 +13,6 @@ import { deleteAiChat, deleteAiRole, getAiChatMessages, - getAiChatRun, getAiChats, getAiRoleCatalog, getAiRoleCatalogBundle, @@ -26,7 +25,6 @@ import { import { IAiChat, IAiChatMessageRow, - IAiChatRunResponse, IAiRole, IAiRoleCatalog, IAiRoleCatalogBundle, @@ -37,7 +35,6 @@ import { IAiRoleUpdateFromCatalogResult, } from "@/features/ai-chat/types/ai-chat.types.ts"; import { IPagination } from "@/lib/types.ts"; -import { runPollInterval } from "@/features/ai-chat/utils/run-polling.ts"; export const AI_CHATS_RQ_KEY = ["ai-chats"]; export const AI_ROLES_RQ_KEY = ["ai-roles"]; @@ -55,7 +52,6 @@ export const AI_CHAT_MESSAGES_RQ_KEY = (chatId: string) => [ "ai-chat-messages", chatId, ]; -export const AI_CHAT_RUN_RQ_KEY = (chatId: string) => ["ai-chat-run", chatId]; /** Paginated list of the current user's chats (auto-loads further pages). */ export function useAiChatsQuery() { @@ -89,7 +85,15 @@ export function useAiChatsQuery() { * Load all persisted messages of a chat (oldest first), flattening the * paginated server response. Used to seed `useChat` initial messages. */ -export function useAiChatMessagesQuery(chatId: string | undefined) { +export function useAiChatMessagesQuery( + chatId: string | undefined, + // #184 phase 1.5: the degraded-poll fallback. When a tab could not attach to a + // still-running run (the attach returned 204 / the resumed stream ended with no + // terminal row), the window arms a dumb timed poll of the message history to + // follow the detached run to settle. The callback form lives in AiChatWindow; + // threaded here verbatim so this query owns the polling. Undefined => no poll. + refetchInterval?: number | false | (() => number | false), +) { const query = useInfiniteQuery({ queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatId ?? ""), queryFn: ({ pageParam }) => @@ -100,6 +104,7 @@ export function useAiChatMessagesQuery(chatId: string | undefined) { ? (lastPage.meta.nextCursor ?? undefined) : undefined, enabled: !!chatId, + refetchInterval, }); // useInfiniteQuery only fetches the first page on its own. The hook's contract @@ -139,34 +144,6 @@ export function useAiChatMessagesQuery(chatId: string | undefined) { }; } -/** - * Reconnect to a chat's latest agent run and LIVE-FOLLOW it (#184). While the run - * is active the query re-polls every {@link runPollInterval} ms (driven off the - * fetched `run.status`, the same status-keyed refetchInterval pattern as the - * embeddings reindex polling); once the run reaches a terminal status — or there - * is no run — the interval returns `false` and polling stops on its own. Polling - * is thus naturally bounded by the run terminating; no separate timeout cap. - * - * `enabled` gates the whole thing: callers pass `false` when the autonomous-runs - * feature is off (the endpoint is NOT flag-gated server-side, but with the feature - * off the chat has no runs, so polling would only ever return `{ run: null }`) OR - * when THIS tab is the one actively streaming the run (the live SSE owns the view, - * so we must not also poll/merge). The global `retry: false` means a failed fetch - * leaves `data` undefined, so refetchInterval(undefined run) returns false — a - * failed fetch can never spin a tight loop. - */ -export function useAiChatRunQuery( - chatId: string | undefined, - enabled: boolean, -) { - return useQuery({ - queryKey: AI_CHAT_RUN_RQ_KEY(chatId ?? ""), - queryFn: () => getAiChatRun(chatId as string), - enabled: !!chatId && enabled, - refetchInterval: (query) => runPollInterval(query.state.data?.run), - }); -} - export function useRenameAiChatMutation() { const queryClient = useQueryClient(); const { t } = useTranslation(); diff --git a/apps/client/src/features/ai-chat/queries/ai-chat-run-query.test.tsx b/apps/client/src/features/ai-chat/queries/ai-chat-run-query.test.tsx deleted file mode 100644 index db09683a..00000000 --- a/apps/client/src/features/ai-chat/queries/ai-chat-run-query.test.tsx +++ /dev/null @@ -1,92 +0,0 @@ -import { describe, it, expect, vi, beforeEach } from "vitest"; -import React from "react"; -import { renderHook, waitFor } from "@testing-library/react"; -import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; -import type { IAiChatRunResponse } from "@/features/ai-chat/types/ai-chat.types.ts"; - -// react-i18next is pulled in transitively by ai-chat-query.ts (the mutation hooks -// use it); stub it so the module imports cleanly in this hook test. -vi.mock("react-i18next", () => ({ - useTranslation: () => ({ t: (key: string) => key }), -})); - -vi.mock("@mantine/notifications", () => ({ - notifications: { show: vi.fn() }, -})); - -// Mock the whole service module; only getAiChatRun is exercised here, but the -// other named exports must exist so ai-chat-query.ts imports resolve. -vi.mock("@/features/ai-chat/services/ai-chat-service.ts", () => ({ - getAiChatRun: vi.fn(), - getAiChatMessages: vi.fn(), - getAiChats: vi.fn(), - getAiRoleCatalog: vi.fn(), - getAiRoleCatalogBundle: vi.fn(), - getAiRoles: vi.fn(), - importAiRolesFromCatalog: vi.fn(), - createAiRole: vi.fn(), - deleteAiChat: vi.fn(), - deleteAiRole: vi.fn(), - renameAiChat: vi.fn(), - updateAiRole: vi.fn(), - updateAiRoleFromCatalog: vi.fn(), -})); - -import { getAiChatRun } from "@/features/ai-chat/services/ai-chat-service.ts"; -import { useAiChatRunQuery } from "@/features/ai-chat/queries/ai-chat-query.ts"; - -function createWrapper() { - const queryClient = new QueryClient({ - defaultOptions: { queries: { retry: false } }, - }); - return function Wrapper({ children }: { children: React.ReactNode }) { - return ( - {children} - ); - }; -} - -const runningResponse: IAiChatRunResponse = { - run: { id: "run-1", chatId: "c1", status: "running" }, - message: { - id: "a1", - role: "assistant", - content: "working...", - createdAt: "2026-01-01T00:00:00Z", - }, -}; - -describe("useAiChatRunQuery — enable gating", () => { - beforeEach(() => { - vi.clearAllMocks(); - }); - - it("fetches the run when enabled (passive observer, feature on)", async () => { - vi.mocked(getAiChatRun).mockResolvedValue(runningResponse); - const { result } = renderHook(() => useAiChatRunQuery("c1", true), { - wrapper: createWrapper(), - }); - await waitFor(() => expect(result.current.isSuccess).toBe(true)); - expect(getAiChatRun).toHaveBeenCalledWith("c1"); - expect(result.current.data?.run?.status).toBe("running"); - }); - - it("does NOT fetch when disabled (this tab is the streamer / feature off)", async () => { - vi.mocked(getAiChatRun).mockResolvedValue(runningResponse); - renderHook(() => useAiChatRunQuery("c1", false), { - wrapper: createWrapper(), - }); - // Give any errant fetch a chance to fire, then assert none did. - await new Promise((r) => setTimeout(r, 20)); - expect(getAiChatRun).not.toHaveBeenCalled(); - }); - - it("does NOT fetch when there is no chat id", async () => { - vi.mocked(getAiChatRun).mockResolvedValue(runningResponse); - renderHook(() => useAiChatRunQuery(undefined, true), { - wrapper: createWrapper(), - }); - await new Promise((r) => setTimeout(r, 20)); - expect(getAiChatRun).not.toHaveBeenCalled(); - }); -}); diff --git a/apps/client/src/features/ai-chat/services/ai-chat-service.ts b/apps/client/src/features/ai-chat/services/ai-chat-service.ts index 025e8069..9946248a 100644 --- a/apps/client/src/features/ai-chat/services/ai-chat-service.ts +++ b/apps/client/src/features/ai-chat/services/ai-chat-service.ts @@ -5,7 +5,6 @@ import { IAiChatListParams, IAiChatMessageRow, IAiChatMessagesParams, - IAiChatRunResponse, IAiRole, IAiRoleCatalog, IAiRoleCatalogBundle, @@ -43,23 +42,6 @@ export async function getAiChatMessages( return req.data; } -/** - * Reconnect to the latest agent run of a chat (#184). Returns the run's - * persisted lifecycle state and the assistant message it materializes (the - * partial output while the run is in-flight, the final output once it finished). - * The DB is the source of truth, so this works for an in-flight run (the browser - * dropped, the run kept going) and a finished one alike; `{ run: null }` when the - * chat has never had a run. Owner-gated server-side (the requesting user must own - * the chat); it is NOT flag-gated — when the feature is off the chat simply has no - * runs, so the endpoint returns `{ run: null }`. - */ -export async function getAiChatRun( - chatId: string, -): Promise { - const req = await api.post("/ai-chat/run", { chatId }); - return req.data; -} - /** * Explicitly STOP the active agent run of a chat (#184). This is the ONLY thing * that ends a DETACHED run — a mere browser disconnect (aborting the local SSE) diff --git a/apps/client/src/features/ai-chat/types/ai-chat.types.ts b/apps/client/src/features/ai-chat/types/ai-chat.types.ts index 893e65b4..2c987263 100644 --- a/apps/client/src/features/ai-chat/types/ai-chat.types.ts +++ b/apps/client/src/features/ai-chat/types/ai-chat.types.ts @@ -210,41 +210,14 @@ export interface IAiChatMessageRow { // renders a "stopped" marker on interrupted turns. finishReason?: string; } | null; + // Persisted lifecycle status of the row's turn, carried on the wire by + // `baseFields`. 'streaming' marks a still-in-progress assistant row (used by + // the resume machinery to decide whether a tail is a live stream to attach to + // or a settled row that must not be replayed). + status?: string; createdAt: string; } -/** - * A persisted agent-run row (#184), mirroring the `ai_chat_runs` fields the - * client reads from `POST /ai-chat/run`. Only `status` is load-bearing for the - * reconnect-and-live-update UX (it drives the poll cadence); the rest are carried - * for display/diagnostics. The DB is the source of truth, so this resolves for an - * in-flight run (the browser dropped, the run kept going) and a finished one. - */ -export interface IAiChatRun { - id: string; - chatId: string; - // 'pending' | 'running' | 'succeeded' | 'failed' | 'aborted'. The first two are - // ACTIVE (keep polling); the rest are TERMINAL (stop polling). - status: "pending" | "running" | "succeeded" | "failed" | "aborted" | string; - error?: string | null; - stepCount?: number; - assistantMessageId?: string | null; - startedAt?: string | null; - finishedAt?: string | null; - createdAt?: string; - updatedAt?: string; -} - -/** - * Response of `POST /ai-chat/run` (#184): the latest run of a chat and the - * assistant message it materializes (the partial/final output, projected from the - * persisted rows). Both are `null` when the chat has never had a run. - */ -export interface IAiChatRunResponse { - run: IAiChatRun | null; - message: IAiChatMessageRow | null; -} - export interface IAiChatListParams extends QueryParams {} export interface IAiChatMessagesParams { diff --git a/apps/client/src/features/ai-chat/utils/resume-helpers.test.ts b/apps/client/src/features/ai-chat/utils/resume-helpers.test.ts new file mode 100644 index 00000000..8a4dc7ef --- /dev/null +++ b/apps/client/src/features/ai-chat/utils/resume-helpers.test.ts @@ -0,0 +1,112 @@ +import { describe, it, expect } from "vitest"; +import type { UIMessage } from "@ai-sdk/react"; +import type { IAiChatMessageRow } from "@/features/ai-chat/types/ai-chat.types.ts"; +import { + isStreamingTail, + isSettledAssistantTail, + seedRows, + mergeById, +} from "./resume-helpers.ts"; + +function row( + id: string, + role: string, + status?: string, +): IAiChatMessageRow { + return { id, role, content: "", status, createdAt: "2026-01-01T00:00:00Z" }; +} + +function makeMsg(id: string, text: string): UIMessage { + return { + id, + role: "assistant", + parts: [{ type: "text", text }], + } as UIMessage; +} + +describe("isStreamingTail", () => { + it("is true when the last row is a streaming assistant row", () => { + expect( + isStreamingTail([row("u1", "user"), row("a1", "assistant", "streaming")]), + ).toBe(true); + }); + + it("is false for a settled assistant tail", () => { + expect(isStreamingTail([row("a1", "assistant", "succeeded")])).toBe(false); + expect(isStreamingTail([row("a1", "assistant")])).toBe(false); + }); + + it("is false when the tail is a user row or the list is empty", () => { + expect(isStreamingTail([row("u1", "user")])).toBe(false); + expect(isStreamingTail([])).toBe(false); + }); +}); + +describe("isSettledAssistantTail", () => { + it("is true for an assistant tail whose status is not streaming", () => { + expect(isSettledAssistantTail([row("a1", "assistant", "succeeded")])).toBe( + true, + ); + expect(isSettledAssistantTail([row("a1", "assistant")])).toBe(true); + expect(isSettledAssistantTail([row("a1", "assistant", "aborted")])).toBe( + true, + ); + }); + + it("is false for a streaming assistant tail", () => { + expect(isSettledAssistantTail([row("a1", "assistant", "streaming")])).toBe( + false, + ); + }); + + it("is false when the tail is a user row or the list is empty", () => { + expect(isSettledAssistantTail([row("u1", "user")])).toBe(false); + expect(isSettledAssistantTail([])).toBe(false); + }); +}); + +describe("seedRows", () => { + const rows = [row("u1", "user"), row("a1", "assistant", "streaming")]; + + it("returns the rows unchanged when not stripping", () => { + expect(seedRows(rows, false)).toBe(rows); + }); + + it("drops the last row when stripping", () => { + const seeded = seedRows(rows, true); + expect(seeded).toHaveLength(1); + expect(seeded[0].id).toBe("u1"); + }); + + it("returns an empty list when stripping a single-row list", () => { + expect(seedRows([row("a1", "assistant", "streaming")], true)).toHaveLength( + 0, + ); + }); +}); + +describe("mergeById", () => { + it("replaces the message with the same id in place (per-step growth)", () => { + const prev = [makeMsg("u1", "hi"), makeMsg("a1", "step 1")]; + const incoming = makeMsg("a1", "step 1\nstep 2"); + const next = mergeById(prev, incoming); + expect(next).toHaveLength(2); + expect(next[1]).toBe(incoming); + expect(next[0]).toBe(prev[0]); // untouched + expect(next).not.toBe(prev); // new array (never mutates input) + }); + + it("appends when the incoming message is not yet present", () => { + const prev = [makeMsg("u1", "hi")]; + const incoming = makeMsg("a1", "first token"); + const next = mergeById(prev, incoming); + expect(next).toHaveLength(2); + expect(next[1]).toBe(incoming); + }); + + it("returns the original list unchanged when there is nothing to merge", () => { + const prev = [makeMsg("u1", "hi")]; + expect(mergeById(prev, null)).toBe(prev); + expect(mergeById(prev, undefined)).toBe(prev); + }); +}); diff --git a/apps/client/src/features/ai-chat/utils/resume-helpers.ts b/apps/client/src/features/ai-chat/utils/resume-helpers.ts new file mode 100644 index 00000000..bddb4d01 --- /dev/null +++ b/apps/client/src/features/ai-chat/utils/resume-helpers.ts @@ -0,0 +1,62 @@ +import type { UIMessage } from "@ai-sdk/react"; +import type { IAiChatMessageRow } from "@/features/ai-chat/types/ai-chat.types.ts"; + +/** + * Pure decisions for the resumable-SSE resume machinery (#184 phase 1.5). A tab + * that reopens a chat whose agent run is still going attaches to the server's + * run-stream registry (replay + live tail) instead of polling snapshots; these + * small predicates decide WHICH tail is safe to resume and how to seed the store, + * extracted so they can be unit-tested in isolation. + */ + +/** + * A STREAMING tail: the last persisted row is an assistant row still marked + * `status === 'streaming'`. Such a tail is stripped from the seed and rebuilt by + * the replay (`expect=live`), since the SDK's `text-start` always pushes a new + * part and replaying over a seeded in-progress row would duplicate its text. + */ +export function isStreamingTail(rows: IAiChatMessageRow[]): boolean { + const tail = rows[rows.length - 1]; + return !!tail && tail.role === "assistant" && tail.status === "streaming"; +} + +/** + * A SETTLED assistant tail: the last row is an assistant row whose status is + * anything OTHER than 'streaming'. A settled assistant tail must NEVER resume — + * replaying a finished run into a store that already holds its message duplicates + * parts (`text-start` always pushes a new part). + */ +export function isSettledAssistantTail(rows: IAiChatMessageRow[]): boolean { + const tail = rows[rows.length - 1]; + return !!tail && tail.role === "assistant" && tail.status !== "streaming"; +} + +/** + * Seed rows for `useChat`: return the rows unchanged, or without the last row when + * `strip` is set (the streaming tail is stripped so the live replay rebuilds it + * without duplicating parts). + */ +export function seedRows( + rows: IAiChatMessageRow[], + strip: boolean, +): IAiChatMessageRow[] { + return strip ? rows.slice(0, -1) : rows; +} + +/** + * Merge an assistant message into the rendered list by id: replace the message + * with the same id in place (the in-progress assistant row is already seeded from + * history, so per-step growth replaces it), or append it when absent. Returns a + * new array; the input is never mutated. + */ +export function mergeById( + messages: UIMessage[], + incoming: UIMessage | null | undefined, +): UIMessage[] { + if (!incoming) return messages; + const idx = messages.findIndex((m) => m.id === incoming.id); + if (idx === -1) return [...messages, incoming]; + const next = messages.slice(); + next[idx] = incoming; + return next; +} 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 deleted file mode 100644 index b93f08fb..00000000 --- a/apps/client/src/features/ai-chat/utils/run-polling.test.ts +++ /dev/null @@ -1,303 +0,0 @@ -import { describe, it, expect } from "vitest"; -import type { UIMessage } from "@ai-sdk/react"; -import type { IAiChatRun } from "@/features/ai-chat/types/ai-chat.types.ts"; -import { - RUN_POLL_INTERVAL_MS, - isRunActive, - runPollInterval, - shouldObserveRun, - shouldClearStoppingLatch, - shouldClearLatchOnQueryError, - mergeObservedMessage, -} from "./run-polling.ts"; - -function makeRun(status: string): IAiChatRun { - return { id: "run-1", chatId: "c1", status }; -} - -function makeMsg(id: string, text: string): UIMessage { - return { - id, - role: "assistant", - parts: [{ type: "text", text }], - } as UIMessage; -} - -describe("isRunActive", () => { - it("treats pending and running as active", () => { - expect(isRunActive(makeRun("pending"))).toBe(true); - expect(isRunActive(makeRun("running"))).toBe(true); - }); - - it("treats terminal / unknown / nullish as not active", () => { - expect(isRunActive(makeRun("succeeded"))).toBe(false); - expect(isRunActive(makeRun("failed"))).toBe(false); - expect(isRunActive(makeRun("aborted"))).toBe(false); - expect(isRunActive(makeRun("weird-future-status"))).toBe(false); - expect(isRunActive(null)).toBe(false); - expect(isRunActive(undefined)).toBe(false); - }); -}); - -describe("runPollInterval (the refetchInterval helper)", () => { - it("returns 2000ms while the run is pending/running", () => { - expect(runPollInterval(makeRun("pending"))).toBe(RUN_POLL_INTERVAL_MS); - expect(runPollInterval(makeRun("running"))).toBe(RUN_POLL_INTERVAL_MS); - expect(RUN_POLL_INTERVAL_MS).toBe(2000); - }); - - it("returns false (stop polling) once the run is terminal", () => { - expect(runPollInterval(makeRun("succeeded"))).toBe(false); - expect(runPollInterval(makeRun("failed"))).toBe(false); - expect(runPollInterval(makeRun("aborted"))).toBe(false); - }); - - it("returns false (no polling) when there is no run", () => { - expect(runPollInterval(null)).toBe(false); - expect(runPollInterval(undefined)).toBe(false); - }); -}); - -describe("shouldObserveRun (observer-vs-streamer decision)", () => { - it("observes an active run when this tab is NOT the local streamer", () => { - expect(shouldObserveRun(makeRun("running"), false)).toBe(true); - expect(shouldObserveRun(makeRun("pending"), false)).toBe(true); - }); - - it("observes a terminal run too (so the final output shows on reopen)", () => { - expect(shouldObserveRun(makeRun("succeeded"), false)).toBe(true); - }); - - it("does NOT observe when this tab IS the streamer (no double-render)", () => { - expect(shouldObserveRun(makeRun("running"), true)).toBe(false); - expect(shouldObserveRun(makeRun("succeeded"), true)).toBe(false); - }); - - it("does NOT observe when there is no run", () => { - expect(shouldObserveRun(null, false)).toBe(false); - expect(shouldObserveRun(undefined, false)).toBe(false); - }); -}); - -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("shouldClearLatchOnQueryError (#234 F7 error-safety-net decision)", () => { - // This guards the REAL anti-flash decision the component's run-query-error - // safety-net effect uses (ai-chat-window.tsx wires the effect to THIS helper, - // not a copy — so the test is non-vacuous vs the live code). - - // (b) The F7 hole: a TRANSIENT run-query error while `run` is STILL ACTIVE must - // NOT clear the latch. TanStack Query v5 retains `data` on error, so - // runQueryFailed can be true while the held run is still pending/running. - // Against the PRE-F7 condition (without `!isRunActive(run)`) this would return - // true — so this assertion fails on the buggy code (non-vacuous). - it("does NOT clear on a transient error while the run is still ACTIVE (F7)", () => { - expect( - shouldClearLatchOnQueryError({ - stoppingRun: true, - isLocalStreaming: false, - runQueryFailed: true, - run: makeRun("running"), - }), - ).toBe(false); - expect( - shouldClearLatchOnQueryError({ - stoppingRun: true, - isLocalStreaming: false, - runQueryFailed: true, - run: makeRun("pending"), - }), - ).toBe(false); - }); - - // (a) The genuine permanent-null-freeze: run cache cleared by removeQueries + - // the refetch keeps ERRORING, so `run === null`. This is the ONLY case the - // safety-net exists to cure — it MUST clear so the frozen view resumes. - it("clears on a permanent error when the run is null (permanent-null-freeze)", () => { - expect( - shouldClearLatchOnQueryError({ - stoppingRun: true, - isLocalStreaming: false, - runQueryFailed: true, - run: null, - }), - ).toBe(true); - expect( - shouldClearLatchOnQueryError({ - stoppingRun: true, - isLocalStreaming: false, - runQueryFailed: true, - run: undefined, - }), - ).toBe(true); - }); - - // A TERMINAL run also satisfies `!isRunActive`; clearing then is harmless — the - // terminal effect (shouldClearStoppingLatch) already clears for a terminal run, - // so this only ever agrees with it. Asserted so the (c) reasoning is pinned. - it("clears on an error when the run is terminal (harmless, agrees with terminal effect)", () => { - expect( - shouldClearLatchOnQueryError({ - stoppingRun: true, - isLocalStreaming: false, - runQueryFailed: true, - run: makeRun("aborted"), - }), - ).toBe(true); - }); - - it("does NOT clear without an actual query error", () => { - expect( - shouldClearLatchOnQueryError({ - stoppingRun: true, - isLocalStreaming: false, - runQueryFailed: false, - run: null, - }), - ).toBe(false); - }); - - it("does NOT clear while this tab is the local streamer", () => { - expect( - shouldClearLatchOnQueryError({ - stoppingRun: true, - isLocalStreaming: true, - runQueryFailed: true, - run: null, - }), - ).toBe(false); - }); - - it("does NOT clear when no stop was requested", () => { - expect( - shouldClearLatchOnQueryError({ - stoppingRun: false, - isLocalStreaming: false, - runQueryFailed: true, - run: null, - }), - ).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")]; - const observed = makeMsg("a1", "step 1\nstep 2"); - const next = mergeObservedMessage(prev, observed); - expect(next).toHaveLength(2); - expect(next[1]).toBe(observed); - expect(next[0]).toBe(prev[0]); // untouched - expect(next).not.toBe(prev); // new array (never mutates input) - }); - - it("appends when the observed message is not yet present", () => { - const prev = [makeMsg("u1", "hi")]; - const observed = makeMsg("a1", "first token"); - const next = mergeObservedMessage(prev, observed); - expect(next).toHaveLength(2); - expect(next[1]).toBe(observed); - }); - - it("returns the original list unchanged when there is nothing to merge", () => { - const prev = [makeMsg("u1", "hi")]; - expect(mergeObservedMessage(prev, null)).toBe(prev); - expect(mergeObservedMessage(prev, undefined)).toBe(prev); - }); -}); diff --git a/apps/client/src/features/ai-chat/utils/run-polling.ts b/apps/client/src/features/ai-chat/utils/run-polling.ts deleted file mode 100644 index 560b675f..00000000 --- a/apps/client/src/features/ai-chat/utils/run-polling.ts +++ /dev/null @@ -1,151 +0,0 @@ -import type { UIMessage } from "@ai-sdk/react"; -import type { IAiChatRun } from "@/features/ai-chat/types/ai-chat.types.ts"; - -/** - * Reconnect-and-live-follow helpers (#184). When a chat is reopened while its - * agent run is STILL going, this tab is a PASSIVE OBSERVER: it did not start the - * run here (no local SSE stream), so it catches up by POLLING the reconnect - * endpoint (`POST /ai-chat/run`) and merging the run's incrementally-persisted - * assistant message into the rendered thread. These are the small pure decisions - * that machinery hangs off, extracted so they can be unit-tested in isolation - * (mirrors how reindex polling / editor-sync-state are tested). - */ - -/** How often to re-poll the reconnect endpoint while a run is ACTIVE. */ -export const RUN_POLL_INTERVAL_MS = 2000; - -// 'pending' and 'running' are the two ACTIVE statuses; 'succeeded' | 'failed' | -// 'aborted' are TERMINAL (and any unknown future status is treated as terminal, -// so a stale/odd value never polls forever). -const ACTIVE_STATUSES = new Set(["pending", "running"]); - -/** Whether a run is still going (worth polling / merging live updates from). */ -export function isRunActive(run: IAiChatRun | null | undefined): boolean { - return !!run && ACTIVE_STATUSES.has(run.status); -} - -/** - * The TanStack Query `refetchInterval` value for the run query: poll every - * {@link RUN_POLL_INTERVAL_MS} while the run is active, and `false` (stop) once - * it is terminal or there is no run. Polling is thus naturally bounded by the run - * reaching a terminal status — no separate timeout cap is needed. - */ -export function runPollInterval( - run: IAiChatRun | null | undefined, -): number | false { - return isRunActive(run) ? RUN_POLL_INTERVAL_MS : false; -} - -/** - * Observer-vs-streamer decision. We render the polled run message (catch up + - * keep advancing) ONLY when this tab is a passive observer: there IS a run AND - * this tab is NOT the one locally streaming it (we reconnected, we didn't start - * it here). When this tab is the streamer, the live SSE stream owns the view, so - * we neither poll nor merge — avoiding a double-render fight. Terminal runs still - * merge (so the final persisted output is shown on reopen); the poll itself is - * stopped separately by {@link runPollInterval}. - */ -export function shouldObserveRun( - run: IAiChatRun | null | undefined, - localStreaming: boolean, -): boolean { - 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); -} - -/** - * Should the "stopping" latch be RELEASED by the run-query ERROR safety-net? - * (#234 F7 — a NEW path of the same re-stream flash the F4 latch exists to - * prevent.) After Stop, `handleServerStop` clears the run cache; the terminal - * effect then holds the latch via `if (!run) return` until the CURRENT turn's run - * is fetched fresh. If that refetch instead ERRORS permanently, `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. This - * safety-net cures ONLY that genuine permanent-null-freeze. - * - * All four must hold: - * - `stoppingRun`: we actually requested a stop (otherwise nothing to release); - * - `!isLocalStreaming`: this tab is NOT the local streamer (same reason as - * {@link shouldClearStoppingLatch}); - * - `runQueryFailed`: the run query is in its error state (TanStack Query v5 with - * retry:false — isError); - * - `!isRunActive(run)`: the observed `run` is NOT an active (pending/running) - * held run. This is the F7 gate. In TanStack Query v5 the query's `data` is - * RETAINED on error, so `runQueryFailed` can be true while `run` is STILL an - * ACTIVE run (a single transient GET-run failure in the window between Stop and - * settle). Without this gate a transient error would release the latch early — - * re-opening the observer merge and flashing the growing detached run over the - * frozen row (exactly the F4 flash). Gating on the run NOT being active means we - * only ever cure the permanent-null-freeze (`run === null`, so - * `isRunActive(null)` is false), never release against an active run. - * - * (A terminal `run` also satisfies `!isRunActive(run)`; clearing then is harmless - * — the terminal effect's {@link shouldClearStoppingLatch} already clears the - * latch for a terminal run, so this only ever agrees with it, never conflicts.) - * - * INVARIANT (do not break): clearing the latch on the `run === null` branch is safe - * ONLY because the run query's `refetchInterval` (see {@link runPollInterval}) stops - * polling when the data is empty — so after we clear on null+error there is no - * subsequent auto-poll that could return a still-active detached run and re-open the - * merge. If `refetchInterval` is ever changed to keep polling on `run === null`/on - * error, this null-branch clear would re-open the F7 flash through the null path. - * Do not change the run query's refetchInterval without re-checking this path. - */ -export function shouldClearLatchOnQueryError(args: { - stoppingRun: boolean; - isLocalStreaming: boolean; - runQueryFailed: boolean; - run: IAiChatRun | null | undefined; -}): boolean { - const { stoppingRun, isLocalStreaming, runQueryFailed, run } = args; - return ( - stoppingRun && !isLocalStreaming && runQueryFailed && !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 - * history, so per-step growth replaces it), or append it when absent. Returns a - * new array; the input is never mutated. - */ -export function mergeObservedMessage( - messages: UIMessage[], - observed: UIMessage | null | undefined, -): UIMessage[] { - if (!observed) return messages; - const idx = messages.findIndex((m) => m.id === observed.id); - if (idx === -1) return [...messages, observed]; - const next = messages.slice(); - next[idx] = observed; - return next; -}