char-fork/llama.cpp

git clone https://git.t4t.associates/char-fork/llama.cpp

Aleksander Grygierui: Stores split refactor (#27240)521a64cd02026-08-20

/**
 * ChatService - Stateless chat completion and streaming API layer
 *
 * Wraps the /chat/completions and /stream endpoints: request building, SSE
 * parsing, streaming callbacks, resume/probe logic and pre-encode KV-cache
 * warming. No reactive state; consumed by chatStore and its managers.
 */

import { getAudioInputFormat } from '../utils/audio-format';
import { capImageDataURLSize } from '../utils/cap-img-size';
import {
	API_CHAT,
	API_SLOTS,
	API_STREAM,
	CONTROL_ACTION,
	HEADERS,
	LEGACY_AGENTIC_REGEX,
	REASONING_EFFORT_TOKENS,
	SETTINGS_KEYS,
	SSE_DATA_PREFIX,
	SSE_DONE_MARKER,
	SSE_LINE_SEPARATOR,
	STREAM_QUERY_PARAMS,
	STREAM_RESUME_LOCALSTORAGE_KEY_PREFIX,
	STREAM_VISIBILITY_KICK_MS
} from '$lib/constants';
import {
	AttachmentLabel,
	AttachmentType,
	ContentPartType,
	MessageRole,
	ReasoningFormat,
	StreamConnectionState
} from '$lib/enums';
import { modelsStore } from '$lib/stores/models/index.svelte';
import { settingsStore } from '$lib/stores/settings/index.svelte';
import type { DatabaseMessageExtraMcpPrompt, DatabaseMessageExtraMcpResource } from '$lib/types';
import type {
	ApiChatCompletionToolCall,
	ApiChatMessageContentPart,
	ApiChatMessageData,
	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';

interface ResumableStreamState {
	bytesReceived: number;
	updatedAt: number;

	// model frozen at POST time, lets a reload rebuild the exact conv::model identity the
	// server keyed the session under. null when the POST carried no explicit model
	model?: string | null;
}

function streamStorageKey(conversationId: string): string {
	return STREAM_RESUME_LOCALSTORAGE_KEY_PREFIX + conversationId;
}

export class ChatService {
	// Per-chunk localStorage writes are throttled to at most one per
	// conversation per interval (saveStreamStateThrottled). The resume offset
	// only needs to be roughly current: on resume the server retransmits from
	// a line boundary and the client discards its partial line. Guaranteed
	// immediate writes happen at stream start, at resume boundaries and when
	// the page goes hidden or away (pagehide/visibilitychange), so a reload
	// always finds a usable offset.
	private static readonly STREAM_STATE_SAVE_INTERVAL_MS = 500;

	private static streamStateSaveTrackers = new Map<
		string,
		{ lastSavedAt: number; model: string | null; pendingBytes: number | null }
	>();

	/**
	 * Checks whether all server slots are currently idle (not processing any requests).
	 * Queries the /slots endpoint (requires --slots flag on the server).
	 * Returns true if all slots are idle, false if any is processing.
	 * If the endpoint is unavailable or errors out, returns true (best-effort fallback).
	 *
	 * @param signal - Optional AbortSignal to cancel the request if needed
	 * @param model - Optional model name to check slots for (required in ROUTER mode)
	 * @returns {Promise<boolean>} Promise that resolves to true if all slots are idle, false if any is processing
	 */
	static async areAllSlotsIdle(model?: string | null, signal?: AbortSignal): Promise<boolean> {
		try {
			const url = model ? `${API_SLOTS.LIST}?model=${encodeURIComponent(model)}` : API_SLOTS.LIST;
			const res = await fetch(url, { signal });

			if (!res.ok) return true;

			const slots: { is_processing: boolean }[] = await res.json();

			return slots.every((s) => !s.is_processing);
		} catch {
			return true;
		}
	}

	/**
	 * Cancels the server-side replay buffer for a conversation, freeing its slot.
	 */
	static async cancelServerStream(conversationId: string, model?: string | null): Promise<void> {
		if (!conversationId) return;

		try {
			const id = streamIdentity(conversationId, model);

			await fetch(ChatService.buildStreamUrl(id), {
				headers: getAuthHeaders(),
				method: 'DELETE'
			});
		} catch (e) {
			console.warn('cancelServerStream failed:', e);
		}
	}

	static clearStreamState(conversationId: string): void {
		if (!conversationId) return;

		ChatService.streamStateSaveTrackers.delete(conversationId);

		try {
			localStorage.removeItem(streamStorageKey(conversationId));
		} catch {
			// nothing to do
		}
	}

	/**
	 * Converts a database message with attachments to API chat message format.
	 * Processes various attachment types (images, text files, PDFs) and formats them
	 * as content parts suitable for the chat completion API.
	 */
	static async convertDbMessageToApiChatMessageData(
		message: DatabaseMessage & { extra?: DatabaseMessageExtra[] }
	): Promise<ApiChatMessageData> {
		// Handle tool result messages (role: 'tool')
		if (message.role === MessageRole.TOOL && message.toolCallId) {
			return {
				content: message.content,
				role: MessageRole.TOOL,
				tool_call_id: message.toolCallId
			};
		}

		// Parse tool calls for assistant messages
		let toolCalls: ApiChatCompletionToolCall[] | undefined;

		if (message.toolCalls) {
			try {
				toolCalls = JSON.parse(message.toolCalls);
			} catch {
				// Ignore parse errors for malformed tool calls
			}
		}

		if (!message.extra || message.extra.length === 0) {
			const result: ApiChatMessageData = {
				content: message.content,
				role: message.role as MessageRole
			};

			if (message.reasoningContent) {
				result.reasoning_content = message.reasoningContent;
			}

			if (toolCalls && toolCalls.length > 0) {
				result.tool_calls = toolCalls;
			}

			return result;
		}

		const contentParts: ApiChatMessageContentPart[] = [];
		const textFiles = message.extra.filter(
			(extra: DatabaseMessageExtra): extra is DatabaseMessageExtraTextFile =>
				extra.type === AttachmentType.TEXT
		);

		for (const textFile of textFiles) {
			contentParts.push({
				text: formatAttachmentText(AttachmentLabel.FILE, textFile.name, textFile.content),
				type: ContentPartType.TEXT
			});
		}

		// Handle legacy 'context' type from the old UI (pasted content)
		const legacyContextFiles = message.extra.filter(
			(extra: DatabaseMessageExtra): extra is DatabaseMessageExtraLegacyContext =>
				extra.type === AttachmentType.LEGACY_CONTEXT
		);

		for (const legacyContextFile of legacyContextFiles) {
			contentParts.push({
				text: formatAttachmentText(
					AttachmentLabel.FILE,
					legacyContextFile.name,
					legacyContextFile.content
				),
				type: ContentPartType.TEXT
			});
		}

		const imageFiles = message.extra.filter(
			(extra: DatabaseMessageExtra): extra is DatabaseMessageExtraImageFile =>
				extra.type === AttachmentType.IMAGE
		);

		for (const image of imageFiles) {
			const maxImageResolution = settingsStore.getConfig(SETTINGS_KEYS.MAX_IMAGE_RESOLUTION);
			// Caps the resolution and bakes the jpeg exif orientation in one pass,
			// untouched images pass through as is
			const base64Url = await capImageDataURLSize(image.base64Url, maxImageResolution);

			contentParts.push({
				image_url: { url: base64Url },
				type: ContentPartType.IMAGE_URL
			});
		}

		const audioFiles = message.extra.filter(
			(extra: DatabaseMessageExtra): extra is DatabaseMessageExtraAudioFile =>
				extra.type === AttachmentType.AUDIO
		);

		for (const audio of audioFiles) {
			contentParts.push({
				input_audio: {
					data: audio.base64Data,
					format: getAudioInputFormat(audio.mimeType)
				},
				type: ContentPartType.INPUT_AUDIO
			});
		}

		if (message.content) {
			contentParts.push({
				text: message.content,
				type: ContentPartType.TEXT
			});
		}

		const videoFiles = message.extra.filter(
			(extra: DatabaseMessageExtra): extra is DatabaseMessageExtraVideoFile =>
				extra.type === AttachmentType.VIDEO
		);

		for (const video of videoFiles) {
			contentParts.push({
				input_video: {
					data: video.base64Data,
					format: video.mimeType.includes('mp4')
						? 'mp4'
						: video.mimeType.includes('ogg')
							? 'ogg'
							: 'auto'
				},
				type: ContentPartType.INPUT_VIDEO
			});
		}

		const pdfFiles = message.extra.filter(
			(extra: DatabaseMessageExtra): extra is DatabaseMessageExtraPdfFile =>
				extra.type === AttachmentType.PDF
		);

		for (const pdfFile of pdfFiles) {
			if (pdfFile.processedAsImages && pdfFile.images) {
				for (let i = 0; i < pdfFile.images.length; i++) {
					contentParts.push({
						image_url: { url: pdfFile.images[i] },
						type: ContentPartType.IMAGE_URL
					});
				}
			} else {
				contentParts.push({
					text: formatAttachmentText(AttachmentLabel.PDF_FILE, pdfFile.name, pdfFile.content),
					type: ContentPartType.TEXT
				});
			}
		}

		const mcpPrompts = message.extra.filter(
			(extra: DatabaseMessageExtra): extra is DatabaseMessageExtraMcpPrompt =>
				extra.type === AttachmentType.MCP_PROMPT
		);

		for (const mcpPrompt of mcpPrompts) {
			contentParts.push({
				text: formatAttachmentText(
					AttachmentLabel.MCP_PROMPT,
					mcpPrompt.name,
					mcpPrompt.content,
					mcpPrompt.serverName
				),
				type: ContentPartType.TEXT
			});
		}

		const mcpResources = message.extra.filter(
			(extra: DatabaseMessageExtra): extra is DatabaseMessageExtraMcpResource =>
				extra.type === AttachmentType.MCP_RESOURCE
		);

		for (const mcpResource of mcpResources) {
			contentParts.push({
				text: formatAttachmentText(
					AttachmentLabel.MCP_RESOURCE,
					mcpResource.name,
					mcpResource.content,
					mcpResource.serverName
				),
				type: ContentPartType.TEXT
			});
		}

		const result: ApiChatMessageData = {
			content: contentParts,
			role: message.role as MessageRole
		};

		if (message.reasoningContent) {
			result.reasoning_content = message.reasoningContent;
		}

		if (toolCalls && toolCalls.length > 0) {
			result.tool_calls = toolCalls;
		}

		return result;
	}

	/**
	 * 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;
	}

	// write a throttled-but-not-yet-persisted offset immediately; used at
	// resume boundaries and on pagehide/visibilitychange so the persisted
	// offset is the freshest one when it matters
	static flushStreamState(conversationId: string): void {
		const tracker = ChatService.streamStateSaveTrackers.get(conversationId);

		if (!tracker || tracker.pendingBytes === null) return;

		const { model, pendingBytes } = tracker;

		tracker.lastSavedAt = Date.now();
		tracker.pendingBytes = null;

		ChatService.writeStreamState(conversationId, pendingBytes, model);
	}

	/**
	 * Sends a streaming chat completion request for generating a chat title.
	 * Delegates to `sendMessage` for fetch, SSE parsing, and error handling.
	 *
	 * @param message - The single message to send (a user message containing the title generation prompt)
	 * @param model - Optional model name to use (required in ROUTER mode)
	 * @param signal - Optional AbortSignal to cancel the request
	 * @returns {Promise<string>} The aggregated title text, or empty string if request failed
	 * @static
	 */
	static async generateTitle(
		message: ApiChatMessageData,
		model?: string | null,
		signal?: AbortSignal
	): Promise<string> {
		let titleResponse = '';

		try {
			await ChatService.sendMessage(
				[message],
				{
					custom: { chat_template_kwargs: { enable_thinking: false } },
					model: model || undefined,
					onChunk: (chunk: string) => {
						titleResponse += chunk;
					},
					stream: true
				},
				undefined,
				signal
			);
		} catch {
			return '';
		}

		return titleResponse;
	}

	static getStreamState(conversationId: string): ResumableStreamState | null {
		if (!conversationId) return null;

		try {
			const raw = localStorage.getItem(streamStorageKey(conversationId));

			if (!raw) return null;

			const parsed = JSON.parse(raw) as ResumableStreamState;

			if (!parsed || typeof parsed.bytesReceived !== 'number') return null;

			return parsed;
		} catch {
			return null;
		}
	}

	/**
	 * Handles streaming response from the chat completion API.
	 */
	static async handleStreamResponse(
		response: Response,
		onChunk?: (chunk: string) => void,
		onComplete?: (
			response: string,
			reasoningContent?: string,
			timings?: ChatMessageTimings,
			toolCalls?: string
		) => void,
		onError?: (error: Error) => void,
		onReasoningChunk?: (chunk: string) => void,
		onToolCallChunk?: (chunk: string) => void,
		onModel?: (model: string) => void,
		onCompletionId?: (id: string) => void,
		onTimings?: (timings?: ChatMessageTimings, promptProgress?: ChatMessagePromptProgress) => void,
		conversationId?: string,
		abortSignal?: AbortSignal,
		onConnectionState?: (state: StreamConnectionState) => void,
		streamModel?: string | null
	): Promise<void> {
		let reader = response.body?.getReader();

		if (!reader) {
			throw new Error('No response body');
		}

		// bytesParsed is the absolute server side buffer offset of the next byte to parse
		// segmentStartOffset is the absolute offset where the current reader started, reset on resume
		// segmentBytesRead is wire bytes read by the current reader
		let bytesParsed = 0;
		let segmentStartOffset = 0;
		let segmentBytesRead = 0;
		let lastByteAt = Date.now();
		// each resume must produce at least one byte to be retried again
		// if a resume returns 200 but yields nothing, we abandon
		// since the session has a bounded size, the total number of retries is bounded by construction
		let madeProgress = true;

		const encoder = new TextEncoder();

		if (conversationId) {
			ChatService.saveStreamState(conversationId, 0, streamModel);
		}

		onConnectionState?.(StreamConnectionState.STREAMING);

		let decoder = new TextDecoder();
		let aggregatedContent = '';
		let fullReasoningContent = '';
		let aggregatedToolCalls: ApiChatCompletionToolCall[] = [];
		let lastTimings: ChatMessageTimings | undefined;
		let streamFinished = false;
		let modelEmitted = false;
		let idEmitted = false;
		let toolCallIndexOffset = 0;
		let hasOpenToolCallBatch = false;

		const finalizeOpenToolCallBatch = () => {
			if (!hasOpenToolCallBatch) {
				return;
			}

			toolCallIndexOffset = aggregatedToolCalls.length;
			hasOpenToolCallBatch = false;
		};
		const processToolCallDelta = (toolCalls?: ApiChatCompletionToolCallDelta[]) => {
			if (!toolCalls || toolCalls.length === 0) {
				return;
			}

			aggregatedToolCalls = ChatService.mergeToolCallDeltas(
				aggregatedToolCalls,
				toolCalls,
				toolCallIndexOffset
			);

			if (aggregatedToolCalls.length === 0) {
				return;
			}

			hasOpenToolCallBatch = true;

			const serializedToolCalls = JSON.stringify(aggregatedToolCalls);

			if (import.meta.env.DEV && import.meta.env.VITE_DEBUG) {
				console.log('[ChatService] Aggregated tool calls:', serializedToolCalls);
			}

			if (!serializedToolCalls) {
				return;
			}

			if (!abortSignal?.aborted) {
				onToolCallChunk?.(serializedToolCalls);
			}
		};
		const onVisibilityChange = () => {
			if (typeof document === 'undefined') return;

			if (document.visibilityState === 'hidden') {
				// the tab is going to the background and the OS may throttle or
				// drop the socket shortly; persist the freshest resume offset now
				if (conversationId) ChatService.flushStreamState(conversationId);

				return;
			}

			if (streamFinished) return;

			if (!conversationId) return;

			// the bytes have been quiet for too long, the OS likely killed the socket
			// kicking the reader unblocks reader.read with done=true so the outer loop can resume
			if (Date.now() - lastByteAt > STREAM_VISIBILITY_KICK_MS) {
				reader!.cancel().catch(() => {});
			}
		};
		const onPageHide = () => {
			// a reload or navigation is about to happen; make sure the resume
			// offset that getStreamState() will read is not a stale throttled one
			if (conversationId) ChatService.flushStreamState(conversationId);
		};

		if (typeof document !== 'undefined') {
			document.addEventListener('visibilitychange', onVisibilityChange);
			window.addEventListener('pagehide', onPageHide);
		}

		try {
			let chunk = '';

			// outer loop drives the resume cycle, swaps reader on premature end of stream
			while (true) {
				while (true) {
					if (abortSignal?.aborted) break;

					let done: boolean;
					let value: Uint8Array | undefined;

					try {
						const r = await reader.read();

						done = r.done;
						value = r.value;
					} catch (readErr) {
						// reader.read() rejects with TypeError when the underlying connection drops
						// instead of just resolving with done=true. treat it like done so the outer
						// loop swaps reader via the resume path
						if (isAbortError(readErr)) {
							throw readErr;
						}

						console.warn('reader.read() rejected, treating as premature end:', readErr);
						done = true;
						value = undefined;
					}

					if (done) break;

					if (abortSignal?.aborted) break;

					if (value && value.byteLength > 0) {
						segmentBytesRead += value.byteLength;
						lastByteAt = Date.now();

						if (!madeProgress) {
							madeProgress = true;
							onConnectionState?.(StreamConnectionState.STREAMING);
						}
					}

					chunk += decoder.decode(value, { stream: true });
					const lines = chunk.split(SSE_LINE_SEPARATOR);

					chunk = lines.pop() || '';

					// the persisted offset must point right after the last fully parsed line,
					// the trailing `chunk` is partial bytes still waiting for a newline
					if (conversationId) {
						const tailBytes = encoder.encode(chunk).byteLength;

						bytesParsed = segmentStartOffset + segmentBytesRead - tailBytes;
						ChatService.saveStreamStateThrottled(conversationId, bytesParsed, streamModel);
					}

					for (const line of lines) {
						if (abortSignal?.aborted) break;

						if (line.startsWith(SSE_DATA_PREFIX)) {
							const data = line.slice(SSE_DATA_PREFIX.length).trim();

							if (data === SSE_DONE_MARKER) {
								streamFinished = true;

								continue;
							}

							try {
								const parsed: ApiChatCompletionStreamChunk = JSON.parse(data);
								const choice = parsed.choices?.[0];
								const content = choice?.delta?.content;
								const reasoningContent = choice?.delta?.reasoning_content;
								const toolCalls = choice?.delta?.tool_calls;
								const timings = parsed.timings;
								const promptProgress = parsed.prompt_progress;
								const chunkModel = ChatService.extractModelName(parsed);

								if (chunkModel && !modelEmitted) {
									modelEmitted = true;
									onModel?.(chunkModel);
								}

								if (parsed.id && !idEmitted) {
									idEmitted = true;
									onCompletionId?.(parsed.id);
								}

								if (promptProgress) {
									ChatService.notifyTimings(undefined, promptProgress, onTimings);
								}

								if (timings) {
									ChatService.notifyTimings(timings, promptProgress, onTimings);
									lastTimings = timings;
								}

								if (content) {
									finalizeOpenToolCallBatch();
									aggregatedContent += content;

									if (!abortSignal?.aborted) {
										onChunk?.(content);
									}
								}

								if (reasoningContent) {
									finalizeOpenToolCallBatch();
									fullReasoningContent += reasoningContent;

									if (!abortSignal?.aborted) {
										onReasoningChunk?.(reasoningContent);
									}
								}

								processToolCallDelta(toolCalls);
							} catch (e) {
								console.error('Error parsing JSON chunk:', e);
							}
						}
					}

					if (abortSignal?.aborted) break;

					if (streamFinished) break;
				}

				// inner reader done, decide whether to try a resume
				if (abortSignal?.aborted) break;

				if (streamFinished) break;

				if (!conversationId) break;

				if (!madeProgress) {
					onConnectionState?.(StreamConnectionState.LOST);
					onError?.(new Error('Stream resume produced no new bytes, giving up'));

					break;
				}

				onConnectionState?.(StreamConnectionState.RESUMING);
				madeProgress = false;

				// the server resends starting at bytesParsed, discard any partial line we held, it
				// will be retransmitted from a clean line boundary. reuse the frozen model, not the
				// live dropdown
				// resumeStream reads the offset from localStorage, so persist the
				// freshest bytesParsed before asking the server to replay from it
				ChatService.flushStreamState(conversationId);
				const resumeResp = await ChatService.resumeStream(
					conversationId,
					abortSignal,
					streamModel
				).catch(() => null);

				// an abort landing during the resume request is intentional, not a lost connection
				if (abortSignal?.aborted) break;

				if (!resumeResp || resumeResp.status !== 200) {
					onConnectionState?.(StreamConnectionState.LOST);
					onError?.(new Error('Stream connection lost and could not be resumed'));

					break;
				}

				const newReader = resumeResp.body?.getReader();

				if (!newReader) break;

				try {
					reader.releaseLock();
				} catch {
					/* ignore */
				}
				reader = newReader;
				decoder = new TextDecoder();
				chunk = '';
				segmentStartOffset = bytesParsed;
				segmentBytesRead = 0;
				lastByteAt = Date.now();
			}

			if (abortSignal?.aborted) return;

			if (streamFinished) {
				finalizeOpenToolCallBatch();

				if (conversationId) {
					ChatService.clearStreamState(conversationId);
				}

				const finalToolCalls =
					aggregatedToolCalls.length > 0 ? JSON.stringify(aggregatedToolCalls) : undefined;

				onComplete?.(
					aggregatedContent,
					fullReasoningContent || undefined,
					lastTimings,
					finalToolCalls
				);
			}
		} catch (error) {
			const err = error instanceof Error ? error : new Error('Stream error');

			onError?.(err);

			throw err;
		} finally {
			if (typeof document !== 'undefined') {
				document.removeEventListener('visibilitychange', onVisibilityChange);
				window.removeEventListener('pagehide', onPageHide);
			}

			try {
				reader.releaseLock();
			} catch {
				/* ignore */
			}
		}
	}

	/**
	 * 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[];
	}

	/**
	 * Normalizes an array of messages (database or already-API-shaped) into
	 * API chat message data, converting DB messages and dropping empty system
	 * messages. Shared by sendMessage, preEncode and the agentic flow.
	 */
	static async normalizeMessagesForApi(
		messages: ApiChatMessageData[] | (DatabaseMessage & { extra?: DatabaseMessageExtra[] })[]
	): Promise<ApiChatMessageData[]> {
		return (
			await Promise.all(
				messages.map((msg) => {
					if ('id' in msg && 'convId' in msg && 'timestamp' in msg) {
						return ChatService.convertDbMessageToApiChatMessageData(
							msg as DatabaseMessage & { extra?: DatabaseMessageExtra[] }
						);
					}

					return msg as ApiChatMessageData;
				})
			)
		).filter((msg: { role: ChatRole; content: string | ApiChatMessageContentPart[] }) => {
			// Filter out empty system messages
			if (msg.role === MessageRole.SYSTEM) {
				const content = typeof msg.content === 'string' ? msg.content : '';

				return content.trim().length > 0;
			}

			return true;
		});
	}

	/**
	 * Fire-and-forget request to pre-encode the conversation in the server's KV cache.
	 * Re-submits the full conversation with n_predict=0 so the server processes the prompt
	 * without generating tokens, warming the cache for the next turn.
	 */
	static async preEncode(
		messages: ApiChatMessageData[] | (DatabaseMessage & { extra?: DatabaseMessageExtra[] })[],
		model?: string | null,
		excludeReasoning?: boolean,
		signal?: AbortSignal
	): Promise<void> {
		const normalizedMessages: ApiChatMessageData[] =
			await ChatService.normalizeMessagesForApi(messages);
		const requestBody: Record<string, unknown> = {
			messages: normalizedMessages.map((msg: ApiChatMessageData) => {
				const mapped: Record<string, unknown> = {
					content: excludeReasoning ? ChatService.stripReasoningContent(msg.content) : msg.content,
					role: msg.role,
					tool_call_id: msg.tool_call_id,
					tool_calls: msg.tool_calls
				};

				if (!excludeReasoning && msg.reasoning_content) {
					mapped.reasoning_content = msg.reasoning_content;
				}

				return mapped;
			}),
			n_predict: 0,
			stream: false
		};

		if (model) {
			requestBody.model = model;
		}

		try {
			await fetch(API_CHAT.COMPLETIONS, {
				body: JSON.stringify(requestBody),
				headers: getJsonHeaders(),
				method: 'POST',
				signal
			});
		} catch (error) {
			if (!isAbortError(error)) {
				console.warn('[ChatService] Pre-encode request failed:', error);
			}
		}
	}

	// probe the resume route status without consuming the stream: the SSE route has no HEAD,
	// so issue the GET and abort it right after the status line. 0 on network error
	static async probeResumeStatus(streamId: string): Promise<number> {
		if (!streamId) return 0;

		const ac = new AbortController();

		try {
			const resp = await fetch(ChatService.buildStreamUrl(streamId, 0), {
				headers: getAuthHeaders(),
				signal: ac.signal
			});

			ac.abort();

			return resp.status;
		} catch {
			return 0;
		}
	}

	static async resumeStream(
		conversationId: string,
		signal?: AbortSignal,
		model?: string | null
	): Promise<Response | null> {
		if (!conversationId) return null;

		const state = ChatService.getStreamState(conversationId);
		const from = state?.bytesReceived ?? 0;
		const id = streamIdentity(conversationId, model);
		const url = ChatService.buildStreamUrl(id, from);

		return await fetch(url, { headers: getAuthHeaders(), method: 'GET', signal });
	}

	/**
	 * Rebuild the stream identity for a resume. The model persisted at POST time wins, including a
	 * stored null which means the POST carried no explicit model so the identity stays the bare conv
	 * id. Only fall back to the caller supplied current model when nothing was persisted.
	 */
	static resumeStreamIdentity(
		conversationId: string,
		state: ResumableStreamState | null,
		fallbackModel: string | null
	): string {
		const model = state && state.model !== undefined ? state.model : fallbackModel;

		return streamIdentity(conversationId, model);
	}

	// persist the running byte count and the frozen model for a conversation, a later visit
	// resumes the SSE replay at the right offset under the same conv::model
	// identity. Writes immediately; the per-chunk read loop uses the throttled
	// variant instead.
	static saveStreamState(
		conversationId: string,
		bytesReceived: number,
		model?: string | null
	): void {
		if (!conversationId) return;

		ChatService.writeStreamState(conversationId, bytesReceived, model);
		// record the write so a throttled save landing inside the interval
		// holds its value pending instead of re-writing
		ChatService.streamStateSaveTrackers.set(conversationId, {
			lastSavedAt: Date.now(),
			model: model ?? null,
			pendingBytes: null
		});
	}

	// throttled variant for the per-chunk read loop: writes at most once per
	// conversation per STREAM_STATE_SAVE_INTERVAL_MS, holding the latest value
	// pending until the interval elapses or flushStreamState() forces it out
	static saveStreamStateThrottled(
		conversationId: string,
		bytesReceived: number,
		model?: string | null
	): void {
		if (!conversationId) return;

		const tracker = ChatService.streamStateSaveTrackers.get(conversationId) ?? {
			lastSavedAt: 0,
			model: null,
			pendingBytes: null
		};

		tracker.model = model ?? null;

		if (Date.now() - tracker.lastSavedAt >= ChatService.STREAM_STATE_SAVE_INTERVAL_MS) {
			tracker.lastSavedAt = Date.now();
			tracker.pendingBytes = null;
			ChatService.writeStreamState(conversationId, bytesReceived, model);
		} else {
			tracker.pendingBytes = bytesReceived;
		}

		ChatService.streamStateSaveTrackers.set(conversationId, tracker);
	}

	/**
	 * 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
	 * to the DB by the original onComplete handler, so attaching to them would replay a buffer that
	 * may not match what the DB holds. A continue session's buffer holds only the appended deltas,
	 * not the pre continue prefix, so replaying it as a fresh generation would erase the original.
	 *
	 * Among running sessions we tie break on the most recent started_at, which covers the case of
	 * multiple inferences left running on the same conversation.
	 */
	static selectActiveStream(
		sessions: ApiStreamSession[] | null | undefined
	): ApiStreamSession | null {
		if (!Array.isArray(sessions) || sessions.length === 0) {
			return null;
		}

		const running = sessions.filter((s) => !s.is_done);

		if (running.length === 0) {
			return null;
		}

		return running.reduce((best, cur) => (cur.started_at > best.started_at ? cur : best));
	}

	/**
	 * Sends a chat completion request to the llama-server.
	 * Supports both streaming and non-streaming responses with comprehensive parameter configuration.
	 * Automatically converts database messages with attachments to the appropriate API format.
	 *
	 * @param messages - Array of chat messages to send to the API (supports both ApiChatMessageData and DatabaseMessage with attachments)
	 * @param options - Configuration options for the chat completion request. See `SettingsChatServiceOptions` type for details.
	 * @returns {Promise<string | void>} that resolves to the complete response string (non-streaming) or void (streaming)
	 * @throws {Error} if the request fails or is aborted
	 */
	static async sendMessage(
		messages: ApiChatMessageData[] | (DatabaseMessage & { extra?: DatabaseMessageExtra[] })[],
		options: SettingsChatServiceOptions = {},
		conversationId?: string,
		signal?: AbortSignal
	): Promise<string | void> {
		const {
			backend_sampling,
			continueFinalMessage,
			custom,
			// Config options
			disableReasoningParsing,
			dry_allowed_length,
			dry_base,
			dry_multiplier,
			dry_penalty_last_n,
			dynatemp_exponent,
			// Sampling parameters
			dynatemp_range,
			enableThinking,
			excludeReasoningFromContext,
			frequency_penalty,
			max_tokens,
			min_p,
			onChunk,
			onComplete,
			onCompletionId,
			onConnectionState,
			onError,
			onModel,
			onReasoningChunk,
			onTimings,
			onToolCallChunk,
			presence_penalty,
			reasoningEffort,
			// Penalty parameters
			repeat_last_n,
			repeat_penalty,
			// Other parameters
			samplers,
			stream,
			// Generation parameters
			temperature,
			timings_per_token,
			// Tools for function calling
			tools,
			top_k,
			top_p,
			typ_p,
			xtc_probability,
			xtc_threshold
		} = options;
		const normalizedMessages: ApiChatMessageData[] =
			await ChatService.normalizeMessagesForApi(messages);

		// Filter out image attachments if the model doesn't support vision
		if (options.model && !modelsStore.props.modelSupportsVision(options.model)) {
			normalizedMessages.forEach((msg) => {
				if (Array.isArray(msg.content)) {
					msg.content = msg.content.filter((part: ApiChatMessageContentPart) => {
						if (part.type === ContentPartType.IMAGE_URL) {
							console.info(
								`[ChatService] Skipping image attachment in message history (model "${options.model}" does not support vision)`
							);

							return false;
						}

						return true;
					});

					// If only text remains and it's a single part, simplify to string
					if (
						msg.content.length === 1 &&
						msg.content[0].type === ContentPartType.TEXT &&
						typeof msg.content[0].text === 'string'
					) {
						msg.content = msg.content[0].text;
					}
				}
			});
		}

		const requestBody: ApiChatCompletionRequest = {
			messages: normalizedMessages.map((msg: ApiChatMessageData) => {
				const mapped: ApiChatCompletionRequest['messages'][0] = {
					content: msg.content,
					role: msg.role,
					tool_call_id: msg.tool_call_id,
					tool_calls: msg.tool_calls
				};

				// Include reasoning_content from the dedicated field
				if (!excludeReasoningFromContext && msg.reasoning_content) {
					mapped.reasoning_content = msg.reasoning_content;
				}

				return mapped;
			}),
			return_progress: stream ? true : undefined,
			sse_ping_interval: stream ? 1 : undefined,
			stream,
			tools: tools && tools.length > 0 ? tools : undefined
		};

		// Include model in request if provided (required in ROUTER mode)
		if (options.model) {
			requestBody.model = options.model;
		}

		requestBody.reasoning_format = disableReasoningParsing
			? ReasoningFormat.NONE
			: ReasoningFormat.AUTO;

		const reasoningBudgetTokens =
			enableThinking && reasoningEffort ? (REASONING_EFFORT_TOKENS[reasoningEffort] ?? -1) : -1;

		// an explicit user choice injects the kwarg, otherwise it is omitted so
		// the server default applies (--reasoning flag or chat template)
		if (enableThinking !== undefined) {
			requestBody.chat_template_kwargs = {
				...(requestBody.chat_template_kwargs ?? {}),
				enable_thinking: enableThinking
			};
		}

		if (reasoningBudgetTokens >= 0) {
			requestBody.thinking_budget_tokens = reasoningBudgetTokens;
		}

		// arms the budget sampler so reasoning can be ended at runtime via the control endpoint
		requestBody.reasoning_control = true;

		if (continueFinalMessage) {
			requestBody.continue_final_message = true;
			requestBody.add_generation_prompt = false;
		}

		if (temperature !== undefined) requestBody.temperature = temperature;

		if (max_tokens !== undefined) {
			// Set max_tokens to -1 (infinite) when explicitly configured as 0 or null
			requestBody.max_tokens = max_tokens !== null && max_tokens !== 0 ? max_tokens : -1;
		}

		if (dynatemp_range !== undefined) requestBody.dynatemp_range = dynatemp_range;

		if (dynatemp_exponent !== undefined) requestBody.dynatemp_exponent = dynatemp_exponent;

		if (top_k !== undefined) requestBody.top_k = top_k;

		if (top_p !== undefined) requestBody.top_p = top_p;

		if (min_p !== undefined) requestBody.min_p = min_p;

		if (xtc_probability !== undefined) requestBody.xtc_probability = xtc_probability;

		if (xtc_threshold !== undefined) requestBody.xtc_threshold = xtc_threshold;

		if (typ_p !== undefined) requestBody.typ_p = typ_p;

		if (repeat_last_n !== undefined) requestBody.repeat_last_n = repeat_last_n;

		if (repeat_penalty !== undefined) requestBody.repeat_penalty = repeat_penalty;

		if (presence_penalty !== undefined) requestBody.presence_penalty = presence_penalty;

		if (frequency_penalty !== undefined) requestBody.frequency_penalty = frequency_penalty;

		if (dry_multiplier !== undefined) requestBody.dry_multiplier = dry_multiplier;

		if (dry_base !== undefined) requestBody.dry_base = dry_base;

		if (dry_allowed_length !== undefined) requestBody.dry_allowed_length = dry_allowed_length;

		if (dry_penalty_last_n !== undefined) requestBody.dry_penalty_last_n = dry_penalty_last_n;

		if (samplers !== undefined) {
			requestBody.samplers =
				typeof samplers === 'string'
					? samplers.split(';').filter((s: string) => s.trim())
					: samplers;
		}

		if (backend_sampling !== undefined) requestBody.backend_sampling = backend_sampling;

		if (timings_per_token !== undefined) requestBody.timings_per_token = timings_per_token;

		if (custom) {
			try {
				const customParams = typeof custom === 'string' ? JSON.parse(custom) : custom;

				Object.assign(requestBody, customParams);
			} catch (error) {
				console.warn('Failed to parse custom parameters:', error);
			}
		}

		try {
			const headers: Record<string, string> = { ...getJsonHeaders() };

			// tag streaming requests with the conversation id, this single header is the opt in for the
			// server side replay buffer and powers discoverActiveStream on tab reopen. with an explicit
			// model the ::model suffix keeps the per model session distinct
			if (stream && conversationId) {
				headers[HEADERS.X_CONVERSATION_ID_HEADER] = streamIdentity(conversationId, options.model);
				// persist the pending stream before the fetch: a reload during the model load or
				// the prompt processing must still find its way back to the session once it exists
				ChatService.saveStreamState(conversationId, 0, options.model ?? null);
			}

			const response = await fetch(API_CHAT.COMPLETIONS, {
				body: JSON.stringify(requestBody),
				headers,
				method: 'POST',
				signal
			});

			if (!response.ok) {
				// a rejected request (including one cancelled by a stop during the model load)
				// leaves nothing to resume
				if (conversationId) {
					ChatService.clearStreamState(conversationId);
				}

				const error = await ChatService.parseErrorResponse(response);

				if (onError) {
					onError(error);
				}

				throw error;
			}

			if (stream) {
				await ChatService.handleStreamResponse(
					response,
					onChunk,
					onComplete,
					onError,
					onReasoningChunk,
					onToolCallChunk,
					onModel,
					onCompletionId,
					onTimings,
					conversationId,
					signal,
					onConnectionState,
					options.model
				);

				return;
			} else {
				return ChatService.handleNonStreamResponse(
					response,
					onComplete,
					onError,
					onToolCallChunk,
					onModel
				);
			}
		} catch (error) {
			if (isAbortError(error)) {
				console.log('Chat completion request was aborted');

				return;
			}

			let userFriendlyError: Error;

			if (error instanceof Error) {
				if (error.name === 'TypeError' && error.message.includes('fetch')) {
					userFriendlyError = new Error(
						'Unable to connect to server - please check if the server is running'
					);
					userFriendlyError.name = 'NetworkError';
				} else if (error.message.includes('ECONNREFUSED')) {
					userFriendlyError = new Error('Connection refused - server may be offline');
					userFriendlyError.name = 'NetworkError';
				} else if (error.message.includes('ETIMEDOUT')) {
					userFriendlyError = new Error('Request timed out - the server took too long to respond');
					userFriendlyError.name = 'TimeoutError';
				} else {
					userFriendlyError = error;
				}
			} else {
				userFriendlyError = new Error('Unknown error occurred while sending message');
			}

			console.error('Error in sendMessage:', error);

			if (onError) {
				onError(userFriendlyError);
			}

			throw userFriendlyError;
		}
	}

	/**
	 * Ends the current reasoning block of a running completion, targeted by its
	 * chat completion id (streamed back as `id`). Matching the completion rather
	 * than a slot index avoids a TOCTOU: a finished completion simply matches
	 * nothing server side. The model is carried so the router forwards to the
	 * right child, single model ignores it. Returns true on success.
	 */
	static async stopReasoning(completionId: string, model?: string | null): Promise<boolean> {
		if (!completionId) {
			console.error(
				'stopReasoning: no completion id for the active message, cannot target the running completion'
			);

			return false;
		}

		const body: Record<string, unknown> = {
			action: CONTROL_ACTION.END_REASONING,
			id: completionId
		};

		if (model) body.model = model;

		try {
			const res = await fetch(API_CHAT.CONTROL, {
				body: JSON.stringify(body),
				headers: getJsonHeaders(),
				method: 'POST'
			});
			const data = await res.json().catch(() => null);

			if (!res.ok || data?.success !== true) {
				console.error('stopReasoning: control request failed', {
					completionId,
					response: data,
					status: res.status
				});

				return false;
			}

			return true;
		} catch (error) {
			console.error('stopReasoning: control request threw', { completionId, error });

			return false;
		}
	}

	// 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}`;
	}

	/**
	 * Extracts model name from Chat Completions API response data.
	 * Handles various response formats including streaming chunks and final responses.
	 *
	 * WORKAROUND: In single model mode, llama-server returns a default/incorrect model name
	 * in the response. We override it with the actual model name from serverStore.
	 *
	 * @param data - Raw response data from the Chat Completions API
	 * @returns Model name string if found, undefined otherwise
	 * @private
	 */
	private static extractModelName(data: unknown): string | undefined {
		const asRecord = (value: unknown): Record<string, unknown> | undefined => {
			return typeof value === 'object' && value !== null
				? (value as Record<string, unknown>)
				: undefined;
		};
		const getTrimmedString = (value: unknown): string | undefined => {
			return typeof value === 'string' && value.trim() ? value.trim() : undefined;
		};
		const root = asRecord(data);

		if (!root) return undefined;

		// 1) root (some implementations provide `model` at the top level)
		const rootModel = getTrimmedString(root.model);

		if (rootModel) {
			return rootModel;
		}

		// 2) streaming choice (delta) or final response (message)
		const firstChoice = Array.isArray(root.choices) ? asRecord(root.choices[0]) : undefined;

		if (!firstChoice) {
			return undefined;
		}

		// priority: delta.model (first chunk) else message.model (final response)
		const deltaModel = getTrimmedString(asRecord(firstChoice.delta)?.model);

		if (deltaModel) {
			return deltaModel;
		}

		const messageModel = getTrimmedString(asRecord(firstChoice.message)?.model);

		if (messageModel) {
			return messageModel;
		}

		// avoid guessing from non-standard locations (metadata, etc.)
		return undefined;
	}

	/**
	 * Handles non-streaming response from the chat completion API.
	 * Parses the JSON response and extracts the generated content.
	 *
	 * @param response - The fetch Response object containing the JSON data
	 * @param onComplete - Optional callback invoked when response is successfully parsed
	 * @param onError - Optional callback invoked if an error occurs while parsing
	 * @returns {Promise<string>} Promise that resolves to the generated content string
	 * @throws {Error} if the response cannot be parsed or is malformed
	 */
	private static async handleNonStreamResponse(
		response: Response,
		onComplete?: (
			response: string,
			reasoningContent?: string,
			timings?: ChatMessageTimings,
			toolCalls?: string
		) => void,
		onError?: (error: Error) => void,
		onToolCallChunk?: (chunk: string) => void,
		onModel?: (model: string) => void
	): Promise<string> {
		try {
			const responseText = await response.text();

			if (!responseText.trim()) {
				const noResponseError = new Error('No response received from server. Please try again.');

				throw noResponseError;
			}

			const data: ApiChatCompletionResponse = JSON.parse(responseText);
			const responseModel = ChatService.extractModelName(data);

			if (responseModel) {
				onModel?.(responseModel);
			}

			const content = data.choices[0]?.message?.content || '';
			const reasoningContent = data.choices[0]?.message?.reasoning_content;
			const toolCalls = data.choices[0]?.message?.tool_calls;

			let serializedToolCalls: string | undefined;

			if (toolCalls && toolCalls.length > 0) {
				const mergedToolCalls = ChatService.mergeToolCallDeltas([], toolCalls);

				if (mergedToolCalls.length > 0) {
					serializedToolCalls = JSON.stringify(mergedToolCalls);

					if (serializedToolCalls) {
						onToolCallChunk?.(serializedToolCalls);
					}
				}
			}

			if (!content.trim() && !serializedToolCalls) {
				const noResponseError = new Error('No response received from server. Please try again.');

				throw noResponseError;
			}

			onComplete?.(content, reasoningContent, undefined, serializedToolCalls);

			return content;
		} catch (error) {
			const err = error instanceof Error ? error : new Error('Parse error');

			onError?.(err);

			throw err;
		}
	}

	/**
	 * Merges tool call deltas into an existing array of tool calls.
	 * Handles both existing and new tool calls, updating existing ones and adding new ones.
	 *
	 * @param existing - The existing array of tool calls to merge into
	 * @param deltas - The array of tool call deltas to merge
	 * @param indexOffset - Optional offset to apply to the index of new tool calls
	 * @returns {ApiChatCompletionToolCall[]} The merged array of tool calls
	 */
	private static mergeToolCallDeltas(
		existing: ApiChatCompletionToolCall[],
		deltas: ApiChatCompletionToolCallDelta[],
		indexOffset = 0
	): ApiChatCompletionToolCall[] {
		const result = existing.map((call) => ({
			...call,
			function: call.function ? { ...call.function } : undefined
		}));

		for (const delta of deltas) {
			const index =
				typeof delta.index === 'number' && delta.index >= 0
					? delta.index + indexOffset
					: result.length;

			while (result.length <= index) {
				result.push({ function: undefined });
			}

			const target = result[index]!;

			if (delta.id) {
				target.id = delta.id;
			}

			if (delta.type) {
				target.type = delta.type;
			}

			if (delta.function) {
				const fn = target.function ? { ...target.function } : {};

				if (delta.function.name) {
					fn.name = delta.function.name;
				}

				if (delta.function.arguments) {
					fn.arguments = (fn.arguments ?? '') + delta.function.arguments;
				}

				target.function = fn;
			}
		}

		return result;
	}

	/**
	 * Calls the onTimings callback with timing data from streaming response.
	 *
	 * @param timings - Timing information from the Chat Completions API response
	 * @param promptProgress - Prompt processing progress data
	 * @param onTimingsCallback - Callback function to invoke with timing data
	 * @private
	 */
	private static notifyTimings(
		timings: ChatMessageTimings | undefined,
		promptProgress: ChatMessagePromptProgress | undefined,
		onTimingsCallback:
			| ((timings?: ChatMessageTimings, promptProgress?: ChatMessagePromptProgress) => void)
			| undefined
	): void {
		if (!onTimingsCallback || (!timings && !promptProgress)) return;

		onTimingsCallback(timings, promptProgress);
	}

	/**
	 * Parses error response and creates appropriate error with context information
	 * @param response - HTTP response object
	 * @returns Promise<Error> - Parsed error with context info if available
	 */
	private static async parseErrorResponse(
		response: Response
	): Promise<Error & { contextInfo?: { n_prompt_tokens: number; n_ctx: number } }> {
		try {
			const errorText = await response.text();
			const errorData: ApiErrorResponse = JSON.parse(errorText);
			const message = errorData.error?.message || 'Unknown server error';
			const error = new Error(message) as Error & {
				contextInfo?: { n_prompt_tokens: number; n_ctx: number };
			};

			error.name = response.status === 400 ? 'ServerError' : 'HttpError';

			if (errorData.error && 'n_prompt_tokens' in errorData.error && 'n_ctx' in errorData.error) {
				error.contextInfo = {
					n_ctx: errorData.error.n_ctx,
					n_prompt_tokens: errorData.error.n_prompt_tokens
				};
			}

			return error;
		} catch {
			const fallback = new Error(
				`Server error (${response.status}): ${response.statusText}`
			) as Error & {
				contextInfo?: { n_prompt_tokens: number; n_ctx: number };
			};

			fallback.name = 'HttpError';

			return fallback;
		}
	}

	/**
	 * Strips legacy inline reasoning content tags from message content.
	 * Handles both plain string content and multipart content arrays.
	 */
	private static stripReasoningContent(
		content: string | ApiChatMessageContentPart[]
	): string | ApiChatMessageContentPart[] {
		const stripFromString = (text: string): string =>
			text.replace(LEGACY_AGENTIC_REGEX.REASONING_BLOCK, '').trim();

		if (typeof content === 'string') {
			return stripFromString(content);
		}

		return content.map((part) => {
			if (part.type === ContentPartType.TEXT && part.text) {
				return { ...part, text: stripFromString(part.text) };
			}

			return part;
		});
	}

	// write the resume state straight to localStorage, bypassing the throttle
	private static writeStreamState(
		conversationId: string,
		bytesReceived: number,
		model?: string | null
	): void {
		try {
			const state: ResumableStreamState = {
				bytesReceived,
				model: model ?? null,
				updatedAt: Date.now()
			};

			localStorage.setItem(streamStorageKey(conversationId), JSON.stringify(state));
		} catch {
			// localStorage may be full or disabled, silently ignore
		}
	}
}