import { LLMOrbiter, OpenRouterProvider } from 'libllmorbiter'; import * as db from '../db/modelconversation.js'; import env from '../config/env.js'; import { emit as emitCompanyEvent } from '../api/sse.js'; import { mkdir, writeFile } from 'node:fs/promises'; import { resolve, extname } from 'node:path'; 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 & Precompiled Helpers const KEEP_ALIVE_MS = 20_000; const ISO_NO_TZ_RE = /^\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}/; const HAS_TZ_RE = /[+-]\d{2}(:?\d{2})?$/; const TIME_PREFIX_RE = /^\s*\[\d{2}:\d{2}[A-Za-z]+\]\s*/; const DATA_MIME_RE = /^data:([^;,]+)/; const PATH_PREFIX_RE = /^.*[\\/]/; const UNSAFE_NAME_CHARS_RE = /[\0\r\n"]/g; // Reuse a single Intl.DateTimeFormat instance instead of re-instantiating per message const BERLIN_DATE_FORMATTER = new Intl.DateTimeFormat('en-GB', { timeZone: 'Europe/Berlin', hour: '2-digit', minute: '2-digit', hour12: false, timeZoneName: 'short', }); // #endregion // #region Message Helpers 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(); const normalized = (ISO_NO_TZ_RE.test(s) && !s.endsWith('Z') && !HAS_TZ_RE.test(s)) ? `${s.replace(' ', 'T')}Z` : s; const parsed = new Date(normalized); if (!isNaN(parsed.getTime())) return parsed; } return new Date(); } export function formatTimePrefix(createdDate) { const parts = BERLIN_DATE_FORMATTER.formatToParts(parseUtcDate(createdDate)); let hour = '00'; let minute = '00'; let tz = 'MEST'; for (const p of parts) { if (p.type === 'hour') hour = p.value.padStart(2, '0'); else if (p.type === 'minute') minute = p.value.padStart(2, '0'); else if (p.type === 'timeZoneName') { tz = p.value === 'CEST' ? 'MEST' : p.value === 'CET' ? 'MET' : p.value; } } return `[${hour}:${minute}${tz}] `; } function tryParseJson(str) { if (typeof str !== 'string') return null; const t = str.trim(); if ((t.startsWith('{') && t.endsWith('}')) || (t.startsWith('[') && t.endsWith(']'))) { try { return JSON.parse(t); } catch { return null; } } return null; } function sanitizeAttachmentName(raw, fallback = 'attachment') { if (!raw || typeof raw !== 'string') return fallback; return raw.replace(PATH_PREFIX_RE, '').replace(UNSAFE_NAME_CHARS_RE, '').trim() || fallback; } function formatAttachmentTag(name, vfsRelPath, mime) { const q = (val) => { const s = String(val); return s.includes(' ') ? `"${s}"` : s; }; return `[ATTACHMENT name=${q(name)} path=${q(vfsRelPath)} type=${q(mime)}]`; } export function parseIncomingUserMessage({ message, attachments: directAttachments, content: directContent } = {}) { let text = ''; const attachments = Array.isArray(directAttachments) ? [...directAttachments] : []; 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)) { const rawInner = typeof candidate.content === 'string' ? candidate.content : typeof candidate.message === 'string' ? candidate.message : null; if (rawInner !== null) { const innerParsed = tryParseJson(rawInner); if (innerParsed && typeof innerParsed === 'object' && !Array.isArray(innerParsed)) { if (typeof innerParsed.content === 'string') text = innerParsed.content; if (Array.isArray(innerParsed.attachments)) attachments.push(...innerParsed.attachments); } else { text = rawInner; } } 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.trimStart().startsWith('{') || rawContent.includes('{"content"'))) { try { const parsed = parseIncomingUserMessage({ message: rawContent }); if (parsed.attachments.length > 0) { const tags = parsed.attachments.map((att) => { const name = sanitizeAttachmentName(att.name || att.filename); const mime = att.mimeType || att.type || getMimeType(name); return formatAttachmentTag(name, `attachments/${name}`, mime); }); const prefix = tags.join(' '); return parsed.content ? `${prefix}\n\n${parsed.content}` : prefix; } if (parsed.content) return parsed.content; } catch { // Fall through to rawContent } } return rawContent; } export function historyToLlmMessages(history) { const messages = []; for (const m of history) { if (m.role === 'usage' || m.role === 'reasoning') continue; let payload; if (typeof m.message === 'string') { try { payload = JSON.parse(m.message); } catch { payload = { content: m.message }; } } else { payload = m.message || {}; } if (m.role === 'user') { const rawContent = payload.content ?? (typeof m.message === 'string' ? m.message : ''); const cleaned = cleanUserMessageContent(rawContent); const content = TIME_PREFIX_RE.test(cleaned) ? cleaned : `${formatTimePrefix(m.created_at)}${cleaned}`; messages.push({ role: 'user', content }); } else if (m.role === 'assistant') { const msg = { role: 'assistant', content: payload.content ?? null }; if (Array.isArray(payload.tool_calls) && payload.tool_calls.length > 0) { msg.tool_calls = payload.tool_calls; } messages.push(msg); } else if (m.role === 'tool') { const toolCallId = payload.tool_call_id ?? ''; const toolName = payload.name ?? ''; const toolParams = payload.parameters ?? payload.arguments ?? '{}'; const callEntry = { id: toolCallId, type: 'function', function: { name: toolName, arguments: typeof toolParams === 'string' ? toolParams : JSON.stringify(toolParams), }, }; const prevMsg = messages[messages.length - 1]; if (!prevMsg || prevMsg.role !== 'assistant' || !prevMsg.tool_calls) { messages.push({ role: 'assistant', content: null, tool_calls: [callEntry] }); } else if (!prevMsg.tool_calls.some((tc) => tc.id === toolCallId)) { prevMsg.tool_calls.push(callEntry); } 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} */ this.subscribers = new Set(); /** @type {NodeJS.Timeout|null} */ this.keepAliveTimer = null; /** @type {LLMOrbiter|null} */ this.orbiter = null; this.destroyed = false; this.isProcessing = false; } // #region Subscription & Broadcast broadcast(event, data) { if (this.destroyed || this.subscribers.size === 0) return; const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; for (const res of this.subscribers) { if (!res.writableEnded && !res.destroyed) { res.write(payload); } } } 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'); this.subscribers.add(res); // Single shared heartbeat timer per conversation instead of one interval per connected client if (!this.keepAliveTimer) { this.keepAliveTimer = setInterval(() => { for (const sub of this.subscribers) { if (!sub.writableEnded && !sub.destroyed) { sub.write(': heartbeat\n\n'); } } }, KEEP_ALIVE_MS); } let cleanedUp = false; const unsubscribe = () => { if (cleanedUp) return; cleanedUp = true; this.subscribers.delete(res); if (this.subscribers.size === 0 && this.keepAliveTimer) { clearInterval(this.keepAliveTimer); this.keepAliveTimer = null; } }; res.once('close', unsubscribe); req?.once('close', unsubscribe); return unsubscribe; } // #endregion // #region Attachments 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 attachmentsDir = resolve(vfsRoot, 'attachments'); try { await mkdir(attachmentsDir, { recursive: true }); } catch (err) { console.error(`[ConversationManager ${this.conversationId}] failed to create attachments directory:`, err); } const tags = []; const usedNames = new Set(); const writePromises = []; for (const att of attachments) { if (!att || typeof att !== 'object') continue; const cleanName = sanitizeAttachmentName(att.name || att.filename, `attachment-${Date.now()}`); 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 (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); fileBuffer = metaPart.includes(';base64') ? Buffer.from(dataPart, 'base64') : Buffer.from(decodeURIComponent(dataPart), 'utf8'); } } else if (att.textContent != null) { fileBuffer = Buffer.from(String(att.textContent), 'utf8'); } else if (att.content != null) { fileBuffer = Buffer.isBuffer(att.content) ? att.content : Buffer.from(String(att.content)); } let mime = att.mimeType || att.type || ''; if (!mime && typeof att.dataUrl === 'string') { mime = att.dataUrl.match(DATA_MIME_RE)?.[1] || ''; } if (!mime) mime = getMimeType(finalName); const vfsRelPath = `attachments/${finalName}`; if (fileBuffer) { const diskPath = resolve(vfsRoot, vfsRelPath); writePromises.push( writeFile(diskPath, fileBuffer).catch((err) => { console.error(`[ConversationManager ${this.conversationId}] failed to write attachment disk file:`, err); }) ); } tags.push(formatAttachmentTag(finalName, vfsRelPath, mime)); } if (writePromises.length > 0) { await Promise.all(writePromises); } return tags; } // #endregion // #region Message Handlers async _enqueueAndRun(googleAuthId, role, payloadObj, initialSseState) { let savedMsg; try { const payload = JSON.stringify(payloadObj); savedMsg = await db.addMessage(googleAuthId, this.conversationId, { message: payload, role }); await db.setState(googleAuthId, this.conversationId, 'busy'); } catch (err) { this.isProcessing = false; throw err; } this.broadcast('state', { state: initialSseState }); emitCompanyEvent('modelconversation', { companyId: this.companyId }); this.runGeneration(googleAuthId).catch((err) => { console.error(`[ConversationManager ${this.conversationId}] unexpected generation error:`, err); }); return savedMsg; } async handleUserMessage(googleAuthId, { message, attachments, content }) { if (this.isProcessing) throw new Error('Conversation is not idle'); this.isProcessing = true; let finalMessage; try { const parsed = parseIncomingUserMessage({ message, attachments, content }); if (!parsed.content && parsed.attachments.length === 0) { throw new Error('Message required'); } 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; } } } catch (err) { this.isProcessing = false; throw err; } return this._enqueueAndRun(googleAuthId, 'user', { content: finalMessage }, 'thinking'); } 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; const rawContent = content !== undefined ? (typeof content === 'string' ? content : JSON.stringify(content)) : JSON.stringify({ answer: answer ?? '' }); return this._enqueueAndRun( googleAuthId, 'tool', { content: rawContent, tool_call_id, name: name || 'ask_user' }, 'working' ); } // #endregion // #region Run async _persistMessageSafe(googleAuthId, role, payloadObj, emitEvent = false) { try { await db.addMessage(googleAuthId, this.conversationId, { role, message: JSON.stringify(payloadObj), }); if (emitEvent) { emitCompanyEvent('modelconversation', { companyId: this.companyId }); } } catch (err) { console.error(`[ConversationManager ${this.conversationId}] failed to save ${role} message:`, err); } } buildTools(googleAuthId, model) { return [ createDashbackRequestTool({ googleAuthId }), createAskUserTool(), 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, }), ]; } async runGeneration(googleAuthId) { let error = null; try { // Parallelize all independent DB, config, and VFS lookups const [ conversation, defaultModel, hasCustomKey, apiKey, history, reasoningEffort, vfsRoot, ] = await Promise.all([ db.get(googleAuthId, this.conversationId), resolveDefaultModel(googleAuthId), hasConfiguredOpenRouterKey(googleAuthId), resolveOpenRouterKey(googleAuthId), db.listMessages(googleAuthId, this.conversationId), resolveModelReasoning(googleAuthId), resolveVfsRoot(googleAuthId).catch((err) => { console.warn('conversation: could not resolve VFS root for skill index:', err.message); return null; }), ]); if (conversation.provider !== 'openrouter') { throw new Error(`Unsupported provider: ${conversation.provider}`); } const storedModel = typeof conversation.model === 'string' ? conversation.model.trim() : ''; const shouldFallback = !storedModel || RETIRED_MODELS.has(storedModel) || (storedModel === env.openrouterFreeModel && (hasCustomKey || defaultModel !== env.openrouterFreeModel)); const model = shouldFallback ? defaultModel : storedModel; const systemPrompt = await buildSystemPrompt(vfsRoot, googleAuthId); const messages = [ { role: 'system', content: systemPrompt }, ...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) { await this._persistMessageSafe(googleAuthId, 'reasoning', { plain, encrypted }); } const trimmedContent = (content || text).trim(); const hasToolCalls = Array.isArray(tool_calls) && tool_calls.length > 0; if (trimmedContent || hasToolCalls) { text = ''; await this._persistMessageSafe( googleAuthId, 'assistant', { content: trimmedContent, ...(hasToolCalls ? { tool_calls } : {}) }, true ); } }, 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 || ''; const toolCallId = id || callItem?.id || ''; this.broadcast('state', { state: 'toolcall', id: toolCallId, name, arguments: rawArgs, content: rawArgs, result, }); await this._persistMessageSafe( googleAuthId, 'tool', { name, parameters: rawArgs, content: typeof result === 'string' ? result : JSON.stringify(result), tool_call_id: toolCallId, }, true ); }, onToolError: async ({ name, error: toolErr, id, callItem }) => { console.error(`[ConversationManager ${this.conversationId}] Tool "${name}" errored:`, toolErr); const rawArgs = callItem?.function?.arguments || ''; const toolCallId = id || callItem?.id || ''; this.broadcast('state', { state: 'toolcall', id: toolCallId, name, arguments: rawArgs, content: rawArgs, error: String(toolErr), }); await this._persistMessageSafe( googleAuthId, 'tool', { name, parameters: rawArgs, content: `Tool error [${name}]: ${toolErr}`, tool_call_id: toolCallId, }, true ); }, onUsage: async (u) => { if (!u) return; const usagePayload = { input: u.input ?? 0, cached: u.cached ?? 0, output: u.output ?? 0, ...(u.cost != null ? { cost: u.cost } : {}), }; this.broadcast('usage', usagePayload); await this._persistMessageSafe(googleAuthId, 'usage', usagePayload, true); 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.` ); } 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 result = await this.orbiter.run({ messages, model, extraBody: reasoningEffort ? { reasoning: { effort: reasoningEffort } } : {}, tools: this.buildTools(googleAuthId, model), }); if (this.destroyed) return; 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}` ); if (result?.directReturn) { console.log(`[conversation ${this.conversationId}] direct return halted loop: ${JSON.stringify(result.data)}`); return; } const hasContent = Boolean(result.content?.trim()); const hasReasoning = Boolean( (Array.isArray(result.reasoning_details) && result.reasoning_details.length > 0) || result.reasoning || currentReasoning.trim() ); if (!hasContent && !hasReasoning) { console.error( `[conversation ${this.conversationId}] generation produced no content after ` + `${result?.iterations ?? 0} iteration(s).` ); 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'); if (this.keepAliveTimer) { clearInterval(this.keepAliveTimer); this.keepAliveTimer = null; } for (const res of this.subscribers) { if (!res.writableEnded && !res.destroyed) { res.end(); } } this.subscribers.clear(); } // #endregion } // #endregion // #region Manager Registry /** @type {Map} */ const activeManagers = new Map(); 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; } export function getActiveManager(conversationId) { return activeManagers.get(conversationId); } export function removeManager(conversationId) { const manager = activeManagers.get(conversationId); if (manager) { manager.destroy(); activeManagers.delete(conversationId); } } export function closeAllManagers() { for (const manager of activeManagers.values()) { manager.destroy(); } activeManagers.clear(); } // #endregion