import { LLMOrbiter, OpenRouterProvider } from 'libllmorbiter'; import * as db from '../db/modelconversation.js'; import { getStore } from '../db/store.js'; import env from '../config/env.js'; import { emit as emitCompanyEvent } from '../api/sse.js'; import { readFileSync, writeFileSync, mkdirSync } from 'fs'; import { resolve, dirname, extname } from 'path'; import { fileURLToPath } from 'url'; import { randomUUID } from 'node:crypto'; import { buildSystemPrompt } from './prompts.js'; import { createDashbackRequestTool } from '../tools/dashback-request.js'; import { createAskUserTool } from '../tools/ask-user.js'; import { createJavascriptExecuteTool } from '../tools/javascript.js'; import { createResetVfsTool } from '../tools/reset-vfs.js'; import { resolveVfsRoot, prepareVfs, getMimeType } from '../tools/vfs.js'; import { createApiSuggestionsOrIssuesTool } from '../tools/apisuggestionsorissues.js'; import { createFindFilesTool } from '../tools/find-files.js'; import { createReadFileTool } from '../tools/read-file.js'; import { createPatchTool } from '../tools/patch.js'; import { createRmTool } from '../tools/rm.js'; import { createSearchImagesTool } from '../tools/search-images.js'; import { createAnalyzeImageTool } from '../tools/analyze-image.js'; import { createEditImageTool } from '../tools/edit-image.js'; import { createWebSearchTool } from '../tools/web-search.js'; import { createSearchWebPriceTool } from '../tools/search-web-price.js'; import { createUrlGetTool } from '../tools/url-get.js'; import { createPdfReadTool } from '../tools/pdf-read.js'; import { createSearchArticlesTool } from '../tools/search-articles.js'; import { createEanChecksumTool } from '../tools/ean-checksum.js'; import { createNetGrossCalcTool } from '../tools/net-gross-calc.js'; import { createSetUserRouteTool } from '../tools/set-user-route.js'; import { createDraftDeliveryManifestTool } from '../tools/draft-delivery-manifest.js'; import { createSubagentTool } from '../tools/subagent.js'; import { resolveOpenRouterKey, resolveDefaultModel, resolveModelReasoning, RETIRED_MODELS, hasConfiguredOpenRouterKey } from '../tools/llm-key.js'; // #region Constants const KEEP_ALIVE_MS = 20_000; // #endregion // #region Message Helpers /** * Per-conversation manager. Owns the SSE subscribers of one conversation and * runs the LLM (via LLMOrbiter) for each posted user message. * * Persistence: the database holds the transcript (`modelmessage`) and a binary * `idle`/`busy` state. The finer SSE states (`thinking`, `working`, `done`) * and the streamed `delta` chunks are SSE-only. */ function parseUtcDate(val) { if (!val) return new Date(); if (val instanceof Date) return isNaN(val.getTime()) ? new Date() : val; if (typeof val === 'string') { const s = val.trim(); if (/^\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}/.test(s) && !s.endsWith('Z') && !/[+-]\d{2}(:?\d{2})?$/.test(s)) { const parsed = new Date(s.replace(' ', 'T') + 'Z'); if (!isNaN(parsed.getTime())) return parsed; } const parsed = new Date(s); if (!isNaN(parsed.getTime())) return parsed; } return new Date(); } export function formatTimePrefix(createdDate) { const d = parseUtcDate(createdDate); const parts = new Intl.DateTimeFormat('en-GB', { timeZone: 'Europe/Berlin', hour: '2-digit', minute: '2-digit', hour12: false, timeZoneName: 'short', }).formatToParts(d); const hour = parts.find((p) => p.type === 'hour')?.value.padStart(2, '0') || '00'; const minute = parts.find((p) => p.type === 'minute')?.value.padStart(2, '0') || '00'; let tz = parts.find((p) => p.type === 'timeZoneName')?.value || 'MEST'; if (tz === 'CEST') tz = 'MEST'; else if (tz === 'CET') tz = 'MET'; return `[${hour}:${minute}${tz}] `; } export function parseIncomingUserMessage({ message, attachments: directAttachments, content: directContent } = {}) { let text = ''; const attachments = Array.isArray(directAttachments) ? [...directAttachments] : []; const tryParseJson = (str) => { if (typeof str !== 'string') return null; const trimmed = str.trim(); if ((trimmed.startsWith('{') && trimmed.endsWith('}')) || (trimmed.startsWith('[') && trimmed.endsWith(']'))) { try { return JSON.parse(trimmed); } catch { return null; } } return null; }; let candidate = message !== undefined ? message : directContent; if (typeof candidate === 'string') { const parsed = tryParseJson(candidate); if (parsed && typeof parsed === 'object') { candidate = parsed; } else { text = candidate; } } if (candidate && typeof candidate === 'object' && !Array.isArray(candidate)) { if (typeof candidate.content === 'string') { const innerParsed = tryParseJson(candidate.content); if (innerParsed && typeof innerParsed === 'object') { if (typeof innerParsed.content === 'string') { text = innerParsed.content; } if (Array.isArray(innerParsed.attachments)) { attachments.push(...innerParsed.attachments); } } else { text = candidate.content; } } else if (typeof candidate.message === 'string') { const innerParsed = tryParseJson(candidate.message); if (innerParsed && typeof innerParsed === 'object') { if (typeof innerParsed.content === 'string') { text = innerParsed.content; } if (Array.isArray(innerParsed.attachments)) { attachments.push(...innerParsed.attachments); } } else { text = candidate.message; } } if (Array.isArray(candidate.attachments)) { attachments.push(...candidate.attachments); } } const uniqueAttachments = []; const seen = new Set(); for (const att of attachments) { if (!att || typeof att !== 'object') continue; const key = att.id || `${att.name || att.filename}_${att.size || ''}`; if (!seen.has(key)) { seen.add(key); uniqueAttachments.push(att); } } return { content: String(text || '').trim(), attachments: uniqueAttachments, }; } export function cleanUserMessageContent(rawContent) { if (!rawContent || typeof rawContent !== 'string') return ''; if (rawContent.includes('"attachments"') && (rawContent.trim().startsWith('{') || rawContent.includes('{"content"'))) { try { const parsed = parseIncomingUserMessage({ message: rawContent }); if (parsed.attachments.length > 0) { const tags = parsed.attachments.map((att) => { const name = (att.name || att.filename || 'attachment').replace(/^.*[\\\/]/, '').replace(/[\0\r\n"]/g, '').trim() || 'attachment'; const mime = att.mimeType || att.type || getMimeType(name); const vfsRelPath = `attachments/${name}`; const formatAttr = (val) => String(val).includes(' ') ? `"${val}"` : String(val); return `[ATTACHMENT name=${formatAttr(name)} path=${formatAttr(vfsRelPath)} type=${formatAttr(mime)}]`; }); const prefix = tags.join(' '); return parsed.content ? `${prefix}\n\n${parsed.content}` : prefix; } if (parsed.content) return parsed.content; } catch { // keep rawContent if parsing fails } } return rawContent; } export function historyToLlmMessages(history) { const messages = []; for (const m of history) { if (m.role === 'usage' || m.role === 'reasoning') continue; let payload; try { payload = typeof m.message === 'string' ? JSON.parse(m.message) : m.message; } catch { payload = { content: m.message }; } if (m.role === 'user') { const rawContent = payload.content ?? (typeof m.message === 'string' ? m.message : ''); const cleanedContent = cleanUserMessageContent(rawContent); const timePrefix = formatTimePrefix(m.created_at); const content = /^\s*\[\d{2}:\d{2}[A-Za-z]+\]\s*/.test(cleanedContent) ? cleanedContent : `${timePrefix}${cleanedContent}`; messages.push({ role: 'user', content }); } else if (m.role === 'assistant') { messages.push({ role: 'assistant', content: payload.content ?? null, ...(Array.isArray(payload.tool_calls) && payload.tool_calls.length > 0 ? { tool_calls: payload.tool_calls } : {}), }); } else if (m.role === 'tool') { const toolCallId = payload.tool_call_id ?? ''; const toolName = payload.name ?? ''; const toolParams = payload.parameters ?? payload.arguments ?? '{}'; const prevMsg = messages[messages.length - 1]; if (!prevMsg || prevMsg.role !== 'assistant' || !prevMsg.tool_calls) { messages.push({ role: 'assistant', content: null, tool_calls: [{ id: toolCallId, type: 'function', function: { name: toolName, arguments: typeof toolParams === 'string' ? toolParams : JSON.stringify(toolParams), }, }], }); } else if (!prevMsg.tool_calls.some((tc) => tc.id === toolCallId)) { prevMsg.tool_calls.push({ id: toolCallId, type: 'function', function: { name: toolName, arguments: typeof toolParams === 'string' ? toolParams : JSON.stringify(toolParams), }, }); } messages.push({ role: 'tool', tool_call_id: toolCallId, content: typeof payload.content === 'string' ? payload.content : JSON.stringify(payload.content ?? ''), ...(toolName ? { name: toolName } : {}), }); } } return messages; } // #endregion // #region ConversationManager export class ConversationManager { /** * @param {number} conversationId * @param {{ companyId: number, googleAuthId: number }} meta */ constructor(conversationId, { companyId, googleAuthId }) { this.conversationId = conversationId; this.companyId = companyId; this.googleAuthId = googleAuthId; /** @type {Set<{ res: import('express').Response, keepAlive: NodeJS.Timeout }>} */ this.subscribers = new Set(); /** @type {LLMOrbiter|null} */ this.orbiter = null; this.destroyed = false; this.isProcessing = false; } // #region Subscription & Broadcast /** * Broadcast an SSE event to all subscribers of this conversation. * @param {string} event * @param {object} data */ broadcast(event, data) { if (this.destroyed) return; const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; for (const client of this.subscribers) { if (!client.res.writableEnded && !client.res.destroyed) { client.res.write(payload); } } } /** * Add an SSE subscriber for this conversation. * @param {import('express').Response} res * @param {import('express').Request} [req] * @returns {() => void} unsubscribe */ subscribe(res, req) { req?.socket?.setTimeout(0); res.setTimeout?.(0); res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache, no-transform', 'Connection': 'keep-alive', 'X-Accel-Buffering': 'no', }); res.flushHeaders?.(); res.write(': ping\n\n'); const client = { res, keepAlive: setInterval(() => { if (res.writableEnded || res.destroyed) return; res.write(': heartbeat\n\n'); }, KEEP_ALIVE_MS), }; this.subscribers.add(client); const unsubscribe = () => { clearInterval(client.keepAlive); this.subscribers.delete(client); }; res.on('close', unsubscribe); req?.on('close', unsubscribe); return unsubscribe; } // #endregion // #region Attachments /** * Saves attachments sent with a user message to the company's VFS * under `attachments/` and registers them in `modelvfs`. * * @param {number} googleAuthId * @param {Array} attachments * @returns {Promise>} Array of `[ATTACHMENT name= path= type=]` tags */ async saveAttachmentsToVfs(googleAuthId, attachments) { if (!Array.isArray(attachments) || attachments.length === 0) return []; let vfsRoot; try { vfsRoot = await prepareVfs({ googleAuthId }); } catch (err) { console.error(`[ConversationManager ${this.conversationId}] prepareVfs error:`, err); return []; } const store = await getStore(); const attachmentsDir = resolve(vfsRoot, 'attachments'); try { mkdirSync(attachmentsDir, { recursive: true }); } catch (err) { console.error(`[ConversationManager ${this.conversationId}] failed to create attachments directory:`, err); } const tags = []; const usedNames = new Set(); for (const att of attachments) { if (!att || typeof att !== 'object') continue; const rawName = (att.name || att.filename || `attachment-${Date.now()}`).replace(/^.*[\\\/]/, ''); const cleanName = rawName.replace(/[\0\r\n"]/g, '').trim() || 'attachment'; let finalName = cleanName; let counter = 1; const ext = extname(cleanName); const base = ext ? cleanName.slice(0, -ext.length) : cleanName; while (usedNames.has(finalName.toLowerCase())) { counter++; finalName = `${base}_${counter}${ext}`; } usedNames.add(finalName.toLowerCase()); let fileBuffer = null; if (att.dataUrl && typeof att.dataUrl === 'string') { const commaIdx = att.dataUrl.indexOf(','); if (commaIdx !== -1) { const metaPart = att.dataUrl.slice(0, commaIdx); const dataPart = att.dataUrl.slice(commaIdx + 1); if (metaPart.includes(';base64')) { fileBuffer = Buffer.from(dataPart, 'base64'); } else { fileBuffer = Buffer.from(decodeURIComponent(dataPart), 'utf8'); } } } if (!fileBuffer && att.textContent != null) { fileBuffer = Buffer.from(String(att.textContent), 'utf8'); } if (!fileBuffer && att.content != null) { fileBuffer = Buffer.isBuffer(att.content) ? att.content : Buffer.from(String(att.content)); } let mime = att.mimeType || att.type || ''; if (!mime && att.dataUrl && typeof att.dataUrl === 'string') { const match = att.dataUrl.match(/^data:([^;,]+)/); if (match) mime = match[1]; } if (!mime) { mime = getMimeType(finalName); } const vfsRelPath = `attachments/${finalName}`; const diskPath = resolve(vfsRoot, vfsRelPath); if (fileBuffer) { try { writeFileSync(diskPath, fileBuffer); } catch (err) { console.error(`[ConversationManager ${this.conversationId}] failed to write attachment disk file:`, err); } } const formatAttr = (val) => String(val).includes(' ') ? `"${val}"` : String(val); tags.push(`[ATTACHMENT name=${formatAttr(finalName)} path=${formatAttr(vfsRelPath)} type=${formatAttr(mime)}]`); } return tags; } // #endregion // #region Message Handlers /** * Handle an incoming user message. * 1. Extracts text content and saves any attachments into the company VFS (`attachments/`). * 2. Prefixes user message with `[ATTACHMENT name= path= type=]` tags. * 3. Persists the user message (DB refuses unless `idle`). * 4. Broadcasts it, marks the conversation `busy` (DB) and `thinking` (SSE). * 5. Starts the LLM run in the background; the HTTP request returns immediately. * * @param {number} googleAuthId * @param {{ message?: any, attachments?: Array, content?: any }} input * @returns {Promise} The created user message * @throws {Error} 'Conversation is not idle' while a run is active. */ async handleUserMessage(googleAuthId, { message, attachments, content }) { if (this.isProcessing) throw new Error('Conversation is not idle'); this.isProcessing = true; let userMsg; try { const parsed = parseIncomingUserMessage({ message, attachments, content }); if (!parsed.content && parsed.attachments.length === 0) { throw new Error('Message required'); } let finalMessage = parsed.content; if (parsed.attachments.length > 0) { const tags = await this.saveAttachmentsToVfs(googleAuthId, parsed.attachments); if (tags.length > 0) { const prefix = tags.join(' '); finalMessage = parsed.content ? `${prefix}\n\n${parsed.content}` : prefix; } } const payload = JSON.stringify({ content: finalMessage }); userMsg = await db.addMessage(googleAuthId, this.conversationId, { message: payload, role: 'user' }); await db.setState(googleAuthId, this.conversationId, 'busy'); } catch (err) { this.isProcessing = false; throw err; } this.broadcast('state', { state: 'thinking' }); emitCompanyEvent('modelconversation', { companyId: this.companyId }); this.runGeneration(googleAuthId).catch((err) => { console.error(`[ConversationManager ${this.conversationId}] unexpected generation error:`, err); }); return userMsg; } /** * Handle an incoming tool response message (e.g. from the user answering ask_user). * 1. Persists the tool response message (DB refuses unless `idle`). * 2. Broadcasts it, marks the conversation `busy` (DB) and `working` (SSE). * 3. Starts the LLM run in the background; the HTTP request returns immediately. * * @param {number} googleAuthId * @param {{ tool_call_id: string, content?: any, answer?: any }} input * @returns {Promise} The created tool message * @throws {Error} 'Conversation is not idle' while a run is active. */ async handleToolMessage(googleAuthId, { tool_call_id, content, answer, name }) { if (!tool_call_id) throw new Error('tool_call_id is required'); if (this.isProcessing) throw new Error('Conversation is not idle'); this.isProcessing = true; let toolMsg; try { const rawContent = content !== undefined ? (typeof content === 'string' ? content : JSON.stringify(content)) : JSON.stringify({ answer: answer ?? '' }); const payload = JSON.stringify({ content: rawContent, tool_call_id, name: name || 'ask_user', }); toolMsg = await db.addMessage(googleAuthId, this.conversationId, { message: payload, role: 'tool' }); await db.setState(googleAuthId, this.conversationId, 'busy'); } catch (err) { this.isProcessing = false; throw err; } this.broadcast('state', { state: 'working' }); emitCompanyEvent('modelconversation', { companyId: this.companyId }); this.runGeneration(googleAuthId).catch((err) => { console.error(`[ConversationManager ${this.conversationId}] unexpected generation error:`, err); }); return toolMsg; } // #endregion // #region Run /** * Run the LLM over the stored transcript, stream the answer over SSE, * persist reasoning, tool calls, and assistant messages, then return to `idle`. * @param {number} googleAuthId */ async runGeneration(googleAuthId) { let error = null; try { const conversation = await db.get(googleAuthId, this.conversationId); if (conversation.provider !== 'openrouter') { throw new Error(`Unsupported provider: ${conversation.provider}`); } // Fall back to the configured default when the stored model is missing, blank, or // retired. If the stored model was defaulted to the free model and the company now has // a custom key configured, transition to the configured paid default model. const defaultModel = await resolveDefaultModel(googleAuthId); const hasCustomKey = await hasConfiguredOpenRouterKey(googleAuthId); const storedModel = typeof conversation.model === 'string' ? conversation.model.trim() : ''; const model = (!storedModel || RETIRED_MODELS.has(storedModel) || (storedModel === env.openrouterFreeModel && (hasCustomKey || defaultModel !== env.openrouterFreeModel))) ? defaultModel : storedModel; const apiKey = await resolveOpenRouterKey(googleAuthId); // Transcript comes from the database (includes user, assistant, tool, reasoning). const history = await db.listMessages(googleAuthId, this.conversationId); // Resolve the VFS root so the prompt can list the company's SKILL files. A failure // here must not abort the turn — the prompt simply falls back to "no skills". let vfsRoot = null; try { vfsRoot = await resolveVfsRoot(googleAuthId); } catch (err) { console.warn('conversation: could not resolve VFS root for skill index:', err.message); } const messages = [ { role: 'system', content: await buildSystemPrompt(vfsRoot, googleAuthId) }, ...historyToLlmMessages(history), ]; let text = ''; let currentReasoning = ''; this.orbiter = new LLMOrbiter(new OpenRouterProvider({ apiKey, apiUrl: env.openrouterApiUrl }), { onStatus: (status, meta) => { this.broadcast('state', { state: status, ...(meta || {}) }); }, onReasoning: (reasoning) => { currentReasoning = reasoning; this.broadcast('reasoning', { text: reasoning }); }, onContent: (chunk) => { text += chunk; this.broadcast('delta', { delta: chunk, text }); }, onAssistantMessage: async ({ content, tool_calls, reasoning_details, reasoning }) => { const plain = (reasoning || currentReasoning.trim() || reasoning_details?.find((d) => d?.type === 'reasoning.summary')?.summary || '').trim(); const encrypted = reasoning_details?.find((d) => d?.type === 'reasoning.encrypted')?.data || ''; currentReasoning = ''; if (plain || encrypted) { try { await db.addMessage(googleAuthId, this.conversationId, { role: 'reasoning', message: JSON.stringify({ plain, encrypted }), }); } catch (err) { console.error(`[ConversationManager ${this.conversationId}] failed to save reasoning message:`, err); } } const trimmedContent = (content || text).trim(); const hasToolCalls = Array.isArray(tool_calls) && tool_calls.length > 0; if (trimmedContent || hasToolCalls) { try { await db.addMessage(googleAuthId, this.conversationId, { role: 'assistant', message: JSON.stringify({ content: trimmedContent, ...(hasToolCalls ? { tool_calls } : {}), }), }); text = ''; emitCompanyEvent('modelconversation', { companyId: this.companyId }); } catch (err) { console.error(`[ConversationManager ${this.conversationId}] failed to save assistant message:`, err); } } }, onToolCall: async ({ name, args, id, callItem }) => { const rawArgs = callItem?.function?.arguments || (typeof args === 'string' ? args : JSON.stringify(args || {})); console.log(`[conversation ${this.conversationId}] tool call: ${name} args=${rawArgs}`); this.broadcast('state', { state: 'toolcall', id: id || callItem?.id || '', name, arguments: rawArgs, content: rawArgs, }); }, onToolResult: async ({ name, result, id, callItem }) => { const rawArgs = callItem?.function?.arguments || ''; this.broadcast('state', { state: 'toolcall', id: id || callItem?.id || '', name, arguments: rawArgs, content: rawArgs, result, }); try { await db.addMessage(googleAuthId, this.conversationId, { role: 'tool', message: JSON.stringify({ name, parameters: rawArgs, content: typeof result === 'string' ? result : JSON.stringify(result), tool_call_id: id || callItem?.id || '', }), }); emitCompanyEvent('modelconversation', { companyId: this.companyId }); } catch (err) { console.error(`[ConversationManager ${this.conversationId}] failed to save tool result message:`, err); } }, onToolError: async ({ name, error, id, callItem }) => { console.error(`[ConversationManager ${this.conversationId}] Tool "${name}" errored:`, error); const rawArgs = callItem?.function?.arguments || ''; this.broadcast('state', { state: 'toolcall', id: id || callItem?.id || '', name, arguments: rawArgs, content: rawArgs, error: String(error), }); try { await db.addMessage(googleAuthId, this.conversationId, { role: 'tool', message: JSON.stringify({ name, parameters: rawArgs, content: `Tool error [${name}]: ${error}`, tool_call_id: id || callItem?.id || '', }), }); emitCompanyEvent('modelconversation', { companyId: this.companyId }); } catch (err) { console.error(`[ConversationManager ${this.conversationId}] failed to save tool error message:`, err); } }, onUsage: async (u) => { if (u) { this.broadcast('usage', { input: u.input ?? 0, cached: u.cached ?? 0, output: u.output ?? 0, ...(u.cost != null ? { cost: u.cost } : {}), }); try { await db.addMessage(googleAuthId, this.conversationId, { role: 'usage', message: JSON.stringify({ input: u.input ?? 0, cached: u.cached ?? 0, output: u.output ?? 0, ...(u.cost != null ? { cost: u.cost } : {}), }), }); emitCompanyEvent('modelconversation', { companyId: this.companyId }); } catch (err) { console.error(`[ConversationManager ${this.conversationId}] failed to save usage message:`, err); } } // A call that consumed 0 input tokens AND produced 0 output means the // provider returned nothing at all — not a real generation. LLMOrbiter // treats "no content and no tool calls" as "keep looping", so several of // these in a row burn iterations for nothing. Flag them loudly so the // cause is visible instead of showing up as a silent weird loop. if (u.input === 0 && u.output === 0) { console.warn( `[conversation ${this.conversationId}] EMPTY LLM RESPONSE ` + `(iteration reported no usage at all). The provider returned no content ` + `and no tool calls, so the orchestrator will retry. ` + `If this repeats, the upstream model/API is failing — check the model id ` + `and the key.` ); } console.log( `[conversation ${this.conversationId}] usage: input=${u.input}, cached=${u.cached}, output=${u.output}` + (u.cost != null ? `, cost=$${u.cost.toFixed(4)}` : '') ); }, onLLMCall: (call) => { console.log( `[conversation ${this.conversationId}] LLM call #${call.iteration}: ` + `model=${call.model}, messages=${call.messageCount}, flow=${call.messages}, ` + `promptChars=${call.promptSize}` ); }, }); const reasoningEffort = await resolveModelReasoning(googleAuthId); const extraBody = reasoningEffort ? { reasoning: { effort: reasoningEffort } } : {}; const result = await this.orbiter.run({ messages, model, extraBody, tools: [ createDashbackRequestTool({ googleAuthId }), createAskUserTool(), // VFS tools take no companyId on purpose — each resolves the user's // currently selected company itself, so the on-disk VFS and the database // can never end up pointed at different companies. createJavascriptExecuteTool({ googleAuthId }), createFindFilesTool({ googleAuthId }), createReadFileTool({ googleAuthId }), createPatchTool({ googleAuthId }), createRmTool({ googleAuthId }), createSearchImagesTool({ googleAuthId }), createAnalyzeImageTool({ googleAuthId, model }), createEditImageTool({ googleAuthId, model }), createWebSearchTool({ googleAuthId }), createSearchWebPriceTool({ googleAuthId }), createUrlGetTool({ googleAuthId }), createPdfReadTool({ googleAuthId }), createResetVfsTool({ googleAuthId }), createApiSuggestionsOrIssuesTool({ googleAuthId, companyId: this.companyId }), createSearchArticlesTool({ googleAuthId, companyId: this.companyId }), createEanChecksumTool(), createNetGrossCalcTool(), createSetUserRouteTool({ onRoute: (route) => { this.broadcast('route', { route }); }, }), createDraftDeliveryManifestTool(), createSubagentTool({ googleAuthId, companyId: this.companyId, conversationId: this.conversationId, parentModel: model, depth: 0, }), ], }); if (this.destroyed) return; // Log the shape of every run, including degenerate ones. When nothing comes back // there is nothing persisted and the UI just sits there — this line is the only // clue, so make it explicit about which field was empty. console.log( `[conversation ${this.conversationId}] run finished: ` + `iterations=${result?.iterations}, ` + `content=${result?.content ? `${result.content.length} chars` : 'EMPTY'}, ` + `toolCalls=${result?.toolResults ? Object.keys(result.toolResults).length : 0}, ` + `directReturn=${result?.directReturn === true}` ); // Direct return (e.g. from ask_user) halts without adding an additional assistant message. if (result?.directReturn) { console.log( `[conversation ${this.conversationId}] direct return halted loop: ` + JSON.stringify(result.data) ); return; } const content = (result.content ?? '').trim(); const finalReasoningDetails = (Array.isArray(result.reasoning_details) && result.reasoning_details.length > 0) ? result.reasoning_details : undefined; const finalReasoning = result.reasoning || currentReasoning.trim() || undefined; // Nothing came back at all: no content, no reasoning. Surface it to the user rather // than silently returning to `idle` with an empty turn, which reads as a hang. if (!content && !finalReasoningDetails && !finalReasoning) { console.error( `[conversation ${this.conversationId}] generation produced no content after ` + `${result?.iterations ?? 0} iteration(s). The model returned neither text nor tool calls.` ); error = 'The model returned an empty response. Please try again.'; } } catch (err) { if (this.destroyed) return; error = err.message; console.error(`[ConversationManager ${this.conversationId}] generation failed:`, err); } finally { this.orbiter = null; this.isProcessing = false; } try { await db.setState(googleAuthId, this.conversationId, 'idle'); } catch (err) { console.error(`[ConversationManager ${this.conversationId}] failed to reset state:`, err); } emitCompanyEvent('modelconversation', { companyId: this.companyId }); this.broadcast('state', error ? { state: 'done', error } : { state: 'done' }); } // #endregion // #region Teardown destroy() { this.destroyed = true; this.orbiter?.abort('Conversation manager closed'); for (const client of this.subscribers) { clearInterval(client.keepAlive); if (!client.res.writableEnded && !client.res.destroyed) { client.res.end(); } } this.subscribers.clear(); } // #endregion } // #endregion // #region Manager Registry /** @type {Map} */ const activeManagers = new Map(); /** * Get or create an active ConversationManager for a conversation. * @param {number} conversationId * @param {{ companyId: number, googleAuthId: number }} meta * @returns {ConversationManager} */ export function getOrCreateManager(conversationId, { companyId, googleAuthId }) { let manager = activeManagers.get(conversationId); if (!manager) { manager = new ConversationManager(conversationId, { companyId, googleAuthId }); activeManagers.set(conversationId, manager); } else { if (companyId) manager.companyId = companyId; if (googleAuthId) manager.googleAuthId = googleAuthId; } return manager; } /** * Get an active ConversationManager if one exists. * @param {number} conversationId * @returns {ConversationManager|undefined} */ export function getActiveManager(conversationId) { return activeManagers.get(conversationId); } /** * Close, destroy, and remove an active ConversationManager if one exists. * @param {number} conversationId */ export function removeManager(conversationId) { const manager = activeManagers.get(conversationId); if (manager) { manager.destroy(); activeManagers.delete(conversationId); } } /** * Close and destroy all active conversation managers. */ export function closeAllManagers() { for (const [id, manager] of activeManagers.entries()) { manager.destroy(); activeManagers.delete(id); } } // #endregion