603 lines
28 KiB
JavaScript
603 lines
28 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 { resolveVfsRoot } from '../tools/vfs.js';
|
||
import { listSkillFiles } from '../tools/skill.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 { resolveOpenRouterKey } from '../tools/llm-key.js';
|
||
|
||
const KEEP_ALIVE_MS = 20_000;
|
||
|
||
const ROUTES_DOC_PATH = resolve(dirname(fileURLToPath(import.meta.url)), '../../docs/api/routes.md');
|
||
|
||
/**
|
||
* Builds the SKILL index block for the system prompt.
|
||
*
|
||
* Listing the skills inline means the model knows what guidance exists BEFORE it decides what
|
||
* to do, so it can reach for `read_file` on the right skill instead of guessing a procedure
|
||
* from scratch. Only the filename, title and line count are included — the body is read on
|
||
* demand, which keeps the prompt small no matter how many skills exist.
|
||
*
|
||
* @param {string} vfsRoot
|
||
* @returns {string}
|
||
*/
|
||
function buildSkillIndex(vfsRoot) {
|
||
// A null root means the VFS could not be resolved (no company selected, or a store
|
||
// error). Say so plainly instead of pretending the company has no skills — otherwise the
|
||
// model would conclude no guidance exists and invent its own procedure.
|
||
if (!vfsRoot) {
|
||
return 'AVAILABLE SKILLS: unknown (the virtual filesystem could not be read this turn). Call find_files({ skills: true }) to check.';
|
||
}
|
||
|
||
let skills;
|
||
try {
|
||
skills = listSkillFiles(vfsRoot);
|
||
} catch (err) {
|
||
console.warn('conversation: failed to list skill files:', err.message);
|
||
return 'AVAILABLE SKILLS: could not be read this turn. Call find_files({ skills: true }) to check.';
|
||
}
|
||
|
||
if (skills.length === 0) {
|
||
return [
|
||
'AVAILABLE SKILLS: none. This company has no SKILL files yet.',
|
||
].join('\n');
|
||
}
|
||
|
||
const width = Math.max(...skills.map((s) => s.path.length));
|
||
return [
|
||
`AVAILABLE SKILLS (${skills.length}): stored in the VFS root as flat <slug>.SKILL files.`,
|
||
'Before starting a task that any of these cover, read the matching skill with read_file and follow it.',
|
||
'',
|
||
...skills.map((s) => ` ${s.path.padEnd(width)} ${String(s.lines).padStart(4)} lines | ${s.title}`),
|
||
].join('\n');
|
||
}
|
||
|
||
/**
|
||
* System prompt: base instructions plus the generated API routes reference
|
||
* (docs/api/routes.md), read on each run so it stays current.
|
||
*
|
||
* @param {string} vfsRoot Resolved VFS root for this user's company, used to list SKILL files.
|
||
* @returns {string}
|
||
*/
|
||
async function buildSystemPrompt(vfsRoot) {
|
||
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. Snippets can call Dashback API routes using fetch() (e.g. await fetch("/article/2")). A sandboxed virtual filesystem (fs / vfs) is available to store and retrieve artifacts across runs.',
|
||
'4. find_files: List what is stored in the virtual filesystem (VFS). Call find_files({ skills: true }) to list SKILL files only.',
|
||
'5. read_file: Read a VFS file as text, windowed with seek/limit for large files.',
|
||
'6. patch: Create, overwrite, or search-and-replace text in a VFS file. This is the only text write path — it also creates SKILL files.',
|
||
'7. search_images: Search Google Images and store the candidates in the VFS as AVIF files.',
|
||
'8. analyze_image: Have a vision model look at up to 9 images (VFS paths or "image/<fullHash>") and return its judgement. This is how you "see" — never claim to have inspected an image yourself.',
|
||
'9. reset_vfs: Delete files, images, and artifacts stored in the virtual filesystem (VFS) to free up storage or clean up. SKILL files are exempt and survive.',
|
||
'10. rm: Delete a single file from the VFS. This is how a SKILL file is deleted on purpose, since reset_vfs always spares skills.',
|
||
'11. apisuggestionsorissues: Report missing API endpoints, bugs, or issues to be triaged. Automatically writes an entry to the company history table for review.',
|
||
'',
|
||
'VIRTUAL FILESYSTEM:',
|
||
'- You have a persistent per-company VFS. Paths are relative to the VFS root and always use forward slashes, e.g. "labels/sheet1.png".',
|
||
'- To READ text: find_files (list) then read_file (content). To WRITE text: patch. Binary artifacts (images, PDFs, buffers) go through javascript_execute.',
|
||
'- Never invent a VFS path. Always confirm with find_files before reading or referencing a file.',
|
||
'',
|
||
'SKILL FILES:',
|
||
'- A SKILL file is a reusable instruction document: a file named "<slug>.SKILL" at the VFS root whose FIRST LINE is its title and whose remaining lines are markdown.',
|
||
'- SKILL file names may NOT contain "/" — skills are flat, at the root only. Use find_files({ skills: true }) to discover them; this listing is not affected by pagination or by how many artifacts exist.',
|
||
'- Read a skill with read_file BEFORE starting a task it covers, and follow it. If you learn a better procedure during the task, offer to write it back as a new SKILL file.',
|
||
'- SKILL files are exempt from reset_vfs: artifacts are disposable, skills are not. Never call reset_vfs expecting to clear skills, and never put throwaway output in a .SKILL file.',
|
||
'- To delete a skill (or any single file) use rm — it is the only way to remove one. Deletion is final.',
|
||
'',
|
||
buildSkillIndex(vfsRoot),
|
||
'',
|
||
'IMAGES:',
|
||
'- You cannot see images yourself. To evaluate one, call analyze_image and report what the vision model says.',
|
||
'- Typical flow: search_images("query") → analyze_image({ paths: [...candidates], prompt: "…" }) → upload the winner with dashback_request.',
|
||
'- The `image/` folder holds database images and is READ-ONLY. Generated images go anywhere else (e.g. "images/…").',
|
||
'',
|
||
'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).',
|
||
'- 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)`), reference the exact path you wrote with standard markdown ``, which renders inline. Make the alt text descriptive. Other file types (PDF, JSON, CSV) are NOT rendered inline — link to them instead and offer them as a download.',
|
||
'- 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');
|
||
}
|
||
|
||
/**
|
||
* 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);
|
||
|
||
// 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) },
|
||
...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(),
|
||
// 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: conversation.model }),
|
||
createResetVfsTool({ googleAuthId }),
|
||
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);
|
||
}
|
||
}
|