From 77e20ecb41df6b5088b005fbe72ff90691fd9230 Mon Sep 17 00:00:00 2001 From: agent_coder Date: Sat, 11 Jul 2026 06:44:10 +0300 Subject: [PATCH] =?UTF-8?q?fix(client):=20=D0=BC=D0=B8=D0=B3=D1=80=D0=B0?= =?UTF-8?q?=D1=86=D0=B8=D1=8F=20chat-thread=20=D0=BD=D0=B0=20FSM=20?= =?UTF-8?q?=E2=80=94=20reconnect=C3=97N,=20stalled,=20supersede=20(#488)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Полная миграция chat-thread.tsx на автомат run-fsm: 13 ref-флагов жизненного цикла resume/reconnect/poll/ownership УБРАНЫ (карта ref'ов в run-fsm.spec.md, колонка pending пуста). Коммиты 3/4/5 приезжают одной атомарной миграцией — три фикса делят одну модель состояний (раздельные коммиты давали бы несобираемые промежуточные состояния, что противоречит смыслу единого автомата). Коммит 3 — повторные циклы reconnect: attached→reconnecting разрешён многократно. Различие «live-follow (лестница reconnect) vs mount-resume» вынесено в ctx-поле liveFollow (НЕ новый ref — это и есть смысл FSM): live-follow-обрыв перезаходит в лестницу (сброс счётчика после успешного re-attach), mount-resume-обрыв уходит в poll. Тест «два обрыва подряд → два цикла». Коммит 4 — (a) polling→stalled по idle-капу (баннер+Retry вместо тихого «вечно-полуготового»); кап переехал в тред (idleCapTimerRef, effect-owned, не флаг), окно теперь тупо поллит по armed-флагу. (b) resume армится ТОЛЬКО при серверном подтверждении активного рана: streaming-tail (статус) или POST /run для user-tail — чат без активного рана больше не порождает ~240 req/10мин. Тесты: stalled-баннер; user-tail с/без активного рана. Коммит 5 (supersede) — удалены SUPERSEDE_RETRY_DELAYS_MS/isRunAlreadyActive/ supersedeRetryRef (клиентская лестница ретраев). «Прервать и отправить» идёт через FSM superseding → POST /stream {supersede:{runId}} (runId из start-метаданных, extractRunId). Транспорт разбирает CAS-исход: ok→SUPERSEDE_READY (новый стрим), 409 MISMATCH→verify через /run, TIMEOUT/INVALID→классифицированная ошибка без авто-ретрая; голый 409 A_RUN_ALREADY_ACTIVE→RUN_ALREADY_ACTIVE. pendingSupersedeRef (send-плумбинг data) — единственная замена трёх удалённых one-shot'ов. Инвариант epoch (I1) гейтит каждый command-outcome (attach/reconnect/supersede/ postRun): устаревшее поколение колбэка отбрасывается редьюсером; DISPOSE на unmount инкрементит epoch. mountedRef оставлен как React-liveness (ортогонален lifecycle). Тесты: FSM 37 переходов; chat-thread 35 (переписан на FSM-переходы); все зелёные. E2E (реальный SSE/reconnect/supersede через редиплой) — на staging QA. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../ai-chat/components/ai-chat-window.tsx | 58 +- .../ai-chat/components/chat-thread.test.tsx | 1027 ++++--------- .../ai-chat/components/chat-thread.tsx | 1358 +++++++---------- .../features/ai-chat/state/run-fsm.spec.md | 75 +- .../features/ai-chat/state/run-fsm.test.ts | 41 +- .../src/features/ai-chat/state/run-fsm.ts | 96 +- 6 files changed, 1035 insertions(+), 1620 deletions(-) diff --git a/apps/client/src/features/ai-chat/components/ai-chat-window.tsx b/apps/client/src/features/ai-chat/components/ai-chat-window.tsx index 5e34fbf3..4b8b07c0 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 @@ -86,19 +86,11 @@ const MIN_HEIGHT = 400; // Margin kept between the window and the viewport edges while dragging. const EDGE_MARGIN = 8; -// #184 phase 1.5 / #430: backstop for 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). -// -// #430: measured from RUN ACTIVITY, not from arm-time. A real autonomous run takes -// 11-25 min — longer than a fixed 10-min-from-start cap, which used to cut the poll -// off mid-run. Instead we cap on INACTIVITY: keep polling as long as the run is -// still making progress (its persisted rows keep changing), and only give up after -// this long with NO new activity. A genuinely stuck run produces no row changes, so -// the idle cap still bounds it; a long-but-progressing run polls to completion. -const DEGRADED_POLL_IDLE_MAX_MS = 10 * 60_000; +// #184 phase 1.5 / #430 / #488: the degraded-poll fallback. The window owns only +// a DUMB 2.5s timer, gated by an armed flag; the THREAD's run-lifecycle FSM owns +// arm/disarm AND the inactivity cap that turns a stuck run into a `stalled` banner +// (#488 commit 4a — the cap moved into the thread so polling->stalled is a single +// FSM transition; the window no longer silently stops polling at the cap). /** Compact token formatter: 1.2M / 3.4k / 950. */ function formatTokens(n: number): string { @@ -259,17 +251,13 @@ 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). + // #184 phase 1.5 / #488: degraded-poll fallback. ChatThread's FSM arms this via + // onResumeFallback(true) when it enters a poll-bearing recovery (attach 204 / + // starved finish / stop) and disarms it on settle / local stream / stalled. The + // window owns ONLY the dumb 2.5s timer; the THREAD owns arm/disarm AND the + // inactivity cap (a stuck run -> the thread's `stalled` banner disarms this). const [degradedPoll, setDegradedPoll] = useState(false); - // #430: timestamp of the LAST run activity while the poll is armed — stamped on - // arm and re-stamped whenever the polled rows change (see the effect below). The - // idle cap is measured from this, so a long-but-progressing run keeps polling. - const lastActivityAtRef = useRef(0); const onResumeFallback = useCallback((active: boolean): void => { - if (active) lastActivityAtRef.current = Date.now(); setDegradedPoll(active); }, []); // Reset the degraded poll whenever the open chat changes: it is scoped to the @@ -281,33 +269,17 @@ export default function AiChatWindow() { const { data: messageRows, isLoading: messagesLoading } = useAiChatMessagesQuery( activeChatId ?? undefined, - // DELIBERATELY DUMB (invariant 8 / task 2.4): poll every 2.5s while armed - // and while the run is still active (#430: under the INACTIVITY cap, not a - // fixed-from-start 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 idle cap is the only backstop. - () => - degradedPoll === true && - Date.now() - lastActivityAtRef.current < DEGRADED_POLL_IDLE_MAX_MS - ? 2500 - : false, + // DELIBERATELY DUMB: poll every 2.5s WHILE ARMED, otherwise off. NO error + // checks (TanStack resets fetchFailureCount each fetch; the poll must survive + // a server restart), NO tail checks, NO cap here — the settled/stalled/idle-cap + // semantics all live in ChatThread's FSM, which disarms via onResumeFallback. + () => (degradedPoll === true ? 2500 : false), // #344: gate on windowOpen too — no message history is fetched (and no // degraded poll runs) while the window is closed; it loads when the window // opens with an active chat. windowOpen, ); - // #430: re-stamp the activity clock whenever the polled rows change while the - // poll is armed. TanStack keeps the same `messageRows` reference across refetches - // that return deep-equal data (structural sharing), so a new reference means the - // run genuinely progressed — which extends the inactivity cap above. A stuck run - // yields no reference change, so the cap eventually fires and stops the poll. - useEffect(() => { - if (degradedPoll) lastActivityAtRef.current = Date.now(); - }, [degradedPoll, messageRows]); - // #184 reconnect-and-live-follow. Whether detached agent runs are enabled for // 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 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 c6b8b65a..9aaaed60 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 @@ -15,12 +15,12 @@ import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; const h = vi.hoisted(() => ({ state: { status: "streaming" as string, + messages: [] as unknown[], onFinish: null as null | ((arg: Record) => void), 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: { @@ -33,11 +33,17 @@ const h = vi.hoisted(() => ({ init?: { method?: string; body?: unknown }, ) => Promise; }, + // getRun mock (POST /ai-chat/run run-fact). Default: no active run. + getRun: vi.fn( + async (_chatId?: string) => + ({ run: null, message: null }) as { + run: { id: string; status: string } | null; + message: unknown; + }, + ), }, })); -// Mock useChat: capture onFinish + seeded messages, return the spies and the -// controllable status. vi.mock("@ai-sdk/react", () => ({ useChat: (opts: { messages?: unknown[]; @@ -46,7 +52,7 @@ vi.mock("@ai-sdk/react", () => ({ h.state.onFinish = opts.onFinish ?? null; h.state.seededMessages = opts.messages ?? null; return { - messages: [], + messages: h.state.messages, sendMessage: h.state.sendMessage, status: h.state.status, stop: h.state.stop, @@ -57,8 +63,6 @@ vi.mock("@ai-sdk/react", () => ({ }, })); -// Mock "ai": deterministic ids + a transport that records its options so the test -// can invoke prepareSendMessagesRequest / prepareReconnectToStreamRequest / fetch. vi.mock("ai", () => { let counter = 0; return { @@ -71,15 +75,16 @@ vi.mock("ai", () => { }; }); -// 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". +// The run-fact service (POST /ai-chat/run). Mocked so a user-tail mount and the +// supersede verify do not hit axios. +vi.mock("@/features/ai-chat/services/ai-chat-service.ts", () => ({ + getRun: (chatId: string) => h.state.getRun(chatId), +})); + vi.mock("@/features/ai-chat/components/message-list.tsx", () => ({ default: () =>
, })); @@ -87,14 +92,19 @@ vi.mock("@/features/ai-chat/components/chat-input.tsx", () => ({ default: ({ onQueue, onStop, + onSend, }: { onQueue: (text: string) => void; onStop: () => void; + onSend: (text: string) => void; }) => ( <> + @@ -145,6 +155,7 @@ function renderThread(props?: { function resetState() { h.state.status = "streaming"; + h.state.messages = []; h.state.onFinish = null; h.state.seededMessages = null; h.state.transport = null; @@ -152,70 +163,120 @@ function resetState() { h.state.stop.mockClear(); h.state.setMessages.mockClear(); h.state.resumeStream.mockClear(); + h.state.getRun.mockReset(); + h.state.getRun.mockResolvedValue({ run: null, message: null }); } -describe("ChatThread — send now (#198)", () => { +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")]; + +// ----------------------------------------------------------------------------- +// Send now (#198) — local send + #488 commit 5 CAS supersede. +// ----------------------------------------------------------------------------- +describe("ChatThread — send now", () => { beforeEach(resetState); + afterEach(cleanup); - it("aborts the current turn and resends the queued message on the abort", () => { - renderThread(); - - fireEvent.click(screen.getByTestId("queue-btn")); - const sendNowBtn = screen.getByLabelText("Send now"); - expect(sendNowBtn).toBeTruthy(); - - fireEvent.click(sendNowBtn); - expect(h.state.stop).toHaveBeenCalledTimes(1); - expect(h.state.sendMessage).not.toHaveBeenCalled(); - - act(() => { - h.state.onFinish?.({ - message: { id: "a", role: "assistant", parts: [] }, - isAbort: true, - isDisconnect: false, - isError: false, - }); - }); - expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); - }); - - it("tags exactly the next send as interrupted (one-shot flag)", () => { - renderThread(); - fireEvent.click(screen.getByTestId("queue-btn")); - fireEvent.click(screen.getByLabelText("Send now")); - - const prep = h.state.transport!.prepareSendMessagesRequest!; - expect(prep({ messages: [], body: {} }).body.interrupted).toBe(true); - expect(prep({ messages: [], body: {} }).body.interrupted).toBe(false); - }); - - it("sends immediately without an interrupt when not streaming", () => { + it("sends immediately (no interrupt) when not streaming", () => { h.state.status = "ready"; - renderThread(); - + renderThread({ initialRows: settledTail() }); fireEvent.click(screen.getByTestId("queue-btn")); fireEvent.click(screen.getByLabelText("Send now")); - expect(h.state.stop).not.toHaveBeenCalled(); expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); + }); + + it("#488 commit 5: autonomous live sendNow with a known runId POSTs a supersede body", () => { + // A streaming assistant message carrying the start-metadata runId -> the FSM + // adopts the run-fact; sendNow then interrupts via the server CAS. + h.state.status = "streaming"; + h.state.messages = [ + { + id: "a1", + role: "assistant", + parts: [{ type: "text", text: "x" }], + metadata: { runId: "run-1" }, + }, + ]; + renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); + fireEvent.click(screen.getByTestId("queue-btn")); + fireEvent.click(screen.getByLabelText("Send now")); + // The message is sent... + expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); + // ...and the NEXT POST carries supersede:{runId} (read-and-cleared once). const prep = h.state.transport!.prepareSendMessagesRequest!; - expect(prep({ messages: [], body: {} }).body.interrupted).toBe(false); + const body = prep({ messages: [], body: {} }).body as Record; + expect(body.supersede).toEqual({ runId: "run-1" }); + // One-shot: a subsequent request has no supersede. + expect(prep({ messages: [], body: {} }).body.supersede).toBeUndefined(); + }); + + it("#488 commit 5: a supersede POST that 409s SUPERSEDE_TIMEOUT is returned as-is (NO retry ladder)", async () => { + h.state.status = "streaming"; + h.state.messages = [ + { + id: "a1", + role: "assistant", + parts: [{ type: "text", text: "x" }], + metadata: { runId: "run-1" }, + }, + ]; + renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); + fireEvent.click(screen.getByTestId("queue-btn")); + fireEvent.click(screen.getByLabelText("Send now")); // -> superseding, arms body + // Consume the supersede body (as the SDK would) so the POST is the CAS one. + h.state.transport!.prepareSendMessagesRequest!({ messages: [], body: {} }); + + const fetchMock = vi.fn().mockResolvedValue( + new Response(JSON.stringify({ code: "SUPERSEDE_TIMEOUT" }), { + status: 409, + }), + ); + vi.stubGlobal("fetch", fetchMock); + let res!: Response; + await act(async () => { + res = (await h.state.transport!.fetch!("http://x", { + method: "POST", + body: "{}", + })) as Response; + }); + // Exactly ONE fetch (the old bounded 409 retry ladder is gone). + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(res.status).toBe(409); + }); + + it("Send now is HIDDEN while observing a resumed run and VISIBLE on a local stream", () => { + // Resumed (mount attach) -> observer -> hidden. + renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() }); + fireEvent.click(screen.getByTestId("queue-btn")); + expect(screen.queryByLabelText("Send now")).toBeNull(); + // Local stream (no resume) -> visible. + cleanup(); + resetState(); + renderThread({ initialRows: [] }); + fireEvent.click(screen.getByTestId("queue-btn")); + expect(screen.getByLabelText("Send now")).toBeTruthy(); }); }); -// #486: the final onFinish -> flushNext() must be gated on the live-mount flag. -// A clean onFinish can land AFTER the thread unmounts (New-chat / chat-switch -// mid-stream — the async attach/resume settles late); flushing then dequeues and -// re-POSTs a queued message from an abandoned thread (a "ghost" send). +// ----------------------------------------------------------------------------- +// onFinish flush gated on mount (#486). +// ----------------------------------------------------------------------------- describe("ChatThread — onFinish flush gated on mount (#486)", () => { beforeEach(resetState); afterEach(cleanup); - it("a clean onFinish WHILE MOUNTED flushes the queued message (control)", () => { + it("a clean onFinish WHILE MOUNTED flushes the queued message", () => { renderThread(); - fireEvent.click(screen.getByTestId("queue-btn")); // enqueue "queued text" + fireEvent.click(screen.getByTestId("queue-btn")); expect(h.state.sendMessage).not.toHaveBeenCalled(); - act(() => { h.state.onFinish?.({ message: { id: "a", role: "assistant", parts: [] }, @@ -224,18 +285,14 @@ describe("ChatThread — onFinish flush gated on mount (#486)", () => { isError: false, }); }); - // Mounted: the queue flushes normally. expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); }); it("a clean onFinish AFTER unmount does NOT flush (no ghost send)", () => { const { unmount } = renderThread(); - fireEvent.click(screen.getByTestId("queue-btn")); // enqueue "queued text" + fireEvent.click(screen.getByTestId("queue-btn")); h.state.sendMessage.mockClear(); - - // Chat switched away mid-stream: the streamer unmounts... unmount(); - // ...and a late, clean onFinish lands on the abandoned thread. act(() => { h.state.onFinish?.({ message: { id: "a", role: "assistant", parts: [] }, @@ -244,252 +301,80 @@ describe("ChatThread — onFinish flush gated on mount (#486)", () => { isError: false, }); }); - // Gated on mountedRef: NOTHING is sent from the dead thread. expect(h.state.sendMessage).not.toHaveBeenCalled(); }); }); -// #396: in autonomous mode a live sendNow must additionally request the -// AUTHORITATIVE server stop of the detached run (a local abort is only a client -// disconnect the server ignores) and arm a bounded 409 retry so the re-POST -// converges once the one-active-run slot frees. Legacy mode is unchanged. -describe("ChatThread — send now server-stop + supersede retry (#396)", () => { +// ----------------------------------------------------------------------------- +// Turn-end decision (onFinish). +// ----------------------------------------------------------------------------- +describe("ChatThread — turn-end decision (onFinish)", () => { beforeEach(resetState); afterEach(cleanup); - // A settled assistant tail => no mount resume (attemptResumeRef false), so the - // "Send now" button is visible for the NEW local streaming turn while - // autonomous runs are enabled. - const settledTail = () => [ - row("u1", "user", undefined, "hi"), - row("a1", "assistant", "succeeded", "done"), - ]; - - it("autonomous: sendNow during a live stream calls onServerStop with the chat id", () => { - const { onServerStop } = renderThread({ - autonomousRunsEnabled: true, - initialRows: settledTail(), - }); + function finishWith(flags: { + isAbort?: boolean; + isDisconnect?: boolean; + isError?: boolean; + }) { + const { onTurnFinished } = renderThread(); fireEvent.click(screen.getByTestId("queue-btn")); - fireEvent.click(screen.getByLabelText("Send now")); - - expect(h.state.stop).toHaveBeenCalledTimes(1); - expect(onServerStop).toHaveBeenCalledWith("c1"); - }); - - it("legacy (autonomous off): sendNow does NOT call onServerStop and does NOT retry the send", async () => { - const { onServerStop } = renderThread({ - autonomousRunsEnabled: false, - initialRows: settledTail(), - }); - fireEvent.click(screen.getByTestId("queue-btn")); - fireEvent.click(screen.getByLabelText("Send now")); - expect(onServerStop).not.toHaveBeenCalled(); - - // The supersede retry must NOT be armed: a POST that 409s is returned as-is - // (single fetch, no retry). - const fetchMock = vi - .fn() - .mockResolvedValue( - new Response(JSON.stringify({ code: "A_RUN_ALREADY_ACTIVE" }), { - status: 409, - }), - ); - vi.stubGlobal("fetch", fetchMock); - let res!: Response; - await act(async () => { - res = (await h.state.transport!.fetch!("http://x", { - method: "POST", - body: "{}", - })) as Response; - }); - expect(fetchMock).toHaveBeenCalledTimes(1); - expect(res.status).toBe(409); - }); - - it("armed supersede send retries 409 A_RUN_ALREADY_ACTIVE and succeeds once the slot frees", async () => { - renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); - // Arm the retry by performing a live sendNow (autonomous branch sets the ref). - fireEvent.click(screen.getByTestId("queue-btn")); - fireEvent.click(screen.getByLabelText("Send now")); - - const fetchMock = vi - .fn() - // First POST: the old detached run still holds the slot -> 409. - .mockResolvedValueOnce( - new Response(JSON.stringify({ code: "A_RUN_ALREADY_ACTIVE" }), { - status: 409, - }), - ) - // Retry: the server stop settled the old run -> 200. - .mockResolvedValueOnce(new Response("ok", { status: 200 })); - vi.stubGlobal("fetch", fetchMock); - - let res!: Response; - await act(async () => { - res = (await h.state.transport!.fetch!("http://x", { - method: "POST", - body: "{}", - })) as Response; - }); - expect(fetchMock).toHaveBeenCalledTimes(2); - expect(res.status).toBe(200); - }); - - it("supersede retry is one-shot: a later send (ref cleared) does NOT retry a 409", async () => { - renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); - fireEvent.click(screen.getByTestId("queue-btn")); - fireEvent.click(screen.getByLabelText("Send now")); // arms the one-shot - - // First armed send: immediately succeeds, consuming the arm. - let fetchMock = vi - .fn() - .mockResolvedValue(new Response("ok", { status: 200 })); - vi.stubGlobal("fetch", fetchMock); - await act(async () => { - await h.state.transport!.fetch!("http://x", { - method: "POST", - body: "{}", - }); - }); - expect(fetchMock).toHaveBeenCalledTimes(1); - - // A subsequent send is NOT armed -> a 409 is returned as-is (no retry). - fetchMock = vi.fn().mockResolvedValue( - new Response(JSON.stringify({ code: "A_RUN_ALREADY_ACTIVE" }), { - status: 409, - }), - ); - vi.stubGlobal("fetch", fetchMock); - let res!: Response; - await act(async () => { - res = (await h.state.transport!.fetch!("http://x", { - method: "POST", - body: "{}", - })) as Response; - }); - expect(fetchMock).toHaveBeenCalledTimes(1); - expect(res.status).toBe(409); - }); - - it("supersede retry is bounded: exhaustion surfaces the 409 error", async () => { - renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); - fireEvent.click(screen.getByTestId("queue-btn")); - fireEvent.click(screen.getByLabelText("Send now")); - - // Every attempt 409s -> after 4 attempts the last 409 surfaces. - const fetchMock = vi.fn().mockResolvedValue( - new Response(JSON.stringify({ code: "A_RUN_ALREADY_ACTIVE" }), { - status: 409, - }), - ); - vi.stubGlobal("fetch", fetchMock); - let res!: Response; - await act(async () => { - res = (await h.state.transport!.fetch!("http://x", { - method: "POST", - body: "{}", - })) as Response; - }); - // 4 attempts total (1 immediate + 3 backoff retries), then give up. - expect(fetchMock).toHaveBeenCalledTimes(4); - expect(res.status).toBe(409); - }); - - it("armed supersede send does NOT retry a non-409 status", async () => { - renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); - fireEvent.click(screen.getByTestId("queue-btn")); - fireEvent.click(screen.getByLabelText("Send now")); - const fetchMock = vi - .fn() - .mockResolvedValue(new Response("boom", { status: 500 })); - vi.stubGlobal("fetch", fetchMock); - let res!: Response; - await act(async () => { - res = (await h.state.transport!.fetch!("http://x", { - method: "POST", - body: "{}", - })) as Response; - }); - expect(fetchMock).toHaveBeenCalledTimes(1); - expect(res.status).toBe(500); - }); - - // Strand-path regression: sendNow arms the supersede retry, but if the promoted - // head is removed before the abort's onFinish lands, flushNext() sends nothing - // (returns false) and NO re-POST consumes the arm. The arm must be disarmed on - // that no-send branch so the NEXT unrelated NORMAL send does not inherit it and - // silently retry a genuine 409 (e.g. a legitimate two-tab conflict) 4x instead - // of surfacing it immediately. - it("strand-path: a stranded supersede arm (flushNext no-send) does NOT retry a later normal 409", async () => { - renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); - // Arm the retry via a live autonomous sendNow (promotes the head + arms). - fireEvent.click(screen.getByTestId("queue-btn")); - fireEvent.click(screen.getByLabelText("Send now")); - - // Remove the promoted head BEFORE the abort lands, so flushNext() returns - // false (no POST) and the arm would strand without the disarm fix. - fireEvent.click(screen.getByLabelText("Remove queued message")); - - // The abort's onFinish now takes the flushOnAbortRef branch, calls flushNext() - // which finds an empty queue and returns false -> the no-send disarm must run. act(() => { h.state.onFinish?.({ - message: { id: "a1", role: "assistant", parts: [] }, - isAbort: true, + message: { id: "a", role: "assistant", parts: [] }, + isAbort: false, isDisconnect: false, isError: false, + ...flags, }); }); - // No re-POST was sent (nothing to flush). - expect(h.state.sendMessage).not.toHaveBeenCalled(); + return { onTurnFinished }; + } - // A subsequent NORMAL send that 409s must be returned as-is (exactly 1 fetch): - // the stranded arm must NOT cause the genuine 409 to be retried. - const fetchMock = vi - .fn() - .mockResolvedValue( - new Response(JSON.stringify({ code: "A_RUN_ALREADY_ACTIVE" }), { - status: 409, - }), - ); - vi.stubGlobal("fetch", fetchMock); - let res!: Response; - await act(async () => { - res = (await h.state.transport!.fetch!("http://x", { - method: "POST", - body: "{}", - })) as Response; - }); - expect(fetchMock).toHaveBeenCalledTimes(1); - expect(res.status).toBe(409); + it("CONTINUES — flushes the next queued message on a clean finish", () => { + finishWith({}); + expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); + expect(screen.queryByText("Response stopped.")).toBeNull(); }); - it("armed supersede send does NOT retry a 409 with a different (non-A_RUN_ALREADY_ACTIVE) body", async () => { - renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); - fireEvent.click(screen.getByTestId("queue-btn")); - fireEvent.click(screen.getByLabelText("Send now")); - const fetchMock = vi.fn().mockResolvedValue( - new Response(JSON.stringify({ code: "SOMETHING_ELSE" }), { - status: 409, - }), - ); - vi.stubGlobal("fetch", fetchMock); - let res!: Response; - await act(async () => { - res = (await h.state.transport!.fetch!("http://x", { - method: "POST", - body: "{}", - })) as Response; - }); - expect(fetchMock).toHaveBeenCalledTimes(1); - expect(res.status).toBe(409); + it("ENDS — keeps the queue on a user abort and shows the stopped notice", () => { + finishWith({ isAbort: true }); + expect(h.state.sendMessage).not.toHaveBeenCalled(); + expect(screen.getByText("Response stopped.")).toBeTruthy(); + }); + + it("ENDS — keeps the queue on a disconnect (non-autonomous) and shows the notice", () => { + finishWith({ isDisconnect: true }); + expect(h.state.sendMessage).not.toHaveBeenCalled(); + expect( + screen.getByText("Connection lost — the answer was interrupted."), + ).toBeTruthy(); + }); + + it("ENDS — keeps the queue on a stream error (no notice)", () => { + finishWith({ isError: true }); + expect(h.state.sendMessage).not.toHaveBeenCalled(); + expect(screen.queryByText("Response stopped.")).toBeNull(); + }); + + it("notifies the parent on EVERY terminal outcome", () => { + for (const flags of [ + {}, + { isAbort: true }, + { isDisconnect: true }, + { isError: true }, + ]) { + cleanup(); + resetState(); + const { onTurnFinished } = finishWith(flags); + expect(onTurnFinished).toHaveBeenCalled(); + } }); }); -// #388: the editor selection is snapshotted at send time and nested inside -// openPage on the wire. The getter is read live from a ref, so each send ships a -// fresh snapshot. +// ----------------------------------------------------------------------------- +// editor selection wiring (#388). +// ----------------------------------------------------------------------------- describe("ChatThread — editor selection wiring (#388)", () => { beforeEach(resetState); afterEach(cleanup); @@ -518,7 +403,7 @@ describe("ChatThread — editor selection wiring (#388)", () => { ); } - it("nests the snapshot from the getter into openPage.selection at send time", () => { + it("nests the snapshot into openPage.selection at send time", () => { const selection = { text: "fix this", blockIds: ["b1"], before: "a " }; renderWithSelection({ openPage: { id: "p1", title: "Doc" }, @@ -545,131 +430,36 @@ describe("ChatThread — editor selection wiring (#388)", () => { expect(openPage).toEqual({ id: "p1", title: "Doc", selection: null }); }); - it("does not send selection at all on a non-page route (openPage null)", () => { + it("does not consult the getter on a non-page route (openPage null)", () => { const getter = vi.fn(() => ({ text: "sel" })); renderWithSelection({ openPage: null, getEditorSelection: getter }); const prep = h.state.transport!.prepareSendMessagesRequest!; expect(prep({ messages: [], body: {} }).body.openPage).toBeNull(); - // The getter must not even be consulted when there is no page. expect(getter).not.toHaveBeenCalled(); }); }); -describe("ChatThread — turn-end decision (onFinish)", () => { +// ----------------------------------------------------------------------------- +// Resume (attach) machinery (#184) + #488 commit 4b run-fact gating. +// ----------------------------------------------------------------------------- +describe("ChatThread — resume (attach) machinery", () => { beforeEach(resetState); - - function finishWith(flags: { - isAbort?: boolean; - isDisconnect?: boolean; - isError?: boolean; - }) { - cleanup(); - h.state.sendMessage.mockClear(); - const { onTurnFinished } = renderThread(); - fireEvent.click(screen.getByTestId("queue-btn")); - act(() => { - h.state.onFinish?.({ - message: { id: "a", role: "assistant", parts: [] }, - isAbort: false, - isDisconnect: false, - isError: false, - ...flags, - }); - }); - return { onTurnFinished }; - } - - it("CONTINUES — flushes the next queued message on a clean finish", () => { - finishWith({}); - expect(h.state.sendMessage).toHaveBeenCalledWith({ text: "queued text" }); - expect(screen.queryByText("Response stopped.")).toBeNull(); - }); - - it("ENDS — keeps the queue intact on a user abort and shows the stopped notice", () => { - finishWith({ isAbort: true }); - expect(h.state.sendMessage).not.toHaveBeenCalled(); - expect(screen.getByText("Response stopped.")).toBeTruthy(); - }); - - it("ENDS — keeps the queue intact on a disconnect and shows the connection-lost notice", () => { - finishWith({ isDisconnect: true }); - expect(h.state.sendMessage).not.toHaveBeenCalled(); - expect( - screen.getByText("Connection lost — the answer was interrupted."), - ).toBeTruthy(); - }); - - it("ENDS — keeps the queue intact on a stream error (no auto-retry, no stopped notice)", () => { - finishWith({ isError: true }); - expect(h.state.sendMessage).not.toHaveBeenCalled(); - expect(screen.queryByText("Response stopped.")).toBeNull(); - }); - - it("notifies the parent on EVERY terminal outcome", () => { - for (const flags of [ - {}, - { isAbort: true }, - { isDisconnect: true }, - { isError: true }, - ]) { - const { onTurnFinished } = finishWith(flags); - expect(onTurnFinished).toHaveBeenCalled(); - } - }); -}); - -// #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 + it("resumes on mount for a STREAMING tail (the streaming status is the run-fact)", () => { 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(); + it("does NOT resume for a settled tail, flag off, or no chatId", () => { renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); expect(h.state.resumeStream).not.toHaveBeenCalled(); - - // flag off -> NO resume cleanup(); - h.state.resumeStream.mockClear(); + resetState(); renderThread({ autonomousRunsEnabled: false, initialRows: streamingTail() }); expect(h.state.resumeStream).not.toHaveBeenCalled(); - - // no chatId -> NO resume cleanup(); - h.state.resumeStream.mockClear(); + resetState(); renderThread({ autonomousRunsEnabled: true, chatId: null, @@ -678,61 +468,87 @@ describe("ChatThread — resume (attach) machinery (#184)", () => { expect(h.state.resumeStream).not.toHaveBeenCalled(); }); - 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); - - cleanup(); + it("#488 commit 4b: a USER tail resumes ONLY when POST /run confirms an active run", async () => { + h.state.getRun.mockResolvedValue({ + run: { id: "run-1", status: "running" }, + message: null, + }); + renderThread({ autonomousRunsEnabled: true, initialRows: userTail() }); + await act(async () => { + await Promise.resolve(); + }); + expect(h.state.getRun).toHaveBeenCalledWith("c1"); + expect(h.state.resumeStream).toHaveBeenCalledTimes(1); + }); + + it("#488 commit 4b: a USER tail with NO active run does NOT resume and does NOT arm the poll", async () => { + h.state.getRun.mockResolvedValue({ run: null, message: null }); + const { onResumeFallback } = renderThread({ + autonomousRunsEnabled: true, + initialRows: userTail(), + }); + await act(async () => { + await Promise.resolve(); + }); + expect(h.state.resumeStream).not.toHaveBeenCalled(); + expect(onResumeFallback).not.toHaveBeenCalledWith(true); + }); + + it("strips the streaming tail from the seed, keeps a user tail whole", () => { + renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() }); + expect(h.state.seededMessages).toHaveLength(1); + cleanup(); + resetState(); 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", () => { + it("builds the attach URL with expect=live&anchor only for a stripped streaming tail", () => { renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail() }); expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe( "/api/ai-chat/runs/c1/stream?expect=live&anchor=a1", ); - cleanup(); + resetState(); renderThread({ autonomousRunsEnabled: true, initialRows: userTail() }); expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe( "/api/ai-chat/runs/c1/stream", ); }); - async function fetch204() { + async function attachFetch(response: unknown, reject = false) { vi.stubGlobal( "fetch", - vi.fn().mockResolvedValue({ status: 204, ok: false }), + reject + ? vi.fn().mockRejectedValue(response) + : vi.fn().mockResolvedValue(response), ); await act(async () => { - await h.state.transport!.fetch!("http://x", { method: "GET" }); + await h.state + .transport!.fetch!("http://x", { method: "GET" }) + .catch(() => undefined); }); } - 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. + await attachFetch({ status: 204, ok: false }); + expect(h.state.setMessages).toHaveBeenCalledTimes(1); // restore + expect(invalidateSpy).toHaveBeenCalledWith({ + queryKey: ["ai-chat-messages", "c1"], + }); + expect(onResumeFallback).toHaveBeenCalledWith(true); + }); + + it("F7 restart-survival: a 500 attach failure restores the row AND arms the poll", async () => { + const { onResumeFallback, invalidateSpy } = renderThread({ + autonomousRunsEnabled: true, + initialRows: streamingTail(), + }); + await attachFetch({ status: 500, ok: false }); expect(h.state.setMessages).toHaveBeenCalledTimes(1); expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ["ai-chat-messages", "c1"], @@ -740,39 +556,12 @@ describe("ChatThread — resume (attach) machinery (#184)", () => { expect(onResumeFallback).toHaveBeenCalledWith(true); }); - it("F7 restart-survival: a 500 attach failure restores the stripped row AND arms the poll (not lost)", async () => { + it("F7 restart-survival: a network throw restores the row AND arms the poll", 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 - }); + await attachFetch(new Error("network down"), true); expect(h.state.setMessages).toHaveBeenCalledTimes(1); expect(invalidateSpy).toHaveBeenCalledWith({ queryKey: ["ai-chat-messages", "c1"], @@ -780,7 +569,7 @@ describe("ChatThread — resume (attach) machinery (#184)", () => { expect(onResumeFallback).toHaveBeenCalledWith(true); }); - it("unmount during a pending attach aborts the controller and gates late callbacks", async () => { + it("unmount during a pending attach aborts the controller and gates late callbacks (epoch I1)", async () => { const { onResumeFallback, invalidateSpy, unmount } = renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail(), @@ -798,55 +587,33 @@ describe("ChatThread — resume (attach) machinery (#184)", () => { }); }), ); - // 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(); + unmount(); // DISPOSE aborts the attach + bumps the epoch 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; }); + // The stale (superseded-epoch) outcome is dropped. 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", () => { + it("a resumed (observer) 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, + message: { + id: "a1", + role: "assistant", + parts: [{ type: "text", text: "streamed answer" }], + }, isAbort: false, isDisconnect: false, isError: false, @@ -855,7 +622,7 @@ describe("ChatThread — resume (attach) machinery (#184)", () => { expect(h.state.sendMessage).not.toHaveBeenCalled(); }); - it("a healthy resumed finish (visible content) arms nothing and keeps the store", () => { + it("an empty resumed message (starved replay) restores the row AND arms the poll", () => { h.state.status = "ready"; const { onResumeFallback } = renderThread({ autonomousRunsEnabled: true, @@ -865,53 +632,7 @@ describe("ChatThread — resume (attach) machinery (#184)", () => { 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, + message: { id: "a1", role: "assistant", parts: [] }, isAbort: false, isDisconnect: false, isError: false, @@ -921,67 +642,11 @@ describe("ChatThread — resume (attach) machinery (#184)", () => { 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 () => { + it("handleStop aborts the attach controller and calls onServerStop", () => { const { onServerStop } = renderThread({ autonomousRunsEnabled: true, initialRows: streamingTail(), }); - // Establish an attach controller via a (pending) reconnect GET. let abortSeen = false; vi.stubGlobal( "fetch", @@ -989,7 +654,7 @@ describe("ChatThread — resume (attach) machinery (#184)", () => { init.signal?.addEventListener("abort", () => { abortSeen = true; }); - return new Promise(() => undefined); // never resolves + return new Promise(() => undefined); }), ); act(() => { @@ -1001,45 +666,10 @@ describe("ChatThread — resume (attach) machinery (#184)", () => { }); }); -// 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[] }) => ( - - - - - - ); - const view = render(); - const rerender = (rows: IAiChatMessageRow[]) => - act(() => view.rerender()); - return { rerender, onResumeFallback }; -} - -// #430: auto-reconnect to a DETACHED run after a LIVE SSE disconnect. The mount -// path only resumes on mount/reload; these cover the missing trigger — a live -// `isDisconnect` on onFinish must (backoff-)re-attach WITHOUT a reload, pin+strip -// the live row to avoid duplicates, fall back to the degraded poll on a 204, and -// exhaust to a manual Retry. -describe("ChatThread — live reconnect after isDisconnect (#430)", () => { - // A LIVE local turn that just dropped: the settled tail existed before, and the - // partial assistant row lives only in `messages` (not persisted as a tail). - const settledTail = () => [ - row("u1", "user", undefined, "hi"), - row("a1", "assistant", "succeeded", "done"), - ]; - // The partial assistant message onFinish hands us for the dropped LIVE turn. +// ----------------------------------------------------------------------------- +// Live reconnect (#430) + #488 commits 2/3 + stalled (4a). Fake timers. +// ----------------------------------------------------------------------------- +describe("ChatThread — live reconnect + stalled", () => { const liveMsg = { id: "a2", role: "assistant", @@ -1048,8 +678,6 @@ describe("ChatThread — live reconnect after isDisconnect (#430)", () => { beforeEach(() => { resetState(); - // status "ready": with a live disconnect the mock is not streaming, so the - // status==="streaming" auto-clear effect stays out of the way. h.state.status = "ready"; vi.useFakeTimers(); }); @@ -1058,144 +686,92 @@ describe("ChatThread — live reconnect after isDisconnect (#430)", () => { cleanup(); }); - // Render a NON-resuming mount (settled tail -> no mount resume) with autonomous - // runs on, then simulate a live disconnect via onFinish. - function renderLiveThenDisconnect() { - const view = renderThread({ - autonomousRunsEnabled: true, - initialRows: settledTail(), - }); - // The settled tail must NOT have triggered a mount resume. - expect(h.state.resumeStream).not.toHaveBeenCalled(); + function disconnect(message: unknown = liveMsg) { act(() => { h.state.onFinish?.({ - message: liveMsg, + message, isAbort: false, isDisconnect: true, isError: false, }); }); + } + function renderLive() { + const view = renderThread({ + autonomousRunsEnabled: true, + initialRows: settledTail(), + }); + expect(h.state.resumeStream).not.toHaveBeenCalled(); return view; } - - // Fire the pending (scheduled) attempt for `attempt` (backoff = 1s,2s,4s,...). function advanceToAttempt(attempt: number) { act(() => { vi.advanceTimersByTime(1000 * 2 ** (attempt - 1)); }); } - - // Simulate the reconnect GET returning 204 (nothing live) so the transport's - // no-active-stream recovery runs. - async function reconnect204() { - vi.stubGlobal( - "fetch", - vi.fn().mockResolvedValue({ status: 204, ok: false }), - ); + async function reconnect(response: unknown) { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(response)); await act(async () => { await h.state.transport!.fetch!("http://x", { method: "GET" }); }); } - // Simulate the reconnect GET returning a live 2xx stream. - async function reconnect200() { - vi.stubGlobal( - "fetch", - vi.fn().mockResolvedValue({ status: 200, ok: true }), - ); - await act(async () => { - await h.state.transport!.fetch!("http://x", { method: "GET" }); - }); - } - - it("calls resumeStream POST-mount (a live disconnect triggers a backoff reconnect)", () => { - renderLiveThenDisconnect(); - // The banner shows immediately; the attach itself fires after the first backoff. + it("a live disconnect starts a backoff reconnect (banner + resumeStream after backoff)", () => { + renderLive(); + disconnect(); expect(screen.getByText(/reconnecting/i)).toBeTruthy(); expect(h.state.resumeStream).not.toHaveBeenCalled(); advanceToAttempt(1); - // resumeStream is now called AFTER mount — the bug was it only ever fired once - // on mount. The reconnect URL pins expect=live&anchor to OUR run. expect(h.state.resumeStream).toHaveBeenCalledTimes(1); expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe( "/api/ai-chat/runs/c1/stream?expect=live&anchor=a2", ); }); - // #488 commit 2: a break during the SETUP phase — before the first assistant - // frame (onFinish carries no assistant message) — must STILL reconnect, because - // a detached autonomous run keeps writing to pages. The old gate required - // `message?.role === "assistant"`, so this fell through to neither reconnect nor - // poll. With no anchor row, the attach is a PLAIN live attach (no strip/anchor). - it("#488 commit 2: disconnect BEFORE the first assistant frame reconnects with no anchor", () => { - renderThread({ autonomousRunsEnabled: true, initialRows: settledTail() }); - expect(h.state.resumeStream).not.toHaveBeenCalled(); - act(() => { - h.state.onFinish?.({ - message: undefined, - isAbort: false, - isDisconnect: true, - isError: false, - }); - }); - // The reconnect banner shows (not the dead terminal "connection lost" notice). + it("#488 commit 2: a disconnect BEFORE the first assistant frame reconnects with NO anchor", () => { + renderLive(); + disconnect(null); // no assistant message yet (pre-first-frame break) expect(screen.getByText(/reconnecting/i)).toBeTruthy(); expect( screen.queryByText("Connection lost — the answer was interrupted."), ).toBeNull(); advanceToAttempt(1); expect(h.state.resumeStream).toHaveBeenCalledTimes(1); - // No anchor -> a plain attach URL (no expect=live&anchor pinning a row). expect(h.state.transport!.prepareReconnectToStreamRequest!().api).toBe( "/api/ai-chat/runs/c1/stream", ); }); - it("strips the pinned live row before replay so content is NOT duplicated", () => { - renderLiveThenDisconnect(); - advanceToAttempt(1); - // The attempt strips the anchor row from the store (the live replay rebuilds - // it). Apply the setMessages updater to prove it removes exactly the anchor. - const updater = h.state.setMessages.mock.calls.at(-1)![0] as ( - prev: { id: string }[], - ) => { id: string }[]; - expect(updater([{ id: "u1" }, { id: "a2" }])).toEqual([{ id: "u1" }]); - }); - it("a live re-attach (2xx) clears the reconnect banner", async () => { - renderLiveThenDisconnect(); + renderLive(); + disconnect(); advanceToAttempt(1); - await reconnect200(); + await reconnect({ status: 200, ok: true }); expect(screen.queryByText(/reconnecting/i)).toBeNull(); }); it("a 204 arms the degraded poll and backs off to the next attempt", async () => { - const { onResumeFallback } = renderLiveThenDisconnect(); + const { onResumeFallback } = renderLive(); + disconnect(); advanceToAttempt(1); expect(h.state.resumeStream).toHaveBeenCalledTimes(1); - await reconnect204(); - // Fallback engaged: the degraded poll is armed (204 -> onNoActiveStream). + await reconnect({ status: 204, ok: false }); expect(onResumeFallback).toHaveBeenCalledWith(true); - // Still reconnecting — the banner advanced to attempt 2/5. expect(screen.getByText(/reconnecting.*2\/5/i)).toBeTruthy(); - // The next backoff fires attempt 2 (another resumeStream). advanceToAttempt(2); expect(h.state.resumeStream).toHaveBeenCalledTimes(2); }); it("exhausts the attempt limit into a manual Retry, which restarts the sequence", async () => { - renderLiveThenDisconnect(); - // Drive all 5 attempts, each failing with a 204. + renderLive(); + disconnect(); for (let n = 1; n <= 5; n++) { advanceToAttempt(n); expect(h.state.resumeStream).toHaveBeenCalledTimes(n); - await reconnect204(); + await reconnect({ status: 204, ok: false }); } - // The 5th 204 exhausted the cap -> the manual Retry replaces the banner. expect(screen.queryByText(/reconnecting/i)).toBeNull(); const retry = screen.getByText("Retry"); - expect(retry).toBeTruthy(); - // Retry fires attempt 1 immediately (no backoff) — a 6th resumeStream. act(() => { fireEvent.click(retry); }); @@ -1203,22 +779,43 @@ describe("ChatThread — live reconnect after isDisconnect (#430)", () => { expect(screen.getByText(/reconnecting/i)).toBeTruthy(); }); + it("#488 commit 3: two breaks in a row produce two reconnect cycles", async () => { + renderLive(); + // First break -> reconnect -> re-attach live. + disconnect(); + advanceToAttempt(1); + expect(h.state.resumeStream).toHaveBeenCalledTimes(1); + await reconnect({ status: 200, ok: true }); + expect(screen.queryByText(/reconnecting/i)).toBeNull(); + // The re-attached observer stream drops AGAIN -> a SECOND reconnect cycle + // (the old one-shot !wasResumed gate sent this to silent poll). + disconnect(); + expect(screen.getByText(/reconnecting/i)).toBeTruthy(); + advanceToAttempt(1); + expect(h.state.resumeStream).toHaveBeenCalledTimes(2); + }); + it("does NOT reconnect when autonomous runs are disabled", () => { renderThread({ autonomousRunsEnabled: false, initialRows: settledTail() }); - act(() => { - h.state.onFinish?.({ - message: liveMsg, - isAbort: false, - isDisconnect: true, - isError: false, - }); - }); + disconnect(); expect(screen.queryByText(/reconnecting/i)).toBeNull(); - // The terminal "connection lost" notice is shown instead (unchanged behavior). expect( screen.getByText("Connection lost — the answer was interrupted."), ).toBeTruthy(); advanceToAttempt(1); expect(h.state.resumeStream).not.toHaveBeenCalled(); }); + + it("#488 commit 4a: the poll idle cap surfaces a stalled banner + Retry (not silent)", async () => { + renderLive(); + disconnect(); + advanceToAttempt(1); + await reconnect({ status: 204, ok: false }); // arms the poll (reconnecting) + // No activity for the whole idle cap -> stalled. + act(() => { + vi.advanceTimersByTime(10 * 60_000); + }); + expect(screen.getByText(/the run stopped responding/i)).toBeTruthy(); + expect(screen.getByText("Retry")).toBeTruthy(); + }); }); 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 544e5f3f..8a4d72cb 100644 --- a/apps/client/src/features/ai-chat/components/chat-thread.tsx +++ b/apps/client/src/features/ai-chat/components/chat-thread.tsx @@ -34,7 +34,10 @@ import { shouldResetRolePicked, } 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 { + extractServerChatId, + extractRunId, +} from "@/features/ai-chat/utils/adopt-chat-id.ts"; import { assistantMessageHasVisibleContent } from "@/features/ai-chat/utils/message-content.ts"; import { isStreamingTail, @@ -42,6 +45,7 @@ import { seedRows, mergeById, } from "@/features/ai-chat/utils/resume-helpers.ts"; +import { getRun } from "@/features/ai-chat/services/ai-chat-service.ts"; import { AI_CHAT_MESSAGES_RQ_KEY } from "@/features/ai-chat/queries/ai-chat-query.ts"; import type { EditorSelectionContext } from "@/features/editor/utils/get-editor-selection.ts"; import { @@ -51,6 +55,14 @@ import { removeQueuedById, type QueuedMessage, } from "@/features/ai-chat/utils/queue-helpers.ts"; +import { + reduce, + initialMachine, + RECONNECT_MAX_ATTEMPTS, + type Machine, + type Event as RunEvent, + type Effect as RunEffect, +} from "@/features/ai-chat/state/run-fsm.ts"; import classes from "@/features/ai-chat/components/ai-chat.module.css"; // Throttle how often the streamed `messages` state triggers a re-render. Without @@ -61,42 +73,32 @@ import classes from "@/features/ai-chat/components/ai-chat.module.css"; // from the token rate. const STREAM_THROTTLE_MS = 50; -// #430: auto-reconnect after a LIVE SSE disconnect of a DETACHED (autonomous) run. -// The run keeps executing server-side, so instead of a dead "Lost connection" -// banner we re-attach to the live tail through the SAME resumable machinery the -// mount path uses. Attempts back off exponentially and are capped; on exhaustion -// the user gets a manual Retry (the degraded poll keeps catching up underneath). -const RECONNECT_MAX_ATTEMPTS = 5; -// Backoff before attempt N (1-based): 1s, 2s, 4s, 8s, 16s. -const RECONNECT_BASE_DELAY_MS = 1000; +// #488 commit 4a: how long the degraded poll may run with NO new run activity +// (its persisted rows unchanged) before the FSM declares the run STALLED and +// surfaces a banner + Retry — instead of the old silent `refetchInterval -> false` +// ("forever half-done answer"). Measured from INACTIVITY, so a long-but-progressing +// run keeps polling; only a genuinely stuck run trips it. This cap lives in the +// THREAD now (the FSM owns polling->stalled); the window just polls while armed. +const DEGRADED_POLL_IDLE_MAX_MS = 10 * 60_000; -// #396: bounded retry for the "Interrupt and send now" re-send when it races the -// authoritative server stop of the just-superseded detached run. The re-POST can -// arrive before the old run has released the one-active-run slot, so the server -// returns 409 A_RUN_ALREADY_ACTIVE. The server stop guarantees the slot frees, so -// a few short backoffs converge. 4 total attempts: attempt 1 fires immediately, -// then these are the waits BEFORE attempts 2, 3 and 4 (150ms, 300ms, 600ms). If -// all 4 attempts 409, the last 409 surfaces (the banner) — acceptable per #396. -const SUPERSEDE_RETRY_DELAYS_MS = [150, 300, 600]; -// The server error code that means "another run is already active for this chat". -const A_RUN_ALREADY_ACTIVE = "A_RUN_ALREADY_ACTIVE"; +/** The #487 active (non-terminal) run statuses — mirrors the server's + * ACTIVE_RUN_STATUSES. A run-fact is "active" only for these. */ +function isActiveRunStatus(status: string | null | undefined): boolean { + return status === "pending" || status === "running"; +} -/** - * #396: defensively decide whether a 409 response is the one-active-run gate - * rejection (code A_RUN_ALREADY_ACTIVE) vs. some other 409. Reads a CLONE so the - * original response body stays intact for the caller when it is returned as-is. - * Any parse failure or unexpected shape => false (do NOT retry). - */ -async function isRunAlreadyActive(response: Response): Promise { +/** Read the `{ code, runId }` off a JSON error response WITHOUT consuming the + * original body (reads a clone), so the caller can still return the Response + * untouched to the SDK. Any parse failure => empty. */ +async function read409(response: Response): Promise<{ code?: string; runId?: string }> { try { - const body = (await response.clone().json()) as unknown; - return ( - typeof body === "object" && - body !== null && - (body as { code?: unknown }).code === A_RUN_ALREADY_ACTIVE - ); + const b = (await response.clone().json()) as { code?: unknown; runId?: unknown }; + return { + code: typeof b?.code === "string" ? b.code : undefined, + runId: typeof b?.runId === "string" ? b.runId : undefined, + }; } catch { - return false; + return {}; } } @@ -150,21 +152,18 @@ interface ChatThreadProps { * which fires only at the terminal outcome. */ onServerChatId?: (serverChatId?: string) => 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. */ + * chat's window. Called `true` when the FSM enters a poll-bearing recovery + * (attach 204 / starved-or-torn resumed finish / a stop), `false` the moment a + * local stream starts or the run settles. The window owns only the dumb 2.5s + * timer; the THREAD's FSM owns arm/disarm + the stalled cap (#488). */ 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, * which the server deliberately ignores, so the detached run would keep going. */ autonomousRunsEnabled?: boolean; - /** #184: request the server-side stop of this chat's active run (the parent owns - * the endpoint call + the "stopping" latch that keeps observer-polling from - * immediately re-streaming the stopping run's output). Called with the resolved - * chat id when the user presses Stop in autonomous mode. */ + /** #184: request the server-side stop of this chat's active run. Called with the + * resolved chat id when the user presses Stop in autonomous mode. */ onServerStop?: (chatId: string) => void; } @@ -199,6 +198,14 @@ function rowToUiMessage(row: IAiChatMessageRow): UIMessage { * Owns the AI SDK `useChat` lifecycle for ONE chat. The parent remounts this * with a `key` when the selected chat changes, so initial messages re-seed * cleanly (the v6 transport-based hook keeps its state per mount). + * + * #488: the resume/reconnect/poll/stop/supersede lifecycle is driven by the pure + * FSM in `state/run-fsm.ts` (see `run-fsm.spec.md`). The component is the RUNTIME: + * it dispatches typed events (from SDK callbacks / HTTP outcomes) and executes the + * reducer's command EFFECTS (attach GET, POST /run, POST /stop, POST /stream + * supersede, backoff timers, poll arm/disarm). The FSM lives in this thread (not + * the window) so a late SDK callback dies with the owner (#161). The one-shot-ref + * zoo is gone: the epoch counter (I1) drops stale-generation outcomes. */ export default function ChatThread({ chatId, @@ -219,186 +226,204 @@ export default function ChatThread({ 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); - // #430: called from the transport's reconnect-GET success branch when a live - // stream re-attached (2xx, not 204) — clears the reconnect banner. Kept in a ref - // because the transport's fetch closure (useMemo([])) reads it live. - const onReconnectAttachedRef = 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); - }, []); + // --- The run-lifecycle FSM (source of truth) ----------------------------- + // Stored in state (drives UI re-render on phase/ctx change) + mirrored in a ref + // for the transport/timer closures (useMemo([])-stable) that read it live. + const [machine, setMachine] = useState(() => initialMachine()); + const machineRef = useRef(machine); + // I1: the current generation. Command-transitions bump it; async OUTCOME events + // carry the epoch they were issued under; a mismatch is dropped by the reducer. + const epochRef = useRef(0); + // The generation an in-flight attach GET was issued under (for stamping its + // 2xx/204/throw outcome). Replaces the one-shot `noStreamHandledRef` guard. + const pendingAttachEpochRef = useRef(0); + // Stable dispatch handle for closures defined before `dispatch` (transport, + // timers) — assigned once `dispatch` exists below. + const dispatchRef = useRef<(e: RunEvent) => void>(() => undefined); - // 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). + // --- Effect-dependency refs (live values for the stable effect runner) ----- + // These are useChat/prop values that change identity across renders; the effect + // runner (useCallback([])) reads them live so it never captures a stale closure. + const resumeStreamRef = useRef<(() => Promise | void) | null>(null); + const setMessagesRef = useRef< + ((updater: (prev: UIMessage[]) => UIMessage[]) => void) | null + >(null); + const sendMessageRef = useRef<((m: { text: string }) => void) | null>(null); + const stopFnRef = useRef<(() => void) | null>(null); + const onResumeFallbackRef = useRef(onResumeFallback); + onResumeFallbackRef.current = onResumeFallback; + const onServerStopRef = useRef(onServerStop); + onServerStopRef.current = onServerStop; + + // Live mount flag: the attach GET / a resumed onFinish are async and can land + // AFTER unmount (the parent remounts per chat). The epoch (I1) drops stale FSM + // OUTCOMES, but the imperative onFinish side-effects (parent flush, invalidate) + // still need a plain React-liveness bit — orthogonal to the run-lifecycle, so it + // is NOT one of the lifecycle flags the FSM replaced. + const mountedRef = useRef(true); + + // attachStrategy DATA (behind the resumeStream effect; #491 swaps it to tail-only + // WITHOUT touching the FSM). The controller is effect-owned (aborted in cleanup, + // I5). `stripRef`/`strippedRowRef` are the current full-replay+strip anchor. + const attachAbortRef = useRef(null); 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, ); + // Effect-owned backoff timers (not lifecycle flags): the reconnect ladder and the + // stalled inactivity cap. Cleared by the cancelReconnect effect / the cap effect. + const reconnectTimerRef = useRef | null>(null); + const idleCapTimerRef = useRef | null>(null); const initialMessages = useMemo( () => seedRows( initialRows ?? [], - attemptResumeRef.current && stripRef.current, + stripRef.current && autonomousRunsEnabled === true, ).map(rowToUiMessage), [initialRows], ); - // The server resolves/creates the chat from the `chatId` in the request body. - // A new chat starts as null; we keep the id in a ref so the SAME hook instance - // can keep streaming to a chat once it exists (the parent adopts the id on - // finish, but within this mount the body carries whatever we know). + // Identity/data mirrors (NOT lifecycle flags — they stay as data). const chatIdRef = useRef(chatId); chatIdRef.current = chatId; - - // Keep the currently-open page in a ref, updated each render, so the LATEST - // open page is sent on every send WITHOUT re-creating the `useMemo([])`-stable - // transport (and thus without re-creating the useChat store mid-stream — see - // the `chatStoreId` note below). Read live inside `prepareSendMessagesRequest`. const openPageRef = useRef(openPage ?? null); openPageRef.current = openPage ?? null; - - // Keep the selection snapshotter in a ref, same rationale as openPageRef: the - // transport useMemo([]) closes it over, so prop-identity churn must not matter. - // Called at send time inside prepareSendMessagesRequest (#388). const getEditorSelectionRef = useRef< (() => EditorSelectionContext | null) | undefined >(getEditorSelection); getEditorSelectionRef.current = getEditorSelection; - - // Keep the selected role id in a ref, same rationale as openPageRef. Only the - // FIRST request of a brand-new chat uses it (the server persists it then and - // ignores it for existing chats), but sending it on every send is harmless. const roleIdRef = useRef(roleId ?? null); roleIdRef.current = roleId ?? null; - - // Stable `useChat` store key for the lifetime of THIS mount. - // - // CRITICAL: `useChat` (@ai-sdk/react) re-creates its internal `Chat` store - // whenever the `id` option no longer equals the store's current id - // (`"id" in options && chatRef.current.id !== options.id`). For a brand-new - // chat (`chatId === null`) we previously passed `id: undefined`; the store - // then generated its OWN random id internally, so `store.id !== undefined` - // stayed true on EVERY render and the store was re-created on every render — - // wiping the optimistic user message, the "submitted" status, and every - // streamed delta until the turn fully finished (then the parent adopts the - // new chat id and remounts with the persisted history, making everything - // "appear at once"). Passing a STABLE non-undefined id keeps one store for - // the whole turn, so the user message shows immediately and tokens stream - // live. This id is purely the client store key; the server still resolves the - // real chat from `chatId` in the request body (see `prepareSendMessagesRequest`). - // The id only needs to be stable per mount — the parent remounts this via - // `key` on chat switch, which re-seeds cleanly. + // Stable useChat store key for this mount (see the long #137/#174 note that used + // to live here — recreating the store mid-stream wipes the live turn). const stableIdRef = useRef(chatId ?? `new-${generateId()}`); - // Stable for the LIFETIME of this mount. When a brand-new chat adopts its - // server id, the parent now updates the `chatId` prop WITHOUT remounting this - // thread, so the store id must NOT follow `chatId`: recreating the useChat - // store would wipe the live (just-finished) turn. The server still resolves - // the real chat from `chatId` in the request body (see chatIdRef / - // prepareSendMessagesRequest), so this purely-client store key can stay fixed. const chatStoreId = stableIdRef.current; - // Pending messages the user composed WHILE a turn was streaming. They are sent - // automatically, FIFO, on successful turn completion (`onFinish`). The queue is - // LOCAL state so it is scoped to this conversation: it is cleared when the user - // deliberately switches chat / starts a new chat (the parent remounts this via - // `key`), but it SURVIVES in-place new-chat id adoption (no remount), so a - // message queued during a brand-new chat's first turn is not lost. On Stop or - // error the queue is intentionally preserved (onFinish does not fire then) so - // the user decides what to do with the pending messages. + // Pending messages composed while a turn streams; flushed FIFO on a clean finish. const [queued, setQueued] = useState([]); - // Mirror the queue in a ref so the `onFinish` flush always reads the latest - // queue without a stale closure; `setQueue` updates BOTH the ref and the state. const queuedRef = useRef([]); const setQueue = useCallback((next: QueuedMessage[]) => { queuedRef.current = next; setQueued(next); }, []); - // Capture the latest `sendMessage` (returned by useChat below) so the flush - // helper can call the current instance from the stable `onFinish` callback. - const sendMessageRef = useRef<((m: { text: string }) => void) | null>(null); + // #488 commit 5: the runId to inject into the NEXT POST as `supersede:{runId}` + // (read-and-cleared by prepareSendMessagesRequest). This is send-plumbing DATA + // (like chatIdRef/roleIdRef), the single replacement for the three REMOVED + // one-shot flags flushOnAbortRef/interruptNextSendRef/supersedeRetryRef. + const pendingSupersedeRef = useRef(null); - // "Send now" single-flight flags. Kept in refs (not state) so they are read - // inside the stable `onFinish` callback and the transport closure WITHOUT a - // re-render or a stale closure. Both are one-shot (read-and-clear). - // - flushOnAbortRef: flush the promoted head on the abort WE triggered, even - // though an aborted turn normally keeps the queue intact. - // - interruptNextSendRef: tag the next send as a user interrupt so the server - // injects the "your previous answer was interrupted" note for that turn only. - const flushOnAbortRef = useRef(false); - const interruptNextSendRef = useRef(false); + // --- Effect runner: executes the reducer's command effects --------------- + const runEffect = useCallback( + (eff: RunEffect, epoch: number) => { + switch (eff.type) { + case "resumeStream": { + // The attach GET. Stamp the outcome's generation (I1). A reconnect + // attempt filters the pinned live row from the store first (the mount + // seed already stripped it), so the live replay's text-start rebuilds it + // without duplicating parts (#430). + pendingAttachEpochRef.current = epoch; + if (machineRef.current.phase.name === "reconnecting") { + const anchor = strippedRowRef.current; + if (anchor) + setMessagesRef.current?.((prev) => + prev.filter((m) => m.id !== anchor.id), + ); + } + void resumeStreamRef.current?.(); + break; + } + case "scheduleReconnect": { + if (reconnectTimerRef.current) clearTimeout(reconnectTimerRef.current); + const attempt = eff.attempt; + const scheduledEpoch = epoch; + reconnectTimerRef.current = setTimeout(() => { + dispatchRef.current({ + type: "RECONNECT_ATTEMPT", + attempt, + epoch: scheduledEpoch, + }); + }, eff.delayMs); + break; + } + case "cancelReconnect": + if (reconnectTimerRef.current) { + clearTimeout(reconnectTimerRef.current); + reconnectTimerRef.current = null; + } + break; + case "armPoll": + onResumeFallbackRef.current?.(true); + break; + case "disarmPoll": + onResumeFallbackRef.current?.(false); + break; + case "abortAttach": + attachAbortRef.current?.abort(); + break; + case "stopRun": + // Authoritative stop. When the chat id is not known yet (brand-new chat's + // first moment), it is deferred: the chat-id adoption effect fires + // onServerStop the moment the id lands while still in `stopping`. + if (chatIdRef.current) + onServerStopRef.current?.(chatIdRef.current); + break; + case "postRun": { + // (Re)establish / verify the run-fact from POST /ai-chat/run. Stamp the + // RUN_FACT outcome with the issuing generation so a superseded verify is + // dropped (I1). + const cid = chatIdRef.current; + if (!cid) break; + const ep = epoch; + void getRun(cid) + .then((res) => { + const active = + res.run && isActiveRunStatus(res.run.status) + ? { runId: res.run.id } + : null; + dispatchRef.current({ type: "RUN_FACT", runFact: active, epoch: ep }); + }) + .catch(() => undefined); + break; + } + case "supersede": + // Arm the next POST to carry `supersede:{runId}`; the send itself is + // triggered by the caller (sendNow) via sendMessage. + pendingSupersedeRef.current = eff.targetRunId; + break; + } + }, + // Reads only refs + the stable queryClient — safe as an empty-dep callback. + [], + ); - // #396: one-shot arm for the bounded 409 A_RUN_ALREADY_ACTIVE retry on the - // "Interrupt and send now" re-send in autonomous mode. sendNow triggers the - // authoritative server stop of the detached run, but that stop and the - // onFinish->flushNext re-POST race: the new POST can hit the one-active-run - // gate before the old detached run has settled, yielding a spurious 409. When - // this ref is armed, the transport's send path retries that 409 with a short - // bounded backoff (the server stop guarantees convergence). A normal send (ref - // not armed) must STILL fail a 409 instantly (e.g. a genuine two-tab conflict). - // - // INVARIANT: sendNow arms this only to be consumed by the ONE re-POST that - // flushNext fires from onFinish. But that re-POST does not always happen (the - // promoted head may be gone, the finish may be a resumed turn, or the arm may - // race a stale finish). To keep the arm strictly one-shot it is disarmed on - // EVERY path where the paired interrupt one-shots (flushOnAbortRef / - // interruptNextSendRef) are cleared without a POST: the transport POST branch - // consumes it (read-and-clear), the onFinish `!flushNext()` no-send branch - // clears it, and the isStreaming-defuse effect clears it symmetrically. So it - // can never leak into a later, unrelated send and retry that send's genuine 409. - const supersedeRetryRef = useRef(false); + // Dispatch: reduce, run effects, re-render on a real state change. + const dispatch = useCallback( + (event: RunEvent) => { + const prev = machineRef.current; + const next = reduce(prev, event); + machineRef.current = next; + epochRef.current = next.ctx.epoch; + if (next.phase !== prev.phase || next.ctx !== prev.ctx) setMachine(next); + for (const eff of next.effects) runEffect(eff, next.ctx.epoch); + }, + [runEffect], + ); + dispatchRef.current = dispatch; - // #234 F5: the user pressed Stop while streaming a BRAND-NEW chat whose server - // chat id has not been adopted yet (the `start` chunk carrying it hadn't landed - // when Stop was pressed). A local SSE abort alone does NOT stop the DETACHED - // autonomous run — it keeps burning tokens and WRITING TO PAGES — so we cannot - // just no-op. We latch the stop as PENDING and fire the authoritative server - // stop the moment onServerChatId adopts the id (below). Read-and-cleared there; - // also defused on every new turn start so it can never fire against a later, - // unrelated turn's run. - const stopPendingRef = useRef(false); - - // FIFO dequeue + send the next queued message (no-op when the queue is empty). - // Returns whether a message was actually sent, so callers can tell an empty - // dequeue (nothing to flush) from a real send. + // FIFO dequeue + local send of the next queued message (no-op when empty). + const localSend = useCallback((text: string) => { + dispatchRef.current({ type: "SEND_LOCAL" }); + sendMessageRef.current?.({ text }); + }, []); const flushNext = useCallback(() => { 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 }); + localSend(head.text); return true; - }, [setQueue, setResumedTurnPair]); + }, [setQueue, localSend]); const enqueue = useCallback( (text: string) => { @@ -419,12 +444,11 @@ export default function ChatThread({ 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. + // Build the attach URL from the REAL chat id. ?expect=live&anchor= + // only when a streaming tail was stripped: expect=live opts into a + // finished-retained replay (safe only because the row is stripped and the + // replay rebuilds it), and the anchor pins the replay to OUR run — a + // mismatching (newer) run 204s into the restore+poll path instead. api: `/api/ai-chat/runs/${chatIdRef.current}/stream${ stripRef.current ? `?expect=live&anchor=${strippedRowRef.current!.id}` @@ -433,102 +457,77 @@ export default function ChatThread({ }), fetch: async (input: RequestInfo | URL, init: RequestInit = {}) => { if ((init.method ?? "GET") !== "GET") { - // Send path (POST). #396: read-and-clear the one-shot supersede arm - // here so it is strictly scoped to THIS send. When unarmed, behave - // exactly as before — a single fetch, a 409 surfaces instantly (a - // genuine two-tab conflict must NOT be retried). - const supersede = supersedeRetryRef.current; - supersedeRetryRef.current = false; - if (!supersede) return fetch(input, init); - // Buffer a ReadableStream body once so each retry can replay it. - // DefaultChatTransport sends the body as a JSON STRING (replayable as - // is), but guard defensively in case a future SDK streams it. - let sendInit = init; - if (init.body instanceof ReadableStream) { - const buffered = await new Response(init.body).arrayBuffer(); - sendInit = { ...init, body: buffered }; - } - // Bounded retry: attempt 1 fires immediately, then wait between - // attempts per SUPERSEDE_RETRY_DELAYS_MS. Retry ONLY on a real - // 409 A_RUN_ALREADY_ACTIVE; any other status/body is returned as-is. - for (let attempt = 0; ; attempt++) { - const response = await fetch(input, sendInit); - if ( - response.status !== 409 || - attempt >= SUPERSEDE_RETRY_DELAYS_MS.length || - !(await isRunAlreadyActive(response)) - ) { - return response; + // Send path (POST). #488 commit 5: NO client 409 retry ladder anymore + // (the CAS supersede replaces it). We DO peek 409/CAS outcomes to drive + // the FSM, then return the Response untouched to the SDK (a 409 body is + // surfaced by the classified error banner via error-message.ts). + const wasSupersede = + machineRef.current.phase.name === "superseding"; + const ep = epochRef.current; + const response = await fetch(input, init); + if (wasSupersede) { + if (response.ok) { + dispatchRef.current({ type: "SUPERSEDE_READY", epoch: ep }); + } else if (response.status === 409) { + const { code, runId } = await read409(response); + if (code === "SUPERSEDE_TARGET_MISMATCH") + dispatchRef.current({ + type: "SUPERSEDE_MISMATCH", + currentRunId: runId, + epoch: ep, + }); + else if (code === "SUPERSEDE_TIMEOUT") + dispatchRef.current({ type: "SUPERSEDE_TIMEOUT", epoch: ep }); + else if (code === "SUPERSEDE_INVALID") + dispatchRef.current({ type: "SUPERSEDE_INVALID", epoch: ep }); } - // The old detached run has not released the one-active-run slot - // yet; the server stop we requested guarantees it will, so back off - // and re-POST (the 409 fired before the user message was persisted, - // so re-POSTing is safe — no duplicate rows). - await new Promise((r) => - setTimeout(r, SUPERSEDE_RETRY_DELAYS_MS[attempt]), - ); + } else if (response.status === 409) { + const { code } = await read409(response); + if (code === "A_RUN_ALREADY_ACTIVE") + dispatchRef.current({ type: "RUN_ALREADY_ACTIVE" }); } + return response; } - // Reconnect GET: the SDK passes no AbortSignal, so wire our own controller - // for observer Stop / unmount abort. + // Reconnect/attach GET: the SDK passes no AbortSignal, so wire our own + // controller for observer Stop / unmount abort (effect-owned, I5). const controller = new AbortController(); attachAbortRef.current = controller; + const ep = pendingAttachEpochRef.current; + const wasReconnecting = + machineRef.current.phase.name === "reconnecting"; 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?.(); - // #430: a 2xx stream re-attached (live tail or finished-replay). Signal - // the reconnect controller to clear its banner. No-op outside an active - // reconnect sequence (e.g. the mount attach), so it is safe here. - else onReconnectAttachedRef.current?.(); + handleAttachOutcome(ep, wasReconnecting, false); + else handleAttachOutcome(ep, wasReconnecting, true); 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?.(); + handleAttachOutcome(ep, wasReconnecting, false); 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 - // are read live from refs so changing chats/pages does NOT recreate the - // transport. `openPage` is null on a non-page route. prepareSendMessagesRequest: ({ messages, body }) => { - // Read-and-clear the interrupt flag so the "you were interrupted" note - // is carried by ONLY this request (the one resending the promoted - // message right after we aborted the previous turn). The server still - // confirms it against history before acting on it. - const interrupted = interruptNextSendRef.current; - interruptNextSendRef.current = false; // one-shot + // #488 commit 5: read-and-clear the supersede runId so `supersede:{runId}` + // is carried by ONLY this request (the CAS "interrupt and send now"). + const supersedeRunId = pendingSupersedeRef.current; + pendingSupersedeRef.current = null; return { body: { ...body, chatId: chatIdRef.current, - // Attach the live editor selection to the open-page context at send - // time — "this"/"here" in the user's message means THIS selection. - // Nested inside openPage so it dies with the page when the server - // rejects the page id (#388). Null when nothing is selected. openPage: openPageRef.current ? { ...openPageRef.current, selection: getEditorSelectionRef.current?.() ?? null, } : null, - // Honoured by the server only when creating a new chat; null => - // universal assistant. roleId: roleIdRef.current, - interrupted, + ...(supersedeRunId + ? { supersede: { runId: supersedeRunId } } + : {}), messages, }, }; @@ -537,6 +536,37 @@ export default function ChatThread({ [], ); + // Attach GET outcome -> FSM event. The epoch guard replaces BOTH the one-shot + // 204 guard (noStreamHandledRef) and the unmount gate: a stale/superseded or + // post-DISPOSE outcome is dropped (I1). For a NONE outcome the attachStrategy + // recovery (restore the stripped row + invalidate for a fresh poll) runs first. + const handleAttachOutcome = useCallback( + (ep: number, wasReconnecting: boolean, live: boolean) => { + if (ep !== epochRef.current) return; // stale generation — drop + if (live) { + dispatchRef.current( + wasReconnecting + ? { type: "RECONNECT_ATTACHED", epoch: ep } + : { type: "ATTACH_LIVE", epoch: ep }, + ); + return; + } + if (strippedRowRef.current) + setMessagesRef.current?.((prev) => + mergeById(prev, rowToUiMessage(strippedRowRef.current!)), + ); + queryClient.invalidateQueries({ + queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current), + }); + dispatchRef.current( + wasReconnecting + ? { type: "RECONNECT_NONE", epoch: ep } + : { type: "ATTACH_NONE", epoch: ep }, + ); + }, + [queryClient], + ); + const { messages, sendMessage, @@ -546,579 +576,318 @@ export default function ChatThread({ 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). id: chatStoreId, messages: initialMessages, transport, - // See STREAM_THROTTLE_MS — bounds re-render/markdown-reparse frequency. experimental_throttle: STREAM_THROTTLE_MS, - // `onFinish` (ai@6 useChat) fires from a `finally` on EVERY terminal outcome - // — success, user Stop/abort (`isAbort`), network drop (`isDisconnect`), and - // stream error (`isError`). Keep calling `onTurnFinished()` on all of them - // (chat-list refresh + new-chat id adoption must happen even on a failed - // first turn), but flush the pending queue ONLY on a clean finish: auto- - // sending after the user hit Stop — or blindly retrying after a failure — - // 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!)), - ); - } - } - // (2b) #430/#488: a LIVE (non-resumed) detached run whose SSE just dropped. - // The server run keeps executing, so instead of a dead "Lost connection" - // banner start a reconnect sequence and re-attach to the live tail via the - // resumable machinery. - // - // #488 commit 2 — enter reconnect by the RUN-FACT (an active detached run), - // NOT by the presence of an assistant message. The old gate additionally - // required `message?.role === "assistant"`, so a break during the SETUP phase - // — before the first assistant frame, INCLUDING an up-to-60s MCP-toolset - // assembly — fell through to neither reconnect NOR poll while the detached - // run kept writing to pages (a silent data-integrity hole). In autonomous - // mode a run is active for the whole turn, so `autonomousRunsEnabled` is the - // run-fact signal here; the richer server run-fact (POST /run / start-metadata - // runId) is modelled + tested in the FSM (run-fsm.ts `FINISH_DISCONNECT`, - // gated on `ctx.runFact`) and lands with the full component migration. - // - // When there IS an assistant row, pin it as the strip/anchor so the live - // replay rebuilds it without duplicating parts; when there is NONE (the - // pre-first-frame break), reconnect WITHOUT an anchor (a plain live attach — - // there is nothing on screen to rebuild). - const canReconnect = - isDisconnect && - !wasResumed && - autonomousRunsEnabled === true && - mountedRef.current; - if (canReconnect) { - const hasAnchor = - message?.role === "assistant" && typeof message.id === "string"; - beginReconnect( - hasAnchor - ? { - id: message.id, - role: "assistant", - content: "", - status: "streaming", - createdAt: new Date().toISOString(), - // Preserve the partial parts so a 204 restore (onNoActiveStream) - // re-shows what was on screen while the degraded poll catches the - // run up to terminal (rowToUiMessage prefers metadata.parts). - metadata: { parts: message.parts }, - } - : null, - ); - } - // (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 - // session ignore this finish if it belongs to a thread abandoned by New chat - // mid-stream (#161). + // Ownership (I2) is the FSM ctx: a resumed/attached/reconnected turn is an + // OBSERVER; a local send is the owner. The queue flushes ONLY under local + // ownership; an observer never flushes (invariant 7). + const wasObserver = machineRef.current.ctx.ownership === "observer"; + // Notify the parent on EVERY terminal outcome (threadKey-guarded downstream + // for #161); fires even while unmounting. onTurnFinished(extractServerChatId(message), threadKey); - // Show a neutral "stopped" marker for an aborted turn; the red error banner - // (via `error`) already covers isError, and a clean finish clears any marker. - // On a live disconnect that STARTED a reconnect, suppress the terminal - // "connection lost" notice — the reconnect banner takes over (#430). - if (isError) setStopNotice(null); - else if (isAbort) setStopNotice("manual"); - else if (isDisconnect) setStopNotice(canReconnect ? null : "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 - // interrupt note travels with this send via interruptNextSendRef. - if (flushOnAbortRef.current) { - flushOnAbortRef.current = false; - // Suppress the "Response stopped." flash for an intentional interrupt. + + if (isError) { + dispatch({ type: "FINISH_ERROR", kind: "stream" }); setStopNotice(null); - // If the promoted head vanished (e.g. the user removed it before the - // abort landed) flushNext sends nothing — clear the one-shot interrupt - // tag AND the #396 supersede arm so neither can leak onto the next - // unrelated send (no re-POST will consume the arm here). On a real send - // the tag is consumed by prepareSendMessagesRequest and the arm by the - // transport POST branch, so both stay untouched then. - if (!flushNext()) { - interruptNextSendRef.current = false; - supersedeRetryRef.current = false; + return; + } + if (isAbort) { + // A user Stop / an interrupt abort finished. The FSM stopping/idle exit is + // by DATA (this terminal outcome, I4). + dispatch({ type: "FINISH_ABORT" }); + setStopNotice("manual"); + return; + } + // A missing message (a pre-first-frame break) has no visible content. + const msgHasVisible = message + ? assistantMessageHasVisibleContent(message) + : false; + if (isDisconnect) { + if (wasObserver) { + // A resumed/attached OBSERVER stream dropped. Recover via the degraded + // poll (restore the stripped row only when there is no visible content; + // never clobber a fuller on-screen tail, invariant 9). The FSM decides + // reconnect-vs-poll from liveFollow (a live-follow drop reconnects again, + // #488 commit 3; a mount-resume drop polls). + if (mountedRef.current) { + const hasVisible = msgHasVisible; + if (!hasVisible && strippedRowRef.current) + setMessages((prev) => + mergeById(prev, rowToUiMessage(strippedRowRef.current!)), + ); + queryClient.invalidateQueries({ + queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current), + }); + dispatch({ + type: "FINISH_DISCONNECT", + hasVisibleContent: hasVisible, + }); + } + setStopNotice(null); + return; + } + // A LOCAL live turn dropped. #488 commit 2: recover by the RUN-FACT, not by + // the presence of an assistant message — a setup-phase break (before the + // first frame) still leaves a detached run writing to pages. In autonomous + // mode a run is active for the whole turn, so seed the run-fact from the + // start-metadata runId when known, else a sentinel (the attach GET goes by + // chatId, not runId). Pin the assistant row as the strip/anchor when present. + if (autonomousRunsEnabled === true && mountedRef.current) { + const hasAnchor = + message?.role === "assistant" && typeof message.id === "string"; + if (hasAnchor) { + strippedRowRef.current = { + id: message.id, + role: "assistant", + content: "", + status: "streaming", + createdAt: new Date().toISOString(), + metadata: { parts: message.parts }, + }; + stripRef.current = true; + } else { + strippedRowRef.current = null; + stripRef.current = false; + } + dispatch({ + type: "RUN_FACT", + runFact: { runId: extractRunId(message) ?? "pending" }, + }); + dispatch({ + type: "FINISH_DISCONNECT", + hasVisibleContent: msgHasVisible, + }); + setStopNotice(null); + } else { + dispatch({ type: "FINISH_DISCONNECT", hasVisibleContent: false }); + setStopNotice("disconnect"); } return; } - if (isAbort || isDisconnect || isError) return; - // Gate the final flush on the live-mount flag (#486): a clean onFinish can - // land AFTER this thread unmounted (a New-chat / chat-switch mid-stream — - // the async attach/resume settles late). Flushing then dequeues and POSTs a - // queued message from an abandoned thread — a "ghost" send / ghost chat. - // Every other queue side effect already guards on mountedRef; this last one - // was the gap. + // Clean finish. + if (wasObserver) { + if (mountedRef.current) { + const hasVisible = msgHasVisible; + if (!hasVisible) { + // Starved replay: restore the stripped row + poll to the real terminal. + if (strippedRowRef.current) + setMessages((prev) => + mergeById(prev, rowToUiMessage(strippedRowRef.current!)), + ); + queryClient.invalidateQueries({ + queryKey: AI_CHAT_MESSAGES_RQ_KEY(chatIdRef.current), + }); + dispatch({ type: "STREAM_INCOMPLETE", reason: "starved" }); + } else { + // Healthy resumed finish — nothing to restore/arm, just settle. + dispatch({ type: "FINISH_CLEAN" }); + } + } + setStopNotice(null); + return; + } + // Local clean finish: settle + flush the queue (gated on liveness, #486). + dispatch({ type: "FINISH_CLEAN" }); + setStopNotice(null); if (mountedRef.current) flushNext(); }, - // `onError` runs in addition to `onFinish` (which ai@6 also calls on error). - // Log the raw failure here for devtools; the UI shows a friendly classified - // banner via `error` below. We still call `onTurnFinished()` with NO server id - // (idempotent with the onFinish call): for a brand-new chat that ARMS the - // bounded list-refetch fallback (adopt the single newly-appeared chat once the - // refetch lands); for an existing chat it just refreshes the chat list - // immediately rather than after a manual refresh. onError: (streamError) => { - // Surface the raw failure in the browser console (devtools) for debugging; - // the UI separately shows a friendly classified banner (see errorView). console.error("AI chat stream error:", streamError); onTurnFinished(undefined, threadKey); }, }); - // Keep the flush helper pointed at the latest sendMessage instance. + // Publish the live useChat handles to the effect-runner refs. + resumeStreamRef.current = resumeStream; + setMessagesRef.current = setMessages as unknown as ( + updater: (prev: UIMessage[]) => UIMessage[], + ) => void; sendMessageRef.current = sendMessage; + stopFnRef.current = stop; - // Mirror the live turn status in a ref so event handlers (sendNow) branch on the - // CURRENT status rather than a value captured in a stale render closure — a turn - // can finish between render and click, and arming the interrupt refs against a - // no-op stop() would leave them set to leak into a later, unrelated Stop. const statusRef = useRef(status); statusRef.current = status; - // EARLY chat-id adoption (#174): the server streams the authoritative chat id - // on the assistant message metadata at the `start` chunk (message.metadata. - // chatId — see adopt-chat-id.ts / chatStreamMetadata). Forward it to the parent - // AS SOON AS it appears (mid-stream), so a brand-new chat adopts its real id - // WHILE the first turn is still streaming and activeChatId-gated affordances - // (the Copy/export button) light up immediately, instead of only at onFinish. - // Keyed by the last-seen id so we forward each distinct id exactly once. The - // parent's onServerChatId is idempotent and a no-op once the chat has an id. + // EARLY chat-id (#174) + runId (#488) adoption from the streaming assistant + // message's start metadata. Forward the chat id once; adopt the runId into the + // FSM run-fact; and (#234 F5) fire a DEFERRED server stop the moment the id lands + // while still `stopping` — no stopPendingRef needed (the FSM phase is the latch). const lastForwardedChatIdRef = useRef(undefined); useEffect(() => { - if (!onServerChatId) return; const tail = messages[messages.length - 1]; if (tail?.role !== "assistant") return; const serverChatId = extractServerChatId(tail); - if (!serverChatId || serverChatId === lastForwardedChatIdRef.current) - return; - lastForwardedChatIdRef.current = serverChatId; - onServerChatId(serverChatId); - // #234 F5: if Stop was pressed before the id was known, the authoritative - // server stop was deferred to this adoption point — fire it now with the - // just-adopted id. One-shot (read-and-clear) so it can't fire twice. - if (stopPendingRef.current) { - stopPendingRef.current = false; - onServerStop?.(serverChatId); + if ( + onServerChatId && + serverChatId && + serverChatId !== lastForwardedChatIdRef.current + ) { + lastForwardedChatIdRef.current = serverChatId; + onServerChatId(serverChatId); + if (machineRef.current.phase.name === "stopping") + onServerStopRef.current?.(serverChatId); } - }, [messages, onServerChatId, onServerStop]); + const runId = extractRunId(tail); + if (runId && machineRef.current.ctx.runFact?.runId !== runId) + dispatch({ type: "RUN_FACT", runFact: { runId } }); + }, [messages, onServerChatId, dispatch]); - // Live "turn was interrupted" marker for the CURRENT session. The red error - // banner (driven by `error`) covers the error case; this covers an aborted - // turn, distinguishing a manual Stop (`isAbort`) from a dropped connection - // (`isDisconnect`) — a distinction only available live (the server persists - // both as finishReason 'aborted'). Cleared when the next turn starts. + // Live "turn was interrupted" marker (a manual Stop vs a dropped connection — a + // distinction only available live). Cleared when the next turn starts. const [stopNotice, setStopNotice] = useState( null, ); const isStreaming = status === "submitted" || status === "streaming"; + const phase = machine.phase; + const isObserver = machine.ctx.ownership === "observer"; - // #430: live-disconnect reconnect controller. `null` = idle; `{ trying, attempt }` - // = a backoff sequence is running (drives the "reconnecting… (N/max)" banner); - // `{ failed }` = attempts exhausted (drives the manual Retry). Mirrored into a ref - // so the transport/onNoActiveStream closures branch on the LIVE value. - type ReconnectState = - | null - | { phase: "trying"; attempt: number } - | { phase: "failed" }; - const [reconnectState, setReconnectState] = useState(null); - const reconnectStateRef = useRef(null); - const setReconnectStatePair = useCallback((s: ReconnectState) => { - reconnectStateRef.current = s; - setReconnectState(s); - }, []); - const reconnectTimerRef = useRef | null>(null); - const clearReconnectTimer = useCallback(() => { - if (reconnectTimerRef.current) { - clearTimeout(reconnectTimerRef.current); - reconnectTimerRef.current = null; - } - }, []); - - // One reconnect attempt — MIRRORS the mount strip/anchor path for the LIVE case. - // beginReconnect pinned strippedRowRef/stripRef to the run's assistant row, so: - // - remove that row from the store (the mount path strips it from the SEED; here - // it is already shown, so filter it out) — the live replay's `text-start` then - // rebuilds it without DUPLICATING parts (the main dedup risk, #430); - // - reset the one-shot 204 guard so onNoActiveStream can fire for THIS attempt; - // - mark the turn resumed (invariant 7/8) so onFinish runs the recovery block and - // never flushes the queue; - // - resumeStream() -> prepareReconnectToStreamRequest builds - // ?expect=live&anchor=, pinning the replay to OUR run (invariant 6). - const attemptReconnectOnce = useCallback( - (attempt: number) => { - if (!mountedRef.current) return; - const anchor = strippedRowRef.current; - if (anchor) { - setMessages((prev) => prev.filter((m) => m.id !== anchor.id)); - } - noStreamHandledRef.current = false; - setResumedTurnPair(true); - setReconnectStatePair({ phase: "trying", attempt }); - void resumeStream(); - }, - [setMessages, setResumedTurnPair, setReconnectStatePair, resumeStream], - ); - - // Schedule attempt `attempt` after an exponential backoff. - const scheduleReconnectAttempt = useCallback( - (attempt: number) => { - clearReconnectTimer(); - setReconnectStatePair({ phase: "trying", attempt }); - reconnectTimerRef.current = setTimeout( - () => attemptReconnectOnce(attempt), - RECONNECT_BASE_DELAY_MS * 2 ** (attempt - 1), - ); - }, - [clearReconnectTimer, setReconnectStatePair, attemptReconnectOnce], - ); - - // Start a fresh reconnect sequence. `anchorRow` (the live run's assistant row) is - // pinned as the strip/anchor reused by every attempt; `null` (a pre-first-frame - // break, #488 commit 2) means there is nothing on screen to rebuild, so attach - // WITHOUT a strip/anchor (a plain live attach to whatever the run is emitting). - const beginReconnect = useCallback( - (anchorRow: IAiChatMessageRow | null) => { - if (!autonomousRunsEnabled || !mountedRef.current) return; - strippedRowRef.current = anchorRow; - stripRef.current = anchorRow !== null; - scheduleReconnectAttempt(1); - }, - [autonomousRunsEnabled, scheduleReconnectAttempt], - ); - - // Manual Retry (shown once attempts are exhausted): restart at attempt 1 and fire - // immediately (the user asked for it now — no backoff). - const retryReconnect = useCallback(() => { - clearReconnectTimer(); - attemptReconnectOnce(1); - }, [clearReconnectTimer, attemptReconnectOnce]); - - // Live SSE re-attached (the reconnect GET returned a 2xx stream): clear the - // banner + any pending backoff. No-op outside a sequence (e.g. the mount attach). - const onReconnectAttached = useCallback(() => { - if (!mountedRef.current || !reconnectStateRef.current) return; - clearReconnectTimer(); - setReconnectStatePair(null); - }, [clearReconnectTimer, setReconnectStatePair]); - onReconnectAttachedRef.current = onReconnectAttached; - - // The reconnect GET could not attach (204 / error). onNoActiveStream has already - // armed the degraded poll (the robust fallback that drives the row to terminal - // from the DB), so this only decides the LIVE-attach retry: back off and try - // again up to the cap, else surface the manual Retry. - const onReconnectNoStream = useCallback(() => { - const s = reconnectStateRef.current; - if (s?.phase !== "trying") return; - if (s.attempt < RECONNECT_MAX_ATTEMPTS) - scheduleReconnectAttempt(s.attempt + 1); - else setReconnectStatePair({ phase: "failed" }); - }, [scheduleReconnectAttempt, setReconnectStatePair]); - - // 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); - // (e) #430: if this 204/error landed during a live-disconnect reconnect - // sequence, back off and retry the live attach (or give up to the manual - // Retry). The degraded poll armed in (c) is the fallback either way. - onReconnectNoStream(); - }, [ - setMessages, - queryClient, - onResumeFallback, - setResumedTurnPair, - onReconnectNoStream, - ]); - onNoActiveStreamRef.current = onNoActiveStream; - - // 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. + // Mount effect: arm the resume attempt — ONLY on a server-confirmed active run + // (#488 commit 4b). A STREAMING tail IS the run-fact (status streaming) -> attach + // now. A USER tail (ended on a user message) may have NO active run, so confirm + // via POST /run first: a chat with no active run must NOT arm a poll (the old code + // attached -> 204 -> ~240 req/10min storm). A SETTLED assistant tail never resumes. useEffect(() => { - // 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(); + if (autonomousRunsEnabled === true && chatId !== null) { + const rows = initialRows ?? []; + if (isStreamingTail(rows)) { + dispatch({ type: "ATTACH_START" }); + } else if (!isSettledAssistantTail(rows)) { + void getRun(chatId) + .then((res) => { + if (!mountedRef.current) return; + if (res.run && isActiveRunStatus(res.run.status)) + dispatch({ type: "ATTACH_START", runId: res.run.id }); + }) + .catch(() => undefined); + } } - // 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(); - // #430: drop any pending reconnect backoff so it can't fire against the next - // chat this thread's refs are reused for. - if (reconnectTimerRef.current) clearTimeout(reconnectTimerRef.current); + dispatch({ type: "DISPOSE" }); // aborts attach + timers, bumps epoch (I5) }; // 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]. + // Reconciliation + degraded-merge (invariant 8): while a poll-bearing recovery is + // active (polling / reconnecting / stopping), merge the polled assistant tail on + // every initialRows update and settle to terminal via POLL_TERMINAL. 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; + if (isStreaming) return; // a local stream owns the view + const p = machineRef.current.phase.name; + if (p !== "polling" && p !== "reconnecting" && p !== "stopping") 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). + // Anchor-mismatch coherence: a restored stripped row A that a DIFFERENT run's + // row B has replaced as the tail would linger as an orphan — settle A from + // fresh history so no phantom row survives. 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); - // #430: the run reached its terminal state via the degraded poll — there is - // no live tail left to reconnect to, so drop any reconnect banner / Retry. - clearReconnectTimer(); - setReconnectStatePair(null); - } - // onResumeFallback intentionally omitted (parent-stable callback); deps are - // fixed by the resume design. + if (tail.status !== "streaming") dispatch({ type: "POLL_TERMINAL" }); // eslint-disable-next-line react-hooks/exhaustive-deps }, [initialRows, isStreaming, setMessages]); - // #430: a real stream is live again — the reconnect re-attached to the live tail - // (status -> "streaming") OR the user started a new local turn. Either way clear - // the reconnect banner + any pending backoff. Gated on "streaming" (not the - // broader "submitted") so a still-pending attach GET does not clear prematurely. + // #488 commit 4a: the stalled inactivity cap. While polling/reconnecting, if no + // new run activity (initialRows unchanged) for DEGRADED_POLL_IDLE_MAX_MS, fire + // POLL_IDLE_CAP -> the FSM goes `stalled` (banner + Retry) instead of silent. + // Reset on every activity (initialRows change) and on phase change. useEffect(() => { - if (status === "streaming") { - clearReconnectTimer(); - setReconnectStatePair(null); + if (idleCapTimerRef.current) { + clearTimeout(idleCapTimerRef.current); + idleCapTimerRef.current = null; } - }, [status, clearReconnectTimer, setReconnectStatePair]); + const p = phase.name; + if (p !== "polling" && p !== "reconnecting") return; + idleCapTimerRef.current = setTimeout(() => { + dispatchRef.current({ type: "POLL_IDLE_CAP" }); + }, DEGRADED_POLL_IDLE_MAX_MS); + return () => { + if (idleCapTimerRef.current) clearTimeout(idleCapTimerRef.current); + }; + }, [phase, initialRows]); - // "Send now" on a queued message: interrupt the current turn and immediately - // send THIS message, keeping the agent's partial output. Other queued messages - // stay queued and flush normally after the new turn. Reuses the existing - // queue/flush machinery: promote the target to the head, then abort — the - // onFinish flush-on-abort branch sends exactly that head, tagged as an - // interrupt so the server notes the previous answer was cut off. - const sendNow = useCallback( - (id: string) => { - // Branch on the LIVE status (statusRef), NOT the closure-captured isStreaming: - // the turn may have finished between this render and the click, in which case - // stop() is a no-op and arming the interrupt refs would strand them for a - // later, unrelated Stop. Reading the ref always sees the current status. - const liveStreaming = - statusRef.current === "submitted" || statusRef.current === "streaming"; - if (liveStreaming) { - // Promote to head so the onFinish -> flushNext path sends exactly it. - setQueue(promoteToHead(queuedRef.current, id)); - flushOnAbortRef.current = true; - interruptNextSendRef.current = true; - // #396: in autonomous mode the turn is a DETACHED run — a local stop() - // is only a client disconnect the server ignores, so the run keeps going. - // The onFinish->flushNext re-POST would then hit the one-active-run gate - // and get a spurious 409 A_RUN_ALREADY_ACTIVE. Mirror handleStop: request - // the AUTHORITATIVE server stop so the detached run settles, and arm the - // one-shot bounded 409 retry BEFORE stop() so the re-send converges once - // the slot frees. Read chatId live from chatIdRef (adopted at the `start` - // chunk). If it is not known yet (brand-new chat, first moment of its - // first turn), defer the server stop via stopPendingRef exactly as - // handleStop does — the onServerChatId adoption effect fires it once the - // id lands; the retry stays armed so the re-send still converges then. - if (autonomousRunsEnabled) { - supersedeRetryRef.current = true; // arm the bounded 409 retry - if (chatIdRef.current) { - onServerStop?.(chatIdRef.current); - } else { - // Same #234-F5 sub-window limitation documented in handleStop: if the - // local abort below cancels the reader before the `start` chunk lands, - // the adoption effect never runs and the deferred stop never fires. Not - // a regression; at minimum we don't strand refs (the isStreaming effect - // defuses stopPendingRef on the next turn start). - stopPendingRef.current = true; - } - } - stop(); // -> onFinish({ isAbort: true }) flushes the promoted head - } else { - // Nothing to interrupt: just send it now (no interrupt note). - 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, setResumedTurnPair, autonomousRunsEnabled, onServerStop], - ); - - // Stop the current turn. ALWAYS abort the local SSE (`stop()`) so the composer - // returns to idle immediately. In AUTONOMOUS mode the turn is a DETACHED run: - // aborting the local SSE is only a client disconnect, which the server ignores, - // so the run would keep executing — we ADDITIONALLY request the authoritative - // server-side stop (the parent owns that call + the "stopping" latch that keeps - // observer-polling from re-streaming the stopping run's output). The chat id is - // read live from chatIdRef (adopted early at the stream's `start` chunk); if it - // 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(); - // #430: pressing Stop also cancels an in-progress reconnect sequence. - clearReconnectTimer(); - setReconnectStatePair(null); - if (!autonomousRunsEnabled) return; - if (chatIdRef.current) { - onServerStop?.(chatIdRef.current); - } else { - // #234 F5: no chat id yet (brand-new chat in the first moment of its first - // turn, before the `start` chunk adopted the id). Latch the stop as pending; - // the onServerChatId adoption effect fires the deferred server stop as soon - // as the id appears, so the detached run is still authoritatively stopped - // instead of left running by a silent local-only abort. - // - // KNOWN LIMITATION (#234 F5 review): `stop()` above has already aborted the - // local SSE reader. In the rare sub-window where Stop is pressed while still - // `submitted` (request sent, not one chunk read yet), that abort can cancel - // the reader BEFORE the `start` chunk is applied to `messages`, so the - // adoption effect never runs and this pending stop never fires. The detached - // run then keeps going for that turn. This is not a regression (the pre-fix - // behavior sent no server stop at all); closing it fully would require - // deferring the local abort until adoption, which is riskier and out of scope - // for this fix. Documented so a future change can address the abort-ordering. - stopPendingRef.current = true; - } - }, [ - stop, - autonomousRunsEnabled, - onServerStop, - clearReconnectTimer, - setReconnectStatePair, - ]); - - // Clear the stopped marker as soon as a new turn begins streaming, and drop any - // stale "Send now" interrupt flags. On the legit interrupt path both refs are - // already consumed synchronously (onFinish + prepareSendMessagesRequest) before - // this effect runs, so clearing here is a no-op for it; its purpose is to defuse - // the race where a flag was armed but the expected abort never fired (the turn - // finished in the same tick as the click), so it cannot leak into a later turn. + // Clear the stopped marker as soon as a new turn begins streaming. useEffect(() => { - if (isStreaming) { - setStopNotice(null); - flushOnAbortRef.current = false; - interruptNextSendRef.current = false; - // #396: symmetric with the other one-shot interrupt flags — defuse a stale - // supersede arm that was set but whose expected re-POST never fired (the - // turn finished in the same tick as the click, or the promoted head was - // gone), so it can never leak into this (or a later) turn's send and retry - // that send's genuine 409. A legit arm is consumed by the transport POST - // branch before this new turn streams, so this does not clobber it. - supersedeRetryRef.current = false; - // #234 F5: a new turn is starting — drop any pending deferred-stop from a - // previous turn that never adopted an id, so it can never fire against this - // (or a later) unrelated turn's run. A deferred stop for the CURRENT turn is - // set AFTER this effect (on the Stop click), so this does not clobber it. - stopPendingRef.current = false; - } + if (isStreaming) setStopNotice(null); }, [isStreaming]); - // Classify the turn error into a heading + detail so the banner names the cause - // (connection reset, timeout, rate limit, context overflow, quota, ...) instead - // of a generic "Something went wrong". Computed here (not only in the JSX) so - // the SAME on-screen banner text can be mirrored into the export (issue #160). + // "Send now" on a queued message: #488 commit 5 — interrupt the active run and + // send THIS message via the server CAS supersede (POST /stream {supersede:{runId}}) + // when the run id is known. No local stop/re-POST dance, no client retry ladder. + const sendNow = useCallback( + (id: string) => { + const msg = queuedRef.current.find((m) => m.id === id); + if (!msg) return; + const liveStreaming = + statusRef.current === "submitted" || statusRef.current === "streaming"; + const runId = machineRef.current.ctx.runFact?.runId; + if ( + liveStreaming && + autonomousRunsEnabled === true && + runId && + runId !== "pending" + ) { + // CAS supersede: the server stops the old run + starts this one atomically. + setQueue(removeQueuedById(queuedRef.current, id)); + dispatch({ type: "SUPERSEDE_REQUESTED", targetRunId: runId }); + sendMessageRef.current?.({ text: msg.text }); // POST carries supersede body + return; + } + if (liveStreaming) { + // No CAS possible (legacy without a run, or the runId not adopted yet): + // promote to head and abort; the local turn's abort finishes and the queue + // holds the promoted head for the user (a blind re-POST would 409 -> the + // classified "already answering — interrupt and send" banner guides them). + setQueue(promoteToHead(queuedRef.current, id)); + stopFnRef.current?.(); + return; + } + // Nothing to interrupt: send it now. + setQueue(removeQueuedById(queuedRef.current, id)); + localSend(msg.text); + }, + [setQueue, autonomousRunsEnabled, dispatch, localSend], + ); + + // Stop the current turn. Abort the local SSE + the attach GET; in AUTONOMOUS mode + // additionally request the AUTHORITATIVE server stop (a local abort is only a + // client disconnect the server ignores). The FSM `stopping` exits by DATA (I4): + // the local turn's onFinish (FINISH_ABORT) or the poll reaching terminal. + const handleStop = useCallback(() => { + stopFnRef.current?.(); + if (autonomousRunsEnabled) { + dispatch({ type: "STOP_REQUESTED" }); + } else { + // Legacy: no server run to stop from the client (the server's onClose issues + // requestStop on disconnect). Just reset the FSM recovery. + attachAbortRef.current?.abort(); + if (reconnectTimerRef.current) clearTimeout(reconnectTimerRef.current); + dispatch({ type: "FINISH_ABORT" }); + } + }, [autonomousRunsEnabled, dispatch]); + + // Manual Retry from the reconnect-failed OR stalled banner. + const onRetry = useCallback(() => { + dispatch({ type: "RETRY" }); + }, [dispatch]); + + // Classify the turn error into a heading + detail (connection reset, timeout, the + // #487 409 codes, ...). Computed here so the SAME text can mirror into the export. const errorView = error ? describeChatError(error.message ?? "", t) : null; - // A role was picked with autoStart=false: the role is bound but NOTHING was - // sent, so chatId stays null and the empty state would keep showing the cards. - // This flag hides the cards and reveals the composer (with the role indicated) - // so the user can type the first message themselves. roleIdRef is already set, - // so that first manual message carries the roleId. + // Role-picker empty state (unchanged from #149). const [rolePickedNoSend, setRolePickedNoSend] = useState(false); - - // Clicking a role card always binds the role to THIS new chat. Whether it also - // auto-starts the conversation is per-role (autoStart). roleIdRef is set - // synchronously here because the parent's selectedRoleId state update would - // only reach roleIdRef on the next render — after this synchronous sendMessage - // has already read it. const handleRolePick = (role: IAiRole): void => { roleIdRef.current = role.id; onRolePicked?.(role); @@ -1127,20 +896,11 @@ export default function ChatThread({ t("Take a look at the current document"), ); if (launch !== null) { - sendMessage({ text: launch }); + localSend(launch); } else { - // autoStart=false -> bind only: hide the cards, show the composer. setRolePickedNoSend(true); } }; - // Reset the "picked, not sent" flag when the thread returns to a truly empty, - // role-less state — e.g. the user hit "New chat" after picking an autoStart=false - // role. That path clears the parent's selectedRoleId (roleId -> null) but leaves - // chatId null, so the thread never remounts and the flag would stay set, hiding - // the cards forever. A picked-and-bound role keeps roleId non-null, so the cards - // correctly stay hidden then. Render-phase reset (React "adjust state on prop - // change"): one-shot — it re-renders with the flag false and the guard no longer - // matches, so it cannot loop. (Review of #149.) if (shouldResetRolePicked(chatId, roleId, rolePickedNoSend)) { setRolePickedNoSend(false); } @@ -1160,29 +920,19 @@ export default function ChatThread({ /> {errorView ? ( - - ) : reconnectState ? ( - // #430: while auto-reconnecting to a detached run's live tail, show progress - // instead of a dead "Lost connection" banner; once attempts are exhausted, - // offer a manual Retry (the degraded poll keeps catching up underneath). - + + ) : phase.name === "reconnecting" ? ( + // #430/#488: while auto-reconnecting to a detached run's live tail, show + // progress; once attempts are exhausted, offer a manual Retry (the degraded + // poll keeps catching up underneath). + - {reconnectState.phase === "trying" ? ( + {!phase.failed ? ( <> {t("Connection lost — reconnecting…")} - {` (${reconnectState.attempt}/${RECONNECT_MAX_ATTEMPTS})`} + {` (${phase.attempt}/${RECONNECT_MAX_ATTEMPTS})`} ) : ( @@ -1194,7 +944,7 @@ export default function ChatThread({ size="compact-xs" variant="light" color="gray" - onClick={retryReconnect} + onClick={onRetry} > {t("Retry")} @@ -1202,6 +952,24 @@ export default function ChatThread({ )} + ) : phase.name === "stalled" ? ( + // #488 commit 4a: the degraded poll hit the inactivity cap — instead of a + // silent "forever half-done answer", tell the user and offer Retry. + + + + {t("The answer may be incomplete — the run stopped responding.")} + + + + ) : stopNotice ? ( {m.text} - {/* "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 && ( + {/* "Send now" (interrupt) is hidden while OBSERVING a detached run + (ownership observer): a local stop() does not abort the attach + fetch, so the interrupt would be swallowed. Only the remove + affordance stays. */} + {!isObserver && ( )} { - // Local send: clear any resume-suppression flag (invariant 8). - setResumedTurnPair(false); - sendMessage({ text }); - }} + onSend={(text) => localSend(text)} onQueue={enqueue} onStop={handleStop} isStreaming={isStreaming} diff --git a/apps/client/src/features/ai-chat/state/run-fsm.spec.md b/apps/client/src/features/ai-chat/state/run-fsm.spec.md index d7c7557c..f6a84e01 100644 --- a/apps/client/src/features/ai-chat/state/run-fsm.spec.md +++ b/apps/client/src/features/ai-chat/state/run-fsm.spec.md @@ -18,7 +18,10 @@ the observable property) — see `run-fsm.test.ts`. Phases: `idle | sending | streaming | attaching | reconnecting(attempt,failed) | polling(reason) | stalled | stopping | superseding | error(kind)`. -Context (orthogonal): `epoch`, `ownership: local|observer`, `runFact: {runId}|null`. +Context (orthogonal): `epoch`, `ownership: local|observer`, `runFact: {runId}|null`, +`liveFollow` (are we following a live run we locally streamed — the reconnect +ladder — vs a one-shot mount-attach resume? both are `observer`, but a live-follow +drop RE-ENTERS the ladder (#488 commit 3) while a mount-resume drop polls). Legend: **†** = command-transition (bumps `epoch`, I1). Effects in `[…]`. @@ -28,8 +31,10 @@ Legend: **†** = command-transition (bumps `epoch`, I1). Effects in `[…]`. | `STREAM_START{runId}` (SDK `start` metadata) | sending, attaching, reconnecting, superseding | streaming | `[cancelReconnect, disarmPoll]`, runFact←runId | | `FINISH_CLEAN` (onFinish clean) | streaming, … | idle | `[disarmPoll, cancelReconnect]`, runFact←null | | `FINISH_ABORT` (onFinish isAbort) | streaming, stopping | idle | `[disarmPoll, cancelReconnect]`, runFact←null (I4 exits stopping by this DATA) | -| `FINISH_DISCONNECT{hasVisibleContent}` (onFinish isDisconnect) | streaming | reconnecting(1) **†** *iff runFact* | `[scheduleReconnect(1)]` (+`armPoll(disconnect-visible)` if visible), ownership=observer | -| `FINISH_DISCONNECT` (no runFact) | streaming | idle | runFact←null (plain terminal "connection lost") | +| `FINISH_DISCONNECT` (observer, NOT liveFollow) | streaming(observer) | polling(disconnect-visible) | `[armPoll]` (a mount-resume drop polls) | +| `FINISH_DISCONNECT{hasVisibleContent}` (local drop OR liveFollow) | streaming | reconnecting(1) **†** *iff runFact\|liveFollow* | `[scheduleReconnect(1)]` (+`armPoll` if visible), ownership=observer, liveFollow=true (commit 3: repeatable) | +| `FINISH_DISCONNECT` (no runFact, not liveFollow) | streaming | idle | runFact←null (plain terminal "connection lost") | +| `STREAM_INCOMPLETE{reason}` (observer starved/torn clean finish) | streaming(observer) | polling(reason) | `[armPoll(reason)]` | | `FINISH_ERROR{kind}` (onFinish isError) | any | error(kind) | `[disarmPoll, cancelReconnect]`, runFact←null | | `ATTACH_START{runId}` (mount resume) | idle | attaching **†** | `[resumeStream]`, ownership=observer, runFact←runId | | `ATTACH_LIVE` (attach GET 2xx) | attaching | streaming | — | @@ -74,44 +79,40 @@ disconnects) carry no epoch. --- -## 2. Ref-map — every `chat-thread.tsx` ref → its new home +## 2. Ref-map — every `chat-thread.tsx` ref → its new home (MIGRATION RESOLVED) -(develop@363f20ab counted "26"; the branch already collapsed one during #486/#487, -so 25 refs are present today. Each is classified below; the run-lifecycle FLAGS -move into the FSM, the identity/data mirrors STAY as data — the post-merge rule -forbids only new **lifecycle-flag** refs.) +The migration is COMPLETE: the 13 run-lifecycle FLAGS below are GONE from +`chat-thread.tsx` (collapsed into FSM phase/ctx/effects, or deleted). What remains +are identity/data mirrors, effect-owned controllers/timers, and ONE React-liveness +bit — none of which is a run-lifecycle flag, so the post-merge "no new flags" rule +holds. **Pending column: empty.** -| # | Ref | Classification | Notes | +| # | Old ref | Resolved to | Where now | |---|---|---|---| -| 1 | `reconcileTailRef` | **FSM** (polling phase) | "poll is driving the tail" is `phase=polling` | -| 2 | `noStreamHandledRef` | **FSM ctx (epoch)** | one-shot 204 guard → epoch drops the stale second outcome | -| 3 | `onNoActiveStreamRef` | **FSM effect** | the transport dispatches `ATTACH_NONE`; no ref-callback | -| 4 | `onReconnectAttachedRef` | **FSM effect** | transport dispatches `RECONNECT_ATTACHED` / `ATTACH_LIVE` | -| 5 | `resumedTurnRef` | **FSM ctx (ownership)** | `ownership==='observer'` ⇒ resumed turn ⇒ never flush | -| 6 | `reconnectStateRef` | **FSM** (reconnecting phase) | `{trying,attempt}`/`{failed}` = `reconnecting(attempt,failed)` | -| 7 | `reconnectTimerRef` | **FSM effect** | `scheduleReconnect`/`cancelReconnect` own the timer | -| 8 | `flushOnAbortRef` | **FSM** (superseding/queue) | flush-on-abort is the superseding→READY / interrupt transition | -| 9 | `interruptNextSendRef` | **FSM ctx** (interrupt tag) | tag carried by the supersede/interrupt transition, one-shot via epoch | -| 10 | `supersedeRetryRef` | **REMOVED** (commit 5) | the client 409 retry ladder is deleted; CAS supersede replaces it | -| 11 | `stopPendingRef` | **FSM** (stopping deferral) | deferred stop is a `STOP_REQUESTED` pended on run-fact adoption | -| 12 | `mountedRef` | **FSM ctx (epoch)** + `DISPOSE` | unmount → `DISPOSE` bumps epoch; late callbacks dropped by I1 | -| 13 | `attemptResumeRef` | **FSM** (ATTACH_START decision) | armed ONLY on a server-confirmed run-fact (commit 4b) | -| 14 | `stripRef` | **data** (attachStrategy) | strip+replay strategy detail; effect-owned | -| 15 | `strippedRowRef` | **data** (attachStrategy) | the anchor row; effect-owned, aborts in cleanup | -| 16 | `attachAbortRef` | **FSM effect** (`abortAttach`) | controller owned by the attach effect, aborted in cleanup (I5) | -| 17 | `chatIdRef` | **data** (identity mirror) | stays; live chat id for the transport body | -| 18 | `openPageRef` | **data** | stays; live open-page for the send body | -| 19 | `getEditorSelectionRef` | **data** | stays; live selection snapshotter | -| 20 | `roleIdRef` | **data** | stays; live role id for the first send | -| 21 | `stableIdRef` | **data** | stays; the useChat store key (mount-stable) | -| 22 | `queuedRef` | **data** (queue) | stays; the queue is a data structure (decision #8) | -| 23 | `sendMessageRef` | **data** | stays; latest `sendMessage` for the flush | -| 24 | `statusRef` | **data** | stays; live SDK status mirror | -| 25 | `lastForwardedChatIdRef` | **data** (one-shot forward) | stays; dedupes the chat-id forward | +| 1 | `reconcileTailRef` | **FSM phase** | reconcile-merge gated on `phase ∈ {polling, reconnecting, stopping}` | +| 2 | `noStreamHandledRef` | **FSM epoch (I1)** | the attach outcome's epoch guard drops the stale/second outcome | +| 3 | `onNoActiveStreamRef` | **FSM event** | transport → `handleAttachOutcome` dispatches `ATTACH_NONE`/`RECONNECT_NONE` | +| 4 | `onReconnectAttachedRef` | **FSM event** | transport dispatches `ATTACH_LIVE` / `RECONNECT_ATTACHED` | +| 5 | `resumedTurnRef` + `resumedTurn` state | **FSM ctx `ownership`** | `ownership==='observer'` ⇒ never flush; hides "Send now" | +| 6 | `reconnectStateRef` + `reconnectState` state | **FSM phase** | `reconnecting(attempt,failed)` renders the banner | +| 7 | `reconnectTimerRef` | **effect-owned timer** | owned by `scheduleReconnect`/`cancelReconnect` effects (not a flag) | +| 8 | `flushOnAbortRef` | **DELETED** | the stop→flush dance is replaced by the CAS supersede (commit 5) | +| 9 | `interruptNextSendRef` | **DELETED** | the server injects the interrupt note from the supersede itself | +| 10 | `supersedeRetryRef` | **DELETED** (commit 5) | the client 409 retry ladder is gone; CAS supersede replaces it | +| 11 | `stopPendingRef` | **FSM phase `stopping`** | the deferred stop fires from the chat-id adoption effect while `stopping` | +| 12 | `mountedRef` | **retained (React liveness)** | orthogonal to run-lifecycle; gates imperative onFinish side-effects post-unmount. Epoch (I1) handles stale COMMAND-outcomes; DISPOSE bumps it | +| 13 | `attemptResumeRef` | **FSM `ATTACH_START` + run-fact** | mount arms attach ONLY on a confirmed active run (commit 4b: streaming-tail status, or POST /run for a user tail) | +| 14 | `stripRef` | **data** (attachStrategy) | strip+replay detail; the `resumeStream` effect reads it | +| 15 | `strippedRowRef` | **data** (attachStrategy) | the anchor row | +| 16 | `attachAbortRef` | **effect-owned controller** | aborted by the `abortAttach` effect in cleanup (I5) | +| 17–25 | `chatIdRef`, `openPageRef`, `getEditorSelectionRef`, `roleIdRef`, `stableIdRef`, `queuedRef`, `sendMessageRef`, `statusRef`, `lastForwardedChatIdRef` | **data** (identity/send mirrors) | unchanged — not lifecycle flags | +| NEW | `pendingSupersedeRef` | **data** (send-plumbing) | the runId injected into the next `POST /stream {supersede}`; the single replacement for the 3 DELETED one-shots (#8/#9/#10) — net −2 refs | +| NEW | `idleCapTimerRef` | **effect-owned timer** | the stalled inactivity cap → `POLL_IDLE_CAP` (commit 4a); not a flag | -Run-lifecycle FLAGS eliminated by the FSM: #1–#13 (10 collapse into phase/ctx/effects; -#10 is deleted). Identity/data mirrors (#14–#25 minus the two attach-strategy/effect -items) intentionally stay — they are not lifecycle flags. +Net: the 13 lifecycle flags (#1–#13) are eliminated (7 → FSM phase/ctx/epoch/event, +3 deleted, `reconnectTimerRef`/`attachAbortRef` become effect-owned controllers, +`mountedRef` retained as React liveness). Two effect-owned timers + one send-plumbing +data ref are added — none is a boolean lifecycle latch. --- diff --git a/apps/client/src/features/ai-chat/state/run-fsm.test.ts b/apps/client/src/features/ai-chat/state/run-fsm.test.ts index e297095b..62833afe 100644 --- a/apps/client/src/features/ai-chat/state/run-fsm.test.ts +++ b/apps/client/src/features/ai-chat/state/run-fsm.test.ts @@ -14,7 +14,10 @@ function run(m: Machine, ...events: Event[]): Machine { return events.reduce(reduce, m); } function withRunFact(runId = "run-1"): Machine { - return { ...initialMachine(), ctx: { epoch: 0, ownership: "local", runFact: { runId } } }; + return { + ...initialMachine(), + ctx: { epoch: 0, ownership: "local", runFact: { runId }, liveFollow: false }, + }; } function effectTypes(m: Machine): string[] { return m.effects.map((e) => e.type); @@ -140,6 +143,38 @@ describe("run-fsm — commit 3: repeated reconnect cycles", () => { expect(hasEffect(m, "scheduleReconnect")).toBe(true); }); + it("a MOUNT-attach observer drop falls to POLL, not the reconnect ladder", () => { + // Distinguishes commit 3 from a one-shot resume: an observer that never + // live-followed (liveFollow false) polls on a drop. + let m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" }); + m = reduce(m, { type: "ATTACH_LIVE", epoch: m.ctx.epoch }); + expect(m.ctx.ownership).toBe("observer"); + expect(m.ctx.liveFollow).toBe(false); + m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: true, epoch: m.ctx.epoch }); + expect(m.phase.name).toBe("polling"); + expect(hasEffect(m, "armPoll")).toBe(true); + }); + + it("STREAM_INCOMPLETE (observer starved/torn finish) → polling", () => { + let m = reduce(initialMachine(), { type: "ATTACH_START", runId: "r" }); + m = reduce(m, { type: "ATTACH_LIVE", epoch: m.ctx.epoch }); + m = reduce(m, { type: "STREAM_INCOMPLETE", reason: "starved", epoch: m.ctx.epoch }); + expect(m.phase).toEqual({ name: "polling", reason: "starved" }); + expect(hasEffect(m, "armPoll")).toBe(true); + }); + + it("liveFollow is set on the first local drop and kept across a re-attach", () => { + let m = withRunFact("run-3"); + m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: false, epoch: m.ctx.epoch }); + expect(m.ctx.liveFollow).toBe(true); + m = reduce(m, { type: "RECONNECT_ATTEMPT", attempt: 1, epoch: m.ctx.epoch }); + m = reduce(m, { type: "RECONNECT_ATTACHED", epoch: m.ctx.epoch }); + expect(m.ctx.liveFollow).toBe(true); // kept — so a second drop reconnects + // A clean finish clears it. + m = reduce(m, { type: "FINISH_CLEAN", epoch: m.ctx.epoch }); + expect(m.ctx.liveFollow).toBe(false); + }); + it("RECONNECT_NONE backs off through the ladder, then fails at the cap", () => { let m = withRunFact("run-3"); m = reduce(m, { type: "FINISH_DISCONNECT", hasVisibleContent: false, epoch: m.ctx.epoch }); @@ -276,8 +311,8 @@ describe("run-fsm — commit 5: supersede CAS + error classification", () => { }); it("RUN_SUPERSEDED (observer's run killed) → attaching + postRun(observer-follow)", () => { - const observer = { ...withRunFact("run-dead"), ctx: { epoch: 0, ownership: "observer" as const, runFact: { runId: "run-dead" } } }; - const streaming: Machine = { phase: { name: "streaming" }, ctx: observer.ctx, effects: [] }; + const ctx = { epoch: 0, ownership: "observer" as const, runFact: { runId: "run-dead" }, liveFollow: false }; + const streaming: Machine = { phase: { name: "streaming" }, ctx, effects: [] }; const m = reduce(streaming, { type: "RUN_SUPERSEDED" }); expect(m.phase.name).toBe("attaching"); expect(m.effects.find((e) => e.type === "postRun")).toEqual({ diff --git a/apps/client/src/features/ai-chat/state/run-fsm.ts b/apps/client/src/features/ai-chat/state/run-fsm.ts index 89348384..a9ff9cca 100644 --- a/apps/client/src/features/ai-chat/state/run-fsm.ts +++ b/apps/client/src/features/ai-chat/state/run-fsm.ts @@ -87,6 +87,16 @@ export interface Ctx { ownership: Ownership; /** I3: the server-confirmed active run. */ runFact: RunFact; + /** + * Are we FOLLOWING a live run we were locally streaming (the reconnect ladder), + * as opposed to a one-shot mount-attach resume? Both are `ownership: 'observer'`, + * but they recover DIFFERENTLY on a drop: a live-follow drop RE-ENTERS the + * reconnect ladder (#488 commit 3 — the second break after a successful re-attach + * must reconnect again, not fall to silent poll), while a mount-resume drop falls + * to the degraded poll. This is the ctx bit that separates the two WITHOUT a new + * component ref (it is why commit 3 needs the FSM, not a surgical patch). + */ + liveFollow: boolean; } export interface Machine { @@ -131,6 +141,10 @@ export type Event = // -- local turn -- | { type: "SEND_LOCAL" } | { type: "STREAM_START"; runId?: string; epoch?: number } + /** An OBSERVER's attached stream ended WITHOUT reaching terminal (a starved + * clean replay, or a torn resume) — fall to the degraded poll to drive the row + * to its real terminal state. (A live-follow drop uses FINISH_DISCONNECT.) */ + | { type: "STREAM_INCOMPLETE"; reason: PollReason; epoch?: number } | { type: "FINISH_CLEAN"; epoch?: number } | { type: "FINISH_ABORT"; epoch?: number } | { type: "FINISH_DISCONNECT"; hasVisibleContent: boolean; epoch?: number } @@ -182,7 +196,7 @@ export function reconnectDelayMs(attempt: number): number { export function initialMachine(overrides?: Partial): Machine { return { phase: { name: "idle" }, - ctx: { epoch: 0, ownership: "local", runFact: null, ...overrides }, + ctx: { epoch: 0, ownership: "local", runFact: null, liveFollow: false, ...overrides }, effects: [], }; } @@ -241,9 +255,16 @@ export function reduce(m: Machine, event: Event): Machine { m, { name: "sending" }, [{ type: "cancelReconnect" }, { type: "disarmPoll" }], - { ownership: "local" }, + { ownership: "local", liveFollow: false }, ); + case "STREAM_INCOMPLETE": + // An OBSERVER's attached stream ended incomplete (starved / torn) — follow + // the run to terminal via the degraded poll. + return to(m, { name: "polling", reason: event.reason }, { + effects: [{ type: "armPoll", reason: event.reason }], + }); + case "STREAM_START": { // First frame arrived. Adopt the run-fact runId if present. sending -> // streaming; a reconnect/attach that just went live also lands here. @@ -259,7 +280,7 @@ export function reduce(m: Machine, event: Event): Machine { // idle. (The queue flush is a component concern gated by ownership; the // FSM only models the phase.) return to(m, { name: "idle" }, { - ctx: { runFact: null }, + ctx: { runFact: null, liveFollow: false }, effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }], }); @@ -267,34 +288,44 @@ export function reduce(m: Machine, event: Event): Machine { // A user Stop / intentional abort finished. If we were stopping, the // terminal data has now arrived (I4) — go idle. The run-fact is cleared. return to(m, { name: "idle" }, { - ctx: { runFact: null }, + ctx: { runFact: null, liveFollow: false }, effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }], }); case "FINISH_DISCONNECT": - // A LIVE SSE drop. If a run is (or may be) active, recover: - // - visible content already on screen -> keep it, poll to terminal - // (a full replay could clobber the fuller live tail), AND begin the - // live re-attach ladder; + // A LIVE SSE drop. Recovery depends on WHO we are (I2 + liveFollow): + // - a mount-attach OBSERVER (a one-shot resume, NOT live-follow) that drops + // -> the degraded poll drives the row to terminal from the DB. + if (m.ctx.ownership === "observer" && !m.ctx.liveFollow) { + return to(m, { name: "polling", reason: "disconnect-visible" }, { + effects: [{ type: "armPoll", reason: "disconnect-visible" }], + }); + } + // - a LOCAL live turn (first drop) OR a live-follow re-attach (a SUBSEQUENT + // drop) -> (re-)enter the reconnect ladder. #488 commit 3: allowed + // REPEATEDLY — `liveFollow` is kept across a successful re-attach, so the + // second break reconnects again instead of falling to silent poll. + // #488 commit 2: gated on the RUN-FACT (or an existing live-follow), NOT on + // the presence of an assistant message — a setup-phase break still recovers. + // - visible content already on screen -> keep it, ALSO poll to terminal + // (a full replay could clobber the fuller live tail); // - no visible content -> the reconnect ladder rebuilds it. - // #488 commit 2: recovery is gated on the RUN-FACT, not on the presence - // of an assistant message — a break during the setup phase (before the - // first assistant frame) still has an active detached run to pick up. - if (m.ctx.runFact) { + if (m.ctx.runFact || m.ctx.liveFollow) { const effects: Effect[] = [ { type: "scheduleReconnect", attempt: 1, delayMs: reconnectDelayMs(1) }, ]; if (event.hasVisibleContent) effects.push({ type: "armPoll", reason: "disconnect-visible" }); return command(m, { name: "reconnecting", attempt: 1, failed: false }, effects, { ownership: "observer", + liveFollow: true, }); } // No run to recover: a plain disconnect. Surface the terminal notice. - return to(m, { name: "idle" }, { ctx: { runFact: null } }); + return to(m, { name: "idle" }, { ctx: { runFact: null, liveFollow: false } }); case "FINISH_ERROR": return to(m, { name: "error", kind: event.kind }, { - ctx: { runFact: null }, + ctx: { runFact: null, liveFollow: false }, effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }], }); @@ -399,7 +430,7 @@ export function reduce(m: Machine, event: Event): Machine { // idle and disarm everything (I4: this is a DATA-driven exit, incl. exit // from `stopping`). return to(m, { name: "idle" }, { - ctx: { runFact: null }, + ctx: { runFact: null, liveFollow: false }, effects: [{ type: "disarmPoll" }, { type: "cancelReconnect" }], }); @@ -424,7 +455,7 @@ export function reduce(m: Machine, event: Event): Machine { m.phase.name === "stopping" ) { return to(m, { name: "idle" }, { - ctx: { runFact: null }, + ctx: { runFact: null, liveFollow: false }, effects: [{ type: "cancelReconnect" }, { type: "disarmPoll" }], }); } @@ -438,12 +469,20 @@ export function reduce(m: Machine, event: Event): Machine { // ---- stop ---------------------------------------------------------- case "STOP_REQUESTED": // Authoritative stop of a detached run. Enter `stopping` and fire stopRun + - // abort the local/attach reader. Exit is by DATA (I4), never by the HTTP - // response — so no transition keys off stopRun's return. + // abort the local/attach reader. ALSO arm the poll so the terminal row is + // observed — the exit is by DATA (I4: a terminal row / negative run-fact), + // never by the stopRun HTTP response (which returns after abort, before + // finalization). For a local turn the onFinish FINISH_ABORT exits first and + // disarms; for an observer the poll drives to the aborted terminal. return command( m, { name: "stopping" }, - [{ type: "stopRun" }, { type: "abortAttach" }, { type: "cancelReconnect" }], + [ + { type: "stopRun" }, + { type: "abortAttach" }, + { type: "cancelReconnect" }, + { type: "armPoll", reason: "attach-none" }, + ], ); // ---- supersede (CAS) ---------------------------------------------- @@ -460,7 +499,9 @@ export function reduce(m: Machine, event: Event): Machine { // CAS succeeded (old run stopped/settled, slot taken, new run begun). We // are now the local streamer of the NEW run. Adopt its runId if provided. const runFact = event.runId ? { runId: event.runId } : m.ctx.runFact; - return to(m, { name: "streaming" }, { ctx: { ownership: "local", runFact } }); + return to(m, { name: "streaming" }, { + ctx: { ownership: "local", runFact, liveFollow: false }, + }); } case "SUPERSEDE_MISMATCH": @@ -497,11 +538,16 @@ export function reduce(m: Machine, event: Event): Machine { case "DISPOSE": // Unmount: abort in-flight controllers, drop timers, and bump the epoch so // NO late callback can drive this (now dead) machine (I5). - return command(m, { name: "idle" }, [ - { type: "abortAttach" }, - { type: "cancelReconnect" }, - { type: "disarmPoll" }, - ]); + return command( + m, + { name: "idle" }, + [ + { type: "abortAttach" }, + { type: "cancelReconnect" }, + { type: "disarmPoll" }, + ], + { liveFollow: false }, + ); default: { // Exhaustiveness guard.