buzzert/Sybil-2
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix for PWA background/foreground

Browse Source
This commit is contained in:
James Magahern
2026-07-23 16:32:20 -07:00
parent 4721717022
commit abc1124d27
2 changed files with 301 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files

382
web/src/App.tsx
View File

@@ -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<Provider, string | null>;
@@ -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<void>((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<void>((resolve) => window.setTimeout(resolve, delayMs));
}
async function retryAfterAppResume<T>(operation: () => Promise<T>) {
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<string>, 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<SidebarSelection | null>(null);
const pendingTitleGenerationRef = useRef<Set<string>>(new Set());
const chatStreamAbortRefs = useRef<Map<string, AbortController>>(new Map());
const backgroundSuspendedChatStreamsRef = useRef<Set<string>>(new Set());
const searchRunAbortRefs = useRef<Map<string, AbortController>>(new Map());
const quickQuestionAbortRef = useRef<AbortController | null>(null);
const searchRunCountersRef = useRef<Map<string, number>>(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;

162
web/src/lib/api.ts
View File

@@ -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 }) {