Merge pull request #1601 from CREDO23/feat/resumable-streaming-and-timestamps

[Feat] Chat : Resumable streaming across in-app navigation + optional message timestamps

surfsense_web/app/dashboard/[workspace_id]/user-settings/appearance/page.tsx

@@ -0,0 +1,5 @@
+import { AppearanceContent } from "../components/AppearanceContent";
+
+export default function Page() {
+	return <AppearanceContent />;
+}

surfsense_web/app/dashboard/[workspace_id]/user-settings/components/AppearanceContent.tsx

@@ -0,0 +1,43 @@
+"use client";
+
+import { useAtom } from "jotai";
+import { showMessageTimestampsAtom } from "@/atoms/chat/show-timestamps.atom";
+import { Label } from "@/components/ui/label";
+import { Switch } from "@/components/ui/switch";
+
+export function AppearanceContent() {
+	const [showTimestamps, setShowTimestamps] = useAtom(showMessageTimestampsAtom);
+
+	return (
+		<div className="flex flex-col gap-4 md:gap-6">
+			<section>
+				<div className="pb-2 md:pb-3">
+					<h2 className="text-base md:text-lg font-semibold">Chat</h2>
+					<p className="text-xs md:text-sm text-muted-foreground">
+						Control how messages are displayed in your conversations.
+					</p>
+				</div>
+				<div className="flex flex-col gap-3">
+					<div className="flex items-center justify-between rounded-lg bg-accent p-4">
+						<div className="space-y-0.5">
+							<Label
+								htmlFor="show-timestamps-toggle"
+								className="text-sm font-medium cursor-pointer"
+							>
+								Show message timestamps
+							</Label>
+							<p className="text-xs text-muted-foreground">
+								Display the time under each message in a chat. Saved on this device.
+							</p>
+						</div>
+						<Switch
+							id="show-timestamps-toggle"
+							checked={showTimestamps}
+							onCheckedChange={setShowTimestamps}
+						/>
+					</div>
+				</div>
+			</section>
+		</div>
+	);
+}

surfsense_web/app/dashboard/[workspace_id]/user-settings/layout-shell.tsx

@@ -7,6 +7,7 @@ import {
 	Library,
 	MessageCircle,
 	Monitor,
+	Palette,
 	ReceiptText,
 	ShieldCheck,
 	WandSparkles,
@@ -20,6 +21,7 @@ import { usePlatform } from "@/hooks/use-platform";
 
 export type UserSettingsTab =
 	| "profile"
+	| "appearance"
 	| "api-key"
 	| "prompts"
 	| "community-prompts"
@@ -49,6 +51,12 @@ export function UserSettingsLayoutShell({ workspaceId, children }: UserSettingsL
 				href: `/dashboard/${workspaceId}/user-settings/profile`,
 				icon: <CircleUser className="h-4 w-4" />,
 			},
+			{
+				value: "appearance" as const,
+				label: "Appearance",
+				href: `/dashboard/${workspaceId}/user-settings/appearance`,
+				icon: <Palette className="h-4 w-4" />,
+			},
 			{
 				value: "api-key" as const,
 				label: t("api_key_nav_label"),

surfsense_web/atoms/chat/show-timestamps.atom.ts

@@ -0,0 +1,11 @@
+import { atomWithStorage } from "jotai/utils";
+
+/**
+ * Per-device preference: show a timestamp under each chat message.
+ *
+ * Off by default to match streaming-AI chat convention (ChatGPT/Claude keep
+ * the message stream clean and put time in the conversation list). Persisted
+ * in localStorage, so it does not sync across devices — acceptable for a
+ * cosmetic display toggle.
+ */
+export const showMessageTimestampsAtom = atomWithStorage<boolean>("chat-show-timestamps:v1", false);

surfsense_web/components/assistant-ui/assistant-message.tsx

@@ -35,6 +35,7 @@ import {
 	useAllCitationMetadata,
 } from "@/components/assistant-ui/citation-metadata-context";
 import { MarkdownText } from "@/components/assistant-ui/markdown-text";
+import { MessageTimestamp } from "@/components/assistant-ui/message-timestamp";
 import { ReasoningMessagePart } from "@/components/assistant-ui/reasoning-message-part";
 import { RevertTurnButton } from "@/components/assistant-ui/revert-turn-button";
 import {
@@ -62,6 +63,7 @@ import { withArtifactAnchor } from "@/features/chat-artifacts";
 import { useComments } from "@/hooks/use-comments";
 import { useMediaQuery } from "@/hooks/use-media-query";
 import { useElectronAPI } from "@/hooks/use-platform";
+import { formatMessageTimestamp } from "@/lib/format-date";
 import { getProviderIcon } from "@/lib/provider-icons";
 import { tryGetHostname } from "@/lib/url";
 import { cn } from "@/lib/utils";
@@ -249,16 +251,6 @@ export const MessageError: FC = () => {
 	);
 };
 
-function formatMessageDate(date: Date): string {
-	return date.toLocaleDateString(undefined, {
-		month: "short",
-		day: "numeric",
-		hour: "numeric",
-		minute: "2-digit",
-		hour12: true,
-	});
-}
-
 /**
  * Format provider USD cost (in micro-USD) for inline display next to a
  * token count. Falls back to ``"<$0.001"`` for sub-tenth-of-a-cent
@@ -367,7 +359,7 @@ const MessageInfoDropdown: FC<{ chatTurnId: string | null | undefined }> = ({ ch
 			>
 				{createdAt && (
 					<DropdownMenuLabel className="text-xs text-muted-foreground font-normal select-none">
-						{formatMessageDate(createdAt)}
+						{formatMessageTimestamp(createdAt)}
 					</DropdownMenuLabel>
 				)}
 				{hasUsage && (
@@ -463,6 +455,8 @@ const AssistantMessageInner: FC = () => {
 				<MessageError />
 			</div>
 
+			<MessageTimestamp className="ml-2 mt-2" />
+
 			{isMobile && (
 				<div className="ml-2 mt-2">
 					<MobileCitationDrawer />

surfsense_web/components/assistant-ui/message-timestamp.tsx

@@ -0,0 +1,24 @@
+import { useAuiState } from "@assistant-ui/react";
+import { useAtomValue } from "jotai";
+import type { FC } from "react";
+import { showMessageTimestampsAtom } from "@/atoms/chat/show-timestamps.atom";
+import { formatMessageTimestamp } from "@/lib/format-date";
+import { cn } from "@/lib/utils";
+
+/**
+ * Muted, always-visible timestamp under a chat message. Renders only when the
+ * user has opted in via {@link showMessageTimestampsAtom} and the message
+ * carries a ``createdAt`` (absent on optimistic pre-persist messages).
+ */
+export const MessageTimestamp: FC<{ className?: string }> = ({ className }) => {
+	const show = useAtomValue(showMessageTimestampsAtom);
+	const createdAt = useAuiState(({ message }) => message?.createdAt);
+
+	if (!show || !createdAt) return null;
+
+	return (
+		<div className={cn("select-none text-[11px] text-muted-foreground", className)}>
+			{formatMessageTimestamp(createdAt)}
+		</div>
+	);
+};

surfsense_web/components/assistant-ui/user-message.tsx

@@ -22,6 +22,7 @@ import { currentThreadAtom } from "@/atoms/chat/current-thread.atom";
 import { messageDocumentsMapAtom } from "@/atoms/chat/mentioned-documents.atom";
 import { openEditorPanelAtom } from "@/atoms/editor/editor-panel.atom";
 import { MentionChip } from "@/components/assistant-ui/mention-chip";
+import { MessageTimestamp } from "@/components/assistant-ui/message-timestamp";
 import { TooltipIconButton } from "@/components/assistant-ui/tooltip-icon-button";
 import { getConnectorIcon } from "@/contracts/enums/connectorIcons";
 import { getMentionDocKey } from "@/lib/chat/mention-doc-key";
@@ -182,6 +183,7 @@ export const UserMessage: FC = () => {
 						</div>
 					)}
 				</div>
+				<MessageTimestamp className="mt-1 pl-1" />
 			</div>
 		</MessagePrimitive.Root>
 	);

