Search: async answer/results

2026-02-14 01:53:34 -08:00
parent bec25aa943
commit 769cd6966a
4 changed files with 540 additions and 66 deletions
--- a/web/src/App.tsx
+++ b/web/src/App.tsx
@@ -17,7 +17,7 @@ import {
  listChats,
  listSearches,
  runCompletion,
-  runSearch,
+  runSearchStream,
  type ChatDetail,
  type ChatSummary,
  type CompletionRequestMessage,
@@ -119,6 +119,8 @@ export default function App() {
  const [error, setError] = useState<string | null>(null);
  const transcriptEndRef = useRef<HTMLDivElement>(null);
  const contextMenuRef = useRef<HTMLDivElement>(null);
+  const searchRunAbortRef = useRef<AbortController | null>(null);
+  const searchRunCounterRef = useRef(0);
  const [contextMenu, setContextMenu] = useState<ContextMenuState | null>(null);

  const sidebarItems = useMemo(() => buildSidebarItems(chats, searches), [chats, searches]);
@@ -241,6 +243,13 @@ export default function App() {
    transcriptEndRef.current?.scrollIntoView({ behavior: "smooth", block: "end" });
  }, [draftKind, selectedChat?.messages.length, isSending, selectedItem?.kind]);

+  useEffect(() => {
+    return () => {
+      searchRunAbortRef.current?.abort();
+      searchRunAbortRef.current = null;
+    };
+  }, []);
+
  const messages = selectedChat?.messages ?? [];

  const selectedChatSummary = useMemo(() => {
@@ -411,6 +420,11 @@ export default function App() {
  };

  const handleSendSearch = async (query: string) => {
+    const runId = ++searchRunCounterRef.current;
+    searchRunAbortRef.current?.abort();
+    const abortController = new AbortController();
+    searchRunAbortRef.current = abortController;
+
    let searchId = draftKind === "search" ? null : selectedItem?.kind === "search" ? selectedItem.id : null;

    if (!searchId) {
@@ -460,15 +474,76 @@ export default function App() {
      };
    });

-    const search = await runSearch(searchId, {
-      query,
-      title: query.slice(0, 80),
-      type: "auto",
-      numResults: 10,
-    });
+    try {
+      await runSearchStream(
+        searchId,
+        {
+          query,
+          title: query.slice(0, 80),
+          type: "auto",
+          numResults: 10,
+        },
+        {
+          onSearchResults: (payload) => {
+            if (runId !== searchRunCounterRef.current) return;
+            setSelectedSearch((current) => {
+              if (!current || current.id !== searchId) return current;
+              return {
+                ...current,
+                requestId: payload.requestId ?? current.requestId,
+                error: null,
+                results: payload.results,
+              };
+            });
+          },
+          onSearchError: (payload) => {
+            if (runId !== searchRunCounterRef.current) return;
+            setSelectedSearch((current) => {
+              if (!current || current.id !== searchId) return current;
+              return { ...current, error: payload.error };
+            });
+          },
+          onAnswer: (payload) => {
+            if (runId !== searchRunCounterRef.current) return;
+            setSelectedSearch((current) => {
+              if (!current || current.id !== searchId) return current;
+              return {
+                ...current,
+                answerText: payload.answerText,
+                answerRequestId: payload.answerRequestId,
+                answerCitations: payload.answerCitations,
+                answerError: null,
+              };
+            });
+          },
+          onAnswerError: (payload) => {
+            if (runId !== searchRunCounterRef.current) return;
+            setSelectedSearch((current) => {
+              if (!current || current.id !== searchId) return current;
+              return { ...current, answerError: payload.error };
+            });
+          },
+          onDone: (payload) => {
+            if (runId !== searchRunCounterRef.current) return;
+            setSelectedSearch(payload.search);
+            setSelectedChat(null);
+          },
+          onError: (payload) => {
+            if (runId !== searchRunCounterRef.current) return;
+            setError(payload.message);
+          },
+        },
+        { signal: abortController.signal }
+      );
+    } catch (err) {
+      if (abortController.signal.aborted) return;
+      throw err;
+    } finally {
+      if (runId === searchRunCounterRef.current) {
+        searchRunAbortRef.current = null;
+      }
+    }

-    setSelectedSearch(search);
-    setSelectedChat(null);
    await refreshCollections({ kind: "search", id: searchId });
  };

--- a/web/src/lib/api.ts
+++ b/web/src/lib/api.ts
@@ -68,6 +68,15 @@ export type SearchDetail = {
  results: SearchResultItem[];
 };

+export type SearchRunRequest = {
+  query?: string;
+  title?: string;
+  type?: "auto" | "fast" | "deep" | "instant";
+  numResults?: number;
+  includeDomains?: string[];
+  excludeDomains?: string[];
+};
+
 export type CompletionRequestMessage = {
  role: "system" | "user" | "assistant" | "tool";
  content: string;
@@ -172,22 +181,117 @@ export async function deleteSearch(searchId: string) {
  await api<{ deleted: true }>(`/v1/searches/${searchId}`, { method: "DELETE" });
 }

-export async function runSearch(
+type RunSearchStreamHandlers = {
+  onSearchResults?: (payload: { requestId: string | null; results: SearchResultItem[] }) => void;
+  onSearchError?: (payload: { error: string }) => void;
+  onAnswer?: (payload: { answerText: string | null; answerRequestId: string | null; answerCitations: SearchDetail["answerCitations"] }) => void;
+  onAnswerError?: (payload: { error: string }) => void;
+  onDone?: (payload: { search: SearchDetail }) => void;
+  onError?: (payload: { message: string }) => void;
+};
+
+export async function runSearchStream(
  searchId: string,
-  body: {
-    query?: string;
-    title?: string;
-    type?: "auto" | "fast" | "deep" | "instant";
-    numResults?: number;
-    includeDomains?: string[];
-    excludeDomains?: string[];
-  }
+  body: SearchRunRequest,
+  handlers: RunSearchStreamHandlers,
+  options?: { signal?: AbortSignal }
 ) {
-  const data = await api<{ search: SearchDetail }>(`/v1/searches/${searchId}/run`, {
-    method: "POST",
-    body: JSON.stringify(body),
+  const headers = new Headers({
+    Accept: "text/event-stream",
+    "Content-Type": "application/json",
  });
-  return data.search;
+  if (authToken) {
+    headers.set("Authorization", `Bearer ${authToken}`);
+  }
+
+  const response = await fetch(`${API_BASE_URL}/v1/searches/${searchId}/run/stream`, {
+    method: "POST",
+    headers,
+    body: JSON.stringify(body),
+    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 };
+    }
+
+    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 runCompletion(body: {
--- a/web/src/pages/search-route-page.tsx
+++ b/web/src/pages/search-route-page.tsx
@@ -4,7 +4,7 @@ import { AuthScreen } from "@/components/auth/auth-screen";
 import { SearchResultsPanel } from "@/components/search/search-results-panel";
 import { Button } from "@/components/ui/button";
 import { Input } from "@/components/ui/input";
-import { createSearch, runSearch, type SearchDetail } from "@/lib/api";
+import { createSearch, runSearchStream, type SearchDetail } from "@/lib/api";
 import { useSessionAuth } from "@/hooks/use-session-auth";

 function readQueryFromUrl() {
@@ -41,6 +41,7 @@ export default function SearchRoutePage() {
  const [isRunning, setIsRunning] = useState(false);
  const [error, setError] = useState<string | null>(null);
  const requestCounterRef = useRef(0);
+  const streamAbortRef = useRef<AbortController | null>(null);

  useEffect(() => {
    const onPopState = () => {
@@ -52,6 +53,13 @@ export default function SearchRoutePage() {
    return () => window.removeEventListener("popstate", onPopState);
  }, []);

+  useEffect(() => {
+    return () => {
+      streamAbortRef.current?.abort();
+      streamAbortRef.current = null;
+    };
+  }, []);
+
  const runQuery = async (query: string) => {
    const trimmed = query.trim();
    if (!trimmed) {
@@ -61,6 +69,9 @@ export default function SearchRoutePage() {
    }

    const requestId = ++requestCounterRef.current;
+    streamAbortRef.current?.abort();
+    const abortController = new AbortController();
+    streamAbortRef.current = abortController;
    setError(null);
    setIsRunning(true);

@@ -86,16 +97,78 @@ export default function SearchRoutePage() {
        query: trimmed,
        title: trimmed.slice(0, 80),
      });
-      const result = await runSearch(created.id, {
-        query: trimmed,
-        title: trimmed.slice(0, 80),
-        type: "auto",
-        numResults: 10,
-      });
-      if (requestId === requestCounterRef.current) {
-        setSearch(result);
-      }
+      if (requestId !== requestCounterRef.current) return;
+
+      setSearch((current) =>
+        current
+          ? {
+              ...current,
+              id: created.id,
+              title: created.title,
+              query: created.query,
+              createdAt: created.createdAt,
+              updatedAt: created.updatedAt,
+            }
+          : current
+      );
+
+      await runSearchStream(
+        created.id,
+        {
+          query: trimmed,
+          title: trimmed.slice(0, 80),
+          type: "auto",
+          numResults: 10,
+        },
+        {
+          onSearchResults: (payload) => {
+            if (requestId !== requestCounterRef.current) return;
+            setSearch((current) =>
+              current
+                ? {
+                    ...current,
+                    requestId: payload.requestId ?? current.requestId,
+                    error: null,
+                    results: payload.results,
+                  }
+                : current
+            );
+          },
+          onSearchError: (payload) => {
+            if (requestId !== requestCounterRef.current) return;
+            setSearch((current) => (current ? { ...current, error: payload.error } : current));
+          },
+          onAnswer: (payload) => {
+            if (requestId !== requestCounterRef.current) return;
+            setSearch((current) =>
+              current
+                ? {
+                    ...current,
+                    answerText: payload.answerText,
+                    answerRequestId: payload.answerRequestId,
+                    answerCitations: payload.answerCitations,
+                    answerError: null,
+                  }
+                : current
+            );
+          },
+          onAnswerError: (payload) => {
+            if (requestId !== requestCounterRef.current) return;
+            setSearch((current) => (current ? { ...current, answerError: payload.error } : current));
+          },
+          onDone: (payload) => {
+            if (requestId !== requestCounterRef.current) return;
+            setSearch(payload.search);
+          },
+          onError: (payload) => {
+            if (requestId !== requestCounterRef.current) return;
+            setError(payload.message);
+          },
+        },
+        { signal: abortController.signal }
+      );
    } catch (err) {
+      if (abortController.signal.aborted) return;
      const message = err instanceof Error ? err.message : String(err);
      if (message.includes("bearer token")) {
        handleAuthFailure(message);
@@ -104,6 +177,7 @@ export default function SearchRoutePage() {
      }
    } finally {
      if (requestId === requestCounterRef.current) {
+        streamAbortRef.current = null;
        setIsRunning(false);
      }
    }