ui: Services consolidation refactor (#27239)

* ui: Move stream lookup and replay fetches into ChatService

chatStore called fetch() directly for /v1/streams/lookup and the
/v1/stream replay. These now live next to the other stream-session
methods in ChatService, so services stay the only API I/O layer.

* ui: Move /models/sse feed reader into ModelsService

ModelsService.watchModelEvents owns the byte stream, reconnect loop
and SSE record parsing; modelsStore keeps only event routing and
state.

* ui: Extract conversation import/export into ConversationTransferService

The JSONL session format, ZIP archiving and browser downloads are
pure I/O with no store state, so they move out of
conversationsStore. The store keeps the DB orchestration
(bulkExportConversations, downloadConversation,
importConversationsData) and delegates the format work.

* ui: Consolidate active model resolution into modelsStore.activeModelId

The same resolution chain was duplicated in useChatScreenActiveModel,
ChatForm, ChatFormActionModels and contextStatsStore, with slight
drift in the single-model fallback. The canonical getter now lives in
modelsStore, and the shared last-assistant-model lookup moved to
utils as getConversationModel.

* ui: Initialize stores explicitly via initStores()

Store constructors and module-level side effects ran migrations and
localStorage reads in import order. Migrations rename and rewrite
localStorage keys, so a settings load racing ahead of them could
clobber migrated values. initStores() is called once from the root
layout and runs migrations first, then the stores that read
localStorage, then the conversations DB load.

* refactor: Constants for stream query params

tools/ui/src/lib/services/chat.service.ts

@@ -13,6 +13,7 @@ import {
 	SSE_DATA_PREFIX,
 	SSE_DONE_MARKER,
 	SSE_LINE_SEPARATOR,
+	STREAM_QUERY_PARAMS,
 	STREAM_RESUME_LOCALSTORAGE_KEY_PREFIX,
 	STREAM_VISIBILITY_KICK_MS
 } from '$lib/constants';
@@ -33,6 +34,7 @@ import type {
 	ApiStreamSession
 } from '$lib/types/api';
 import { isAbortError } from '$lib/utils/abort';
+import { ApiError } from '$lib/utils/api-fetch';
 import { getAuthHeaders, getJsonHeaders } from '$lib/utils/api-headers';
 import { formatAttachmentText } from '$lib/utils/formatters';
 import { streamIdentity } from '$lib/utils/stream-identity';
@@ -529,7 +531,7 @@ export class ChatService {
 		try {
 			const id = streamIdentity(conversationId, model);
 
-			await fetch(`${API_STREAM.BASE}?conv_id=${encodeURIComponent(id)}`, {
+			await fetch(ChatService.buildStreamUrl(id), {
 				headers: getAuthHeaders(),
 				method: 'DELETE'
 			});
@@ -538,6 +540,46 @@ export class ChatService {
 		}
 	}
 
+	/**
+	 * Look up server-side stream sessions for the given conversation ids. Ids carry the frozen
+	 * conv::model identity when a model was bound at POST time.
+	 */
+	static async lookupStreamSessions(conversationIds: string[]): Promise<ApiStreamSession[]> {
+		const resp = await fetch(API_STREAM.LOOKUP, {
+			body: JSON.stringify({ conversation_ids: conversationIds }),
+			headers: getJsonHeaders(),
+			method: 'POST'
+		});
+
+		if (!resp.ok) {
+			throw new ApiError(`Stream lookup failed with HTTP ${resp.status}`, resp.status);
+		}
+
+		const body = (await resp.json()) as unknown;
+
+		if (!Array.isArray(body)) {
+			throw new Error('Stream lookup returned a non-array response');
+		}
+
+		return body as ApiStreamSession[];
+	}
+
+	/**
+	 * Fetch the full replay of a server-side stream from byte 0. Returns the raw Response so the
+	 * caller can pipe it through the SSE parser like a fresh stream.
+	 */
+	static async fetchStreamReplay(streamId: string): Promise<Response> {
+		const resp = await fetch(ChatService.buildStreamUrl(streamId, 0), {
+			headers: getAuthHeaders()
+		});
+
+		if (!resp.ok) {
+			throw new ApiError(`Stream replay failed with HTTP ${resp.status}`, resp.status);
+		}
+
+		return resp;
+	}
+
 	/**
 	 * Pick the running session to splice into when discoverActiveStream lists candidates for a
 	 * conversation. Finalized sessions are not candidates: their final content was already written
@@ -629,6 +671,15 @@ export class ChatService {
 		return streamIdentity(conversationId, model);
 	}
 
+	// build the replay route url for a stream identity, from is the resume byte offset, omitted
+	// for the cancel route
+	private static buildStreamUrl(streamId: string, from?: number): string {
+		const query = `${STREAM_QUERY_PARAMS.CONV_ID}=${encodeURIComponent(streamId)}`;
+		const offset = from === undefined ? '' : `&${STREAM_QUERY_PARAMS.FROM}=${from}`;
+
+		return `${API_STREAM.BASE}?${query}${offset}`;
+	}
+
 	/**
 	 * Reconnect to an interrupted stream for this conversation. Returns the fetch Response so the
 	 * existing SSE parser drains it like a fresh stream. The server returns 200 on success, 404 if
@@ -642,13 +693,10 @@ export class ChatService {
 		const ac = new AbortController();
 
 		try {
-			const resp = await fetch(
-				`${API_STREAM.BASE}?conv_id=${encodeURIComponent(streamId)}&from=0`,
-				{
-					headers: getAuthHeaders(),
-					signal: ac.signal
-				}
-			);
+			const resp = await fetch(ChatService.buildStreamUrl(streamId, 0), {
+				headers: getAuthHeaders(),
+				signal: ac.signal
+			});
 
 			ac.abort();
 
@@ -668,7 +716,7 @@ export class ChatService {
 		const state = ChatService.getStreamState(conversationId);
 		const from = state?.bytesReceived ?? 0;
 		const id = streamIdentity(conversationId, model);
-		const url = `${API_STREAM.BASE}?conv_id=${encodeURIComponent(id)}&from=${from}`;
+		const url = ChatService.buildStreamUrl(id, from);
 
 		return await fetch(url, { headers: getAuthHeaders(), method: 'GET', signal });
 	}

tools/ui/src/lib/services/conversation-transfer.service.ts

@@ -0,0 +1,279 @@
+/**
+ * ConversationTransferService - Stateless conversation import/export layer
+ *
+ * Owns the session file format (one JSONL record per line: a SESSION header
+ * followed by MESSAGE records), ZIP archiving and browser downloads.
+ * DB access and store refreshes stay in conversationsStore.
+ */
+
+import { EXPORT_CONV, NEWLINE, ZIP_MAGIC } from '$lib/constants';
+import {
+	FileExtensionText,
+	MimeTypeApplication,
+	MimeTypeText,
+	SessionRecordType
+} from '$lib/enums';
+import { strFromU8, strToU8, unzipSync, zipSync } from 'fflate';
+
+export class ConversationTransferService {
+	/**
+	 *
+	 *
+	 * JSONL Session Format
+	 *
+	 *
+	 */
+
+	/**
+	 * Serializes a session (a conversation with its messages) as JSONL.
+	 * The first line is the session header (a `SessionRecordType.SESSION` record
+	 * carrying the conversation properties); each subsequent line is a single message.
+	 * @param data - The exported conversation payload
+	 * @returns The JSONL string (one record per line)
+	 */
+	static serializeSessionToJsonl(data: ExportedConversation): string {
+		const { conv, messages } = data;
+		const sessionLine = JSON.stringify({
+			harness: EXPORT_CONV.HARNESS,
+			type: SessionRecordType.SESSION,
+			...conv
+		});
+		const messageLines = messages.map((message: DatabaseMessage) => {
+			// `toolCalls` is stored as a JSON string; drop it when empty, otherwise parse it.
+			const { toolCalls, ...rest } = message;
+			const normalized = toolCalls ? { ...rest, toolCalls: JSON.parse(toolCalls) } : rest;
+
+			return JSON.stringify({ message: normalized, type: SessionRecordType.MESSAGE });
+		});
+
+		return [sessionLine, ...messageLines].join(NEWLINE);
+	}
+
+	/**
+	 * Parses the JSONL session format produced by {@link serializeSessionToJsonl}.
+	 * A `SessionRecordType.SESSION` line starts a new session; following
+	 * `SessionRecordType.MESSAGE` lines are appended to it. Supports multiple
+	 * sessions in a single file.
+	 * @param text - The JSONL file contents
+	 * @returns The parsed conversations with their messages
+	 */
+	static parseSessionsJsonl(text: string): ExportedConversation[] {
+		const sessions: ExportedConversation[] = [];
+
+		let current: ExportedConversation | null = null;
+
+		for (const line of text.split(NEWLINE)) {
+			const trimmed = line.trim();
+
+			if (!trimmed) continue;
+
+			const record = JSON.parse(trimmed);
+
+			if (record.type === SessionRecordType.SESSION) {
+				// Drop the discriminator and harness marker; the rest is the conversation.
+				const conv = { ...record };
+
+				delete conv.type;
+				delete conv.harness;
+				current = { conv: conv as DatabaseConversation, messages: [] };
+				sessions.push(current);
+			} else if (record.type === SessionRecordType.MESSAGE) {
+				if (!current) {
+					throw new Error('Invalid JSONL: message record before any session record');
+				}
+
+				const message = record.message as DatabaseMessage;
+
+				// `toolCalls` is parsed to an array on export; the DB stores it as a string.
+				if (message.toolCalls !== undefined && typeof message.toolCalls !== 'string') {
+					message.toolCalls = JSON.stringify(message.toolCalls);
+				}
+
+				current.messages.push(message);
+			}
+			// Ignore unknown record types for forward compatibility.
+		}
+
+		return sessions;
+	}
+
+	/**
+	 * Reports whether the text is the JSONL session format, whose first non-empty
+	 * line is a `SessionRecordType.SESSION` record. A legacy JSON export starts
+	 * with an array or an object that has no such discriminator.
+	 * @param text - The file contents
+	 */
+	private static isSessionsJsonl(text: string): boolean {
+		const trimmed = text.trimStart();
+		const lineEnd = trimmed.indexOf(NEWLINE);
+		const firstLine = lineEnd === -1 ? trimmed : trimmed.slice(0, lineEnd);
+
+		try {
+			return JSON.parse(firstLine).type === SessionRecordType.SESSION;
+		} catch {
+			// Not a standalone JSON record, so not the JSONL format.
+			return false;
+		}
+	}
+
+	/**
+	 * Parses an import file into conversations, accepting the current JSONL and
+	 * ZIP formats as well as the legacy JSON format. The format comes from the
+	 * contents, so an import works whatever the file is named.
+	 * @param file - The user-selected file
+	 * @returns The parsed conversations with their messages
+	 */
+	static async parseImportFile(file: File): Promise<ExportedConversation[]> {
+		const bytes = new Uint8Array(await file.arrayBuffer());
+
+		if (ZIP_MAGIC.every((byte, index) => bytes[index] === byte)) {
+			const entries = unzipSync(bytes);
+			const sessions: ExportedConversation[] = [];
+
+			for (const [entryName, entryBytes] of Object.entries(entries)) {
+				if (!entryName.toLowerCase().endsWith(FileExtensionText.JSONL)) continue;
+
+				sessions.push(...ConversationTransferService.parseSessionsJsonl(strFromU8(entryBytes)));
+			}
+
+			return sessions;
+		}
+
+		const text = strFromU8(bytes);
+
+		if (ConversationTransferService.isSessionsJsonl(text)) {
+			return ConversationTransferService.parseSessionsJsonl(text);
+		}
+
+		// Legacy JSON format: an array of conversations or a single conversation object.
+		const parsed = JSON.parse(text);
+
+		if (Array.isArray(parsed)) {
+			return parsed;
+		}
+
+		if (parsed && typeof parsed === 'object' && 'conv' in parsed && 'messages' in parsed) {
+			return [parsed];
+		}
+
+		throw new Error(
+			'Invalid file format: expected array of conversations or single conversation object'
+		);
+	}
+
+	/**
+	 *
+	 *
+	 * Downloads
+	 *
+	 *
+	 */
+
+	/**
+	 * Generates a sanitized filename for a conversation export
+	 * @param conversation - The conversation metadata
+	 * @param msgs - Optional array of messages belonging to the conversation
+	 * @returns The generated filename string
+	 */
+	static generateConversationFilename(
+		conversation: { id?: string; name?: string },
+		msgs?: DatabaseMessage[]
+	): string {
+		const conversationName = (conversation.name ?? '').trim().toLowerCase();
+		const sanitizedName = conversationName
+			.replace(EXPORT_CONV.NON_ALPHANUMERIC_REGEX, EXPORT_CONV.NONALNUM_REPLACEMENT)
+			.replace(EXPORT_CONV.MULTIPLE_UNDERSCORE_REGEX, '_')
+			.substring(0, EXPORT_CONV.NAME_SUFFIX_MAX_LENGTH);
+		// If we have messages, use the timestamp of the newest message
+		const referenceDate = msgs?.length
+			? new Date(Math.max(...msgs.map((m) => m.timestamp)))
+			: new Date();
+		const iso = referenceDate.toISOString().slice(0, EXPORT_CONV.ISO_TIMESTAMP_SLICE);
+		const formattedDate = iso
+			.replace(EXPORT_CONV.ISO_DATE_TIME_SEPARATOR, EXPORT_CONV.ISO_DATE_TIME_SEPARATOR_REPLACEMENT)
+			.replaceAll(EXPORT_CONV.ISO_TIME_SEPARATOR, EXPORT_CONV.ISO_TIME_SEPARATOR_REPLACEMENT);
+		const trimmedConvId = conversation.id?.slice(0, EXPORT_CONV.ID_TRIM_LENGTH) ?? '';
+
+		return `${formattedDate}_conv_${trimmedConvId}_${sanitizedName}${FileExtensionText.JSONL}`;
+	}
+
+	/**
+	 * Triggers a browser download of the provided exported conversation data
+	 * @param data - The exported conversation payload (a single conversation with its messages)
+	 * @param filename - Filename; if omitted, a deterministic name is generated
+	 */
+	static downloadConversationFile(data: ExportedConversation, filename?: string): void {
+		const { conv: conversation, messages: msgs } = data;
+
+		if (!conversation) {
+			console.error('Invalid data: missing conversation');
+
+			return;
+		}
+
+		const downloadFilename =
+			filename ?? ConversationTransferService.generateConversationFilename(conversation, msgs);
+		const jsonl = ConversationTransferService.serializeSessionToJsonl(data);
+		const blob = new Blob([jsonl], { type: MimeTypeText.JSONL });
+
+		ConversationTransferService.triggerDownload(blob, downloadFilename);
+	}
+
+	/**
+	 * Triggers a browser download of multiple conversations as a `.zip`, one
+	 * `.jsonl` file per conversation.
+	 * @param data - The conversations to export
+	 */
+	static downloadConversationsArchive(data: ExportedConversation[]): void {
+		if (data.length === 0) {
+			console.error('Invalid data: no conversations to export');
+
+			return;
+		}
+
+		const usedNames = new Set<string>();
+		const files: Record<string, Uint8Array> = {};
+
+		for (const session of data) {
+			const baseName = ConversationTransferService.generateConversationFilename(
+				session.conv,
+				session.messages
+			);
+
+			// Disambiguate any duplicate filenames within the archive.
+			let entryName = baseName;
+			let suffix = 1;
+
+			while (usedNames.has(entryName)) {
+				entryName = baseName.replace(
+					new RegExp(`${FileExtensionText.JSONL}$`),
+					`_${suffix++}${FileExtensionText.JSONL}`
+				);
+			}
+			usedNames.add(entryName);
+
+			files[entryName] = strToU8(ConversationTransferService.serializeSessionToJsonl(session));
+		}
+
+		const archiveName = `${new Date().toISOString().split(EXPORT_CONV.ISO_DATE_TIME_SEPARATOR)[0]}_conversations${FileExtensionText.ZIP}`;
+		const zipped = zipSync(files);
+		const blob = new Blob([zipped], { type: MimeTypeApplication.ZIP });
+
+		ConversationTransferService.triggerDownload(blob, archiveName);
+	}
+
+	/**
+	 * Triggers a browser download of a blob under the given filename.
+	 */
+	private static triggerDownload(blob: Blob, filename: string): void {
+		const url = URL.createObjectURL(blob);
+		const a = document.createElement('a');
+
+		a.href = url;
+		a.download = filename;
+		document.body.appendChild(a);
+		a.click();
+		document.body.removeChild(a);
+		URL.revokeObjectURL(url);
+	}
+}

tools/ui/src/lib/services/index.ts

@@ -103,6 +103,15 @@ export { ChatService } from './chat.service';
  */
 export { DatabaseService } from './database.service';
 
+/**
+ * **ConversationTransferService** - Conversation import/export format layer
+ *
+ * Owns the JSONL session format (SESSION header + MESSAGE records), ZIP
+ * archiving and browser downloads. Stateless; DB access and store refreshes
+ * stay in conversationsStore.
+ */
+export { ConversationTransferService } from './conversation-transfer.service';
+
 /**
  * **ModelsService** - Model management API communication
  *

tools/ui/src/lib/services/models.service.ts

@@ -1,7 +1,15 @@
-import { API_MODELS, MODEL_ID } from '$lib/constants';
+import { base } from '$app/paths';
+import {
+	API_MODELS,
+	MODEL_ID,
+	SSE_DATA_PREFIX,
+	SSE_LINE_SEPARATOR,
+	SSE_RECORD_SEPARATOR
+} from '$lib/constants';
 import { ServerModelStatus } from '$lib/enums';
 import type { ParsedModelId } from '$lib/types/models';
 import { apiFetch, apiPost, normalizeModelName } from '$lib/utils';
+import { getAuthHeaders } from '$lib/utils/api-headers';
 
 export class ModelsService {
 	/**
@@ -100,6 +108,89 @@ export class ModelsService {
 		return model.status.value === ServerModelStatus.LOADING;
 	}
 
+	/**
+	 *
+	 *
+	 * Status Feed
+	 *
+	 *
+	 */
+
+	private static readonly SSE_RECONNECT_MS = 1000;
+
+	/**
+	 * Read the /models/sse feed and invoke onEvent for each parsed envelope.
+	 * Reconnects on network drops until the signal aborts. Splits the byte
+	 * stream into SSE records on the blank line boundary; the payload rides in
+	 * the data lines as a JSON envelope with its own model, event and data fields.
+	 */
+	static async watchModelEvents(
+		signal: AbortSignal,
+		onEvent: (event: ApiModelsSseEvent) => void
+	): Promise<void> {
+		const decoder = new TextDecoder();
+
+		while (!signal.aborted) {
+			try {
+				const response = await fetch(`${base}${API_MODELS.SSE}`, {
+					headers: getAuthHeaders(),
+					signal
+				});
+
+				if (response.ok && response.body) {
+					const reader = response.body.getReader();
+
+					let buffer = '';
+
+					while (!signal.aborted) {
+						const { done, value } = await reader.read();
+
+						if (done) break;
+
+						buffer += decoder.decode(value, { stream: true });
+
+						let boundary = buffer.indexOf(SSE_RECORD_SEPARATOR);
+
+						while (boundary !== -1) {
+							const event = ModelsService.parseStatusRecord(buffer.slice(0, boundary));
+
+							if (event) onEvent(event);
+
+							buffer = buffer.slice(boundary + SSE_RECORD_SEPARATOR.length);
+							boundary = buffer.indexOf(SSE_RECORD_SEPARATOR);
+						}
+					}
+				}
+			} catch {
+				// network drop or abort falls through to the reconnect delay
+			}
+
+			if (signal.aborted) return;
+
+			await new Promise((resolve) => setTimeout(resolve, ModelsService.SSE_RECONNECT_MS));
+		}
+	}
+
+	/**
+	 * Parse one SSE record into its JSON envelope, or null when the record
+	 * carries no data payload or malformed JSON.
+	 */
+	private static parseStatusRecord(record: string): ApiModelsSseEvent | null {
+		const payload = record
+			.split(SSE_LINE_SEPARATOR)
+			.filter((line) => line.startsWith(SSE_DATA_PREFIX))
+			.map((line) => line.slice(SSE_DATA_PREFIX.length).trim())
+			.join(SSE_LINE_SEPARATOR);
+
+		if (payload.length === 0) return null;
+
+		try {
+			return JSON.parse(payload) as ApiModelsSseEvent;
+		} catch {
+			return null;
+		}
+	}
+
 	/**
 	 *
 	 *