surfsense_web/components/chat/active-chat-stream-runner.tsx

@@ -0,0 +1,23 @@
+"use client";
+
+import { useEffect } from "react";
+import { chatStreamStore } from "@/lib/chat/stream-engine/store";
+
+/**
+ * Persistent, render-null host that scopes the in-flight chat turn's lifetime
+ * to the workspace shell, not the chat page.
+ *
+ * Mounted in ``LayoutDataProvider``, it survives in-app navigation between
+ * workspace routes and aborts the single active turn only on workspace/app
+ * teardown, so ordinary navigation disconnects the view without stopping the
+ * stream.
+ */
+export function ActiveChatStreamRunner() {
+	useEffect(() => {
+		return () => {
+			chatStreamStore.abortActive();
+		};
+	}, []);
+
+	return null;
+}

surfsense_web/components/layout/providers/LayoutDataProvider.tsx

@@ -18,6 +18,7 @@ import { workspacesAtom } from "@/atoms/workspaces/workspace-query.atoms";
 import { ActionLogDialog } from "@/components/agent-action-log/action-log-dialog";
 import { AnnouncementSpotlight } from "@/components/announcements/AnnouncementSpotlight";
 import { AnnouncementsDialog } from "@/components/announcements/AnnouncementsDialog";
+import { ActiveChatStreamRunner } from "@/components/chat/active-chat-stream-runner";
 import {
 	AlertDialog,
 	AlertDialogAction,
@@ -644,6 +645,9 @@ export function LayoutDataProvider({ workspaceId, children }: LayoutDataProvider
 
 	return (
 		<>
+			{/* Persistent host: keeps an in-flight chat turn streaming across
+			    in-app navigation and aborts it only on workspace teardown. */}
+			<ActiveChatStreamRunner />
 			<LayoutShell
 				workspaces={workspaces}
 				activeWorkspaceId={Number(workspaceId)}

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<void> {
+	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<string, number>,
+	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<string, number>,
+	contentParts: Array<{
+		type: string;
+		toolName?: string;
+		result?: unknown;
+	}>,
+	actionRequests: ReadonlyArray<{ name: string }>
+): Array<string | null> {
+	const claimed = new Set<string>();
+	const paired: Array<string | null> = [];
+	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<string, unknown> | 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<MentionedDocumentInfo>((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 [];
+}

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<number, ThreadStreamState>();
+	private listeners = new Set<Listener>();
+
+	/** 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();

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);
+}

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,
+	});
+}