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 dc08e756..7fa218e8 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,11 +86,19 @@ const MIN_HEIGHT = 400; // Margin kept between the window and the viewport edges while dragging. const EDGE_MARGIN = 8; -// #184 phase 1.5: hard cap on the degraded-poll fallback. The poll is armed when -// a resume attempt could not attach to the live run and disarmed by the thread on -// settle / local stream; this cap is the ONLY backstop against an endless tick -// (a stuck 'streaming' row before the boot-sweep, or a user-tail 204 with no run). -const DEGRADED_POLL_MAX_MS = 10 * 60_000; +// #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; /** Compact token formatter: 1.2M / 3.4k / 950. */ function formatTokens(n: number): string { @@ -254,9 +262,12 @@ export default function AiChatWindow() { // onResumeFallback(true); the thread disarms it on settle / local stream. The // window only OWNS the timer (armedAtRef stamps when it was armed for the cap). const [degradedPoll, setDegradedPoll] = useState(false); - const armedAtRef = useRef(0); + // #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) armedAtRef.current = Date.now(); + if (active) lastActivityAtRef.current = Date.now(); setDegradedPoll(active); }, []); // Reset the degraded poll whenever the open chat changes: it is scoped to the @@ -269,18 +280,28 @@ export default function AiChatWindow() { useAiChatMessagesQuery( activeChatId ?? undefined, // DELIBERATELY DUMB (invariant 8 / task 2.4): poll every 2.5s while armed - // and under the 10-min cap; otherwise off. NO error checks (TanStack v5 - // resets fetchFailureCount each fetch, so consecutive errors are not - // expressible — and the poll must survive a server restart) and NO tail - // checks (the settled/local-stream semantics live in ChatThread, which - // disarms via onResumeFallback(false)). The time cap is the only backstop. + // 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() - armedAtRef.current < DEGRADED_POLL_MAX_MS + Date.now() - lastActivityAtRef.current < DEGRADED_POLL_IDLE_MAX_MS ? 2500 : false, ); + // #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 20b0a3ec..557b4d41 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 @@ -739,3 +739,170 @@ function renderResumable(initialRows: 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. + const liveMsg = { + id: "a2", + role: "assistant", + parts: [{ type: "text", text: "partial live answer" }], + }; + + 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(); + }); + afterEach(() => { + vi.useRealTimers(); + 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(); + act(() => { + h.state.onFinish?.({ + message: liveMsg, + isAbort: false, + isDisconnect: true, + isError: false, + }); + }); + 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 }), + ); + 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. + 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", + ); + }); + + 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(); + advanceToAttempt(1); + await reconnect200(); + expect(screen.queryByText(/reconnecting/i)).toBeNull(); + }); + + it("a 204 arms the degraded poll and backs off to the next attempt", async () => { + const { onResumeFallback } = renderLiveThenDisconnect(); + advanceToAttempt(1); + expect(h.state.resumeStream).toHaveBeenCalledTimes(1); + await reconnect204(); + // Fallback engaged: the degraded poll is armed (204 -> onNoActiveStream). + 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. + for (let n = 1; n <= 5; n++) { + advanceToAttempt(n); + expect(h.state.resumeStream).toHaveBeenCalledTimes(n); + await reconnect204(); + } + // 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); + }); + expect(h.state.resumeStream).toHaveBeenCalledTimes(6); + expect(screen.getByText(/reconnecting/i)).toBeTruthy(); + }); + + 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, + }); + }); + 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(); + }); +}); 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 5ef54585..60b27db8 100644 --- a/apps/client/src/features/ai-chat/components/chat-thread.tsx +++ b/apps/client/src/features/ai-chat/components/chat-thread.tsx @@ -1,7 +1,17 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useQueryClient } from "@tanstack/react-query"; import { generateId } from "ai"; -import { ActionIcon, Box, Group, Stack, Text, Tooltip } from "@mantine/core"; +import { + ActionIcon, + Alert, + Box, + Button, + Group, + Loader, + Stack, + Text, + Tooltip, +} from "@mantine/core"; import { IconClockHour4, IconPlayerPlayFilled, @@ -51,6 +61,15 @@ 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; + /** The page the user is currently viewing, sent as chat context. */ export interface OpenPageContext { id: string; @@ -175,6 +194,10 @@ export default function ChatThread({ 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 @@ -378,6 +401,10 @@ export default function ChatThread({ // 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?.(); return response; } catch (err) { // Network throw: same no-onFinish recovery, then rethrow so the SDK @@ -481,6 +508,31 @@ export default function ChatThread({ ); } } + // (2b) #430: 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: pin the CURRENT streaming assistant row as the + // strip/anchor (the live tail is the already-shown partial in `messages`, not + // a persistent row) and re-attach to the live tail via the resumable machinery. + const startedReconnect = + isDisconnect && + !wasResumed && + autonomousRunsEnabled === true && + mountedRef.current && + message?.role === "assistant" && + typeof message.id === "string"; + if (startedReconnect) { + beginReconnect({ + 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 }, + }); + } // (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 @@ -490,9 +542,11 @@ export default function ChatThread({ 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("disconnect"); + else if (isDisconnect) setStopNotice(startedReconnect ? 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 @@ -579,6 +633,106 @@ export default function ChatThread({ const isStreaming = status === "submitted" || status === "streaming"; + // #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, pinning `anchorRow` (the live run's assistant + // row) as the strip/anchor reused by every attempt. + const beginReconnect = useCallback( + (anchorRow: IAiChatMessageRow) => { + if (!autonomousRunsEnabled || !mountedRef.current) return; + strippedRowRef.current = anchorRow; + stripRef.current = true; + 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 @@ -610,7 +764,17 @@ export default function ChatThread({ // (d) 204 means onFinish will NOT fire — clear the suppression flag so it // cannot swallow the NEXT local turn's queue flush. setResumedTurnPair(false); - }, [setMessages, queryClient, onResumeFallback, setResumedTurnPair]); + // (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 @@ -628,6 +792,9 @@ export default function ChatThread({ 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); }; // Mount-only by design; the parent remounts per chat via `key`. // eslint-disable-next-line react-hooks/exhaustive-deps @@ -666,12 +833,27 @@ export default function ChatThread({ 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. // 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. + useEffect(() => { + if (status === "streaming") { + clearReconnectTimer(); + setReconnectStatePair(null); + } + }, [status, clearReconnectTimer, setReconnectStatePair]); + // "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 @@ -719,6 +901,9 @@ export default function ChatThread({ // 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); @@ -740,7 +925,13 @@ export default function ChatThread({ // for this fix. Documented so a future change can address the abort-ordering. stopPendingRef.current = true; } - }, [stop, autonomousRunsEnabled, onServerStop]); + }, [ + 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 @@ -825,6 +1016,43 @@ export default function ChatThread({ detail={errorView.detail} mb="xs" /> + ) : 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). + + + {reconnectState.phase === "trying" ? ( + <> + + + {t("Connection lost — reconnecting…")} + {` (${reconnectState.attempt}/${RECONNECT_MAX_ATTEMPTS})`} + + + ) : ( + <> + + {t("Couldn't reconnect to the answer.")} + + + + )} + + ) : stopNotice ? ( 204). */ -export const RUN_STREAM_MAX_BUFFER_BYTES = 4 * 1024 * 1024; +/** + * Per-run replay buffer cap. Past this the buffer is dropped (attach -> 204, and + * the client falls back to its restore + degraded-poll path, #430). + * + * Raised from 4MB to 32MB (#430): marathon autonomous runs (11-25 min observed) + * stream far more than 4MB of SSE frames, so a live disconnect mid-run would find + * an already-overflowed buffer and could only degrade-poll instead of re-attaching + * to the live tail. 32MB comfortably covers those runs while staying bounded. + * + * Memory cost: this is the WORST-CASE retained size PER ACTIVE run (the buffer is + * freed on finish + retention, or dropped immediately on overflow). With the small + * number of concurrent autonomous runs a single workspace realistically has, 32MB + * each is an acceptable ceiling; the overflow->204->degraded-poll fallback remains + * the backstop for anything larger, so correctness never depends on this bound. + */ +export const RUN_STREAM_MAX_BUFFER_BYTES = 32 * 1024 * 1024; -// 2x the replay cap: a just-written 4MB replay burst alone can never trip the +// 2x the replay cap: a just-written full-replay burst alone can never trip the // per-subscriber cap (see controller); only a genuinely stalled socket can. export const SUBSCRIBER_MAX_BUFFERED_BYTES = 2 * RUN_STREAM_MAX_BUFFER_BYTES; diff --git a/apps/server/src/core/ai-chat/ai-chat-stream-registry.spec.ts b/apps/server/src/core/ai-chat/ai-chat-stream-registry.spec.ts index 5d10d0cf..193b76b1 100644 --- a/apps/server/src/core/ai-chat/ai-chat-stream-registry.spec.ts +++ b/apps/server/src/core/ai-chat/ai-chat-stream-registry.spec.ts @@ -2,6 +2,7 @@ import { AiChatStreamRegistryService, RUN_STREAM_MAX_BUFFER_BYTES, RUN_STREAM_RETAIN_FINISHED_MS, + SUBSCRIBER_MAX_BUFFERED_BYTES, RunStreamCallbacks, } from './ai-chat-stream-registry.service'; @@ -210,9 +211,10 @@ describe('AiChatStreamRegistryService', () => { const att = (await registry.attach(CHAT, false, undefined, c.cb))!; att.start(); - const oneMb = 'x'.repeat(1024 * 1024); - // 5 x 1MB = 5MB > 4MB cap; the 5th frame is the one that crosses. - for (let i = 0; i < 5; i++) src.push(oneMb + i); + // Cap-relative so it survives a buffer-cap change (#430): a quarter-cap frame + // means 5 frames comfortably exceed the replay cap; the last one crosses. + const chunk = 'x'.repeat(Math.floor(RUN_STREAM_MAX_BUFFER_BYTES / 4)); + for (let i = 0; i < 5; i++) src.push(chunk + i); await flush(); const entry = (registry as any).entries.get(CHAT); @@ -220,7 +222,7 @@ describe('AiChatStreamRegistryService', () => { expect(entry.bytes).toBeGreaterThan(RUN_STREAM_MAX_BUFFER_BYTES); // The live subscriber received ALL 5 frames, including the crossing one. expect(c.frames).toHaveLength(5); - expect(c.frames[4]).toBe(oneMb + 4); + expect(c.frames[4]).toBe(chunk + 4); // A NEW attach after overflow gets null (replay buffer is gone). const c2 = collector(); @@ -240,9 +242,11 @@ describe('AiChatStreamRegistryService', () => { const attB = (await registry.attach(CHAT, false, undefined, b.cb))!; attB.start(); - const oneMb = 'x'.repeat(1024 * 1024); - // 9 x 1MB = 9MB > 8MB per-subscriber cap; A's pending overflows, B streams live. - for (let i = 0; i < 9; i++) src.push(oneMb + i); + // Cap-relative so it survives a buffer-cap change (#430): a quarter-of-the- + // per-subscriber-cap frame means 5 frames exceed A's paused-pending cap while + // B streams every frame live. + const chunk = 'x'.repeat(Math.floor(SUBSCRIBER_MAX_BUFFERED_BYTES / 4)); + for (let i = 0; i < 5; i++) src.push(chunk + i); await flush(); const entry = (registry as any).entries.get(CHAT); @@ -250,7 +254,7 @@ describe('AiChatStreamRegistryService', () => { expect(entry.subscribers.size).toBe(1); expect(a.frames).toEqual([]); // paused + overflowed: nothing was delivered // B received every frame live (delivery unaffected by A's overflow). - expect(b.frames).toHaveLength(9); + expect(b.frames).toHaveLength(5); // A's start() (arriving late) degrades to an immediate end, not a partial replay. attA.start();