u

src/services/conversationManager.js

@@ -0,0 +1,521 @@
+import { LLMOrbiter, OpenRouterProvider } from 'libllmorbiter';
+import * as db from '../db/modelconversation.js';
+import * as dbCompanySetting from '../db/companysetting.js';
+import env from '../config/env.js';
+import { emit as emitCompanyEvent } from '../api/sse.js';
+import { readFileSync } from 'fs';
+import { resolve, dirname } from 'path';
+import { fileURLToPath } from 'url';
+import { randomUUID } from 'crypto';
+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';
+
+const KEEP_ALIVE_MS = 20_000;
+
+const ROUTES_DOC_PATH = resolve(dirname(fileURLToPath(import.meta.url)), '../../docs/api/routes.md');
+
+/**
+ * System prompt: base instructions plus the generated API routes reference
+ * (docs/api/routes.md), read on each run so it stays current.
+ * @returns {string}
+ */
+function buildSystemPrompt() {
+    let routesDoc = '';
+    try {
+        routesDoc = readFileSync(ROUTES_DOC_PATH, 'utf8').trim();
+    } catch (err) {
+        console.error('conversation: failed to read routes doc:', err);
+    }
+    const today = new Date().toISOString().split('T')[0];
+    return [
+        'You are an intelligent assistant specialized in exploring, querying, and managing data from the Dashback REST API.',
+        `Today's date: ${today}`,
+        '',
+        'YOUR TOOLS:',
+        '1. dashback_request: Execute HTTP requests (GET, POST, PUT, PATCH, DELETE) against Dashback endpoints.',
+        '2. ask_user: Ask clarifying questions or request decisions from the user if their intent is ambiguous or requires choices. When you call ask_user, execution pauses immediately and waits for the user to provide their response.',
+        '3. javascript_execute: Run JavaScript snippets in a sandbox to compute, aggregate, transform, or analyze data. Assign module.exports to return a value. With allow_network: true, snippets can directly call Dashback API routes using fetch() (e.g. await fetch("/article/2")) and search images using search_for_images(query). A sandboxed virtual filesystem (fs / vfs) is available to store and retrieve artifacts across runs.',
+        '4. reset_vfs: Delete all files, images, and artifacts stored in the virtual filesystem (VFS) to free up storage or clean up.',
+        '',
+        'OPERATING GUIDELINES:',
+        '- Refer to the DASHBACK API ROUTES documentation below to find appropriate endpoints, parameters, and HTTP methods.',
+        '- SELLING PRICES: In article responses (GET /article/{sku} and returned article), the selling price field is named net (NOT price). The structure of entries in article.prices is { minmultiplier, net, taxRate, customergroupId }.',
+        '- GROSS vs NET PRICES: Users almost always mean GROSS (brutto, incl. VAT) prices when they specify a price (e.g. "1,60 €"), but the API stores and accepts NET prices only. Convert: net = gross / (1 + taxRate/100). Send net with high precision (at least 4–6 decimal places) so rounding back gives the exact gross. Calculate this in javascript_execute.',
+        '- DATA TYPES & WIRE FORMAT: Dashback strictly validates types (numeric values like weight, price, quantity, IDs must be numbers; flags must be booleans).',
+        '- VIRTUAL FILESYSTEM & ARTIFACT PERSISTENCE: Inside javascript_execute, you have access to a persistent virtual filesystem via globals fs and vfs. Artifacts persist across runs for the company.',
+        '- LABEL TEMPLATES (/labeltemplate): Label templates are NOT ZPL. content is Dashback\'s JSON label schema: { "objects": [ ... ] }. Do NOT write or suggest ZPL.',
+        '- Format responses clearly using markdown tables, bullet lists, or bold key-value pairs where appropriate.',
+        ...(routesDoc ? ['', 'DASHBACK API ROUTES:', routesDoc] : []),
+    ].join('\n');
+}
+
+/**
+ * Resolve the OpenRouter API key: company setting `keys.openrouter`, then the
+ * server-wide `OPENROUTER_FREE_KEY` (same order as delivery parsing).
+ * @param {number} googleAuthId
+ * @returns {Promise<string>}
+ */
+async function resolveOpenRouterKey(googleAuthId) {
+    let apiKey;
+    try {
+        const setting = await dbCompanySetting.getByKey(googleAuthId, 'keys.openrouter', { maskWriteonly: false });
+        apiKey = setting?.value?.trim();
+    } catch (err) {
+        console.error('conversation: failed to read OpenRouter key:', err);
+    }
+    if (!apiKey) apiKey = env.openrouterFreeKey?.trim();
+    if (!apiKey) throw new Error('OpenRouter API key not configured (companysetting: keys.openrouter)');
+    return apiKey;
+}
+
+/**
+ * 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.
+ */
+export function toLlmMessage(m) {
+    let payload = {};
+    if (typeof m.message === 'string') {
+        try {
+            payload = JSON.parse(m.message);
+        } catch {
+            payload = { content: m.message };
+        }
+    } else if (m.message && typeof m.message === 'object') {
+        payload = m.message;
+    }
+
+    if (m.role === 'user') {
+        return { role: 'user', content: payload.content ?? (typeof m.message === 'string' ? m.message : '') };
+    }
+    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;
+        }
+        if (Array.isArray(payload.reasoning_details) && payload.reasoning_details.length > 0) {
+            msg.reasoning_details = payload.reasoning_details;
+        }
+        if (payload.reasoning) {
+            msg.reasoning = payload.reasoning;
+        } else if (payload.reasoning_content) {
+            msg.reasoning = payload.reasoning_content;
+        }
+        return msg;
+    }
+    if (m.role === 'tool') {
+        return {
+            role: 'tool',
+            tool_call_id: payload.tool_call_id ?? '',
+            content: typeof payload.content === 'string' ? payload.content : JSON.stringify(payload.content ?? ''),
+            ...(payload.name ? { name: payload.name } : {}),
+        };
+    }
+    return { role: m.role, content: payload.content ?? m.message };
+}
+
+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;
+    }
+
+    /**
+     * 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;
+    }
+
+    /**
+     * Handle an incoming user message.
+     * 1. Persists the user message (DB refuses unless `idle`).
+     * 2. Broadcasts it, marks the conversation `busy` (DB) and `thinking` (SSE).
+     * 3. Starts the LLM run in the background; the HTTP request returns immediately.
+     *
+     * @param {number} googleAuthId
+     * @param {{ message: string }} input
+     * @returns {Promise<object>} The created user message
+     * @throws {Error} 'Conversation is not idle' while a run is active.
+     */
+    async handleUserMessage(googleAuthId, { message }) {
+        if (this.isProcessing) throw new Error('Conversation is not idle');
+        this.isProcessing = true;
+
+        let userMsg;
+        try {
+            const rawContent = typeof message === 'string' ? message : (message?.content || '');
+            const payload = JSON.stringify({ content: rawContent });
+            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('message', userMsg);
+        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<object>} The created tool message
+     * @throws {Error} 'Conversation is not idle' while a run is active.
+     */
+    async handleToolMessage(googleAuthId, { tool_call_id, content, answer }) {
+        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 });
+            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('message', toolMsg);
+        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;
+    }
+
+
+    /**
+     * 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 usage = null;
+        let error = null;
+
+        try {
+            const conversation = await db.get(googleAuthId, this.conversationId);
+            if (conversation.provider !== 'openrouter') {
+                throw new Error(`Unsupported provider: ${conversation.provider}`);
+            }
+            const apiKey = await resolveOpenRouterKey(googleAuthId);
+
+            // Transcript comes from the database (includes user, assistant, tool, reasoning).
+            const history = await db.listMessages(googleAuthId, this.conversationId);
+            const messages = [
+                { role: 'system', content: buildSystemPrompt() },
+                ...history.map(toLlmMessage),
+            ];
+
+            let working = false;
+            let text = '';
+            let currentReasoning = '';
+            let savedToolCallsTurn = null;
+
+            const markWorking = () => {
+                if (working) return;
+                working = true;
+                this.broadcast('state', { state: 'working' });
+            };
+
+            this.orbiter = new LLMOrbiter(new OpenRouterProvider({ apiKey }), {
+                onStatus: (status) => {
+                    if (status === 'working') markWorking();
+                },
+                onReasoning: (reasoning) => {
+                    markWorking();
+                    currentReasoning = reasoning;
+                    this.broadcast('reasoning', { text: reasoning });
+                },
+                onContent: (chunk) => {
+                    markWorking();
+                    text += chunk;
+                    this.broadcast('delta', { delta: chunk, text });
+                },
+                onToolCall: async ({ name, args, id, callItem, tool_calls, reasoning_details, reasoning, content }) => {
+                    markWorking();
+                    try {
+                        const effectiveToolCalls = tool_calls || (callItem ? [callItem] : [{
+                            id: id || randomUUID(),
+                            type: 'function',
+                            function: {
+                                name,
+                                arguments: typeof args === 'string' ? args : JSON.stringify(args || {}),
+                            },
+                        }]);
+
+                        if (savedToolCallsTurn !== effectiveToolCalls) {
+                            savedToolCallsTurn = effectiveToolCalls;
+                            const effectiveReasoningDetails = (Array.isArray(reasoning_details) && reasoning_details.length > 0)
+                                ? reasoning_details
+                                : undefined;
+                            const effectiveReasoning = reasoning || currentReasoning.trim() || undefined;
+                            currentReasoning = '';
+
+                            const assistantPayload = {
+                                content: (content || text).trim() || '',
+                                tool_calls: effectiveToolCalls,
+                                ...(effectiveReasoningDetails ? { reasoning_details: effectiveReasoningDetails } : {}),
+                                ...(effectiveReasoning ? { reasoning: effectiveReasoning } : {}),
+                            };
+
+                            const callMsg = await db.addMessage(googleAuthId, this.conversationId, {
+                                role: 'assistant',
+                                message: JSON.stringify(assistantPayload),
+                            });
+                            text = '';
+                            this.broadcast('message', callMsg);
+                            emitCompanyEvent('modelconversation', { companyId: this.companyId });
+                        }
+                    } catch (err) {
+                        console.error(`[ConversationManager ${this.conversationId}] failed to save tool call message:`, err);
+                    }
+                },
+                onToolResult: async ({ name, result, id, callItem }) => {
+                    try {
+                        const resMsg = await db.addMessage(googleAuthId, this.conversationId, {
+                            role: 'tool',
+                            message: JSON.stringify({
+                                content: typeof result === 'string' ? result : JSON.stringify(result),
+                                tool_call_id: id || callItem?.id || '',
+                            }),
+                        });
+                        this.broadcast('message', resMsg);
+                        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 }) => {
+                    try {
+                        const errMsg = await db.addMessage(googleAuthId, this.conversationId, {
+                            role: 'tool',
+                            message: JSON.stringify({
+                                content: `Tool error [${name}]: ${error}`,
+                                tool_call_id: id || callItem?.id || '',
+                            }),
+                        });
+                        this.broadcast('message', errMsg);
+                        emitCompanyEvent('modelconversation', { companyId: this.companyId });
+                    } catch (err) {
+                        console.error(`[ConversationManager ${this.conversationId}] failed to save tool error message:`, err);
+                    }
+                },
+                onUsage: (u) => {
+                    console.log(
+                        `[conversation ${this.conversationId}] usage: input=${u.input}, cached=${u.cached}, output=${u.output}` +
+                        (u.cost != null ? `, cost=$${u.cost.toFixed(4)}` : '')
+                    );
+                },
+            });
+
+            const result = await this.orbiter.run({
+                messages,
+                model: conversation.model,
+                tools: [
+                    createDashbackRequestTool({ googleAuthId }),
+                    createAskUserTool(),
+                    createJavascriptExecuteTool({ googleAuthId, companyId: this.companyId }),
+                    createResetVfsTool({ googleAuthId, companyId: this.companyId }),
+                ],
+            });
+
+            if (this.destroyed) return;
+
+            // Direct return (e.g. from ask_user) halts without adding an additional assistant message.
+            if (result?.directReturn) {
+                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;
+
+            if (content || finalReasoningDetails || finalReasoning) {
+                const assistantMsg = await db.addMessage(googleAuthId, this.conversationId, {
+                    message: JSON.stringify({
+                        content: content || '',
+                        ...(finalReasoningDetails ? { reasoning_details: finalReasoningDetails } : {}),
+                        ...(finalReasoning ? { reasoning: finalReasoning } : {}),
+                    }),
+                    role: 'assistant',
+                });
+                this.broadcast('message', assistantMsg);
+            }
+
+            usage = {
+                input: result.usage.input,
+                cached: result.usage.cached,
+                output: result.usage.output,
+                cost: result.usage.cost,
+            };
+        } 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', usage });
+    }
+
+    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();
+    }
+}
+
+/** @type {Map<number, ConversationManager>} */
+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);
+    }
+}