feat(ai-chat): авто-реконнект к detached-рану после живого обрыва SSE #432
@@ -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
|
||||
|
||||
@@ -739,3 +739,170 @@ function renderResumable(initialRows: IAiChatMessageRow[]) {
|
||||
act(() => view.rerender(<Wrapper rows={rows} />));
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<ReconnectState>(null);
|
||||
const reconnectStateRef = useRef<ReconnectState>(null);
|
||||
const setReconnectStatePair = useCallback((s: ReconnectState) => {
|
||||
reconnectStateRef.current = s;
|
||||
setReconnectState(s);
|
||||
}, []);
|
||||
const reconnectTimerRef = useRef<ReturnType<typeof setTimeout> | 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=<pinned id>, 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).
|
||||
<Alert
|
||||
variant="light"
|
||||
color="gray"
|
||||
p="xs"
|
||||
mb="xs"
|
||||
style={{ flexShrink: 0 }}
|
||||
>
|
||||
<Group gap={8} wrap="nowrap" align="center">
|
||||
{reconnectState.phase === "trying" ? (
|
||||
<>
|
||||
<Loader size={14} color="gray" style={{ flex: "none" }} />
|
||||
<Text size="sm" lh={1.3} c="dimmed">
|
||||
{t("Connection lost — reconnecting…")}
|
||||
{` (${reconnectState.attempt}/${RECONNECT_MAX_ATTEMPTS})`}
|
||||
</Text>
|
||||
</>
|
||||
) : (
|
||||
<>
|
||||
<Text size="sm" lh={1.3} c="dimmed" style={{ flex: 1 }}>
|
||||
{t("Couldn't reconnect to the answer.")}
|
||||
</Text>
|
||||
<Button
|
||||
size="compact-xs"
|
||||
variant="light"
|
||||
color="gray"
|
||||
onClick={retryReconnect}
|
||||
>
|
||||
{t("Retry")}
|
||||
</Button>
|
||||
</>
|
||||
)}
|
||||
</Group>
|
||||
</Alert>
|
||||
) : stopNotice ? (
|
||||
<ChatStoppedNotice
|
||||
text={
|
||||
|
||||
@@ -17,10 +17,24 @@ import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
|
||||
/** How long a finished entry is retained for late attach (replay + immediate end). */
|
||||
export const RUN_STREAM_RETAIN_FINISHED_MS = 30_000;
|
||||
|
||||
/** Per-run replay buffer cap. Past this the buffer is dropped (attach -> 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;
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user