Allow API keys to access this workspace.
diff --git a/surfsense_web/content/docs/connectors/native/meta.json b/surfsense_web/content/docs/connectors/native/meta.json
index 89c8fad08..824afb980 100644
--- a/surfsense_web/content/docs/connectors/native/meta.json
+++ b/surfsense_web/content/docs/connectors/native/meta.json
@@ -1,5 +1,13 @@
{
"title": "Native Connectors",
- "pages": ["reddit", "youtube", "instagram", "tiktok", "google-maps", "google-search", "web-crawl"],
+ "pages": [
+ "reddit",
+ "youtube",
+ "instagram",
+ "tiktok",
+ "google-maps",
+ "google-search",
+ "web-crawl"
+ ],
"defaultOpen": false
}
diff --git a/surfsense_web/content/docs/connectors/native/tiktok.mdx b/surfsense_web/content/docs/connectors/native/tiktok.mdx
index fc4cf99c3..60506da47 100644
--- a/surfsense_web/content/docs/connectors/native/tiktok.mdx
+++ b/surfsense_web/content/docs/connectors/native/tiktok.mdx
@@ -11,15 +11,14 @@ The TikTok connector pulls structured public data from TikTok across four verbs:
POST /api/v1/workspaces/{workspace_id}/scrapers/tiktok/scrape
```
-Give it URLs (a video, a profile, a hashtag, or a search page) and/or profiles, hashtags, or search terms; returns videos (caption, author, play/like/comment/share counts, music, hashtags, timestamps, and the web URL). At least one of `urls`, `profiles`, `hashtags`, or `search_queries` is required.
+Give it URLs (a video, a profile, a hashtag, or a search page) and/or profiles or hashtags; returns videos (caption, author, play/like/comment/share counts, music, hashtags, timestamps, and the web URL). At least one of `urls`, `profiles`, or `hashtags` is required.
| Field | Default | Description |
|-------|---------|-------------|
| `urls` | — | TikTok URLs: a video, a profile (`/@
`), a hashtag (`/tag/`), or a search URL (max 20 sources per call) |
| `profiles` | — | Profile usernames, with or without a leading `@` |
| `hashtags` | — | Hashtag names, without the `#` |
-| `search_queries` | — | Search terms to run on TikTok |
-| `results_per_page` | `10` | Max videos per profile/hashtag/search target |
+| `results_per_page` | `10` | Max videos per profile/hashtag target |
| `max_items` | `10` | Max total videos returned across all sources (hard cap 100) |
```bash
@@ -30,7 +29,7 @@ curl -X POST "$BASE_URL/api/v1/workspaces/1/scrapers/tiktok/scrape" \
```
-Video and hashtag targets are the reliable video paths. A `profiles` target returns the account's **metadata** (name, followers, bio, verification) reliably, but TikTok often withholds its **video list** from automated clients — so a profile can return metadata with no videos. Keyword **video** search is login-walled and returns a surfaced error; to find accounts by keyword use **user search** below.
+Video and hashtag targets are the reliable video paths. A `profiles` target returns the account's **metadata** (name, followers, bio, verification) reliably, but TikTok often withholds its **video list** from automated clients — so a profile can return metadata with no videos. There is no keyword **video** search (TikTok's own search is login-walled); to find accounts by keyword use **user search** below.
## Comments
diff --git a/surfsense_web/content/docs/docker-installation/index.mdx b/surfsense_web/content/docs/docker-installation/index.mdx
index a2db4e8f5..de65e3699 100644
--- a/surfsense_web/content/docs/docker-installation/index.mdx
+++ b/surfsense_web/content/docs/docker-installation/index.mdx
@@ -143,14 +143,48 @@ To update SurfSense, see [Updating](/docs/docker-installation/updating).
## Building from Source (Contributors)
-If you're contributing to SurfSense and want to build the images from your local checkout, use the dev compose file:
+If you're contributing to SurfSense and want to build the images from your local checkout, use the dev compose file. Unlike the production setup above, there's no bundled Caddy proxy — each service publishes its own port directly, and you need **three** separate `.env` files instead of one.
+
+**Requirements:** Docker with the `buildx` plugin (needed for the multi-stage `Dockerfile`; a plain `docker build` without BuildKit fails). Check with `docker buildx version` — if missing, install `docker-buildx` (or `docker-buildx-plugin` on some distros).
```bash
-cd SurfSense/docker
+git clone https://github.com//SurfSense.git
+cd SurfSense
+cp docker/.env.example docker/.env
+cp surfsense_backend/.env.example surfsense_backend/.env
+cp surfsense_web/.env.example surfsense_web/.env
+```
+
+`docker/.env` only configures Docker Compose variable substitution — it is **not** passed into the containers. The backend and frontend read their own `.env` files instead (wired via `env_file:` in `docker-compose.dev.yml`).
+
+Edit before starting:
+
+- **`SECRET_KEY`** in `surfsense_backend/.env` — required, generate with `openssl rand -base64 32`. Note this is separate from `SECRET_KEY` in `docker/.env`; the dev compose file does not forward one to the other.
+
+Everything else in the three example files (auth type, Stripe, connector OAuth credentials, messaging bots, proxy settings, etc.) is optional — only needed if you're testing that specific integration.
+
+```bash
+cd docker
docker compose -f docker-compose.dev.yml up --build
```
-It builds the backend and frontend from source, publishes raw service ports for debugging, and includes pgAdmin at [http://localhost:5050](http://localhost:5050). There's also a `docker-compose.deps-only.yml` that runs just the dependencies (Postgres, Redis, zero-cache) in Docker while you run the backend and frontend natively — see [Manual Installation](/docs/manual-installation#alternative-let-docker-manage-all-dependencies).
+| Service | Port | Notes |
+|---------|------|-------|
+| `frontend` | `localhost:3000` | Next.js dev server |
+| `backend` | `localhost:8000` | FastAPI, `/ready` for healthcheck |
+| `pgadmin` | `localhost:5050` | Postgres GUI |
+| `zero-cache` | `localhost:4848` (ws) | Real-time sync |
+| `celery_worker` | — | Background task processing, no exposed port |
+| `celery_beat` | — | Periodic task scheduler, no exposed port |
+| `otel-lgtm` | `localhost:3001` | Grafana + Loki + Tempo + Mimir bundle, for tracing/metrics |
+
+Observability (`otel-lgtm`) isn't needed for regular dev work — it's the heaviest non-essential service, so skip it if you're not debugging traces/metrics:
+
+```bash
+docker compose -f docker-compose.dev.yml up --build db redis zero-cache backend celery_worker celery_beat frontend
+```
+
+There's also a `docker-compose.deps-only.yml` that runs just the dependencies (Postgres, Redis, zero-cache) in Docker while you run the backend and frontend natively — see [Manual Installation](/docs/manual-installation#alternative-let-docker-manage-all-dependencies).
## Troubleshooting
diff --git a/surfsense_web/hooks/use-run-stream.ts b/surfsense_web/hooks/use-run-stream.ts
index 6e7f9b766..5d737e059 100644
--- a/surfsense_web/hooks/use-run-stream.ts
+++ b/surfsense_web/hooks/use-run-stream.ts
@@ -89,17 +89,12 @@ export function useRunStream(workspaceId: number) {
setState((s) => ({
...s,
latest: ev,
- events:
- ev.type === "run.progress"
- ? [...s.events.slice(-(MAX_LOG - 1)), ev]
- : s.events,
+ events: ev.type === "run.progress" ? [...s.events.slice(-(MAX_LOG - 1)), ev] : s.events,
}));
}
} catch (e) {
if (signal.aborted) return;
- setState((s) =>
- s.status === "running" ? { ...s, status: "error", error: e } : s
- );
+ setState((s) => (s.status === "running" ? { ...s, status: "error", error: e } : s));
}
},
[workspaceId, queryClient]
@@ -113,12 +108,7 @@ export function useRunStream(workspaceId: number) {
startedAtRef.current = Date.now();
setState({ ...INITIAL, status: "running" });
try {
- const started = await scrapersApiService.runAsync(
- workspaceId,
- platform,
- verb,
- payload
- );
+ const started = await scrapersApiService.runAsync(workspaceId, platform, verb, payload);
runIdRef.current = started.run_id;
trackWeeklyUser("api_run", workspaceId);
setState((s) => ({ ...s, runId: started.run_id }));
diff --git a/surfsense_web/lib/apis/scrapers-api.service.ts b/surfsense_web/lib/apis/scrapers-api.service.ts
index 0de3ef720..69b403d5c 100644
--- a/surfsense_web/lib/apis/scrapers-api.service.ts
+++ b/surfsense_web/lib/apis/scrapers-api.service.ts
@@ -1,15 +1,15 @@
import {
type ListScraperRunsParams,
- type ScraperRunEvent,
- type StartAsyncRunResponse,
listCapabilitiesResponse,
listRunsResponse,
+ type ScraperRunEvent,
+ type StartAsyncRunResponse,
scraperRunDetail,
startAsyncRunResponse,
} from "@/contracts/types/scraper.types";
+import { authenticatedFetch } from "@/lib/auth-fetch";
import { readSSEStream } from "@/lib/chat/streaming-state";
import { buildBackendUrl } from "@/lib/env-config";
-import { authenticatedFetch } from "@/lib/auth-fetch";
import { baseApiService } from "./base-api.service";
const base = (workspaceId: number | string) => `/api/v1/workspaces/${workspaceId}/scrapers`;
diff --git a/surfsense_web/lib/auth-utils.ts b/surfsense_web/lib/auth-utils.ts
index 8c8d919c5..7d9320fb0 100644
--- a/surfsense_web/lib/auth-utils.ts
+++ b/surfsense_web/lib/auth-utils.ts
@@ -40,6 +40,7 @@ const PUBLIC_ROUTE_PREFIXES = [
"/external-mcp-connectors",
"/reddit",
"/instagram",
+ "/tiktok",
"/youtube",
"/google-maps",
"/google-search",
diff --git a/surfsense_web/lib/chat/stream-engine/engine.ts b/surfsense_web/lib/chat/stream-engine/engine.ts
new file mode 100644
index 000000000..57ff47f61
--- /dev/null
+++ b/surfsense_web/lib/chat/stream-engine/engine.ts
@@ -0,0 +1,1432 @@
+import type { AppendMessage, ThreadMessageLike } from "@assistant-ui/react";
+import { getDefaultStore } from "jotai";
+import type { Dispatch, SetStateAction } from "react";
+import { toast } from "sonner";
+import { agentFlagsAtom } from "@/atoms/agent/agent-flags-query.atom";
+import { disabledToolsAtom } from "@/atoms/agent-tools/agent-tools.atoms";
+import {
+ deriveMentionedPayload,
+ type MentionedDocumentInfo,
+ mentionedDocumentsAtom,
+ messageDocumentsMapAtom,
+ submittedMentionsAtom,
+} from "@/atoms/chat/mentioned-documents.atom";
+import { pendingUserImageDataUrlsAtom } from "@/atoms/chat/pending-user-images.atom";
+import { setPremiumAlertForThreadAtom } from "@/atoms/chat/premium-alert.atom";
+import { type AgentCreatedDocument, agentCreatedDocumentsAtom } from "@/atoms/documents/ui.atoms";
+import { updateChatTabTitleAtom } from "@/atoms/tabs/tabs.atom";
+import { currentUserAtom } from "@/atoms/user/user-query.atoms";
+import type { HitlDecision, PendingInterruptState } from "@/features/chat-messages/hitl";
+import {
+ applyActionLogSse,
+ applyActionLogUpdatedSse,
+ markActionRevertedInCache,
+} from "@/hooks/use-agent-actions-query";
+import { getAgentFilesystemSelection } from "@/lib/agent-filesystem";
+import { authenticatedFetch } from "@/lib/auth-fetch";
+import { type ChatFlow, classifyChatError } from "@/lib/chat/chat-error-classifier";
+import { tagPreAcceptSendFailure, toHttpResponseError } from "@/lib/chat/chat-request-errors";
+import { getMentionDocKey } from "@/lib/chat/mention-doc-key";
+import { createStreamFlushHelpers } from "@/lib/chat/stream-flush";
+import { consumeSseEvents, processSharedStreamEvent } from "@/lib/chat/stream-pipeline";
+import {
+ applyTurnIdToAssistantMessageList,
+ mergeChatTurnIdIntoMessage,
+ readStreamedChatTurnId,
+ readStreamedMessageId,
+} from "@/lib/chat/stream-side-effects";
+import {
+ addToolCall,
+ buildContentForUI,
+ type ContentPartsState,
+ type FrameBatchedUpdater,
+ type ThinkingStepData,
+ updateToolCall,
+} from "@/lib/chat/streaming-state";
+import {
+ appendMessage,
+ createThread,
+ getRegenerateUrl,
+ type ThreadListItem,
+ type ThreadListResponse,
+ type ThreadRecord,
+} from "@/lib/chat/thread-persistence";
+import {
+ extractUserTurnForNewChatApi,
+ type NewChatUserImagePayload,
+} from "@/lib/chat/user-turn-api-parts";
+import { buildBackendUrl } from "@/lib/env-config";
+import {
+ trackChatBlocked,
+ trackChatCreated,
+ trackChatErrorDetailed,
+ trackChatMessageSent,
+ trackChatResponseReceived,
+} from "@/lib/posthog/events";
+import { cacheKeys } from "@/lib/query-client/cache-keys";
+import { queryClient } from "@/lib/query-client/client";
+import {
+ computeFallbackTurnCancellingRetryDelay,
+ freshSynthToolCallId,
+ pairBundleToolCallIds,
+ RECENT_CANCEL_WINDOW_MS,
+ sleep,
+ TOOLS_WITH_UI_ALL,
+} from "./helpers";
+import { chatStreamStore } from "./store";
+
+const jotaiStore = getDefaultStore();
+const tokenUsageStore = chatStreamStore.tokenUsage;
+const toolsWithUI = TOOLS_WITH_UI_ALL;
+
+/**
+ * Display-only setters, valid only while the page is mounted. The stream drives
+ * durable state through {@link chatStreamStore}; these just sync the mounted
+ * page's local view and are ignored once it unmounts.
+ */
+export interface EngineView {
+ setThreadId: Dispatch>;
+ setCurrentThread: Dispatch>;
+}
+
+/** Route/view context the page passes into every engine call. */
+export interface EngineContext {
+ workspaceId: number;
+ /** Currently viewed thread id (``activeThreadId`` in the page). */
+ threadId: number | null;
+ /** The page's current displayed messages — history/slice seed. */
+ priorMessages: ThreadMessageLike[];
+ view: EngineView;
+}
+
+// ---------------------------------------------------------------------------
+// Error handling
+// ---------------------------------------------------------------------------
+
+async function persistAssistantErrorMessage({
+ threadId,
+ assistantMsgId,
+ text,
+}: {
+ threadId: number | null;
+ assistantMsgId: string;
+ text: string;
+}): Promise {
+ if (threadId != null) {
+ chatStreamStore.setMessages(threadId, (prev) =>
+ prev.map((m) => (m.id === assistantMsgId ? { ...m, content: [{ type: "text", text }] } : m))
+ );
+ }
+
+ if (!threadId) return;
+
+ // Persist only temporary assistant placeholders to avoid duplicate rows
+ // when the message already has a database-backed ID.
+ if (!assistantMsgId.startsWith("msg-assistant-")) return;
+
+ try {
+ const savedMessage = await appendMessage(threadId, {
+ role: "assistant",
+ content: [{ type: "text", text }],
+ });
+ const newMsgId = `msg-${savedMessage.id}`;
+ tokenUsageStore.rename(assistantMsgId, newMsgId);
+ chatStreamStore.setMessages(threadId, (prev) =>
+ prev.map((m) => (m.id === assistantMsgId ? { ...m, id: newMsgId } : m))
+ );
+ } catch (persistErr) {
+ console.error("Failed to persist assistant error message:", persistErr);
+ }
+}
+
+async function handleChatFailure({
+ error,
+ flow,
+ threadId,
+ assistantMsgId,
+ workspaceId,
+}: {
+ error: unknown;
+ flow: ChatFlow;
+ threadId: number | null;
+ assistantMsgId: string;
+ workspaceId: number;
+}): Promise {
+ const normalized = classifyChatError({
+ error,
+ flow,
+ context: { workspaceId, threadId },
+ });
+
+ const logger =
+ normalized.severity === "error"
+ ? console.error
+ : normalized.severity === "warn"
+ ? console.warn
+ : console.info;
+ logger(`[chat-engine] ${flow} ${normalized.kind}:`, error);
+
+ const telemetryPayload = {
+ flow,
+ kind: normalized.kind,
+ error_code: normalized.errorCode,
+ severity: normalized.severity,
+ is_expected: normalized.isExpected,
+ message: normalized.userMessage,
+ };
+ if (normalized.telemetryEvent === "chat_blocked") {
+ trackChatBlocked(workspaceId, threadId, telemetryPayload);
+ } else {
+ trackChatErrorDetailed(workspaceId, threadId, telemetryPayload);
+ }
+
+ if (normalized.channel === "silent") {
+ return;
+ }
+
+ if (normalized.channel === "pinned_inline") {
+ if (threadId) {
+ const currentUser = jotaiStore.get(currentUserAtom).data;
+ jotaiStore.set(setPremiumAlertForThreadAtom, {
+ threadId,
+ message: normalized.userMessage,
+ userId: currentUser?.id ?? null,
+ });
+ }
+ if (normalized.assistantMessage) {
+ await persistAssistantErrorMessage({
+ threadId,
+ assistantMsgId,
+ text: normalized.assistantMessage,
+ });
+ }
+ return;
+ }
+
+ if (normalized.channel === "inline") {
+ if (normalized.assistantMessage) {
+ await persistAssistantErrorMessage({
+ threadId,
+ assistantMsgId,
+ text: normalized.assistantMessage,
+ });
+ }
+ toast.error(normalized.userMessage);
+ return;
+ }
+
+ toast.error(normalized.userMessage);
+}
+
+async function handleStreamTerminalError({
+ error,
+ flow,
+ threadId,
+ assistantMsgId,
+ accepted,
+ workspaceId,
+ onAbort,
+ onPreAcceptFailure,
+ onAcceptedStreamError,
+}: {
+ error: unknown;
+ flow: ChatFlow;
+ threadId: number | null;
+ assistantMsgId: string;
+ accepted: boolean;
+ workspaceId: number;
+ onAbort?: () => Promise;
+ onPreAcceptFailure?: () => Promise;
+ onAcceptedStreamError?: () => Promise;
+}): Promise {
+ if (error instanceof Error && error.name === "AbortError") {
+ await onAbort?.();
+ return;
+ }
+
+ if (!accepted) {
+ await onPreAcceptFailure?.();
+ } else {
+ await onAcceptedStreamError?.();
+ }
+
+ await handleChatFailure({
+ error: !accepted ? tagPreAcceptSendFailure(error) : error,
+ flow,
+ threadId,
+ assistantMsgId: accepted ? assistantMsgId : "no-persist-assistant",
+ workspaceId,
+ });
+}
+
+async function fetchWithTurnCancellingRetry(runFetch: () => Promise): Promise {
+ const maxAttempts = 4;
+ for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
+ const response = await runFetch();
+ if (response.ok) {
+ return response;
+ }
+ const error = await toHttpResponseError(response);
+ const withMeta = error as Error & { errorCode?: string; retryAfterMs?: number };
+ const isTurnCancelling = withMeta.errorCode === "TURN_CANCELLING";
+ const isRecentThreadBusyAfterCancel =
+ withMeta.errorCode === "THREAD_BUSY" &&
+ Date.now() - chatStreamStore.recentCancelRequestedAt <= RECENT_CANCEL_WINDOW_MS;
+ if ((isTurnCancelling || isRecentThreadBusyAfterCancel) && attempt < maxAttempts) {
+ const waitMs = withMeta.retryAfterMs ?? computeFallbackTurnCancellingRetryDelay(attempt);
+ await sleep(waitMs);
+ continue;
+ }
+ throw error;
+ }
+
+ throw Object.assign(new Error("Turn cancellation retry limit exceeded"), {
+ errorCode: "TURN_CANCELLING",
+ });
+}
+
+// ---------------------------------------------------------------------------
+// Cancel
+// ---------------------------------------------------------------------------
+
+/**
+ * Cancel the single in-flight turn. Targets the active stream's OWNER thread
+ * (not the currently-viewed thread) so cancel works even after navigation and
+ * for the lazy-create case.
+ */
+export async function cancelActiveTurn(): Promise {
+ const threadId = chatStreamStore.activeThreadId;
+ if (threadId) {
+ try {
+ const response = await authenticatedFetch(
+ buildBackendUrl(`/api/v1/threads/${threadId}/cancel-active-turn`),
+ { method: "POST" }
+ );
+ if (response.ok) {
+ const payload = (await response.json()) as { error_code?: string };
+ if (payload.error_code === "TURN_CANCELLING") {
+ chatStreamStore.markRecentCancel();
+ }
+ }
+ } catch (error) {
+ console.warn("[chat-engine] Failed to signal cancel-active-turn:", error);
+ }
+ chatStreamStore.setRunning(threadId, false);
+ }
+ chatStreamStore.abortActive();
+}
+
+// ---------------------------------------------------------------------------
+// New chat turn
+// ---------------------------------------------------------------------------
+
+export async function startNewChat(ctx: EngineContext, message: AppendMessage): Promise {
+ const { workspaceId, threadId, priorMessages, view } = ctx;
+
+ // Supersede any previous in-flight turn.
+ chatStreamStore.abortActive();
+
+ // Prefer the submit-time snapshot; fall back to the live atom for the
+ // send-button path.
+ const submittedSnapshot = jotaiStore.get(submittedMentionsAtom);
+ jotaiStore.set(submittedMentionsAtom, null);
+ const mentionedDocuments = jotaiStore.get(mentionedDocumentsAtom);
+ const activeMentions = submittedSnapshot ?? mentionedDocuments;
+ const mentionPayload = deriveMentionedPayload(activeMentions);
+ if (activeMentions.length > 0) {
+ jotaiStore.set(mentionedDocumentsAtom, []);
+ }
+
+ const pendingUserImageUrls = jotaiStore.get(pendingUserImageDataUrlsAtom);
+ const urlsSnapshot = [...pendingUserImageUrls];
+ const { userQuery, userImages } = extractUserTurnForNewChatApi(message, urlsSnapshot);
+
+ if (!userQuery.trim() && userImages.length === 0) return;
+
+ const localFilesystemEnabled =
+ jotaiStore.get(agentFlagsAtom).data?.enable_desktop_local_filesystem === true;
+ const disabledTools = jotaiStore.get(disabledToolsAtom);
+ const currentUser = jotaiStore.get(currentUserAtom).data;
+
+ // Resolve filesystem selection BEFORE any optimistic UI / lazy thread
+ // creation so an unsatisfied "Local Folder" requirement bails cleanly
+ // with nothing to roll back and no thread left stuck "running".
+ let selection: Awaited>;
+ try {
+ selection = await getAgentFilesystemSelection(workspaceId, { localFilesystemEnabled });
+ } catch (error) {
+ await handleChatFailure({
+ error: tagPreAcceptSendFailure(error),
+ flow: "new",
+ threadId,
+ assistantMsgId: "no-persist-assistant",
+ workspaceId,
+ });
+ return;
+ }
+ if (
+ selection.filesystem_mode === "desktop_local_folder" &&
+ (!selection.local_filesystem_mounts || selection.local_filesystem_mounts.length === 0)
+ ) {
+ toast.error("Select a local folder before using Local Folder mode.");
+ return;
+ }
+
+ // Lazy thread creation: create thread on first message if it doesn't exist.
+ let currentThreadId = threadId;
+ let isNewThread = false;
+ if (!currentThreadId) {
+ try {
+ const newThread = await createThread(workspaceId, "New Chat");
+ currentThreadId = newThread.id;
+ view.setThreadId(currentThreadId);
+ view.setCurrentThread(newThread);
+ queryClient.setQueryData(cacheKeys.threads.detail(newThread.id), newThread);
+ queryClient.setQueryData(cacheKeys.threads.messages(newThread.id), { messages: [] });
+
+ trackChatCreated(workspaceId, currentThreadId);
+
+ isNewThread = true;
+ // Update URL silently using browser API (not router.replace) to avoid
+ // interrupting the ongoing fetch/streaming with React navigation.
+ window.history.replaceState(
+ null,
+ "",
+ `/dashboard/${workspaceId}/new-chat/${currentThreadId}`
+ );
+ } catch (error) {
+ console.error("[chat-engine] Failed to create thread:", error);
+ await handleChatFailure({
+ error: tagPreAcceptSendFailure(error),
+ flow: "new",
+ threadId: currentThreadId,
+ assistantMsgId: "no-persist-assistant",
+ workspaceId,
+ });
+ return;
+ }
+ }
+
+ // Seed the durable per-thread overlay with the pre-turn conversation and
+ // flip it to running so the page renders the live stream.
+ const streamThreadId = currentThreadId;
+ chatStreamStore.begin(streamThreadId, priorMessages);
+
+ if (urlsSnapshot.length > 0) {
+ jotaiStore.set(pendingUserImageDataUrlsAtom, (prev) =>
+ prev.filter((u) => !urlsSnapshot.includes(u))
+ );
+ }
+
+ // Add user message to state. Mutable because the SSE
+ // ``data-user-message-id`` handler renames this optimistic id to the
+ // canonical ``msg-{db_id}`` once ``persist_user_turn`` resolves.
+ let userMsgId = `msg-user-${Date.now()}`;
+
+ const authorMetadata = currentUser
+ ? {
+ custom: {
+ author: {
+ displayName: currentUser.display_name ?? null,
+ avatarUrl: currentUser.avatar_url ?? null,
+ },
+ },
+ }
+ : undefined;
+
+ const existingImageUrls = new Set(
+ message.content
+ .filter(
+ (p): p is { type: "image"; image: string } =>
+ typeof p === "object" && p !== null && "type" in p && p.type === "image" && "image" in p
+ )
+ .map((p) => p.image)
+ );
+ const extraImageParts = urlsSnapshot
+ .filter((u) => !existingImageUrls.has(u))
+ .map((image) => ({ type: "image" as const, image }));
+ const userDisplayContent = [...message.content, ...extraImageParts];
+
+ const userMessage: ThreadMessageLike = {
+ id: userMsgId,
+ role: "user",
+ content: userDisplayContent,
+ createdAt: new Date(),
+ metadata: authorMetadata,
+ };
+ chatStreamStore.setMessages(streamThreadId, (prev) => [...prev, userMessage]);
+
+ trackChatMessageSent(workspaceId, streamThreadId, {
+ hasAttachments: userImages.length > 0,
+ hasMentionedDocuments:
+ mentionPayload.document_ids.length > 0 ||
+ mentionPayload.folder_ids.length > 0 ||
+ mentionPayload.connector_ids.length > 0,
+ messageLength: userQuery.length,
+ });
+
+ // Collect unique mention chips for display & persistence.
+ const allMentionedDocs: MentionedDocumentInfo[] = [];
+ const seenDocKeys = new Set();
+ for (const doc of activeMentions) {
+ const key = getMentionDocKey(doc);
+ if (seenDocKeys.has(key)) continue;
+ seenDocKeys.add(key);
+ allMentionedDocs.push(doc);
+ }
+
+ if (allMentionedDocs.length > 0) {
+ jotaiStore.set(messageDocumentsMapAtom, (prev) => ({
+ ...prev,
+ [userMsgId]: allMentionedDocs,
+ }));
+ }
+
+ const controller = new AbortController();
+ chatStreamStore.beginActive(streamThreadId, controller);
+
+ // Prepare assistant message. Mutable for the same reason as ``userMsgId``.
+ let assistantMsgId = `msg-assistant-${Date.now()}`;
+ const currentThinkingSteps = new Map();
+ const contentPartsState: ContentPartsState = {
+ contentParts: [],
+ currentTextPartIndex: -1,
+ currentReasoningPartIndex: -1,
+ toolCallIndices: new Map(),
+ };
+ const { contentParts } = contentPartsState;
+ let wasInterrupted = false;
+ let newAccepted = false;
+ let streamBatcher: FrameBatchedUpdater | null = null;
+
+ try {
+ // Build message history for context.
+ const messageHistory = priorMessages
+ .filter((m) => m.role === "user" || m.role === "assistant")
+ .map((m) => {
+ let text = "";
+ for (const part of m.content) {
+ if (typeof part === "object" && part.type === "text" && "text" in part) {
+ text += part.text;
+ }
+ }
+ return { role: m.role, content: text };
+ })
+ .filter((m) => m.content.length > 0);
+
+ const hasDocumentIds = mentionPayload.document_ids.length > 0;
+ const hasFolderIds = mentionPayload.folder_ids.length > 0;
+ const hasConnectorIds = mentionPayload.connector_ids.length > 0;
+ const hasThreadIds = mentionPayload.thread_ids.length > 0;
+
+ const response = await fetchWithTurnCancellingRetry(() =>
+ authenticatedFetch(buildBackendUrl("/api/v1/new_chat"), {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify({
+ chat_id: streamThreadId,
+ user_query: userQuery.trim(),
+ workspace_id: workspaceId,
+ filesystem_mode: selection.filesystem_mode,
+ client_platform: selection.client_platform,
+ local_filesystem_mounts: selection.local_filesystem_mounts,
+ messages: messageHistory,
+ mentioned_document_ids: hasDocumentIds ? mentionPayload.document_ids : undefined,
+ mentioned_folder_ids: hasFolderIds ? mentionPayload.folder_ids : undefined,
+ mentioned_connector_ids: hasConnectorIds ? mentionPayload.connector_ids : undefined,
+ mentioned_connectors: hasConnectorIds ? mentionPayload.connectors : undefined,
+ mentioned_thread_ids: hasThreadIds ? mentionPayload.thread_ids : undefined,
+ mentioned_documents: allMentionedDocs.length > 0 ? allMentionedDocs : undefined,
+ disabled_tools: disabledTools.length > 0 ? disabledTools : undefined,
+ ...(userImages.length > 0 ? { user_images: userImages } : {}),
+ }),
+ signal: controller.signal,
+ })
+ );
+
+ if (!response.ok) {
+ throw await toHttpResponseError(response);
+ }
+ newAccepted = true;
+ chatStreamStore.setMessages(streamThreadId, (prev) => [
+ ...prev,
+ {
+ id: assistantMsgId,
+ role: "assistant",
+ content: [{ type: "text", text: "" }],
+ createdAt: new Date(),
+ },
+ ]);
+
+ const flushMessages = () => {
+ chatStreamStore.setMessages(streamThreadId, (prev) =>
+ prev.map((m) =>
+ m.id === assistantMsgId
+ ? { ...m, content: buildContentForUI(contentPartsState, toolsWithUI) }
+ : m
+ )
+ );
+ };
+ const { batcher, scheduleFlush, forceFlush } = createStreamFlushHelpers(flushMessages);
+ streamBatcher = batcher;
+
+ await consumeSseEvents(response, async (parsed) => {
+ if (
+ processSharedStreamEvent(parsed, {
+ contentPartsState,
+ toolsWithUI,
+ currentThinkingSteps,
+ scheduleFlush,
+ forceFlush,
+ onTokenUsage: (data) => {
+ tokenUsageStore.set(assistantMsgId, data);
+ },
+ onTurnStatus: (data) => {
+ if (data.status === "cancelling") {
+ chatStreamStore.markRecentCancel();
+ }
+ },
+ })
+ ) {
+ return;
+ }
+ switch (parsed.type) {
+ case "data-thread-title-update": {
+ const titleData = parsed.data as { threadId: number; title: string };
+ if (titleData?.title && titleData?.threadId === streamThreadId) {
+ view.setCurrentThread((prev) => (prev ? { ...prev, title: titleData.title } : prev));
+ jotaiStore.set(updateChatTabTitleAtom, {
+ chatId: streamThreadId,
+ title: titleData.title,
+ });
+ queryClient.setQueriesData(
+ { queryKey: ["threads", String(workspaceId)] },
+ (old) => {
+ if (!old) return old;
+ const updateTitle = (list: ThreadListItem[]) =>
+ list.map((t) =>
+ t.id === titleData.threadId ? { ...t, title: titleData.title } : t
+ );
+ return {
+ ...old,
+ threads: updateTitle(old.threads),
+ archived_threads: updateTitle(old.archived_threads),
+ };
+ }
+ );
+ }
+ break;
+ }
+
+ case "data-documents-updated": {
+ const docEvent = parsed.data as {
+ action: string;
+ document: AgentCreatedDocument;
+ };
+ if (docEvent?.document?.id) {
+ jotaiStore.set(agentCreatedDocumentsAtom, (prev) => {
+ if (prev.some((d) => d.id === docEvent.document.id)) return prev;
+ return [...prev, docEvent.document];
+ });
+ }
+ break;
+ }
+
+ case "data-interrupt-request": {
+ wasInterrupted = true;
+ const interruptData = parsed.data as Record;
+ const actionRequests = (interruptData.action_requests ?? []) as Array<{
+ name: string;
+ args: Record;
+ }>;
+ const paired = pairBundleToolCallIds(
+ contentPartsState.toolCallIndices,
+ contentPartsState.contentParts,
+ actionRequests
+ );
+ const bundleToolCallIds: string[] = [];
+ for (let i = 0; i < actionRequests.length; i++) {
+ const action = actionRequests[i];
+ let targetTcId = paired[i];
+ if (!targetTcId) {
+ targetTcId = freshSynthToolCallId(contentPartsState.toolCallIndices, action.name, i);
+ addToolCall(
+ contentPartsState,
+ toolsWithUI,
+ targetTcId,
+ action.name,
+ action.args,
+ true
+ );
+ }
+ updateToolCall(contentPartsState, targetTcId, {
+ result: { __interrupt__: true, ...interruptData },
+ });
+ bundleToolCallIds.push(targetTcId);
+ }
+ chatStreamStore.setMessages(streamThreadId, (prev) =>
+ prev.map((m) =>
+ m.id === assistantMsgId
+ ? { ...m, content: buildContentForUI(contentPartsState, toolsWithUI) }
+ : m
+ )
+ );
+ // ``tool_call_id`` is stamped on the backend by
+ // ``checkpointed_subagent_middleware``. Without it we can't
+ // address the paused subagent on resume — skip rather than
+ // fabricate a synthetic key.
+ const interruptId = String(interruptData.tool_call_id ?? "");
+ if (interruptId) {
+ const incoming: PendingInterruptState = {
+ interruptId,
+ threadId: streamThreadId,
+ assistantMsgId,
+ interruptData,
+ bundleToolCallIds,
+ };
+ chatStreamStore.setPendingInterrupts(streamThreadId, (prev) => {
+ const without = prev.filter((p) => p.interruptId !== interruptId);
+ return [...without, incoming];
+ });
+ }
+ break;
+ }
+
+ case "data-action-log": {
+ applyActionLogSse(queryClient, streamThreadId, workspaceId, parsed.data);
+ break;
+ }
+
+ case "data-action-log-updated": {
+ applyActionLogUpdatedSse(
+ queryClient,
+ streamThreadId,
+ parsed.data.id,
+ parsed.data.reversible
+ );
+ break;
+ }
+
+ case "data-turn-info": {
+ const turnId = readStreamedChatTurnId(parsed.data);
+ if (turnId) {
+ chatStreamStore.setMessages(streamThreadId, (prev) =>
+ applyTurnIdToAssistantMessageList(prev, assistantMsgId, turnId)
+ );
+ }
+ break;
+ }
+
+ case "data-user-message-id": {
+ const parsedMsg = readStreamedMessageId(parsed.data);
+ if (!parsedMsg) break;
+ const newUserMsgId = `msg-${parsedMsg.messageId}`;
+ const oldUserMsgId = userMsgId;
+ chatStreamStore.setMessages(streamThreadId, (prev) =>
+ prev.map((m) =>
+ m.id === oldUserMsgId
+ ? mergeChatTurnIdIntoMessage({ ...m, id: newUserMsgId }, parsedMsg.turnId)
+ : m
+ )
+ );
+ if (allMentionedDocs.length > 0) {
+ jotaiStore.set(messageDocumentsMapAtom, (prev) => {
+ if (!(oldUserMsgId in prev)) {
+ return { ...prev, [newUserMsgId]: allMentionedDocs };
+ }
+ const { [oldUserMsgId]: _removed, ...rest } = prev;
+ return { ...rest, [newUserMsgId]: allMentionedDocs };
+ });
+ }
+ userMsgId = newUserMsgId;
+ if (isNewThread) {
+ queryClient.invalidateQueries({
+ queryKey: ["threads", String(workspaceId)],
+ });
+ }
+ break;
+ }
+
+ case "data-assistant-message-id": {
+ const parsedMsg = readStreamedMessageId(parsed.data);
+ if (!parsedMsg) break;
+ const newAssistantMsgId = `msg-${parsedMsg.messageId}`;
+ const oldAssistantMsgId = assistantMsgId;
+ tokenUsageStore.rename(oldAssistantMsgId, newAssistantMsgId);
+ chatStreamStore.setMessages(streamThreadId, (prev) =>
+ prev.map((m) =>
+ m.id === oldAssistantMsgId
+ ? mergeChatTurnIdIntoMessage({ ...m, id: newAssistantMsgId }, parsedMsg.turnId)
+ : m
+ )
+ );
+ chatStreamStore.setPendingInterrupts(streamThreadId, (prev) =>
+ prev.map((p) =>
+ p.assistantMsgId === oldAssistantMsgId
+ ? { ...p, assistantMsgId: newAssistantMsgId }
+ : p
+ )
+ );
+ assistantMsgId = newAssistantMsgId;
+ break;
+ }
+ }
+ });
+
+ batcher.flush();
+
+ if (contentParts.length > 0 && !wasInterrupted) {
+ trackChatResponseReceived(workspaceId, streamThreadId);
+ }
+ } catch (error) {
+ streamBatcher?.dispose();
+ await handleStreamTerminalError({
+ error,
+ flow: "new",
+ threadId: streamThreadId,
+ assistantMsgId,
+ accepted: newAccepted,
+ workspaceId,
+ onPreAcceptFailure: async () => {
+ // Pre-accept failure means the BE never accepted the request — no
+ // server-side persistence ran. Roll back the optimistic UI.
+ chatStreamStore.setMessages(streamThreadId, (prev) =>
+ prev.filter((m) => m.id !== userMsgId)
+ );
+ jotaiStore.set(messageDocumentsMapAtom, (prev) => {
+ if (!(userMsgId in prev)) return prev;
+ const { [userMsgId]: _removed, ...rest } = prev;
+ return rest;
+ });
+ },
+ });
+ } finally {
+ chatStreamStore.setRunning(streamThreadId, false);
+ chatStreamStore.clearActive(controller);
+ void queryClient.invalidateQueries({
+ queryKey: cacheKeys.threads.messages(streamThreadId),
+ });
+ void queryClient.invalidateQueries({
+ queryKey: cacheKeys.threads.detail(streamThreadId),
+ });
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Resume (HITL decisions)
+// ---------------------------------------------------------------------------
+
+export async function resumeChat(
+ ctx: EngineContext,
+ decisions: Array<{
+ type: string;
+ message?: string;
+ edited_action?: { name: string; args: Record };
+ }>
+): Promise {
+ const { workspaceId, threadId } = ctx;
+ if (threadId == null) return;
+ const pendingInterrupts = chatStreamStore.getPendingInterrupts(threadId);
+ if (pendingInterrupts.length === 0) return;
+
+ const resumeThreadId = pendingInterrupts[0].threadId;
+ let assistantMsgId = pendingInterrupts[0].assistantMsgId;
+ const allBundleToolCallIds = pendingInterrupts.flatMap((p) => p.bundleToolCallIds);
+ chatStreamStore.setPendingInterrupts(resumeThreadId, () => []);
+ chatStreamStore.setRunning(resumeThreadId, true);
+
+ const controller = new AbortController();
+ chatStreamStore.beginActive(resumeThreadId, controller);
+
+ const localFilesystemEnabled =
+ jotaiStore.get(agentFlagsAtom).data?.enable_desktop_local_filesystem === true;
+ const disabledTools = jotaiStore.get(disabledToolsAtom);
+
+ const currentThinkingSteps = new Map();
+ const contentPartsState: ContentPartsState = {
+ contentParts: [],
+ currentTextPartIndex: -1,
+ currentReasoningPartIndex: -1,
+ toolCallIndices: new Map(),
+ };
+ const { contentParts, toolCallIndices } = contentPartsState;
+ let resumeAccepted = false;
+ let streamBatcher: FrameBatchedUpdater | null = null;
+
+ const existingMsg = chatStreamStore
+ .getMessages(resumeThreadId)
+ .find((m) => m.id === assistantMsgId);
+ if (existingMsg && Array.isArray(existingMsg.content)) {
+ contentPartsState.suppressStepSeparators = true;
+ for (const part of existingMsg.content) {
+ if (typeof part === "object" && part !== null) {
+ const p = part as Record;
+ if (p.type === "text") {
+ contentParts.push({ type: "text", text: String(p.text ?? "") });
+ contentPartsState.currentTextPartIndex = contentParts.length - 1;
+ } else if (p.type === "tool-call") {
+ toolCallIndices.set(String(p.toolCallId), contentParts.length);
+ contentParts.push({
+ type: "tool-call",
+ toolCallId: String(p.toolCallId),
+ toolName: String(p.toolName),
+ args: (p.args as Record) ?? {},
+ result: p.result as unknown,
+ ...(typeof p.argsText === "string" ? { argsText: p.argsText } : {}),
+ ...(typeof p.langchainToolCallId === "string"
+ ? { langchainToolCallId: p.langchainToolCallId }
+ : {}),
+ ...(p.metadata && typeof p.metadata === "object"
+ ? { metadata: p.metadata as Record }
+ : {}),
+ });
+ contentPartsState.currentTextPartIndex = -1;
+ } else if (p.type === "data-thinking-steps") {
+ const stepsData = p.data as { steps: ThinkingStepData[] } | undefined;
+ contentParts.push({
+ type: "data-thinking-steps",
+ data: { steps: stepsData?.steps ?? [] },
+ });
+ for (const step of stepsData?.steps ?? []) {
+ currentThinkingSteps.set(step.id, step);
+ }
+ }
+ }
+ }
+ }
+
+ // Apply each decision to its own card by toolCallId so mixed bundles
+ // (approve/edit/reject) do not collapse onto ``decisions[0]``.
+ const decisionByTcId = new Map();
+ const tcIds = allBundleToolCallIds;
+ if (decisions.length === tcIds.length) {
+ for (let i = 0; i < tcIds.length; i++) decisionByTcId.set(tcIds[i], decisions[i]);
+ }
+ if (decisionByTcId.size > 0) {
+ for (const part of contentParts) {
+ if (part.type !== "tool-call") continue;
+ const tcId = part.toolCallId as string | undefined;
+ const d = tcId ? decisionByTcId.get(tcId) : undefined;
+ if (!d) continue;
+ if (typeof part.result !== "object" || part.result === null) continue;
+ if (!("__interrupt__" in (part.result as Record))) continue;
+ const decided = d.type;
+ if (decided === "edit" && d.edited_action) {
+ const mergedArgs = { ...part.args, ...d.edited_action.args };
+ part.args = mergedArgs;
+ part.argsText = JSON.stringify(mergedArgs, null, 2);
+ }
+ part.result = {
+ ...(part.result as Record),
+ __decided__: decided,
+ };
+ }
+ }
+
+ try {
+ const selection = await getAgentFilesystemSelection(workspaceId, { localFilesystemEnabled });
+ const response = await fetchWithTurnCancellingRetry(() =>
+ authenticatedFetch(buildBackendUrl(`/api/v1/threads/${resumeThreadId}/resume`), {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify({
+ workspace_id: workspaceId,
+ decisions,
+ disabled_tools: disabledTools.length > 0 ? disabledTools : undefined,
+ filesystem_mode: selection.filesystem_mode,
+ client_platform: selection.client_platform,
+ local_filesystem_mounts: selection.local_filesystem_mounts,
+ }),
+ signal: controller.signal,
+ })
+ );
+
+ if (!response.ok) {
+ throw await toHttpResponseError(response);
+ }
+ resumeAccepted = true;
+
+ const flushMessages = () => {
+ chatStreamStore.setMessages(resumeThreadId, (prev) =>
+ prev.map((m) =>
+ m.id === assistantMsgId
+ ? { ...m, content: buildContentForUI(contentPartsState, toolsWithUI) }
+ : m
+ )
+ );
+ };
+ const { batcher, scheduleFlush, forceFlush } = createStreamFlushHelpers(flushMessages);
+ streamBatcher = batcher;
+
+ await consumeSseEvents(response, async (parsed) => {
+ if (
+ processSharedStreamEvent(parsed, {
+ contentPartsState,
+ toolsWithUI,
+ currentThinkingSteps,
+ scheduleFlush,
+ forceFlush,
+ onTokenUsage: (data) => {
+ tokenUsageStore.set(assistantMsgId, data);
+ },
+ onTurnStatus: (data) => {
+ if (data.status === "cancelling") {
+ chatStreamStore.markRecentCancel();
+ }
+ },
+ })
+ ) {
+ return;
+ }
+ switch (parsed.type) {
+ case "data-interrupt-request": {
+ const interruptData = parsed.data as Record;
+ const actionRequests = (interruptData.action_requests ?? []) as Array<{
+ name: string;
+ args: Record;
+ }>;
+ const paired = pairBundleToolCallIds(
+ contentPartsState.toolCallIndices,
+ contentPartsState.contentParts,
+ actionRequests
+ );
+ const bundleToolCallIds: string[] = [];
+ for (let i = 0; i < actionRequests.length; i++) {
+ const action = actionRequests[i];
+ let targetTcId = paired[i];
+ if (!targetTcId) {
+ targetTcId = freshSynthToolCallId(contentPartsState.toolCallIndices, action.name, i);
+ addToolCall(
+ contentPartsState,
+ toolsWithUI,
+ targetTcId,
+ action.name,
+ action.args,
+ true
+ );
+ }
+ updateToolCall(contentPartsState, targetTcId, {
+ result: { __interrupt__: true, ...interruptData },
+ });
+ bundleToolCallIds.push(targetTcId);
+ }
+ chatStreamStore.setMessages(resumeThreadId, (prev) =>
+ prev.map((m) =>
+ m.id === assistantMsgId
+ ? { ...m, content: buildContentForUI(contentPartsState, toolsWithUI) }
+ : m
+ )
+ );
+ {
+ const interruptId = String(interruptData.tool_call_id ?? "");
+ if (interruptId) {
+ const incoming: PendingInterruptState = {
+ interruptId,
+ threadId: resumeThreadId,
+ assistantMsgId,
+ interruptData,
+ bundleToolCallIds,
+ };
+ chatStreamStore.setPendingInterrupts(resumeThreadId, (prev) => {
+ const without = prev.filter((p) => p.interruptId !== interruptId);
+ return [...without, incoming];
+ });
+ }
+ }
+ break;
+ }
+
+ case "data-action-log": {
+ applyActionLogSse(queryClient, resumeThreadId, workspaceId, parsed.data);
+ break;
+ }
+
+ case "data-action-log-updated": {
+ applyActionLogUpdatedSse(
+ queryClient,
+ resumeThreadId,
+ parsed.data.id,
+ parsed.data.reversible
+ );
+ break;
+ }
+
+ case "data-turn-info": {
+ const turnId = readStreamedChatTurnId(parsed.data);
+ if (turnId) {
+ chatStreamStore.setMessages(resumeThreadId, (prev) =>
+ applyTurnIdToAssistantMessageList(prev, assistantMsgId, turnId)
+ );
+ }
+ break;
+ }
+
+ case "data-assistant-message-id": {
+ const parsedMsg = readStreamedMessageId(parsed.data);
+ if (!parsedMsg) break;
+ const newAssistantMsgId = `msg-${parsedMsg.messageId}`;
+ const oldAssistantMsgId = assistantMsgId;
+ tokenUsageStore.rename(oldAssistantMsgId, newAssistantMsgId);
+ chatStreamStore.setMessages(resumeThreadId, (prev) =>
+ prev.map((m) =>
+ m.id === oldAssistantMsgId
+ ? mergeChatTurnIdIntoMessage({ ...m, id: newAssistantMsgId }, parsedMsg.turnId)
+ : m
+ )
+ );
+ assistantMsgId = newAssistantMsgId;
+ break;
+ }
+ }
+ });
+
+ batcher.flush();
+ } catch (error) {
+ streamBatcher?.dispose();
+ await handleStreamTerminalError({
+ error,
+ flow: "resume",
+ threadId: resumeThreadId,
+ assistantMsgId,
+ accepted: resumeAccepted,
+ workspaceId,
+ });
+ } finally {
+ chatStreamStore.setRunning(resumeThreadId, false);
+ chatStreamStore.clearActive(controller);
+ void queryClient.invalidateQueries({
+ queryKey: cacheKeys.threads.messages(resumeThreadId),
+ });
+ void queryClient.invalidateQueries({
+ queryKey: cacheKeys.threads.detail(resumeThreadId),
+ });
+ }
+}
+
+// ---------------------------------------------------------------------------
+// Regenerate (edit / reload)
+// ---------------------------------------------------------------------------
+
+export async function regenerateChat(
+ ctx: EngineContext,
+ newUserQuery: string | null,
+ editExtras?: {
+ userMessageContent: ThreadMessageLike["content"];
+ userImages: NewChatUserImagePayload[];
+ sourceUserMessageId?: string;
+ },
+ editFromPosition?: {
+ fromMessageId?: number | null;
+ revertActions?: boolean;
+ }
+): Promise {
+ const { workspaceId, threadId, priorMessages } = ctx;
+ if (!threadId) {
+ toast.error("Cannot regenerate: no active chat thread");
+ return;
+ }
+ const streamThreadId = threadId;
+
+ const isEdit = newUserQuery !== null;
+
+ // Supersede any previous in-flight turn.
+ chatStreamStore.abortActive();
+
+ const messageDocumentsMap = jotaiStore.get(messageDocumentsMapAtom);
+ const localFilesystemEnabled =
+ jotaiStore.get(agentFlagsAtom).data?.enable_desktop_local_filesystem === true;
+ const disabledTools = jotaiStore.get(disabledToolsAtom);
+
+ // Extract the original user query BEFORE removing messages (reload mode).
+ let userQueryToDisplay: string | undefined;
+ let originalUserMessageContent: ThreadMessageLike["content"] | null = null;
+ let originalUserMessageMetadata: ThreadMessageLike["metadata"] | undefined;
+ let sourceUserMessageId: string | undefined = editExtras?.sourceUserMessageId;
+
+ if (!isEdit) {
+ const lastUserMessage = [...priorMessages].reverse().find((m) => m.role === "user");
+ if (lastUserMessage) {
+ sourceUserMessageId = lastUserMessage.id;
+ originalUserMessageContent = lastUserMessage.content;
+ originalUserMessageMetadata = lastUserMessage.metadata;
+ for (const part of lastUserMessage.content) {
+ if (typeof part === "object" && part.type === "text" && "text" in part) {
+ userQueryToDisplay = part.text;
+ break;
+ }
+ }
+ }
+ } else {
+ userQueryToDisplay = newUserQuery;
+ }
+
+ // Seed the durable overlay with the pre-regenerate conversation and flip
+ // to running so the page renders the live stream.
+ chatStreamStore.begin(streamThreadId, priorMessages);
+
+ const controller = new AbortController();
+ chatStreamStore.beginActive(streamThreadId, controller);
+
+ let userMsgId = `msg-user-${Date.now()}`;
+ let assistantMsgId = `msg-assistant-${Date.now()}`;
+ const currentThinkingSteps = new Map();
+
+ const contentPartsState: ContentPartsState = {
+ contentParts: [],
+ currentTextPartIndex: -1,
+ currentReasoningPartIndex: -1,
+ toolCallIndices: new Map(),
+ };
+ const { contentParts } = contentPartsState;
+ let regenerateAccepted = false;
+ let streamBatcher: FrameBatchedUpdater | null = null;
+
+ const userMessage: ThreadMessageLike = {
+ id: userMsgId,
+ role: "user",
+ content: isEdit
+ ? (editExtras?.userMessageContent ?? [{ type: "text", text: newUserQuery ?? "" }])
+ : originalUserMessageContent || [{ type: "text", text: userQueryToDisplay || "" }],
+ createdAt: new Date(),
+ metadata: isEdit ? undefined : originalUserMessageMetadata,
+ };
+ const sourceMentionedDocs =
+ sourceUserMessageId && messageDocumentsMap[sourceUserMessageId]
+ ? messageDocumentsMap[sourceUserMessageId]
+ : [];
+ try {
+ const selection = await getAgentFilesystemSelection(workspaceId, { localFilesystemEnabled });
+ const regenerateDocIds = sourceMentionedDocs.filter((d) => d.kind === "doc").map((d) => d.id);
+ const regenerateFolderIds = sourceMentionedDocs
+ .filter((d) => d.kind === "folder")
+ .map((d) => d.id);
+ const regenerateConnectors = sourceMentionedDocs.filter((d) => d.kind === "connector");
+ const regenerateThreadIds = sourceMentionedDocs
+ .filter((d) => d.kind === "thread")
+ .map((d) => d.id);
+
+ const requestBody: Record = {
+ workspace_id: workspaceId,
+ user_query: newUserQuery,
+ disabled_tools: disabledTools.length > 0 ? disabledTools : undefined,
+ filesystem_mode: selection.filesystem_mode,
+ client_platform: selection.client_platform,
+ local_filesystem_mounts: selection.local_filesystem_mounts,
+ mentioned_document_ids: regenerateDocIds.length > 0 ? regenerateDocIds : undefined,
+ mentioned_folder_ids: regenerateFolderIds.length > 0 ? regenerateFolderIds : undefined,
+ mentioned_connector_ids:
+ regenerateConnectors.length > 0 ? regenerateConnectors.map((d) => d.id) : undefined,
+ mentioned_connectors: regenerateConnectors.length > 0 ? regenerateConnectors : undefined,
+ mentioned_thread_ids: regenerateThreadIds.length > 0 ? regenerateThreadIds : undefined,
+ mentioned_documents: sourceMentionedDocs.length > 0 ? sourceMentionedDocs : undefined,
+ };
+ if (isEdit) {
+ requestBody.user_images = editExtras?.userImages ?? [];
+ }
+ if (editFromPosition?.fromMessageId != null) {
+ requestBody.from_message_id = editFromPosition.fromMessageId;
+ if (editFromPosition.revertActions) {
+ requestBody.revert_actions = true;
+ }
+ }
+ const response = await fetchWithTurnCancellingRetry(() =>
+ authenticatedFetch(getRegenerateUrl(streamThreadId), {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify(requestBody),
+ signal: controller.signal,
+ })
+ );
+
+ if (!response.ok) {
+ throw await toHttpResponseError(response);
+ }
+ regenerateAccepted = true;
+
+ chatStreamStore.setMessages(streamThreadId, (prev) => {
+ let base = prev;
+ if (editFromPosition?.fromMessageId != null) {
+ const targetId = `msg-${editFromPosition.fromMessageId}`;
+ const sliceIndex = prev.findIndex((m) => m.id === targetId);
+ if (sliceIndex >= 0) {
+ base = prev.slice(0, sliceIndex);
+ }
+ } else if (prev.length >= 2) {
+ base = prev.slice(0, -2);
+ }
+ return [
+ ...base,
+ userMessage,
+ {
+ id: assistantMsgId,
+ role: "assistant",
+ content: [{ type: "text", text: "" }],
+ createdAt: new Date(),
+ },
+ ];
+ });
+ if (sourceMentionedDocs.length > 0) {
+ jotaiStore.set(messageDocumentsMapAtom, (prev) => ({
+ ...prev,
+ [userMsgId]: sourceMentionedDocs,
+ }));
+ }
+
+ const flushMessages = () => {
+ chatStreamStore.setMessages(streamThreadId, (prev) =>
+ prev.map((m) =>
+ m.id === assistantMsgId
+ ? { ...m, content: buildContentForUI(contentPartsState, toolsWithUI) }
+ : m
+ )
+ );
+ };
+ const { batcher, scheduleFlush, forceFlush } = createStreamFlushHelpers(flushMessages);
+ streamBatcher = batcher;
+
+ await consumeSseEvents(response, async (parsed) => {
+ if (
+ processSharedStreamEvent(parsed, {
+ contentPartsState,
+ toolsWithUI,
+ currentThinkingSteps,
+ scheduleFlush,
+ forceFlush,
+ onTokenUsage: (data) => {
+ tokenUsageStore.set(assistantMsgId, data);
+ },
+ onTurnStatus: (data) => {
+ if (data.status === "cancelling") {
+ chatStreamStore.markRecentCancel();
+ }
+ },
+ })
+ ) {
+ return;
+ }
+ switch (parsed.type) {
+ case "data-action-log": {
+ applyActionLogSse(queryClient, streamThreadId, workspaceId, parsed.data);
+ break;
+ }
+
+ case "data-action-log-updated": {
+ applyActionLogUpdatedSse(
+ queryClient,
+ streamThreadId,
+ parsed.data.id,
+ parsed.data.reversible
+ );
+ break;
+ }
+
+ case "data-turn-info": {
+ const turnId = readStreamedChatTurnId(parsed.data);
+ if (turnId) {
+ chatStreamStore.setMessages(streamThreadId, (prev) =>
+ applyTurnIdToAssistantMessageList(prev, assistantMsgId, turnId)
+ );
+ }
+ break;
+ }
+
+ case "data-user-message-id": {
+ const parsedMsg = readStreamedMessageId(parsed.data);
+ if (!parsedMsg) break;
+ const newUserMsgId = `msg-${parsedMsg.messageId}`;
+ const oldUserMsgId = userMsgId;
+ chatStreamStore.setMessages(streamThreadId, (prev) =>
+ prev.map((m) =>
+ m.id === oldUserMsgId
+ ? mergeChatTurnIdIntoMessage({ ...m, id: newUserMsgId }, parsedMsg.turnId)
+ : m
+ )
+ );
+ if (sourceMentionedDocs.length > 0) {
+ jotaiStore.set(messageDocumentsMapAtom, (prev) => {
+ if (!(oldUserMsgId in prev)) {
+ return { ...prev, [newUserMsgId]: sourceMentionedDocs };
+ }
+ const { [oldUserMsgId]: _removed, ...rest } = prev;
+ return { ...rest, [newUserMsgId]: sourceMentionedDocs };
+ });
+ }
+ userMsgId = newUserMsgId;
+ break;
+ }
+
+ case "data-assistant-message-id": {
+ const parsedMsg = readStreamedMessageId(parsed.data);
+ if (!parsedMsg) break;
+ const newAssistantMsgId = `msg-${parsedMsg.messageId}`;
+ const oldAssistantMsgId = assistantMsgId;
+ tokenUsageStore.rename(oldAssistantMsgId, newAssistantMsgId);
+ chatStreamStore.setMessages(streamThreadId, (prev) =>
+ prev.map((m) =>
+ m.id === oldAssistantMsgId
+ ? mergeChatTurnIdIntoMessage({ ...m, id: newAssistantMsgId }, parsedMsg.turnId)
+ : m
+ )
+ );
+ assistantMsgId = newAssistantMsgId;
+ break;
+ }
+
+ case "data-revert-results": {
+ const summary = parsed.data;
+ const failureCount =
+ summary.failed + summary.not_reversible + (summary.permission_denied ?? 0);
+ if (failureCount > 0) {
+ toast.warning(
+ `Pre-revert: ${summary.reverted}/${summary.total} undone, ${failureCount} could not be rolled back.`
+ );
+ } else if (summary.reverted > 0) {
+ toast.success(
+ summary.reverted === 1
+ ? "Reverted 1 downstream action before regenerating."
+ : `Reverted ${summary.reverted} downstream actions before regenerating.`
+ );
+ }
+ for (const r of summary.results) {
+ if (r.status === "reverted" || r.status === "already_reverted") {
+ markActionRevertedInCache(
+ queryClient,
+ streamThreadId,
+ r.action_id,
+ r.new_action_id ?? null
+ );
+ }
+ }
+ break;
+ }
+ }
+ });
+
+ batcher.flush();
+
+ if (contentParts.length > 0) {
+ trackChatResponseReceived(workspaceId, streamThreadId);
+ }
+ } catch (error) {
+ streamBatcher?.dispose();
+ await handleStreamTerminalError({
+ error,
+ flow: "regenerate",
+ threadId: streamThreadId,
+ assistantMsgId,
+ accepted: regenerateAccepted,
+ workspaceId,
+ });
+ } finally {
+ chatStreamStore.setRunning(streamThreadId, false);
+ chatStreamStore.clearActive(controller);
+ void queryClient.invalidateQueries({
+ queryKey: cacheKeys.threads.messages(streamThreadId),
+ });
+ void queryClient.invalidateQueries({
+ queryKey: cacheKeys.threads.detail(streamThreadId),
+ });
+ }
+}
+
+export type { HitlDecision };
diff --git a/surfsense_web/lib/chat/stream-engine/helpers.ts b/surfsense_web/lib/chat/stream-engine/helpers.ts
new file mode 100644
index 000000000..afa1851e1
--- /dev/null
+++ b/surfsense_web/lib/chat/stream-engine/helpers.ts
@@ -0,0 +1,154 @@
+import { z } from "zod";
+import type { MentionedDocumentInfo } from "@/atoms/chat/mentioned-documents.atom";
+import type { ToolUIGate } from "@/lib/chat/streaming-state";
+
+/**
+ * Every tool call renders a card. The sentinel ``"all"`` matches every tool
+ * — the legacy ``BASE_TOOLS_WITH_UI`` allowlist was dropped so unknown tool
+ * calls route through the generic ``ToolFallback``. Persisted payload size
+ * stays bounded because the backend's ``format_thinking_step`` summarisation
+ * and the ``result_length``-only default for unknown tools keep the JSON
+ * from ballooning.
+ */
+export const TOOLS_WITH_UI_ALL: ToolUIGate = "all";
+
+export const TURN_CANCELLING_INITIAL_DELAY_MS = 200;
+export const TURN_CANCELLING_BACKOFF_FACTOR = 2;
+export const TURN_CANCELLING_MAX_DELAY_MS = 1500;
+export const RECENT_CANCEL_WINDOW_MS = 5_000;
+
+export function sleep(ms: number): Promise {
+ return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+export function computeFallbackTurnCancellingRetryDelay(attempt: number): number {
+ const safeAttempt = Math.max(1, attempt);
+ const raw =
+ TURN_CANCELLING_INITIAL_DELAY_MS * TURN_CANCELLING_BACKOFF_FACTOR ** (safeAttempt - 1);
+ return Math.min(raw, TURN_CANCELLING_MAX_DELAY_MS);
+}
+
+/**
+ * Generate a synthetic ``toolCallId`` for an action_request that has no
+ * matching streamed tool-call card (HITL-blocked subagent calls don't surface
+ * as tool-call events). Suffixes a counter when the base id is already taken
+ * — sequential interrupts for the same tool name otherwise collide on
+ * ``interrupt-${name}-${i}`` and crash assistant-ui with a duplicate-key error.
+ */
+export function freshSynthToolCallId(
+ toolCallIndices: Map,
+ toolName: string,
+ index: number
+): string {
+ const base = `interrupt-${toolName}-${index}`;
+ if (!toolCallIndices.has(base)) return base;
+ let n = 1;
+ while (toolCallIndices.has(`${base}-${n}`)) n++;
+ return `${base}-${n}`;
+}
+
+/**
+ * Pair each ``action_request`` to a unique pending tool-call card, preserving
+ * order so ``decisions[i]`` lines up with ``action_requests[i]`` on the wire.
+ *
+ * Same-name bundles (e.g. three ``create_jira_issue``) used to collapse onto
+ * one card because the matcher keyed by name; this consumes each card via the
+ * ``claimed`` set and walks forward in DOM order.
+ */
+export function pairBundleToolCallIds(
+ toolCallIndices: Map,
+ contentParts: Array<{
+ type: string;
+ toolName?: string;
+ result?: unknown;
+ }>,
+ actionRequests: ReadonlyArray<{ name: string }>
+): Array {
+ const claimed = new Set();
+ const paired: Array = [];
+ for (const action of actionRequests) {
+ let matched: string | null = null;
+ for (const [tcId, idx] of toolCallIndices) {
+ if (claimed.has(tcId)) continue;
+ const part = contentParts[idx];
+ if (!part || part.type !== "tool-call" || part.toolName !== action.name) continue;
+ const result = part.result as Record | undefined | null;
+ if (result == null || (result.__interrupt__ === true && !result.__decided__)) {
+ matched = tcId;
+ claimed.add(tcId);
+ break;
+ }
+ }
+ paired.push(matched);
+ }
+ return paired;
+}
+
+/**
+ * Zod schema for mentioned document info (for type-safe parsing).
+ *
+ * ``kind`` defaults to ``"doc"`` so messages persisted before folder
+ * mentions existed deserialise unchanged.
+ */
+const MentionedDocumentInfoSchema = z.object({
+ id: z.number(),
+ title: z.string(),
+ document_type: z.string().optional(),
+ kind: z
+ .union([z.literal("doc"), z.literal("folder"), z.literal("connector"), z.literal("thread")])
+ .optional()
+ .default("doc"),
+ connector_type: z.string().optional(),
+ account_name: z.string().optional(),
+});
+
+const MentionedDocumentsPartSchema = z.object({
+ type: z.literal("mentioned-documents"),
+ documents: z.array(MentionedDocumentInfoSchema),
+});
+
+/**
+ * Extract mentioned documents from message content (type-safe with Zod).
+ */
+export function extractMentionedDocuments(content: unknown): MentionedDocumentInfo[] {
+ if (!Array.isArray(content)) return [];
+
+ for (const part of content) {
+ const result = MentionedDocumentsPartSchema.safeParse(part);
+ if (result.success) {
+ return result.data.documents.map((doc) => {
+ if (doc.kind === "connector") {
+ return {
+ id: doc.id,
+ title: doc.title,
+ kind: "connector",
+ connector_type: doc.connector_type ?? doc.document_type ?? "UNKNOWN",
+ account_name: doc.account_name ?? doc.title,
+ };
+ }
+ if (doc.kind === "folder") {
+ return {
+ id: doc.id,
+ title: doc.title,
+ kind: "folder",
+ };
+ }
+ if (doc.kind === "thread") {
+ return {
+ id: doc.id,
+ title: doc.title,
+ kind: "thread",
+ };
+ }
+ return {
+ id: doc.id,
+ title: doc.title,
+ document_type: doc.document_type ?? "UNKNOWN",
+ kind: "doc",
+ };
+ });
+ }
+ }
+
+ return [];
+}
diff --git a/surfsense_web/lib/chat/stream-engine/store.ts b/surfsense_web/lib/chat/stream-engine/store.ts
new file mode 100644
index 000000000..44627663f
--- /dev/null
+++ b/surfsense_web/lib/chat/stream-engine/store.ts
@@ -0,0 +1,177 @@
+import type { ThreadMessageLike } from "@assistant-ui/react";
+import { createTokenUsageStore } from "@/components/assistant-ui/token-usage-context";
+import type { PendingInterruptState } from "@/features/chat-messages/hitl";
+
+/**
+ * Durable, per-thread streaming state for a single in-flight chat turn.
+ *
+ * Lives at module scope (see {@link chatStreamStore}) so a running turn's
+ * messages / interrupts survive the chat page unmounting during in-app
+ * navigation. The React tree consumes it through ``useChatStream`` via
+ * ``useSyncExternalStore`` (state lives outside the render cycle).
+ */
+export interface ThreadStreamState {
+ threadId: number;
+ messages: ThreadMessageLike[];
+ isRunning: boolean;
+ pendingInterrupts: PendingInterruptState[];
+}
+
+type Listener = () => void;
+
+/**
+ * Module-level singleton external store for chat streaming.
+ *
+ * Single-active-stream invariant: at most one turn streams at a time,
+ * tracked by {@link active}. Per-thread state is still keyed by threadId so
+ * the currently-streaming thread and the currently-viewed thread can differ
+ * (e.g. a stream finishes while the user is on another chat).
+ *
+ * ponytail: unbounded ``states`` map is bounded in practice by
+ * ``clearInactive`` (called on navigation) + ``clear`` (called after the DB
+ * re-hydrates a finished turn). Upgrade path if that ever leaks: LRU cap.
+ */
+class ChatStreamStore {
+ private states = new Map();
+ private listeners = new Set();
+
+ /** Shared, cross-navigation token-usage store (one instance app-wide). */
+ readonly tokenUsage = createTokenUsageStore();
+
+ /** The one in-flight turn's abort handle, or null when idle. */
+ private active: { threadId: number; controller: AbortController } | null = null;
+
+ /** Timestamp of the last explicit cancel, for the THREAD_BUSY retry window. */
+ recentCancelRequestedAt = 0;
+
+ subscribe = (listener: Listener): (() => void) => {
+ this.listeners.add(listener);
+ return () => {
+ this.listeners.delete(listener);
+ };
+ };
+
+ private notify(): void {
+ for (const l of this.listeners) l();
+ }
+
+ /** Snapshot for ``useSyncExternalStore``; stable ref between mutations. */
+ getSnapshot = (threadId: number | null): ThreadStreamState | null => {
+ if (threadId == null) return null;
+ return this.states.get(threadId) ?? null;
+ };
+
+ isRunning(threadId: number | null): boolean {
+ if (threadId == null) return false;
+ return this.states.get(threadId)?.isRunning ?? false;
+ }
+
+ getMessages(threadId: number): ThreadMessageLike[] {
+ return this.states.get(threadId)?.messages ?? [];
+ }
+
+ getPendingInterrupts(threadId: number): PendingInterruptState[] {
+ return this.states.get(threadId)?.pendingInterrupts ?? [];
+ }
+
+ private ensure(threadId: number): ThreadStreamState {
+ let s = this.states.get(threadId);
+ if (!s) {
+ s = { threadId, messages: [], isRunning: false, pendingInterrupts: [] };
+ this.states.set(threadId, s);
+ }
+ return s;
+ }
+
+ private commit(threadId: number, next: ThreadStreamState): void {
+ this.states.set(threadId, next);
+ this.notify();
+ }
+
+ /** Seed a thread's state at the start of a fresh turn (running=true). */
+ begin(threadId: number, messages: ThreadMessageLike[]): void {
+ this.commit(threadId, { threadId, messages, isRunning: true, pendingInterrupts: [] });
+ }
+
+ setMessages(threadId: number, updater: (prev: ThreadMessageLike[]) => ThreadMessageLike[]): void {
+ const prev = this.ensure(threadId);
+ const messages = updater(prev.messages);
+ if (messages === prev.messages) return;
+ this.commit(threadId, { ...prev, messages });
+ }
+
+ setRunning(threadId: number, running: boolean): void {
+ const prev = this.ensure(threadId);
+ if (prev.isRunning === running) return;
+ this.commit(threadId, { ...prev, isRunning: running });
+ }
+
+ setPendingInterrupts(
+ threadId: number,
+ updater: (prev: PendingInterruptState[]) => PendingInterruptState[]
+ ): void {
+ const prev = this.ensure(threadId);
+ const pendingInterrupts = updater(prev.pendingInterrupts);
+ if (pendingInterrupts === prev.pendingInterrupts) return;
+ this.commit(threadId, { ...prev, pendingInterrupts });
+ }
+
+ /**
+ * A thread whose overlay must survive DB re-hydration / navigation: it is
+ * either streaming or paused awaiting a HITL decision (the pending
+ * interrupts + interrupt cards live only in the overlay).
+ */
+ private isPinned(s: ThreadStreamState): boolean {
+ return s.isRunning || s.pendingInterrupts.length > 0;
+ }
+
+ /** Drop a thread's overlay once the DB is authoritative. No-op while pinned. */
+ clear(threadId: number): void {
+ const s = this.states.get(threadId);
+ if (!s || this.isPinned(s)) return;
+ this.states.delete(threadId);
+ this.notify();
+ }
+
+ /** Evict every non-pinned thread except ``exceptThreadId`` (memory bound). */
+ clearInactive(exceptThreadId: number | null): void {
+ let changed = false;
+ for (const [id, s] of this.states) {
+ if (id === exceptThreadId || this.isPinned(s)) continue;
+ this.states.delete(id);
+ changed = true;
+ }
+ if (changed) this.notify();
+ }
+
+ // ---- active-stream lifecycle -------------------------------------------
+
+ /** Register a new in-flight turn, aborting any previous one first. */
+ beginActive(threadId: number, controller: AbortController): void {
+ this.abortActive();
+ this.active = { threadId, controller };
+ }
+
+ get activeThreadId(): number | null {
+ return this.active?.threadId ?? null;
+ }
+
+ /** Clear the active handle iff it still points at ``controller``. */
+ clearActive(controller: AbortController): void {
+ if (this.active?.controller === controller) this.active = null;
+ }
+
+ /** Abort the in-flight turn's fetch (client disconnect, not a server stop). */
+ abortActive(): void {
+ if (this.active) {
+ this.active.controller.abort();
+ this.active = null;
+ }
+ }
+
+ markRecentCancel(): void {
+ this.recentCancelRequestedAt = Date.now();
+ }
+}
+
+export const chatStreamStore = new ChatStreamStore();
diff --git a/surfsense_web/lib/chat/stream-engine/use-chat-stream.ts b/surfsense_web/lib/chat/stream-engine/use-chat-stream.ts
new file mode 100644
index 000000000..ccfb32481
--- /dev/null
+++ b/surfsense_web/lib/chat/stream-engine/use-chat-stream.ts
@@ -0,0 +1,18 @@
+"use client";
+
+import { useCallback, useSyncExternalStore } from "react";
+import { chatStreamStore, type ThreadStreamState } from "./store";
+
+/**
+ * Subscribe to the durable streaming state for a given thread.
+ *
+ * Returns the live ``ThreadStreamState`` while a turn is streaming (or
+ * pending re-hydration from the DB), or ``null`` when the thread has no
+ * active overlay — in which case the page falls back to its DB-hydrated
+ * messages. State lives in the module-level {@link chatStreamStore}, so it
+ * survives the chat page unmounting during in-app navigation.
+ */
+export function useChatStream(threadId: number | null): ThreadStreamState | null {
+ const getSnapshot = useCallback(() => chatStreamStore.getSnapshot(threadId), [threadId]);
+ return useSyncExternalStore(chatStreamStore.subscribe, getSnapshot, () => null);
+}
diff --git a/surfsense_web/lib/connectors-marketing/instagram.tsx b/surfsense_web/lib/connectors-marketing/instagram.tsx
index d22400b7a..5422a0c90 100644
--- a/surfsense_web/lib/connectors-marketing/instagram.tsx
+++ b/surfsense_web/lib/connectors-marketing/instagram.tsx
@@ -6,28 +6,32 @@ export const instagram: ConnectorPageContent = {
name: "Instagram",
icon: IconBrandInstagram,
- metaTitle: "Instagram Scraper API for Creator Research | SurfSense",
+ metaTitle: "Instagram Scraper API for Profiles and Reels | SurfSense",
metaDescription:
- "Scrape public Instagram posts, reels, and profiles at scale with the SurfSense Instagram Scraper API. No login, no official API, plus a free tier. Start now.",
+ "Instagram scraper API for public profiles, posts, and reels. No login or Graph API review. Structured data for AI agents, plus a free tier. Start now.",
keywords: [
"instagram scraper",
"instagram scraper api",
"instagram api",
"instagram api alternative",
"scrape instagram",
+ "instagram scraping",
"instagram graph api alternative",
"instagram profile scraper",
"instagram post scraper",
+ "instagram posts scraper",
"instagram reel scraper",
+ "instagram reels scraper",
+ "instagram data scraper",
"instagram data api",
"instagram mcp server",
"creator research",
"social listening",
],
- h1: "Instagram Scraper API for Creator Research and Social Listening",
+ h1: "Instagram Scraper API for Profiles, Posts, and Reels",
heroLede:
- "The SurfSense Instagram API extracts public posts, reels, and profile details without logging in or registering for the Instagram Graph API. Give your AI agents a live feed of what creators post, so you spot trends and shifts in engagement first.",
+ "The SurfSense Instagram scraper extracts public profiles, posts, and reels without logging in or registering for the Instagram Graph API. Give your AI agents a live feed of what creators post, so you spot trends and shifts in engagement first.",
transcript: {
prompt: "Pull recent reels from @competitor and summarize what they're posting",
@@ -54,48 +58,48 @@ export const instagram: ConnectorPageContent = {
},
extractIntro:
- "Every call returns structured items keyed by type. Point the API at a public profile, post, or reel URL, or discover creators with a search query.",
+ "Every Instagram scraper call returns structured items keyed by type. Point the API at a public profile, post, or reel URL, or discover creators with a search query.",
extractFields: [
{
- label: "Posts & Reels",
+ label: "Posts and reels",
description:
- "Caption, hashtags, mentions, like and comment counts, media URLs, dimensions, and timestamp.",
+ "Caption, hashtags, mentions, like and comment counts, media URLs, dimensions, and timestamp from any public post or reel.",
},
{
- label: "Profiles",
+ label: "Profile data",
description:
- "Follower, following, and post counts, bio, external URL, verified and business flags.",
+ "Follower, following, and post counts, bio, external URL, verified and business flags for public Instagram profiles.",
},
{
- label: "Owner & Media",
+ label: "Owner and media",
description:
"Owner username and id on every item, plus image and video URLs, alt text, and view counts.",
},
],
- useCasesHeading: "What teams do with the Instagram API",
+ useCasesHeading: "What teams do with the Instagram scraper API",
useCases: [
{
title: "Creator and competitor monitoring",
description:
- "Track what your competitors and target creators post, and how engagement moves. Feed the stream to an agent that flags viral formats, launches, and shifts in cadence the moment they land.",
+ "Scrape Instagram profiles and feeds to track what competitors and target creators post, and how engagement moves. Feed the stream to an agent that flags viral formats, launches, and shifts in cadence the moment they land.",
},
{
title: "Content and format research",
description:
- "Study a creator's recent posts and reels to see which formats earn the most likes and comments, and turn that into a content calendar your team can act on.",
+ "Pull a creator's recent posts and reels to see which formats earn the most likes and comments, and turn that into a content calendar your team can act on.",
},
{
title: "Influencer vetting and outreach",
description:
- "Verify a creator's follower count, post cadence, and real engagement from public data before you pay for a partnership.",
+ "Use the Instagram profile scraper to verify follower count, post cadence, and real engagement from public data before you pay for a partnership.",
},
],
comparison: {
- heading: "An Instagram API alternative built for agents",
+ heading: "An Instagram Graph API alternative built for agents",
intro:
- "The official Instagram Graph API requires a Business account, app review, and access tokens, and it can't read arbitrary public profiles. Here is how SurfSense compares.",
+ "The official Instagram Graph API requires a Business account, app review, and access tokens, and it cannot read arbitrary public profiles. SurfSense is an Instagram API alternative for public data. Here is how it compares.",
columnLabel: "Instagram Graph API",
rows: [
{
@@ -259,7 +263,12 @@ export const instagram: ConnectorPageContent = {
{
question: "Do I need an Instagram account or the Graph API?",
answer:
- "No. This is an independent alternative to the Instagram Graph API, not a wrapper. You do not create a Business account, pass app review, or manage access tokens. You call the SurfSense API with one key, or add the MCP server to your agent, and get structured data back.",
+ "No. This Instagram scraper API is an independent alternative to the Instagram Graph API, not a wrapper. You do not create a Business account, pass app review, or manage access tokens. You call SurfSense with one key, or add the MCP server to your agent, and get structured profile, post, and reel data back.",
+ },
+ {
+ question: "What Instagram data can I scrape?",
+ answer:
+ "Public profiles, posts, and reels. Each item includes captions, hashtags, mentions, engagement counts, media URLs, and owner metadata. Point the Instagram profile scraper at a handle, or pass post and reel URLs directly. Discover creators with search queries when you do not have a URL yet.",
},
{
question: "What are the rate limits?",
@@ -274,11 +283,11 @@ export const instagram: ConnectorPageContent = {
],
related: [
- { label: "Reddit API", href: "/reddit" },
+ { label: "TikTok API", href: "/tiktok" },
{ label: "YouTube API", href: "/youtube" },
+ { label: "Reddit API", href: "/reddit" },
{ label: "Google Maps API", href: "/google-maps" },
{ label: "SERP API", href: "/google-search" },
{ label: "SurfSense MCP Server", href: "/mcp-server" },
- { label: "Read the docs", href: "/docs" },
],
};
diff --git a/surfsense_web/lib/connectors-marketing/tiktok.tsx b/surfsense_web/lib/connectors-marketing/tiktok.tsx
index d77865633..6063fd1ae 100644
--- a/surfsense_web/lib/connectors-marketing/tiktok.tsx
+++ b/surfsense_web/lib/connectors-marketing/tiktok.tsx
@@ -6,30 +6,33 @@ export const tiktok: ConnectorPageContent = {
name: "TikTok",
icon: IconBrandTiktok,
- metaTitle: "TikTok Scraper API for Trend and Creator Research | SurfSense",
+ metaTitle: "TikTok Scraper API for Videos and Comments | SurfSense",
metaDescription:
- "Scrape public TikTok videos, comments, accounts, and trending feeds by hashtag, profile, or URL with the SurfSense TikTok Scraper API. No approval process or research-API gatekeeping, plus a free tier. Start now.",
+ "TikTok scraper API for public videos, comments, hashtags, and profiles. No Research API approval. Structured data for AI agents, plus a free tier. Start now.",
keywords: [
"tiktok scraper",
"tiktok scraper api",
"tiktok api",
"tiktok api alternative",
"scrape tiktok",
+ "tiktok scraping",
+ "tiktok research api alternative",
"tiktok data api",
- "tiktok hashtag scraper",
"tiktok comments scraper",
+ "tiktok comment scraper",
+ "tiktok hashtag scraper",
+ "tiktok profile scraper",
+ "tiktok video scraper",
"tiktok trending scraper",
"tiktok user search",
- "tiktok trend tracking",
"tiktok mcp",
"social listening",
"influencer research tool",
- "short-form video analytics",
],
- h1: "TikTok Scraper API for Trend and Creator Research",
+ h1: "TikTok Scraper API for Videos, Comments, and Trend Research",
heroLede:
- "The SurfSense TikTok API extracts public videos by hashtag, creator profile, or URL without TikTok's approval-gated Research API. Give your AI agents a live feed of what your market watches and shares, so you catch a trend while it is still rising.",
+ "The SurfSense TikTok scraper extracts public videos by hashtag, creator profile, or URL, plus comment threads and trending feeds, without TikTok's approval-gated Research API. Give your AI agents a live feed of what your market watches and shares, so you catch a trend while it is still rising.",
transcript: {
prompt: "Find trending TikToks about meal prep this week",
@@ -55,46 +58,48 @@ export const tiktok: ConnectorPageContent = {
},
extractIntro:
- "Every call returns structured items. Scrape videos from a hashtag, creator profile, or video URL — or switch verbs to pull a video's comments, discover accounts by keyword, or fetch the current trending feed.",
+ "Every TikTok scraper call returns structured items. Scrape videos from a hashtag, creator profile, or video URL, or switch verbs to pull a video's comments, discover accounts by keyword, or fetch the current trending feed.",
extractFields: [
{
label: "Videos",
- description: "Caption text, canonical web URL, duration, and cover image for each video.",
+ description:
+ "Caption text, canonical web URL, duration, and cover image for each TikTok video.",
},
{
label: "Engagement",
description:
- "Play, like, comment, share, and save counts — the signal for what is breaking out.",
+ "Play, like, comment, share, and save counts, the signal for what is breaking out.",
},
{
- label: "Authors",
+ label: "Authors and profiles",
description: "Creator handle, nickname, follower and heart counts, and verified status.",
},
+ {
+ label: "Comments",
+ description:
+ "Public comment threads on any video URL, so you can read audience reaction beyond vanity views.",
+ },
{
label: "Music",
- description: "Track name, artist, and whether the sound is original — the seed of a trend.",
+ description: "Track name, artist, and whether the sound is original, the seed of a trend.",
},
{
label: "Hashtags",
description: "Every hashtag on a video, so you can map a topic cluster or campaign.",
},
- {
- label: "Timestamps",
- description: "Created and scraped times so you can track a video's momentum over runs.",
- },
],
- useCasesHeading: "What teams do with the TikTok API",
+ useCasesHeading: "What teams do with the TikTok scraper API",
useCases: [
{
title: "Trend and hashtag monitoring",
description:
- "Track a hashtag and feed the stream to an agent that flags breakout videos, rising sounds, and formats before they saturate. Catch the wave while it is still rising, not after.",
+ "Use the TikTok hashtag scraper to track a topic and feed the stream to an agent that flags breakout videos, rising sounds, and formats before they saturate. Catch the wave while it is still rising, not after.",
},
{
title: "Creator and influencer discovery",
description:
- "Surface the creators driving a topic, ranked by real engagement, so your team shortlists partners from data instead of a manager's pitch deck.",
+ "Scrape TikTok profiles and search users by keyword to surface the creators driving a topic, ranked by real engagement, so your team shortlists partners from data instead of a manager's pitch deck.",
},
{
title: "Competitor content analysis",
@@ -102,16 +107,16 @@ export const tiktok: ConnectorPageContent = {
"Watch what your category posts and what actually lands. Turn a competitor's best-performing formats and hooks into your own content brief.",
},
{
- title: "Campaign and sentiment tracking",
+ title: "Campaign and comment sentiment",
description:
- "Measure how a launch or branded hashtag spreads across TikTok — video count, reach, and engagement over time — then pull the comments on top videos to read how the audience actually reacts, not just a vanity view count.",
+ "Measure how a launch or branded hashtag spreads across TikTok, then use the TikTok comments scraper on top videos to read how the audience actually reacts, not just a vanity view count.",
},
],
comparison: {
- heading: "A TikTok API alternative built for agents",
+ heading: "A TikTok Research API alternative built for agents",
intro:
- "TikTok's official Research API is approval-gated and largely limited to academic and nonprofit use. If you cannot get access or need it for commercial research, here is how SurfSense compares.",
+ "TikTok's official Research API is approval-gated and largely limited to academic and nonprofit use. If you cannot get access or need commercial TikTok scraping, SurfSense is a TikTok API alternative. Here is how it compares.",
columnLabel: "Official TikTok API",
rows: [
{
@@ -154,7 +159,7 @@ export const tiktok: ConnectorPageContent = {
schema: {
requestNote:
- "Provide at least one source: urls, profiles, hashtags, or search_queries. Up to 20 sources per call.",
+ "Provide at least one source: urls, profiles, or hashtags. Up to 20 sources per call. To find accounts by keyword, use the user search verb.",
request: [
{
name: "urls",
@@ -175,13 +180,6 @@ export const tiktok: ConnectorPageContent = {
defaultValue: "[]",
description: "Hashtag names to scrape, without the # prefix. Max 20.",
},
- {
- name: "search_queries",
- type: "string[]",
- defaultValue: "[]",
- description:
- "Keyword search terms. Keyword video search is login-walled and returns no videos — use hashtags/profiles/urls for videos, or user_search for accounts. Max 20.",
- },
{
name: "results_per_page",
type: "integer",
@@ -253,9 +251,19 @@ export const tiktok: ConnectorPageContent = {
"SurfSense reads only public TikTok data, the same videos any logged-out visitor can see. It never logs in and cannot access private or deleted content. As always, review TikTok's terms and your own compliance needs before you run at scale.",
},
{
- question: "Does this need the official TikTok API?",
+ question: "Does this need the official TikTok Research API?",
answer:
- "No. It is an independent alternative, not a wrapper on the Research API, so there is no application or approval process. You call the SurfSense API with one key, or add the MCP server to your agent, and get structured videos back.",
+ "No. This TikTok scraper API is an independent alternative, not a wrapper on the Research API, so there is no application or approval process. You call SurfSense with one key, or add the MCP server to your agent, and get structured videos, comments, and profile data back.",
+ },
+ {
+ question: "What TikTok data can I scrape?",
+ answer:
+ "Four verbs: scrape (videos by hashtag, profile, or URL), comments (a video's public comment thread), user search (find accounts by keyword), and trending (the current Explore feed). Each returns structured items and is billed per item returned.",
+ },
+ {
+ question: "Can I scrape TikTok comments and hashtags?",
+ answer:
+ "Yes. Pass a video URL to the comments endpoint for the public comment thread. Pass hashtag names or /tag/ URLs to the TikTok hashtag scraper to pull videos under that tag. Keyword video search is login-walled on TikTok, so hashtags and direct URLs are the reliable discovery paths; to find accounts by keyword, use the user search verb.",
},
{
question: "What are the rate limits?",
@@ -265,18 +273,14 @@ export const tiktok: ConnectorPageContent = {
{
question: "Can I scrape a specific creator's videos?",
answer:
- "Pass a profile or profile URL and you always get the account's metadata — name, followers, bio, verification. TikTok often withholds the profile video list from automated clients, so that list can come back empty even for a public account; for reliable video results, scrape by hashtag or by a direct video URL.",
- },
- {
- question: "What TikTok data can I scrape?",
- answer:
- "Four verbs: scrape (videos by hashtag, profile, or URL), comments (a video's public comment thread), user search (find accounts by keyword — the reliable discovery path, since keyword video search is login-walled), and trending (the current Explore feed). Each returns structured items and is billed per item returned.",
+ "Pass a profile or profile URL and you always get the account's metadata: name, followers, bio, verification. TikTok often withholds the profile video list from automated clients, so that list can come back empty even for a public account. For reliable video results, scrape by hashtag or by a direct video URL.",
},
],
related: [
- { label: "Reddit API", href: "/reddit" },
+ { label: "Instagram API", href: "/instagram" },
{ label: "YouTube API", href: "/youtube" },
+ { label: "Reddit API", href: "/reddit" },
{ label: "Google Maps API", href: "/google-maps" },
{ label: "SERP API", href: "/google-search" },
{ label: "Web Crawl API", href: "/web-crawl" },
diff --git a/surfsense_web/lib/format-date.ts b/surfsense_web/lib/format-date.ts
index c2f445537..a062b08fa 100644
--- a/surfsense_web/lib/format-date.ts
+++ b/surfsense_web/lib/format-date.ts
@@ -68,3 +68,17 @@ export function formatRelativeFutureDate(dateString: string): string {
export function formatThreadTimestamp(dateString: string): string {
return format(new Date(dateString), "MMM d, yyyy 'at' h:mm a");
}
+
+/**
+ * Format a chat message timestamp for inline display under a bubble.
+ * Locale-aware, 12-hour clock. Example: "Jul 13, 10:42 PM".
+ */
+export function formatMessageTimestamp(date: Date): string {
+ return date.toLocaleDateString(undefined, {
+ month: "short",
+ day: "numeric",
+ hour: "numeric",
+ minute: "2-digit",
+ hour12: true,
+ });
+}
diff --git a/surfsense_web/lib/playground/code-snippets.selfcheck.ts b/surfsense_web/lib/playground/code-snippets.selfcheck.ts
index 33cb82124..c646d1a6c 100644
--- a/surfsense_web/lib/playground/code-snippets.selfcheck.ts
+++ b/surfsense_web/lib/playground/code-snippets.selfcheck.ts
@@ -23,7 +23,11 @@ assert.deepEqual(payload, {
query: "",
});
-const snippets = buildSnippets("https://api.example.com", "/api/v1/workspaces/1/scrapers/x/y", payload);
+const snippets = buildSnippets(
+ "https://api.example.com",
+ "/api/v1/workspaces/1/scrapers/x/y",
+ payload
+);
// Every popular language is present.
assert.deepEqual(
diff --git a/surfsense_web/lib/playground/format.ts b/surfsense_web/lib/playground/format.ts
index 0e9b75d24..fac6ccef6 100644
--- a/surfsense_web/lib/playground/format.ts
+++ b/surfsense_web/lib/playground/format.ts
@@ -11,9 +11,7 @@ export function formatCost(costMicros: number | null | undefined): string {
/** One meter as a per-1k rate, e.g. 3500 micros/place -> "$3.50 / 1k places". */
export function formatRate(meter: ScraperPricingMeter): string {
const perThousand = (meter.micros_per_unit * 1000) / 1_000_000;
- const dollars = Number.isInteger(perThousand)
- ? perThousand.toString()
- : perThousand.toFixed(2);
+ const dollars = Number.isInteger(perThousand) ? perThousand.toString() : perThousand.toFixed(2);
return `$${dollars} / 1k ${meter.unit}s`;
}
diff --git a/surfsense_web/lib/playground/json-schema.ts b/surfsense_web/lib/playground/json-schema.ts
index 7cbd83be9..0e1e4544e 100644
--- a/surfsense_web/lib/playground/json-schema.ts
+++ b/surfsense_web/lib/playground/json-schema.ts
@@ -8,13 +8,7 @@
type JsonObject = Record;
-export type FieldKind =
- | "string"
- | "string_array"
- | "integer"
- | "number"
- | "boolean"
- | "enum";
+export type FieldKind = "string" | "string_array" | "integer" | "number" | "boolean" | "enum";
export interface FormField {
name: string;
diff --git a/surfsense_web/lib/provider-icons.tsx b/surfsense_web/lib/provider-icons.tsx
index d3e799720..20016fdf2 100644
--- a/surfsense_web/lib/provider-icons.tsx
+++ b/surfsense_web/lib/provider-icons.tsx
@@ -29,6 +29,7 @@ import {
QwenIcon,
RecraftIcon,
ReplicateIcon,
+ RequestyIcon,
SambaNovaIcon,
TogetherAiIcon,
VertexAiIcon,
@@ -117,6 +118,8 @@ export function getProviderIcon(
return ;
case "REPLICATE":
return ;
+ case "REQUESTY":
+ return ;
case "SAMBANOVA":
return ;
case "TOGETHER_AI":
diff --git a/surfsense_web/package.json b/surfsense_web/package.json
index 918b25f33..21894d224 100644
--- a/surfsense_web/package.json
+++ b/surfsense_web/package.json
@@ -1,6 +1,6 @@
{
"name": "surfsense_web",
- "version": "0.0.31",
+ "version": "0.0.32",
"private": true,
"packageManager": "pnpm@10.26.0",
"description": "SurfSense Frontend",