526 lines
23 KiB
JavaScript
526 lines
23 KiB
JavaScript
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';
|
||
import { createApiSuggestionsOrIssuesTool } from '../tools/apisuggestionsorissues.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.',
|
||
'5. apisuggestionsorissues: Report missing API endpoints, bugs, or issues to be triaged. Automatically writes an entry to the company history table for review.',
|
||
'',
|
||
'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.',
|
||
'- SHOWING GENERATED IMAGES: Markdown images ARE supported and render inline in the chat. After writing an image into the virtual filesystem (e.g. `fs.writeFileSync("artifacts/example.png", buf)` or `sharp(...).toFile("images/example.png")`), reference the exact path you wrote with standard markdown ``, which renders inline. The title text is the alt text, so make it descriptive. Use this whenever you generate or modify an image so the user actually sees the result. Other file types (PDF, JSON, CSV) are NOT rendered inline — link to them instead and offer them as a download. NOTE: the `image/` folder is reserved for database images and is read-only; write generated images anywhere else.',
|
||
'- 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 }),
|
||
createApiSuggestionsOrIssuesTool({ 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);
|
||
}
|
||
}
|