diff --git a/README.md b/README.md index 3ec3a97..e6dbc70 100644 --- a/README.md +++ b/README.md @@ -33,9 +33,13 @@ Node.js 22 or newer is required. - `CHAT_LIMIT`: per-account maximum chat sessions. - `CHAT_HISTORY_LIMIT`: number of recent messages included in model prompts. - `MODEL_API_BASE_URL`: base URL of the remote model backend. The server expects `/chat` and `/tags`. -- `MODEL_REQUEST_TIMEOUT_MS`: timeout for chat generation requests. +- `MODEL_REQUEST_TIMEOUT_MS`: legacy fallback for the inactivity timeout (default 180000 ms). +- `MODEL_CONNECT_TIMEOUT_MS`: time to receive Ollama response headers (default 30000 ms). +- `MODEL_INACTIVITY_TIMEOUT_MS`: maximum silence between received stream chunks (default 180000 ms). +- `MODEL_MAX_GENERATION_MS`: overall generation safety limit (default 600000 ms). +- `OLLAMA_KEEP_ALIVE`: Ollama model residency after a request (default `30m`). - `MODEL_LIST_TIMEOUT_MS`: timeout for model list requests. -- `MODEL_UNLOAD_AFTER_MS`: delay before an idle model unload is attempted. +- `MODEL_UNLOAD_AFTER_MS`: delay before idle local request bookkeeping is cleared. Model unloading is managed by Ollama through `OLLAMA_KEEP_ALIVE`. - `MODELS_CACHE_TTL_MS`: model list cache TTL. - `AUTH_RATE_LIMIT_*`: rate-limit window and thresholds for registration/login. - `CHAT_RATE_LIMIT_*`: rate-limit window and thresholds for chat and retry endpoints. @@ -46,12 +50,18 @@ Node.js 22 or newer is required. The server is configured around an Ollama-compatible API surface: -- `POST /chat`: accepts `{ model, messages, stream: true }` and returns newline-delimited JSON stream chunks with `message.content`. +- `POST /chat`: accepts `{ model, messages, stream: true, think: boolean, keep_alive: "30m" }` and returns newline-delimited JSON chunks with optional `message.content` / `message.thinking`, ending in `done: true`. - `GET /tags`: returns the available model list. `qwen3:14b` is the default model for chat and roleplay. The model selector loads available models from `/tags` and selects `qwen3:14b` when available (otherwise the first returned model). Existing chats retain their saved model selection. -Chat replies display Ollama's optional `message.thinking` output in a collapsed “Reasoning” section above the answer, updated through SSE `thinking` events during generation. This text is stored separately from the answer and excluded from conversation prompts. An additive SQLite migration preserves it when reopening chats; exports, imports and branches preserve it too. Models without thinking output show no section. +The composer has independent Web Search and Reasoning toggles, both off by default and reset on server acceptance. It sends `{message, chatId, model, webSearch, reasoning}` using streamed fetch to `POST /chat/stream`. Existing SSE events carry `status` (accepted/searching/generating), `thinking` (chunk/fullThinking), `chunk` (chunk/fullReply), `done` (reply/sources), and `error`. Final answers stream incrementally, including persona responses. Persona validation still runs; a corrective pass resets provisional output before streaming its replacement. Only the validated complete answer is persisted. + +Every Ollama request explicitly sets `stream: true`, `think`, and `keep_alive`. Reasoning is requested and forwarded only when `reasoning === true` and the model supports the boolean thinking API. The shared conservative capability check supports Qwen3, DeepSeek R1 and DeepSeek v3.1 model IDs; unknown models, Qwen coder variants and models requiring thinking levels have the toggle disabled. Internal calls, roleplay openers and regeneration default to `think: false`. There is no extra model call to obtain reasoning. + +Reasoning appears only after actual thinking text arrives, expanded during generation and collapsed when the answer begins. Users can reopen it; stored messages start collapsed. The optional nullable `chat_messages.thinking` field remains separate from answer content and sources, is excluded from future prompts and logs, and survives reloads, exports, imports and branches. Old messages remain valid. + +Ollama reads reset the inactivity deadline on each received chunk; the overall safety limit still applies to active streams. Client disconnects and the existing Stop action abort upstream generation. Missing `done: true`, invalid streams, timeouts and upstream failures use the existing error path and never save a partial assistant response as complete. Browser EOF without a terminal SSE event is likewise an interruption. The existing UI discards temporary output on failure. The refactor adds explicit upstream timeout handling, cached model listing, and `GET /health` for degraded-state visibility. diff --git a/public/app/api.js b/public/app/api.js index 9684c8b..4c27d6f 100644 --- a/public/app/api.js +++ b/public/app/api.js @@ -17,12 +17,20 @@ export const post = (url, data) => request(url, data, "POST"); export const put = (url, data) => request(url, data, "PUT"); export const del = (url) => request(url, null, "DELETE"); -export async function stream(url, data, handlers = {}) { +/** + * @typedef {{event: "status", data: {status: "accepted"|"searching"|"generating", sourceCount?: number}} + * | {event: "chunk", data: {chunk: string, fullReply: string}} + * | {event: "thinking", data: {chunk: string, fullThinking: string}} + * | {event: "done", data: {reply: string, sources?: import("./webSearch.js").WebSource[]}} + * | {event: "error", data: {error: string, code?: string}}} ChatStreamEvent + */ +export async function stream(url, data, handlers = {}, signal = undefined) { const response = await fetch(url, { method: "POST", credentials: "same-origin", headers: {"Content-Type": "application/json"}, - body: JSON.stringify(data || {}) + body: JSON.stringify(data || {}), + signal }); if (!response.ok || !response.body) { @@ -43,25 +51,38 @@ export async function stream(url, data, handlers = {}) { const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ""; + let terminal = false; - while (true) { - const {value, done} = await reader.read(); - if (done) break; - buffer += decoder.decode(value, {stream: true}); - const parts = buffer.split("\n\n"); - buffer = parts.pop() || ""; - for (const part of parts) { - const lines = part.split("\n"); - const eventLine = lines.find((line) => line.startsWith("event:")); - const dataLine = lines.find((line) => line.startsWith("data:")); - if (!eventLine || !dataLine) continue; - const event = eventLine.slice(6).trim(); - const payload = JSON.parse(dataLine.slice(5).trim()); - if (event === "chunk" && handlers.onChunk) handlers.onChunk(payload); - if (event === "thinking" && handlers.onThinking) handlers.onThinking(payload); - if (event === "status" && handlers.onStatus) handlers.onStatus(payload); - if (event === "done" && handlers.onDone) handlers.onDone(payload); - if (event === "error" && handlers.onError) handlers.onError(payload); + try { + while (true) { + const {value, done} = await reader.read(); + if (done) break; + buffer += decoder.decode(value, {stream: true}); + const parts = buffer.split(/\r?\n\r?\n/); + buffer = parts.pop() || ""; + for (const part of parts) { + const lines = part.split("\n"); + const eventLine = lines.find((line) => line.startsWith("event:")); + const dataLine = lines.find((line) => line.startsWith("data:")); + if (!eventLine || !dataLine) continue; + const event = eventLine.slice(6).trim(); + const payload = JSON.parse(dataLine.slice(5).trim()); + if (!payload || typeof payload !== "object") throw new Error("Invalid response stream."); + const textFields = {chunk: "fullReply", thinking: "fullThinking", status: "status", done: "reply", error: "error"}; + if (textFields[event] && typeof payload[textFields[event]] !== "string") throw new Error("Invalid response stream."); + if (terminal) continue; + if (event === "chunk" && handlers.onChunk) handlers.onChunk(payload); + if (event === "thinking" && handlers.onThinking) handlers.onThinking(payload); + if (event === "status" && handlers.onStatus) handlers.onStatus(payload); + if (event === "done" && handlers.onDone) handlers.onDone(payload); + if (event === "error" && handlers.onError) handlers.onError(payload); + if (event === "done" || event === "error") terminal = true; + } + if (terminal) break; } + if (!terminal) throw new Error("Response was interrupted. Please try again."); + } finally { + await reader.cancel().catch(() => {}); + reader.releaseLock(); } } diff --git a/public/app/webSearch.js b/public/app/webSearch.js index 58aae7d..6b1355a 100644 --- a/public/app/webSearch.js +++ b/public/app/webSearch.js @@ -1,6 +1,6 @@ /** @typedef {{title: string, url: string, snippet: string}} WebSource */ -/** @typedef {{message: string, model: string, chatId: number, webSearch?: boolean}} ChatRequest */ -/** @typedef {{role: string, content: string, sources?: WebSource[]}} ChatMessage */ +/** @typedef {{message: string, model: string, chatId: number, webSearch?: boolean, reasoning?: boolean}} ChatRequest */ +/** @typedef {{role: string, content: string, thinking?: string, sources?: WebSource[]}} ChatMessage */ export function createWebSearchToggle(button) { let enabled = false; diff --git a/public/chat.html b/public/chat.html index bdd3e5f..2b94c1a 100644 --- a/public/chat.html +++ b/public/chat.html @@ -200,6 +200,9 @@ + diff --git a/public/script.js b/public/script.js index de6e2a9..9b8a403 100644 --- a/public/script.js +++ b/public/script.js @@ -1,6 +1,6 @@ import {del, get, post, put, stream} from "./app/api.js?v=20260909-account-controls"; import {appendSources, createWebSearchToggle} from "./app/webSearch.js"; -import {updateReasoning} from "./app/reasoning.js"; +import {createReasoningToggle, finishReasoning, updateReasoning} from "./app/reasoning.js"; import {closeChatDrawer, openChatDrawer} from "./app/chatDrawer.js?v=20260909-account-controls"; import {onboardingPrompts, requestedChatId} from "./app/constants.js?v=20260909-account-controls"; import {getModelProfile} from "./app/modelProfiles.js?v=20260909-account-controls"; @@ -106,6 +106,7 @@ import { const headerModelButton = document.getElementById("headerModelButton"); const webSearchButton = document.getElementById("webSearch"); const webSearchToggle = createWebSearchToggle(webSearchButton); +const reasoningToggle = createReasoningToggle(document.getElementById("reasoning")); let isProcessing = false, activeChatId = null, editingPersonaId = null, editingPersonaType = "assistant", currentUsername = ""; let chatSessions = [], currentMessages = [], assistantPersonas = [], userPersonas = []; @@ -911,6 +912,7 @@ function renderDrawerSummary() { } function renderModelSection() { + reasoningToggle.setModel(modelSelect.value); updateModelHelp(); return modelSelect?.selectedOptions[0]?.textContent || ""; } @@ -1361,6 +1363,7 @@ function setLoadingState(loading, overlayOptions = null) { isProcessing = loading; msgInput.disabled = loading; webSearchButton.disabled = loading; + reasoningToggle.setBusy(loading); sendBtn.classList.toggle("loading", loading); messagesDiv.querySelectorAll('[data-message-action="regenerate"]').forEach((button) => { button.disabled = loading || button.dataset.retryLimitReached === "true"; @@ -1811,6 +1814,7 @@ async function sendMessage() { const message = msgInput.value.trim(); if (!message) return; const webSearch = webSearchToggle.enabled; + const reasoning = reasoningToggle.enabled; if (!activeChatId) await createNewChat(); const currentChat = getChatById(activeChatId); if (currentChat && !currentChat.assistant_persona_id && currentChat.title === "New chat") { @@ -1825,33 +1829,44 @@ async function sendMessage() { const loadingBody = loadingMsg.querySelector(".msg-content"); const loadingShell = loadingMsg.querySelector(".loading-shell"); const loadingLabel = loadingMsg.querySelector(".loading-copy strong"); - if (webSearch) loadingLabel.textContent = "Searching the web..."; + loadingLabel.textContent = webSearch ? "Searching the web..." : "Generating response..."; try { let streamError = null; - await stream("/chat/stream", {message, model: modelSelect.value, chatId: activeChatId, webSearch}, { + await stream("/chat/stream", {message, model: modelSelect.value, chatId: activeChatId, webSearch, reasoning}, { onStatus: ({status, sourceCount}) => { - webSearchToggle.reset(); + if (status === "accepted") { + webSearchToggle.reset(); + reasoningToggle.reset(); + return; + } loadingLabel.textContent = status === "searching" ? "Searching the web..." : sourceCount === 0 ? "No web results found. Generating response..." : "Generating response..."; }, onThinking: ({fullThinking}) => { + if (!reasoning) return; + const follow = messagesDiv.scrollHeight - messagesDiv.scrollTop - messagesDiv.clientHeight < 100; updateReasoning(loadingMsg, fullThinking, true); - loadingLabel.textContent = "Thinking…"; + loadingShell?.classList.toggle("hidden", Boolean(fullThinking)); + if (follow) scrollMessagesToBottom(); }, onChunk: ({fullReply}) => { - const reasoningSummary = loadingMsg.querySelector(".msg-reasoning summary"); - if (reasoningSummary) reasoningSummary.textContent = "Reasoning"; + const follow = messagesDiv.scrollHeight - messagesDiv.scrollTop - messagesDiv.clientHeight < 100; + if (fullReply) finishReasoning(loadingMsg); if (loadingBody) { - loadingShell?.classList.add("hidden"); + loadingShell?.classList.toggle("hidden", Boolean(fullReply)); + loadingLabel.textContent = "Generating response..."; loadingBody.classList.add("has-content"); setMessageContent(loadingBody, fullReply); } + if (follow) scrollMessagesToBottom(); }, - onDone: () => { webSearchToggle.reset(); }, + onDone: () => { webSearchToggle.reset(); reasoningToggle.reset(); finishReasoning(loadingMsg); }, onError: ({error}) => { streamError = error || "Failed to generate reply."; } }); + const followOutput = messagesDiv.scrollHeight - messagesDiv.scrollTop - messagesDiv.clientHeight < 100; + const previousScrollTop = messagesDiv.scrollTop; if (messagesDiv.contains(loadingMsg)) messagesDiv.removeChild(loadingMsg); if (streamError) { if (webSearch) { @@ -1865,15 +1880,17 @@ async function sendMessage() { await setActiveChat(activeChatId); await loadSummary(); markOnboardingCompleted(); + if (!followOutput) requestAnimationFrame(() => { messagesDiv.scrollTop = previousScrollTop; }); maybeShowPersonaNudge(); setNotice("Reply received.", "success"); - } catch { + } catch (error) { if (messagesDiv.contains(loadingMsg)) messagesDiv.removeChild(loadingMsg); if (webSearch) { msgInput.value = message; resizeComposerInput(); } - addMessage("[Error: Network error]"); setNotice("Network error while sending message.", "error"); + const errorMessage = error?.message || "Network error while sending message."; + addMessage(`[Error: ${errorMessage}]`); setNotice(errorMessage, "error"); } finally { setLoadingState(false); msgInput.focus(); updateChatActionState(); } diff --git a/public/style.css b/public/style.css index 9249b54..e7523b9 100644 --- a/public/style.css +++ b/public/style.css @@ -456,9 +456,11 @@ input:focus, textarea:focus, select:focus { border-color: var(--brand); } .input-meta, .input-container { width: min(940px, 100%); margin-inline: auto; } .input-meta { margin-bottom: 5px; padding-left: 10px; } .composer-hint { margin: 0; color: var(--text-muted); font-size: .72rem; } -.input-container { padding: 7px; display: grid; grid-template-columns: 44px minmax(0, 1fr) 44px; grid-template-areas: "search message send"; gap: 7px; align-items: end; background: var(--input-bg); border: 1px solid var(--border-strong); border-radius: 24px; box-shadow: var(--shadow-sm); } -#webSearch { grid-area: search; width: 44px; height: 44px; display: grid; place-items: center; border-radius: 17px; } -#webSearch[aria-pressed="true"] { color: var(--brand-strong); background: var(--brand-soft); border-color: var(--brand); box-shadow: inset 0 0 0 1px var(--brand); } +.input-container { padding: 7px; display: grid; grid-template-columns: 44px 44px minmax(0, 1fr) 44px; grid-template-areas: "search reasoning message send"; gap: 7px; align-items: end; background: var(--input-bg); border: 1px solid var(--border-strong); border-radius: 24px; box-shadow: var(--shadow-sm); } +#webSearch, #reasoning { width: 44px; height: 44px; display: grid; place-items: center; border-radius: 17px; } +#webSearch { grid-area: search; } +#reasoning { grid-area: reasoning; } +#webSearch[aria-pressed="true"], #reasoning[aria-pressed="true"] { color: var(--brand-strong); background: var(--brand-soft); border-color: var(--brand); box-shadow: inset 0 0 0 1px var(--brand); } .msg-sources { margin-top: 14px; padding-top: 10px; border-top: 1px solid var(--border); min-width: 0; font-size: .8rem; } .msg-sources-heading { color: var(--text-muted); margin-bottom: 6px; } .msg-source { display: flex; min-width: 0; flex-direction: column; gap: 2px; margin-top: 6px; } diff --git a/scripts/lint.mjs b/scripts/lint.mjs index 97d42c8..20b531a 100644 --- a/scripts/lint.mjs +++ b/scripts/lint.mjs @@ -31,6 +31,8 @@ const files = [ "public/market.js", "public/app/api.js", "public/app/webSearch.js", + "public/app/reasoning.js", + "public/app/modelCapabilities.js", "public/app/chatDrawer.js", "public/app/constants.js", "public/app/dom.js", diff --git a/src/config.js b/src/config.js index 08814ae..46a3664 100644 --- a/src/config.js +++ b/src/config.js @@ -54,6 +54,10 @@ export function createConfig(overrides = {}) { chatMaxRetries: int(process.env.CHAT_MAX_RETRIES, 5), modelApiBaseUrl: process.env.MODEL_API_BASE_URL || "https://ai.krishd.ch/api", modelRequestTimeoutMs: int(process.env.MODEL_REQUEST_TIMEOUT_MS, 180000), + modelConnectTimeoutMs: Math.max(1, int(process.env.MODEL_CONNECT_TIMEOUT_MS, 30000)), + modelInactivityTimeoutMs: Math.max(1, int(process.env.MODEL_INACTIVITY_TIMEOUT_MS, int(process.env.MODEL_REQUEST_TIMEOUT_MS, 180000))), + modelMaxGenerationMs: Math.max(1, int(process.env.MODEL_MAX_GENERATION_MS, 600000)), + ollamaKeepAlive: process.env.OLLAMA_KEEP_ALIVE || "30m", searxngUrl: process.env.SEARXNG_URL || "", webSearchTimeoutMs: Math.max(1, int(process.env.WEB_SEARCH_TIMEOUT_MS, 8000)), modelListTimeoutMs: int(process.env.MODEL_LIST_TIMEOUT_MS, 4000), diff --git a/src/routes/chats.js b/src/routes/chats.js index c43e8b7..10ed5e6 100644 --- a/src/routes/chats.js +++ b/src/routes/chats.js @@ -22,7 +22,8 @@ const sendMessageValidator = (body) => { if (!chatId) return {error: "chatId is required"}; if (!message) return {error: "message is required"}; if (body.webSearch !== undefined && typeof body.webSearch !== "boolean") return {error: "webSearch must be a boolean"}; - return {value: {message, chatId, model: String(body?.model || "qwen3:14b"), webSearch: body.webSearch === true}}; + if (body.reasoning !== undefined && typeof body.reasoning !== "boolean") return {error: "reasoning must be a boolean"}; + return {value: {message, chatId, model: String(body?.model || "qwen3:14b"), webSearch: body.webSearch === true, reasoning: body.reasoning === true}}; }; const organizationValidator = (body) => ({ @@ -236,7 +237,8 @@ export function createChatsRouter({repositories, chatService, modelService, conf targetMessage, content: req.validatedBody.message, model: req.validatedBody.model, - webSearch: req.validatedBody.webSearch + webSearch: req.validatedBody.webSearch, + reasoning: req.validatedBody.reasoning }); if (result.error) return res.status(result.status).json({error: result.error}); return res.json(result); @@ -431,7 +433,8 @@ export function createChatsRouter({repositories, chatService, modelService, conf res.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache, no-transform", - Connection: "keep-alive" + Connection: "keep-alive", + "X-Accel-Buffering": "no" }); const sendEvent = (event, data) => { @@ -449,6 +452,7 @@ export function createChatsRouter({repositories, chatService, modelService, conf onChunk: (chunk, fullReply) => sendEvent("chunk", {chunk, fullReply}), onThinking: (chunk, fullThinking) => sendEvent("thinking", {chunk, fullThinking}), webSearch: req.validatedBody.webSearch, + reasoning: req.validatedBody.reasoning, onStatus: (status) => sendEvent("status", status), signal: generationController.signal }); diff --git a/src/services/chatService.js b/src/services/chatService.js index dfdcdcb..859292d 100644 --- a/src/services/chatService.js +++ b/src/services/chatService.js @@ -19,6 +19,7 @@ import { } from "./chatMessageHelpers.js"; import {SearXNGSearchProvider, WebSearchError, buildWebSearchContext, buildWebSearchQuery, decideWebSearch, selectSearchResults, MAX_CANDIDATES} from "./webSearchService.js"; +import {supportsReasoning} from "../../public/app/modelCapabilities.js"; export function createChatService(repositories, modelService, config, webSearchProvider = new SearXNGSearchProvider(config)) { const chatGenerationQueues = new Map(); @@ -192,21 +193,25 @@ export function createChatService(repositories, modelService, config, webSearchP ], {onThinking}); }; - const generateValidatedRoleplayReply = async ({selectedModel, messagesPayload, userPersona, signal = null, generation = {}, onThinking = null}) => { - let reply = await modelService.generateReply(selectedModel, messagesPayload, {signal, generation, onThinking}); + const generateValidatedRoleplayReply = async ({selectedModel, messagesPayload, userPersona, signal = null, generation = {}, onThinking = null, onChunk = null, reasoning = false}) => { + const generate = (messages) => onChunk + ? modelService.streamReply(selectedModel, messages, onChunk, {signal, generation, onThinking, reasoning}) + : modelService.generateReply(selectedModel, messages, {signal, generation, onThinking, reasoning}); + let reply = await generate(messagesPayload); if (reply && !containsUserVoiceInRoleplayOpener(reply, userPersona)) { return reply; } onThinking?.("", ""); - reply = await modelService.generateReply(selectedModel, [ + onChunk?.("", ""); + reply = await generate([ ...messagesPayload, {role: "assistant", content: reply || ""}, { role: "user", content: "Rewrite your last reply. Hard rule: only the assistant character may speak or act. Never write or imply user speech, thoughts, feelings, reactions, decisions, or actions. Stop before the user's reply." } - ], {signal, generation, onThinking}); + ]); if (!reply || containsUserVoiceInRoleplayOpener(reply, userPersona)) { return null; @@ -320,7 +325,7 @@ export function createChatService(repositories, modelService, config, webSearchP return {fullReply, thinking, promptMessageId: promptMessage.id}; }; - const generateChatTurn = ({user, chatId, message, model, webSearch = false, onStatus = null, onChunk = null, onThinking = null, signal = null, parentMessageId = undefined}) => + const generateChatTurn = ({user, chatId, message, model, webSearch = false, reasoning = false, onStatus = null, onChunk = null, onThinking = null, signal = null, parentMessageId = undefined}) => withChatGenerationLock(chatId, async () => { await (summaryQueues.get(chatId) || Promise.resolve()); const session = repositories.getChat(chatId, user); @@ -337,17 +342,22 @@ export function createChatService(repositories, modelService, config, webSearchP return {error: "Branch point not found", status: 404}; } const {model: effectiveModel, generation} = getGenerationSettings(session, model); + const useReasoning = reasoning === true && supportsReasoning(effectiveModel); + signal?.throwIfAborted(); const inserted = repositories.insertBranchChatMessage(chatId, "user", message, null, null, "pending", null, effectiveParentId); const userMessageId = Number(inserted?.lastInsertRowid); try { let sources = null; let thinking = ""; const captureThinking = (chunk, fullThinking) => { + if (!useReasoning) return; thinking = fullThinking; onThinking?.(chunk, fullThinking); }; let payload; const fullReply = await withActiveGeneration(chatId, signal, async (activeSignal) => { + activeSignal.throwIfAborted(); + onStatus?.({status: "accepted"}); if (webSearch) { const searchDecision = decideWebSearch(message); let searchQuery = null; @@ -364,7 +374,6 @@ export function createChatService(repositories, modelService, config, webSearchP candidateCount = candidates.length; sources = selectSearchResults(message, candidates, searchDecision); selectedCount = sources.length; - onStatus?.({status: "generating", sourceCount: selectedCount}); } if (config.structuredRequestLoggingEnabled) { console.log(JSON.stringify({ @@ -379,6 +388,7 @@ export function createChatService(repositories, modelService, config, webSearchP })); } } + onStatus?.({status: "generating", ...(sources !== null ? {sourceCount: sources.length} : {})}); payload = buildConversationContext({user, chatId, session, message: null, sources}); return payload.activePersona ? generateValidatedRoleplayReply({ @@ -387,18 +397,19 @@ export function createChatService(repositories, modelService, config, webSearchP userPersona: payload.activeUserPersona, signal: activeSignal, generation, - onThinking: captureThinking + onThinking: captureThinking, + onChunk, + reasoning: useReasoning }) : onChunk - ? modelService.streamReply(effectiveModel, payload.messagesPayload, onChunk, {signal: activeSignal, generation, onThinking: captureThinking}) - : modelService.generateReply(effectiveModel, payload.messagesPayload, {signal: activeSignal, generation, onThinking: captureThinking}); + ? modelService.streamReply(effectiveModel, payload.messagesPayload, onChunk, {signal: activeSignal, generation, onThinking: captureThinking, reasoning: useReasoning}) + : modelService.generateReply(effectiveModel, payload.messagesPayload, {signal: activeSignal, generation, onThinking: captureThinking, reasoning: useReasoning}); }); if (!fullReply) { repositories.failChatTurn(chatId, userMessageId, "Failed to generate a valid character reply", webSearch ? userMessageId : effectiveParentId); return {error: "Failed to generate a valid character reply", status: 500}; } - if (payload.activePersona && onChunk) onChunk(fullReply, fullReply); const retryState = { retryVariants: [fullReply], @@ -508,11 +519,11 @@ export function createChatService(repositories, modelService, config, webSearchP containsUserVoiceInRoleplayOpener, persistChatMessageRetryState, buildMessagesPayload: buildConversationContext, - async sendChatMessage({user, chatId, message, model, webSearch = false}) { - return generateChatTurn({user, chatId, message, model, webSearch}); + async sendChatMessage({user, chatId, message, model, webSearch = false, reasoning = false}) { + return generateChatTurn({user, chatId, message, model, webSearch, reasoning}); }, - async streamChatMessage({user, chatId, message, model, webSearch = false, onStatus, onChunk, onThinking, signal = null}) { - return generateChatTurn({user, chatId, message, model, webSearch, onStatus, onChunk, onThinking, signal}); + async streamChatMessage({user, chatId, message, model, webSearch = false, reasoning = false, onStatus, onChunk, onThinking, signal = null}) { + return generateChatTurn({user, chatId, message, model, webSearch, reasoning, onStatus, onChunk, onThinking, signal}); }, async retryMessage({user, chatId, selectedModel, targetMessage, retryStyle = null}) { return withChatGenerationLock(chatId, async () => { diff --git a/src/services/modelService.js b/src/services/modelService.js index bb9cd10..fb71886 100644 --- a/src/services/modelService.js +++ b/src/services/modelService.js @@ -1,4 +1,4 @@ -import {execFile} from "child_process"; +import {supportsReasoning} from "../../public/app/modelCapabilities.js"; class ModelServiceError extends Error { constructor(code, message, status = 500) { @@ -41,9 +41,8 @@ export function createModelService(config, overrides = {}) { if (state.timer) clearTimeout(state.timer); state.timer = setTimeout(() => { if (state.activeRequests > 0) return; - execFile("ollama", ["stop", model], () => { - modelState.delete(model); - }); + // Ollama owns unloading through keep_alive; only clear local bookkeeping. + modelState.delete(model); }, config.modelUnloadAfterMs); state.timer.unref?.(); }; @@ -74,7 +73,37 @@ export function createModelService(config, overrides = {}) { } }; - const readStreamingReply = async (response, onChunk = null, onThinking = null) => { + const withGenerationTimeout = async (callback, externalSignal) => { + const controller = new AbortController(); + let timedOut = false; + let idleTimer; + const expire = () => { timedOut = true; controller.abort(); }; + const touch = (ms = config.modelInactivityTimeoutMs ?? config.modelRequestTimeoutMs ?? 180000) => { + clearTimeout(idleTimer); + idleTimer = setTimeout(expire, ms); + }; + const abort = () => controller.abort(); + if (externalSignal?.aborted) abort(); + else externalSignal?.addEventListener("abort", abort, {once: true}); + touch(config.modelConnectTimeoutMs ?? 30000); + const maxTimer = setTimeout(expire, config.modelMaxGenerationMs ?? 600000); + try { + controller.signal.throwIfAborted(); + return await callback(controller.signal, touch); + } catch (error) { + if (controller.signal.aborted) { + throw new ModelServiceError(timedOut ? "MODEL_TIMEOUT" : "MODEL_CANCELLED", + timedOut ? "Model backend timed out." : "Model request was cancelled.", timedOut ? 504 : 499); + } + throw error; + } finally { + clearTimeout(idleTimer); + clearTimeout(maxTimer); + externalSignal?.removeEventListener("abort", abort); + } + }; + + const readStreamingReply = async (response, onChunk = null, onThinking = null, touch = () => {}, signal = null) => { if (!response.body) { throw new ModelServiceError("MODEL_BAD_RESPONSE", "Model backend returned no body.", 502); } @@ -84,9 +113,18 @@ export function createModelService(config, overrides = {}) { let buffer = ""; let fullReply = ""; let fullThinking = ""; + let completed = false; const consume = (line) => { const data = JSON.parse(line); - if (data.message?.thinking) { + if (!data || typeof data !== "object" || data.error) { + throw new ModelServiceError("MODEL_BAD_RESPONSE", "Model backend returned an invalid stream.", 502); + } + for (const field of ["content", "thinking"]) { + if (data.message?.[field] != null && typeof data.message[field] !== "string") { + throw new ModelServiceError("MODEL_BAD_RESPONSE", "Model backend returned an invalid stream.", 502); + } + } + if (data.message?.thinking && onThinking) { fullThinking += data.message.thinking; onThinking?.(data.message.thinking, fullThinking); } @@ -94,29 +132,43 @@ export function createModelService(config, overrides = {}) { fullReply += data.message.content; onChunk?.(data.message.content, fullReply); } + if (data.done === true) completed = true; }; - while (true) { - const {value, done} = await reader.read(); - if (done) break; - buffer += decoder.decode(value, {stream: true}); - const lines = buffer.split("\n"); - buffer = lines.pop() || ""; - for (const line of lines) { - const trimmed = line.trim(); - if (!trimmed) continue; - consume(trimmed); + const cancel = () => { void reader.cancel().catch(() => {}); }; + signal?.addEventListener("abort", cancel, {once: true}); + try { + while (!completed) { + signal?.throwIfAborted(); + const {value, done} = await reader.read(); + signal?.throwIfAborted(); + if (done) break; + touch(); + buffer += decoder.decode(value, {stream: true}); + const lines = buffer.split("\n"); + buffer = lines.pop() || ""; + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed) continue; + consume(trimmed); + if (completed) break; + } } - } - if (buffer.trim()) { - consume(buffer.trim()); + buffer += decoder.decode(); + if (!completed && buffer.trim()) { + consume(buffer.trim()); + } + if (!completed) throw new ModelServiceError("MODEL_INTERRUPTED", "Model response was interrupted.", 502); + return fullReply.trim(); + } finally { + signal?.removeEventListener("abort", cancel); + await reader.cancel().catch(() => {}); + reader.releaseLock(); } - - return fullReply.trim(); }; - const buildRequestBody = (model, messagesPayload, generation = {}) => { + const buildRequestBody = (model, messagesPayload, generation = {}, reasoning = false) => { const options = {}; if (generation.temperature != null) options.temperature = generation.temperature; if (generation.contextLength != null) options.num_ctx = generation.contextLength; @@ -127,6 +179,8 @@ export function createModelService(config, overrides = {}) { // retries, roleplay openers, and summaries) receives runtime date context. messages: [buildRuntimeDateContext(), ...messagesPayload], stream: true, + think: reasoning === true && supportsReasoning(model), + keep_alive: config.ollamaKeepAlive ?? "30m", ...(Object.keys(options).length ? {options} : {}) }; }; @@ -138,11 +192,11 @@ export function createModelService(config, overrides = {}) { try { const startedAt = Date.now(); - const fullReply = await withTimeout(config.modelRequestTimeoutMs, async (signal) => { + const fullReply = await withGenerationTimeout(async (signal, touch) => { const response = await fetchImpl(`${config.modelApiBaseUrl}/chat`, { method: "POST", headers: {"Content-Type": "application/json"}, - body: JSON.stringify(buildRequestBody(model, messagesPayload, options.generation)), + body: JSON.stringify(buildRequestBody(model, messagesPayload, options.generation, options.reasoning)), signal }); @@ -150,7 +204,8 @@ export function createModelService(config, overrides = {}) { throw new ModelServiceError("MODEL_UPSTREAM_ERROR", `Model backend returned ${response.status}.`, 502); } - return readStreamingReply(response, null, options.onThinking); + touch(); + return readStreamingReply(response, null, options.reasoning === true && supportsReasoning(model) ? options.onThinking : null, touch, signal); }, options.signal); lastHealth = {ok: true, checkedAt: new Date().toISOString(), error: null, latencyMs: Date.now() - startedAt}; return fullReply; @@ -176,11 +231,11 @@ export function createModelService(config, overrides = {}) { try { const startedAt = Date.now(); - const fullReply = await withTimeout(config.modelRequestTimeoutMs, async (signal) => { + const fullReply = await withGenerationTimeout(async (signal, touch) => { const response = await fetchImpl(`${config.modelApiBaseUrl}/chat`, { method: "POST", headers: {"Content-Type": "application/json"}, - body: JSON.stringify(buildRequestBody(model, messagesPayload, options.generation)), + body: JSON.stringify(buildRequestBody(model, messagesPayload, options.generation, options.reasoning)), signal }); @@ -188,7 +243,8 @@ export function createModelService(config, overrides = {}) { throw new ModelServiceError("MODEL_UPSTREAM_ERROR", `Model backend returned ${response.status}.`, 502); } - return readStreamingReply(response, onChunk, options.onThinking); + touch(); + return readStreamingReply(response, onChunk, options.reasoning === true && supportsReasoning(model) ? options.onThinking : null, touch, signal); }, options.signal); lastHealth = {ok: true, checkedAt: new Date().toISOString(), error: null, latencyMs: Date.now() - startedAt}; return fullReply; diff --git a/test/webSearch.test.js b/test/webSearch.test.js index d7b6134..8bf4f4c 100644 --- a/test/webSearch.test.js +++ b/test/webSearch.test.js @@ -82,7 +82,7 @@ test("every Ollama request includes dynamic server date context", async () => { }, { fetch: async (_url, options) => { requestBody = JSON.parse(options.body); - return new Response('{"message":{"content":"ok"}}\n', {status: 200}); + return new Response('{"message":{"content":"ok"},"done":true}\n', {status: 200}); } }); assert.equal(await modelService.generateReply("test:model", [{role: "user", content: "hello"}]), "ok"); @@ -192,9 +192,10 @@ test("chat search integrates with streaming, persistence, personas, settings, re const [event, data] = block.split("\n"); events.push({event: event.slice(7), data: JSON.parse(data.slice(6))}); } - assert.deepEqual(events.map((event) => event.event), ["status", "status", "chunk", "chunk", "done"]); - assert.equal(events[0].data.status, "searching"); - assert.deepEqual(events[1].data, {status: "generating", sourceCount: 1}); + assert.deepEqual(events.map((event) => event.event), ["status", "status", "status", "chunk", "chunk", "done"]); + assert.equal(events[0].data.status, "accepted"); + assert.equal(events[1].data.status, "searching"); + assert.deepEqual(events[2].data, {status: "generating", sourceCount: 1}); assert.deepEqual(events.at(-1).data.sources, [source]); assert.equal(searchCount, 1); assert.equal(lastSearchQuery, "major news events September 10, 2025 2025-09-10");