From abc1124d27fe6a9c5a764d60024b0d7b41f9fef6 Mon Sep 17 00:00:00 2001 From: James Magahern Date: Thu, 23 Jul 2026 16:32:20 -0700 Subject: [PATCH] fix for PWA background/foreground --- web/src/App.tsx | 382 ++++++++++++++++++++++++++++++++++----------- web/src/lib/api.ts | 162 ++----------------- 2 files changed, 301 insertions(+), 243 deletions(-) diff --git a/web/src/App.tsx b/web/src/App.tsx index b4f6af7..698535e 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -139,6 +139,7 @@ const ALL_PROVIDERS: Provider[] = [...BASE_PROVIDERS, "hermes-agent"]; const MODEL_PREFERENCES_STORAGE_KEY = "sybil:modelPreferencesByProvider"; const QUICK_QUESTION_MODEL_SELECTION_STORAGE_KEY = "sybil:quickQuestionModelSelection"; +const STREAM_RESUME_RETRY_DELAYS_MS = [0, 250, 750, 1500]; type ProviderModelPreferences = Record; @@ -159,6 +160,73 @@ const EMPTY_ACTIVE_RUNS: ActiveRunsState = { searches: {}, }; +function waitForAppForeground() { + if (typeof document === "undefined" || typeof window === "undefined") return Promise.resolve(); + if (document.visibilityState !== "hidden" && navigator.onLine !== false) return Promise.resolve(); + + return new Promise((resolve) => { + const finishIfReady = () => { + if (document.visibilityState === "hidden" || navigator.onLine === false) return; + document.removeEventListener("visibilitychange", finishIfReady); + window.removeEventListener("pageshow", finishIfReady); + window.removeEventListener("online", finishIfReady); + resolve(); + }; + + document.addEventListener("visibilitychange", finishIfReady); + window.addEventListener("pageshow", finishIfReady); + window.addEventListener("online", finishIfReady); + }); +} + +function waitForRetry(delayMs: number) { + if (!delayMs) return Promise.resolve(); + return new Promise((resolve) => window.setTimeout(resolve, delayMs)); +} + +async function retryAfterAppResume(operation: () => Promise) { + await waitForAppForeground(); + let lastError: unknown = new Error("Unable to reconnect"); + + for (const delayMs of STREAM_RESUME_RETRY_DELAYS_MS) { + await waitForRetry(delayMs); + try { + return await operation(); + } catch (err) { + lastError = err; + const message = err instanceof Error ? err.message : String(err); + if (message.includes("bearer token")) throw err; + } + } + + throw lastError; +} + +function getActiveRunsAfterResume() { + return retryAfterAppResume(getActiveRuns); +} + +function isRecoverableStreamDisconnect(error: unknown) { + if (error instanceof TypeError) return true; + const message = (error instanceof Error ? error.message : String(error)).toLowerCase(); + return ( + message.includes("failed to fetch") || + message.includes("load failed") || + message.includes("network error") || + message.includes("networkerror") || + message.includes("no response stream") || + message.includes("stream disconnected before completion") + ); +} + +function hasNewPersistedUserMessage(chat: ChatDetail, previousMessageIds: Set, content: string, attachments: ChatAttachment[]) { + return chat.messages.some((message) => { + if (previousMessageIds.has(message.id) || message.role !== "user" || message.content !== content) return false; + const persistedAttachmentIds = new Set(getMessageAttachments(message.metadata).map((attachment) => attachment.id)); + return attachments.length === persistedAttachmentIds.size && attachments.every((attachment) => persistedAttachmentIds.has(attachment.id)); + }); +} + const TRANSCRIPT_BOTTOM_GAP = 20; const REPLY_SCROLL_BUFFER_MIN = 288; const REPLY_SCROLL_BUFFER_MAX = 576; @@ -870,6 +938,7 @@ export default function App() { const selectedItemRef = useRef(null); const pendingTitleGenerationRef = useRef>(new Set()); const chatStreamAbortRefs = useRef>(new Map()); + const backgroundSuspendedChatStreamsRef = useRef>(new Set()); const searchRunAbortRefs = useRef>(new Map()); const quickQuestionAbortRef = useRef(null); const searchRunCountersRef = useRef>(new Map()); @@ -977,6 +1046,7 @@ export default function App() { controller.abort(); } chatStreamAbortRefs.current.clear(); + backgroundSuspendedChatStreamsRef.current.clear(); for (const controller of searchRunAbortRefs.current.values()) { controller.abort(); } @@ -1143,6 +1213,36 @@ export default function App() { return () => window.clearInterval(interval); }, [isAuthenticated]); + useEffect(() => { + if (!isAuthenticated) return; + + const suspendChatStreams = () => { + for (const [chatId, controller] of chatStreamAbortRefs.current) { + backgroundSuspendedChatStreamsRef.current.add(chatId); + controller.abort(); + } + }; + const handleVisibilityChange = () => { + if (document.visibilityState === "hidden") { + suspendChatStreams(); + return; + } + void refreshActiveRuns(); + }; + const handlePageShow = () => { + void refreshActiveRuns(); + }; + + document.addEventListener("visibilitychange", handleVisibilityChange); + window.addEventListener("pagehide", suspendChatStreams); + window.addEventListener("pageshow", handlePageShow); + return () => { + document.removeEventListener("visibilitychange", handleVisibilityChange); + window.removeEventListener("pagehide", suspendChatStreams); + window.removeEventListener("pageshow", handlePageShow); + }; + }, [isAuthenticated]); + useEffect(() => { const onPopState = () => { setContextMenu(null); @@ -2051,6 +2151,7 @@ export default function App() { if (!baseChat || baseChat.id !== chatId) { baseChat = await getChat(chatId); } + const previousMessageIds = new Set(baseChat.messages.map((message) => message.id)); setPendingChatStates((current) => ({ ...current, @@ -2104,6 +2205,9 @@ export default function App() { } let streamErrorMessage: string | null = null; + const abortController = new AbortController(); + chatStreamAbortRefs.current.set(chatId, abortController); + const target: SidebarSelection = { kind: "chat", id: chatId }; try { await runCompletionStream( @@ -2161,13 +2265,15 @@ export default function App() { onError: (payload) => { streamErrorMessage = payload.message; }, - } + }, + { signal: abortController.signal } ); if (streamErrorMessage) { throw new Error(streamErrorMessage); } + backgroundSuspendedChatStreamsRef.current.delete(chatId); await refreshCollections(); const currentSelection = selectedItemRef.current; if (currentSelection?.kind === "chat" && currentSelection.id === chatId) { @@ -2178,13 +2284,62 @@ export default function App() { if (currentSelection?.kind === "chat" && currentSelection.id === chatId) { requestSettleTranscriptTailSpacer(); } - return { kind: "chat", id: chatId }; + return target; } catch (err) { + const wasBackgroundSuspended = + abortController.signal.aborted && backgroundSuspendedChatStreamsRef.current.delete(chatId); + const shouldResumeStream = + wasBackgroundSuspended || + (!abortController.signal.aborted && streamErrorMessage === null && isRecoverableStreamDisconnect(err)); + + if (shouldResumeStream) { + if (chatStreamAbortRefs.current.get(chatId) === abortController) { + chatStreamAbortRefs.current.delete(chatId); + } + + try { + const resumedRuns = await getActiveRunsAfterResume(); + setActiveRuns(buildActiveRunsState(resumedRuns)); + + if (resumedRuns.chats.includes(chatId)) { + removePendingChatState(chatId); + await attachToActiveChatStream(chatId, true); + return target; + } + + const persistedChat = await retryAfterAppResume(() => getChat(chatId)); + await refreshCollections(target); + if (isCurrentSelection(target)) { + setSelectedChat(persistedChat); + setSelectedSearch(null); + } + removePendingChatState(chatId); + removeActiveRun("chat", chatId); + + if (hasNewPersistedUserMessage(persistedChat, previousMessageIds, content, attachments)) { + requestSettleTranscriptTailSpacer(); + return target; + } + + err = new Error( + wasBackgroundSuspended + ? "The app was backgrounded before the message reached the server. Please send it again." + : "The connection dropped before the message reached the server. Please send it again." + ); + } catch (resumeError) { + err = resumeError; + } + } + removePendingChatState(chatId); removeActiveRun("chat", chatId); pendingReplyScrollRef.current = false; setTranscriptTailSpacer(TRANSCRIPT_BOTTOM_GAP); throw err; + } finally { + if (chatStreamAbortRefs.current.get(chatId) === abortController) { + chatStreamAbortRefs.current.delete(chatId); + } } }; @@ -2329,113 +2484,154 @@ export default function App() { return target; }; - const attachToActiveChatStream = async (chatId: string) => { + async function attachToActiveChatStream(chatId: string, resetFromServer = false) { if (chatStreamAbortRefs.current.has(chatId)) return; const target: SidebarSelection = { kind: "chat", id: chatId }; - const abortController = new AbortController(); - chatStreamAbortRefs.current.set(chatId, abortController); addActiveRun("chat", chatId); - - let streamErrorMessage: string | null = null; + let shouldResetFromServer = resetFromServer; try { - const baseChat = selectedChat?.id === chatId ? selectedChat : await getChat(chatId); - setPendingChatStates((current) => { - if (current[chatId]) return current; - return { - ...current, - [chatId]: { - messages: baseChat.messages.concat({ - id: `temp-assistant-attach-${chatId}-${Date.now()}`, - createdAt: new Date().toISOString(), - role: "assistant", - content: "", - name: null, - metadata: null, - }), - }, - }; - }); + while (true) { + const abortController = new AbortController(); + chatStreamAbortRefs.current.set(chatId, abortController); + let streamErrorMessage: string | null = null; - await attachCompletionStream( - chatId, - { - onToolCall: (payload) => { - setPendingChatStates((current) => { - const pendingState = current[chatId]; - if (!pendingState) return current; - return { - ...current, - [chatId]: { - messages: upsertOptimisticToolMessage(pendingState.messages, payload, "temp-assistant-"), - }, - }; - }); - }, - onDelta: (payload) => { - if (!payload.text) return; - setPendingChatStates((current) => { - const pendingState = current[chatId]; - if (!pendingState) return current; - let updated = false; - const nextMessages = pendingState.messages.map((message, index, all) => { - const isTarget = index === all.length - 1 && message.id.startsWith("temp-assistant-"); - if (!isTarget) return message; - updated = true; - return { ...message, content: message.content + payload.text }; - }); - return updated ? { ...current, [chatId]: { messages: nextMessages } } : current; - }); - }, - onDone: (payload) => { - setPendingChatStates((current) => { - const pendingState = current[chatId]; - if (!pendingState) return current; - let updated = false; - const nextMessages = pendingState.messages.map((message, index, all) => { - const isTarget = index === all.length - 1 && message.id.startsWith("temp-assistant-"); - if (!isTarget) return message; - updated = true; - return { ...message, content: payload.text }; - }); - return updated ? { ...current, [chatId]: { messages: nextMessages } } : current; - }); - }, - onError: (payload) => { - streamErrorMessage = payload.message; - }, - }, - { signal: abortController.signal } - ); + try { + const baseChat = await getChat(chatId); + const resetPendingState = shouldResetFromServer; + shouldResetFromServer = false; + setPendingChatStates((current) => { + if (!resetPendingState && current[chatId]) return current; + return { + ...current, + [chatId]: { + messages: baseChat.messages.concat({ + id: `temp-assistant-attach-${chatId}-${Date.now()}`, + createdAt: new Date().toISOString(), + role: "assistant", + content: "", + name: null, + metadata: null, + }), + }, + }; + }); - if (streamErrorMessage) { - throw new Error(streamErrorMessage); - } + await attachCompletionStream( + chatId, + { + onToolCall: (payload) => { + setPendingChatStates((current) => { + const pendingState = current[chatId]; + if (!pendingState) return current; + return { + ...current, + [chatId]: { + messages: upsertOptimisticToolMessage(pendingState.messages, payload, "temp-assistant-"), + }, + }; + }); + }, + onDelta: (payload) => { + if (!payload.text) return; + setPendingChatStates((current) => { + const pendingState = current[chatId]; + if (!pendingState) return current; + let updated = false; + const nextMessages = pendingState.messages.map((message, index, all) => { + const isTarget = index === all.length - 1 && message.id.startsWith("temp-assistant-"); + if (!isTarget) return message; + updated = true; + return { ...message, content: message.content + payload.text }; + }); + return updated ? { ...current, [chatId]: { messages: nextMessages } } : current; + }); + }, + onDone: (payload) => { + setPendingChatStates((current) => { + const pendingState = current[chatId]; + if (!pendingState) return current; + let updated = false; + const nextMessages = pendingState.messages.map((message, index, all) => { + const isTarget = index === all.length - 1 && message.id.startsWith("temp-assistant-"); + if (!isTarget) return message; + updated = true; + return { ...message, content: payload.text }; + }); + return updated ? { ...current, [chatId]: { messages: nextMessages } } : current; + }); + }, + onError: (payload) => { + streamErrorMessage = payload.message; + }, + }, + { signal: abortController.signal } + ); - await refreshCollections(); - if (isCurrentSelection(target)) { - await refreshChat(chatId); - } - } catch (err) { - if (abortController.signal.aborted) return; - const message = err instanceof Error ? err.message : String(err); - if (message.includes("active chat stream not found")) { - await refreshActiveRuns(); - if (isCurrentSelection(target)) await refreshChat(chatId); - } else if (message.includes("bearer token")) { - handleAuthFailure(message); - } else if (isCurrentSelection(target)) { - setError(message); + if (streamErrorMessage) { + throw new Error(streamErrorMessage); + } + + backgroundSuspendedChatStreamsRef.current.delete(chatId); + await refreshCollections(); + if (isCurrentSelection(target)) { + await refreshChat(chatId); + } + return; + } catch (err) { + const wasBackgroundSuspended = + abortController.signal.aborted && backgroundSuspendedChatStreamsRef.current.delete(chatId); + const shouldResumeStream = + wasBackgroundSuspended || + (!abortController.signal.aborted && streamErrorMessage === null && isRecoverableStreamDisconnect(err)); + + if (shouldResumeStream) { + shouldResetFromServer = true; + try { + const resumedRuns = await getActiveRunsAfterResume(); + setActiveRuns(buildActiveRunsState(resumedRuns)); + if (resumedRuns.chats.includes(chatId)) { + continue; + } + + const persistedChat = await retryAfterAppResume(() => getChat(chatId)); + if (isCurrentSelection(target)) { + setSelectedChat(persistedChat); + setSelectedSearch(null); + } + void refreshCollections(target); + return; + } catch (resumeError) { + err = resumeError; + } + } + if (abortController.signal.aborted && !shouldResumeStream) return; + + const message = err instanceof Error ? err.message : String(err); + if (message.includes("active chat stream not found")) { + await refreshActiveRuns(); + if (isCurrentSelection(target)) await refreshChat(chatId); + } else if (message.includes("bearer token")) { + handleAuthFailure(message); + } else if (isCurrentSelection(target)) { + setError(message); + } + return; + } finally { + if (chatStreamAbortRefs.current.get(chatId) === abortController) { + chatStreamAbortRefs.current.delete(chatId); + } + } } } finally { - chatStreamAbortRefs.current.delete(chatId); + backgroundSuspendedChatStreamsRef.current.delete(chatId); removePendingChatState(chatId); removeActiveRun("chat", chatId); if (isCurrentSelection(target)) { requestSettleTranscriptTailSpacer(); } } - }; + } const attachToActiveSearchStream = async (searchId: string) => { if (searchRunAbortRefs.current.has(searchId)) return; diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 904bc92..9185daa 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -450,6 +450,7 @@ async function readSseStream(response: Response, dispatch: (eventName: string, p let buffer = ""; let eventName = "message"; let dataLines: string[] = []; + let sawTerminalEvent = false; const flushEvent = () => { if (!dataLines.length) { @@ -466,6 +467,9 @@ async function readSseStream(response: Response, dispatch: (eventName: string, p } dispatch(eventName, payload); + if (eventName === "done" || eventName === "error") { + sawTerminalEvent = true; + } dataLines = []; eventName = "message"; @@ -505,6 +509,10 @@ async function readSseStream(response: Response, dispatch: (eventName: string, p } } flushEvent(); + + if (!sawTerminalEvent) { + throw new Error("Stream disconnected before completion"); + } } export async function runSearchStream( @@ -528,87 +536,14 @@ export async function runSearchStream( signal: options?.signal, }); - if (!response.ok) { - const fallback = `${response.status} ${response.statusText}`; - let message = fallback; - try { - const body = (await response.json()) as { message?: string }; - if (body.message) message = body.message; - } catch { - // keep fallback message - } - throw new Error(message); - } - - if (!response.body) { - throw new Error("No response stream"); - } - - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; - let eventName = "message"; - let dataLines: string[] = []; - - const flushEvent = () => { - if (!dataLines.length) { - eventName = "message"; - return; - } - - const dataText = dataLines.join("\n"); - let payload: any = null; - try { - payload = JSON.parse(dataText); - } catch { - payload = { message: dataText }; - } - + await readSseStream(response, (eventName, payload) => { if (eventName === "search_results") handlers.onSearchResults?.(payload); else if (eventName === "search_error") handlers.onSearchError?.(payload); else if (eventName === "answer") handlers.onAnswer?.(payload); else if (eventName === "answer_error") handlers.onAnswerError?.(payload); else if (eventName === "done") handlers.onDone?.(payload); else if (eventName === "error") handlers.onError?.(payload); - - dataLines = []; - eventName = "message"; - }; - - while (true) { - const { value, done } = await reader.read(); - if (done) break; - - buffer += decoder.decode(value, { stream: true }); - let newlineIndex = buffer.indexOf("\n"); - - while (newlineIndex >= 0) { - const rawLine = buffer.slice(0, newlineIndex); - buffer = buffer.slice(newlineIndex + 1); - const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; - - if (!line) { - flushEvent(); - } else if (line.startsWith("event:")) { - eventName = line.slice("event:".length).trim(); - } else if (line.startsWith("data:")) { - dataLines.push(line.slice("data:".length).trimStart()); - } - - newlineIndex = buffer.indexOf("\n"); - } - } - - buffer += decoder.decode(); - if (buffer.length) { - const line = buffer.endsWith("\r") ? buffer.slice(0, -1) : buffer; - if (line.startsWith("event:")) { - eventName = line.slice("event:".length).trim(); - } else if (line.startsWith("data:")) { - dataLines.push(line.slice("data:".length).trimStart()); - } - } - flushEvent(); + }); } export async function attachSearchStream(searchId: string, handlers: RunSearchStreamHandlers, options?: { signal?: AbortSignal }) { @@ -679,86 +614,13 @@ export async function runCompletionStream( signal: options?.signal, }); - if (!response.ok) { - const fallback = `${response.status} ${response.statusText}`; - let message = fallback; - try { - const body = (await response.json()) as { message?: string }; - if (body.message) message = body.message; - } catch { - // keep fallback message - } - throw new Error(message); - } - - if (!response.body) { - throw new Error("No response stream"); - } - - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; - let eventName = "message"; - let dataLines: string[] = []; - - const flushEvent = () => { - if (!dataLines.length) { - eventName = "message"; - return; - } - - const dataText = dataLines.join("\n"); - let payload: any = null; - try { - payload = JSON.parse(dataText); - } catch { - payload = { message: dataText }; - } - + await readSseStream(response, (eventName, payload) => { if (eventName === "meta") handlers.onMeta?.(payload); else if (eventName === "tool_call") handlers.onToolCall?.(payload); else if (eventName === "delta") handlers.onDelta?.(payload); else if (eventName === "done") handlers.onDone?.(payload); else if (eventName === "error") handlers.onError?.(payload); - - dataLines = []; - eventName = "message"; - }; - - while (true) { - const { value, done } = await reader.read(); - if (done) break; - - buffer += decoder.decode(value, { stream: true }); - let newlineIndex = buffer.indexOf("\n"); - - while (newlineIndex >= 0) { - const rawLine = buffer.slice(0, newlineIndex); - buffer = buffer.slice(newlineIndex + 1); - const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; - - if (!line) { - flushEvent(); - } else if (line.startsWith("event:")) { - eventName = line.slice("event:".length).trim(); - } else if (line.startsWith("data:")) { - dataLines.push(line.slice("data:".length).trimStart()); - } - - newlineIndex = buffer.indexOf("\n"); - } - } - - buffer += decoder.decode(); - if (buffer.length) { - const line = buffer.endsWith("\r") ? buffer.slice(0, -1) : buffer; - if (line.startsWith("event:")) { - eventName = line.slice("event:".length).trim(); - } else if (line.startsWith("data:")) { - dataLines.push(line.slice("data:".length).trimStart()); - } - } - flushEvent(); + }); } export async function attachCompletionStream(chatId: string, handlers: CompletionStreamHandlers, options?: { signal?: AbortSignal }) {