openvibers/openvibe.live Commits

AI: llm.js single completion layer (system role, per-role models, prompt caching, JSON schema, timeouts/retry, image downscale, per-model pricing, cached-token metering); ai-analysis wrappers; BYO through the same path; usage summary by role/source/owner/model

OP OpenVibers committed 27 Aug 2026 · 7b6f6e9c6f · 5 files, +454 −154


Changes

ddiff --git a/server/ai/ai-analysis.js b/server/ai/ai-analysis.js
iindex d92a00de..6d241ab7 100644
---- a/server/ai/ai-analysis.js
++++ b/server/ai/ai-analysis.js
@@@ -15,7 +15,8 @@ function s(k) { return (db.getSetting(k) || '').toString().trim(); }
function b(k) { const v = db.getSetting(k); return v === true || v === 'true' || v === 1 || v === '1'; }
function num(k, d) { const v = parseFloat(db.getSetting(k)); return Number.isFinite(v) ? v : d; }
--function isEnabled() { return b('ai_enabled') && !!s('ai_api_key'); }
++const llm = require('./llm');
++function isEnabled() { return llm.isEnabled(); }
function pasteAnalysisEnabled() { return isEnabled() && b('ai_paste_analysis_enabled'); }
function streamMemoryEnabled() { return isEnabled() && b('ai_stream_memory_enabled'); }
// Local whisper.cpp transcription (default on when installed). Independent of the
@@@ -26,109 +27,21 @@ function transcriptionEnabled() {
try { return on && require('./transcribe').available(); } catch { return false; }
}
function captureIntervalSec() { return Math.max(30, num('ai_stream_capture_interval_sec', 120)); }
--function model() { return s('ai_model') || (s('ai_provider') === 'openai' ? 'gpt-4o-mini' : 'claude-sonnet-5'); }
--
--function withinBudget() {
-- const cap = num('ai_max_cost_usd_per_day', 0);
-- if (!cap || cap <= 0) return true;
-- try { return db.getAiCostToday() < cap; } catch { return true; }
--}
--function estimateCost(inTok, outTok) {
-- return (inTok / 1e6) * num('ai_input_cost_per_mtok', 3) + (outTok / 1e6) * num('ai_output_cost_per_mtok', 15);
++function model() { return llm.defaultModel(); }
++function withinBudget() { return llm.withinBudget(); }
++// Kept for callers that estimate before/without a real usage object (BYO fallbacks).
++function estimateCost(inTok, outTok, cachedTok = 0, modelId = null) {
++ return llm.estimateCost(modelId || model(), { input: inTok, output: outTok, cached: cachedTok });
}
--// Normalize an image input (data URL, raw base64, http(s) URL, or file path)
--// → {base64, mediaType}.
--//
--// URL support matters: paste screenshots live in Media, not on this disk, and their
--// image is only reachable over HTTP. Without this branch a URL fell through to the
--// file-path case, readFileSync threw, and the caller silently got null — which is why
--// every screenshot paste went un-analysed while text pastes worked.
--async function _normImage(image) {
-- if (!image) return null;
-- if (typeof image === 'string' && image.startsWith('data:')) {
-- const m = image.match(/^data:([^;]+);base64,(.*)$/);
-- if (m) return { base64: m[2], mediaType: m[1] };
-- }
-- if (typeof image === 'string' && /^https?:\/\//i.test(image)) {
-- try {
-- const res = await fetch(image, { signal: AbortSignal.timeout(20000) });
-- if (!res.ok) { console.warn(`[AI] image fetch ${res.status} for ${image}`); return null; }
-- const buf = Buffer.from(await res.arrayBuffer());
-- if (!buf.length) return null;
-- const ct = String(res.headers.get('content-type') || 'image/jpeg').split(';')[0].trim();
-- return { base64: buf.toString('base64'), mediaType: /^image\//.test(ct) ? ct : 'image/jpeg' };
-- } catch (e) { console.warn('[AI] image fetch failed:', e.message); return null; }
-- }
-- if (typeof image === 'string' && /^[A-Za-z0-9+/=]+$/.test(image.slice(0, 40))) {
-- return { base64: image, mediaType: 'image/jpeg' };
-- }
-- // treat as a file path
-- try {
-- const buf = fs.readFileSync(image);
-- const ext = String(image).toLowerCase();
-- const mediaType = ext.endsWith('.png') ? 'image/png' : ext.endsWith('.webp') ? 'image/webp' : 'image/jpeg';
-- return { base64: buf.toString('base64'), mediaType };
-- } catch { return null; }
--}
++// Provider transport now lives in ./llm (real system/user roles, prompt caching, per-role
++// models, structured JSON, timeouts + retry, uniform metering). These wrappers keep the
++// existing call shapes for the analysis features below.
--// ── Provider calls (return { text, input_tokens, output_tokens }) ──
--async function _anthropic(prompt, img, maxTokens, temperature) {
-- const content = [{ type: 'text', text: prompt }];
-- if (img) content.push({ type: 'image', source: { type: 'base64', media_type: img.mediaType, data: img.base64 } });
-- const body = { model: model(), max_tokens: maxTokens, messages: [{ role: 'user', content }] };
-- if (temperature != null) body.temperature = temperature;
-- const res = await fetch('https://api.anthropic.com/v1/messages', {
-- method: 'POST',
-- headers: { 'x-api-key': s('ai_api_key'), 'anthropic-version': '2023-06-01', 'content-type': 'application/json' },
-- body: JSON.stringify(body),
-- });
-- const j = await res.json().catch(() => ({}));
-- if (!res.ok) throw new Error((j.error && j.error.message) || `anthropic ${res.status}`);
-- const text = (j.content || []).filter(c => c.type === 'text').map(c => c.text).join('').trim();
-- const u = j.usage || {};
-- return { text, input_tokens: u.input_tokens || 0, output_tokens: u.output_tokens || 0 };
--}
--async function _openai(prompt, img, maxTokens, temperature) {
-- const base = (s('ai_base_url') || 'https://api.openai.com/v1').replace(/\/+$/, '');
-- const m = model();
-- const content = [{ type: 'text', text: prompt }];
-- if (img) content.push({ type: 'image_url', image_url: { url: `data:${img.mediaType};base64,${img.base64}` } });
-- const body = { model: m, messages: [{ role: 'user', content }] };
-- if (temperature != null && !/^(gpt-5|o\d)/i.test(m)) body.temperature = temperature;
-- // GPT-5 / o-series are reasoning models: they reject `max_tokens` (need
-- // `max_completion_tokens`) and spend hidden reasoning tokens, so give the
-- // output headroom and keep reasoning effort low for these short tasks —
-- // otherwise the whole budget is consumed by reasoning and content is empty.
-- if (/^(gpt-5|o\d)/i.test(m)) {
-- body.max_completion_tokens = Math.max(maxTokens, 256) + 512;
-- body.reasoning_effort = /^gpt-5/i.test(m) ? 'minimal' : 'low';
-- } else {
-- body.max_tokens = maxTokens;
-- }
-- const res = await fetch(`${base}/chat/completions`, {
-- method: 'POST',
-- headers: { Authorization: `Bearer ${s('ai_api_key')}`, 'Content-Type': 'application/json' },
-- body: JSON.stringify(body),
-- });
-- const j = await res.json().catch(() => ({}));
-- if (!res.ok) throw new Error((j.error && j.error.message) || `openai ${res.status}`);
-- const text = (j.choices && j.choices[0] && j.choices[0].message && j.choices[0].message.content || '').trim();
-- const u = j.usage || {};
-- return { text, input_tokens: u.prompt_tokens || 0, output_tokens: u.completion_tokens || 0 };
--}
--
--/** Core call: dispatches by provider, records usage/cost. Returns text or null. */
--async function _complete({ prompt, image = null, maxTokens = 400, kind, temperature = null, ownerUserId = null, source = null }) {
-- if (!isEnabled() || !withinBudget()) return null;
-- const img = image ? await _normImage(image) : null;
-- if (image && !img) return null;
-- const provider = s('ai_provider') === 'openai' ? _openai : _anthropic;
-- let r;
-- try { r = await provider(prompt, img, maxTokens, temperature); }
-- catch (e) { console.warn('[AI] analysis failed:', e.message); return null; }
-- try { db.recordAiUsage({ kind, model: model(), input_tokens: r.input_tokens, output_tokens: r.output_tokens, cost_usd: estimateCost(r.input_tokens, r.output_tokens), owner_user_id: ownerUserId, source }); } catch { /* */ }
-- return r.text || null;
++/** Core call: legacy single-prompt shape. Returns text or null. */
++async function _complete({ prompt, image = null, maxTokens = 400, kind, temperature = null, ownerUserId = null, source = null, role = 'legacy', imageMaxWidth = 1280 }) {
++ const r = await llm.complete({ role, user: prompt, image, imageMaxWidth, maxTokens, temperature, kind, ownerUserId, source });
++ return r && r.text ? r.text : null;
}
/** Generic text completion (used by media analysis to synthesize overviews). */
@@@ -143,8 +56,8 @@ async function summarizeText(prompt, maxTokens = 350, kind = 'media_overview') {
* disabled / over the global budget. Supports an optional image (vision) input.
*/
async function viewerComplete({ system = '', user = '', image = null, maxTokens = 80, temperature = 1.0, ownerUserId = null }) {
-- const prompt = system ? `${system}\n\n${user}` : user;
-- return _complete({ prompt, image, maxTokens, temperature, kind: 'ai_viewers', source: 'ai_viewers', ownerUserId });
++ const r = await llm.complete({ role: 'chat', system, user, image, imageMaxWidth: 768, maxTokens, temperature, kind: 'ai_viewers', source: 'ai_viewers', ownerUserId });
++ return r && r.text ? r.text : null;
}
/** Is the shared admin AI key usable right now (enabled + within global budget)? */
@@@ -202,24 +115,15 @@ function _extractTags(text, maxTags) {
/** Re-encode any image (path/data) to a vision-friendly JPEG data URL. Handles
* avif/gif/webp/huge images that the vision API otherwise rejects. */
--async function _toVisionJpeg(image) {
-- try {
-- const sharp = require('sharp');
-- const buf = await sharp(image, { failOn: 'none', animated: false })
-- .rotate()
-- .resize({ width: 1280, height: 1280, fit: 'inside', withoutEnlargement: true })
-- .jpeg({ quality: 82 })
-- .toBuffer();
-- return `data:image/jpeg;base64,${buf.toString('base64')}`;
-- } catch { return image; }
++async function _toVisionJpeg(image, opts = {}) {
++ return (await llm.toVisionJpeg(image, { maxWidth: 1280, quality: 82, ...opts })) || image;
}
/** Describe an image paste → { description, tags }. */
--async function analyzeImagePaste(image, title) {
++async function analyzeImagePaste(image, title, kind = 'paste_image') {
const prompt = `You are describing an uploaded image/screenshot for a paste titled "${(title || '').slice(0, 120)}".
Reply ONLY with compact JSON: {"description":"1-2 sentence description of what the image shows","tags":["3-6","short","lowercase","tags"]}.`;
-- const img = await _toVisionJpeg(image);
-- const text = await _complete({ prompt, image: img, maxTokens: 300, kind: 'paste_image' });
++ const text = await _complete({ prompt, image, maxTokens: 300, kind, role: 'vision', imageMaxWidth: 1280 });
if (!text) return null;
const description = _extractDescription(text, 600);
return description ? { description, tags: _extractTags(text, 8) } : null;
@@@ -236,7 +140,7 @@ async function analyzeTextPaste(content, title) {
/** Analyze a live-stream frame → { description, tags }. */
async function analyzeStreamFrame(image) {
const prompt = `This is a frame from a live stream. Reply ONLY with compact JSON: {"description":"one concise sentence describing what is happening on screen right now","tags":["2-5","short","tags"]}.`;
-- const text = await _complete({ prompt, image, maxTokens: 200, kind: 'stream_memory' });
++ const text = await _complete({ prompt, image, maxTokens: 200, kind: 'stream_memory', role: 'vision', imageMaxWidth: 768 });
if (!text) return null;
const description = _extractDescription(text, 400);
return description ? { description, tags: _extractTags(text, 6) } : null;
@@@ -567,4 +471,5 @@ module.exports = {
generateStreamerOverview, generateVodOverview, generateClipOverview, ensureVodTimeline,
generateVodTranscript, generateClipTranscript, summarizeText, testStatus,
viewerComplete, sharedKeyReady, estimateCost,
++ complete: llm.complete, llm,
};
ddiff --git a/server/ai/ai-moments-job.js b/server/ai/ai-moments-job.js
iindex 470207a5..d601ce8c 100644
---- a/server/ai/ai-moments-job.js
++++ b/server/ai/ai-moments-job.js
@@@ -360,7 +360,7 @@ async function tick(opts = {}) {
// tags, and it confirms we didn't grab a black/loading screen.
if (_aiOn() && ai.analyzeImagePaste) {
try {
-- const vis = await ai.analyzeImagePaste(outPath, title);
++ const vis = await ai.analyzeImagePaste(outPath, title, 'moment_frame');
if (vis && vis.description && vis.description.length > 25) {
desc = _cleanText(vis.description, 500);
if (Array.isArray(vis.tags) && vis.tags.length) tags = vis.tags.slice(0, 8);
ddiff --git a/server/ai/llm.js b/server/ai/llm.js
nnew file mode 100644
iindex 00000000..6824fed2
---- /dev/null
++++ b/server/ai/llm.js
@@@ -0,0 +1,360 @@
++/**
++ * llm.js — the ONE chat-completion entry point for OpenVibe.Live.
++ *
++ * Every feature (AI viewers, overviews, chat insights, vision, paste analysis) calls
++ * complete(); it is the single place that:
++ * - gates on the admin master switch + global daily budget (shared key), or routes
++ * to a streamer's BYO provider (uniformly metered either way);
++ * - picks the model per ROLE (chat / vision / director / summary) with per-role admin
++ * overrides and BYO equivalents, falling back to the default model;
++ * - talks to Anthropic or any OpenAI-compatible endpoint with real system/user roles
++ * (the old code flattened everything into one user message, which made prompt
++ * caching impossible);
++ * - uses prompt caching: Anthropic cache_control on the stable system blocks, OpenAI
++ * automatic prefix caching + prompt_cache_key;
++ * - supports structured JSON output (OpenAI json_schema, Anthropic forced tool), with a
++ * prose-JSON repair fallback for gateways that lack it;
++ * - always downscales images before sending them;
++ * - applies a timeout and ONE retry on transient failures;
++ * - records usage (incl. cached tokens, role, provider, latency) with per-model pricing.
++ */
++'use strict';
++const fs = require('fs');
++const db = require('../db/database');
++
++const ROLES = ['chat', 'vision', 'director', 'summary', 'legacy'];
++const DEFAULT_TIMEOUT_MS = { chat: 20000, vision: 30000, director: 25000, summary: 30000, legacy: 30000 };
++
++function s(k) { return (db.getSetting(k) || '').toString().trim(); }
++function b(k) { const v = db.getSetting(k); return v === true || v === 'true' || v === 1 || v === '1'; }
++function num(k, d) { const v = parseFloat(db.getSetting(k)); return Number.isFinite(v) ? v : d; }
++
++// ── Gates ────────────────────────────────────────────────────
++function isEnabled() { return b('ai_enabled') && !!s('ai_api_key'); }
++function withinBudget() {
++ const cap = num('ai_max_cost_usd_per_day', 0);
++ if (!cap || cap <= 0) return true;
++ try { return db.getAiCostToday() < cap; } catch { return true; }
++}
++function defaultModel() { return s('ai_model') || (s('ai_provider') === 'anthropic' ? 'claude-sonnet-5' : 'gpt-4o-mini'); }
++
++/** Model for a role: admin per-role override → default model. */
++function modelForRole(role) {
++ const r = ROLES.includes(role) ? role : 'legacy';
++ if (r !== 'legacy') { const m = s(`ai_model_${r}`); if (m) return m; }
++ return defaultModel();
++}
++
++/**
++ * Resolve the provider for a call.
++ * override = { kind?: 'openai'|'anthropic', baseUrl?, apiKey?, model?, models?: {chat,vision,...} }
++ */
++function resolveProvider(role, override = null) {
++ if (override && (override.apiKey || override.baseUrl)) {
++ const baseUrl = String(override.baseUrl || '').trim().replace(/\/+$/, '');
++ const kind = override.kind || (/anthropic\.com/i.test(baseUrl) ? 'anthropic' : 'openai');
++ const model = (override.models && override.models[role]) || override.model || 'gpt-4o-mini';
++ return { kind, baseUrl: baseUrl || (kind === 'anthropic' ? 'https://api.anthropic.com/v1' : 'https://api.openai.com/v1'), apiKey: override.apiKey || '', model, shared: false };
++ }
++ const baseUrl = (s('ai_base_url') || '').replace(/\/+$/, '');
++ // Anthropic only when explicitly chosen (and not pointed at an OpenAI-compatible
++ // gateway); everything else — openai, openrouter, groq, ollama, blank — is OpenAI-shaped.
++ const kind = s('ai_provider') === 'anthropic' && (!baseUrl || /anthropic\.com/i.test(baseUrl)) ? 'anthropic' : 'openai';
++ return {
++ kind,
++ baseUrl: baseUrl || (kind === 'anthropic' ? 'https://api.anthropic.com/v1' : 'https://api.openai.com/v1'),
++ apiKey: s('ai_api_key'),
++ model: modelForRole(role),
++ shared: true,
++ };
++}
++
++// ── Pricing ──────────────────────────────────────────────────
++// site_settings.ai_pricing_json = { "gpt-5-nano": {"in":0.05,"out":0.4,"cached":0.005}, "default": {...} }
++// (USD per million tokens). Longest key-prefix match on the model id; falls back to the
++// flat ai_input/output_cost_per_mtok settings, cached input = 10% of input.
++let _pricingCache = { at: 0, table: null };
++function pricingTable() {
++ if (Date.now() - _pricingCache.at < 30000 && _pricingCache.table) return _pricingCache.table;
++ let table = {};
++ try {
++ const raw = db.getSetting('ai_pricing_json'); // type 'json' → already parsed
++ table = (raw && typeof raw === 'object') ? raw : (raw ? JSON.parse(String(raw)) : {}) || {};
++ } catch { table = {}; }
++ _pricingCache = { at: Date.now(), table };
++ return table;
++}
++function priceFor(model) {
++ const table = pricingTable();
++ const m = String(model || '').toLowerCase();
++ let best = null, bestLen = -1;
++ for (const [k, v] of Object.entries(table)) {
++ const key = String(k).toLowerCase();
++ if (key === 'default') continue;
++ if (m.startsWith(key) && key.length > bestLen && v && typeof v === 'object') { best = v; bestLen = key.length; }
++ }
++ if (!best && table.default && typeof table.default === 'object') best = table.default;
++ const inRate = best && Number.isFinite(Number(best.in)) ? Number(best.in) : num('ai_input_cost_per_mtok', 3);
++ const outRate = best && Number.isFinite(Number(best.out)) ? Number(best.out) : num('ai_output_cost_per_mtok', 15);
++ const cachedRate = best && Number.isFinite(Number(best.cached)) ? Number(best.cached) : inRate * 0.1;
++ return { in: inRate, out: outRate, cached: cachedRate };
++}
++/** usage = { input (total prompt tokens incl. cached), output, cached } */
++function estimateCost(model, usage) {
++ const p = priceFor(model);
++ const input = Math.max(0, (usage.input || 0) - (usage.cached || 0));
++ return (input / 1e6) * p.in + ((usage.cached || 0) / 1e6) * p.cached + ((usage.output || 0) / 1e6) * p.out;
++}
++
++// ── Images ───────────────────────────────────────────────────
++async function _loadImage(image) {
++ if (!image) return null;
++ if (Buffer.isBuffer(image)) return image;
++ if (typeof image !== 'string') return null;
++ if (image.startsWith('data:')) {
++ const m = image.match(/^data:([^;]+);base64,(.*)$/);
++ return m ? Buffer.from(m[2], 'base64') : null;
++ }
++ if (/^https?:\/\//i.test(image)) {
++ try {
++ const res = await fetch(image, { signal: AbortSignal.timeout(20000) });
++ if (!res.ok) { console.warn(`[AI] image fetch ${res.status} for ${image}`); return null; }
++ const buf = Buffer.from(await res.arrayBuffer());
++ return buf.length ? buf : null;
++ } catch (e) { console.warn('[AI] image fetch failed:', e.message); return null; }
++ }
++ if (/^[A-Za-z0-9+/=]+$/.test(image.slice(0, 40)) && !fs.existsSync(image)) {
++ try { return Buffer.from(image, 'base64'); } catch { return null; }
++ }
++ try { return fs.readFileSync(image); } catch { return null; }
++}
++/** Any image input → downscaled JPEG data URL (sharp). Falls back to the original bytes. */
++async function toVisionJpeg(image, { maxWidth = 1280, quality = 82 } = {}) {
++ const buf = await _loadImage(image);
++ if (!buf) return null;
++ try {
++ const sharp = require('sharp');
++ const out = await sharp(buf, { failOn: 'none', animated: false })
++ .rotate()
++ .resize({ width: maxWidth, height: maxWidth, fit: 'inside', withoutEnlargement: true })
++ .jpeg({ quality })
++ .toBuffer();
++ return `data:image/jpeg;base64,${out.toString('base64')}`;
++ } catch {
++ return `data:image/jpeg;base64,${buf.toString('base64')}`;
++ }
++}
++function _splitDataUrl(dataUrl) {
++ const m = String(dataUrl || '').match(/^data:([^;]+);base64,(.*)$/);
++ return m ? { mediaType: m[1], base64: m[2] } : null;
++}
++
++// ── Message shaping ──────────────────────────────────────────
++function _systemBlocks(system) {
++ if (!system) return [];
++ if (typeof system === 'string') return system.trim() ? [{ text: system, cache: false }] : [];
++ return system.filter(x => x && String(x.text || '').trim()).map(x => ({ text: String(x.text), cache: !!x.cache }));
++}
++function _normMessages(messages, user) {
++ const out = [];
++ for (const m of (messages || [])) {
++ if (!m || !m.role) continue;
++ out.push({ role: m.role === 'assistant' ? 'assistant' : 'user', content: m.content });
++ }
++ if (user) out.push({ role: 'user', content: user });
++ if (!out.length) out.push({ role: 'user', content: '' });
++ return out;
++}
++
++function _isReasoningModel(m) { return /^(gpt-5|o\d)/i.test(m || ''); }
++
++function _openaiBody({ p, system, messages, img, json, maxTokens, temperature, cacheKey, jsonMode }) {
++ const msgs = [];
++ const sysText = system.map(x => x.text).join('\n\n');
++ if (sysText) msgs.push({ role: 'system', content: sysText });
++ messages.forEach((m, i) => {
++ const last = i === messages.length - 1;
++ if (last && img && m.role === 'user') {
++ const parts = Array.isArray(m.content) ? m.content : [{ type: 'text', text: String(m.content || '') }];
++ msgs.push({ role: 'user', content: [...parts, { type: 'image_url', image_url: { url: `data:${img.mediaType};base64,${img.base64}` } }] });
++ } else msgs.push({ role: m.role, content: Array.isArray(m.content) ? m.content : String(m.content || '') });
++ });
++ const body = { model: p.model, messages: msgs };
++ if (_isReasoningModel(p.model)) {
++ body.max_completion_tokens = Math.max(maxTokens, 256) + 512;
++ body.reasoning_effort = /^gpt-5/i.test(p.model) ? 'minimal' : 'low';
++ } else {
++ body.max_tokens = maxTokens;
++ if (temperature != null) body.temperature = temperature;
++ }
++ if (json && jsonMode === 'schema') body.response_format = { type: 'json_schema', json_schema: { name: json.name || 'result', schema: json.schema, strict: json.strict !== false } };
++ else if (json && jsonMode === 'object') body.response_format = { type: 'json_object' };
++ if (cacheKey && /api\.openai\.com/i.test(p.baseUrl)) body.prompt_cache_key = String(cacheKey).slice(0, 64);
++ return body;
++}
++function _anthropicBody({ p, system, messages, img, json, maxTokens, temperature }) {
++ const body = { model: p.model, max_tokens: Math.max(1, maxTokens) };
++ if (system.length) {
++ body.system = system.map(x => x.cache ? { type: 'text', text: x.text, cache_control: { type: 'ephemeral' } } : { type: 'text', text: x.text });
++ }
++ body.messages = messages.map((m, i) => {
++ const last = i === messages.length - 1;
++ const content = Array.isArray(m.content)
++ ? m.content.map(c => (c.type === 'text' ? { type: 'text', text: c.text } : c))
++ : [{ type: 'text', text: String(m.content || '') }];
++ if (last && img && m.role === 'user') content.push({ type: 'image', source: { type: 'base64', media_type: img.mediaType, data: img.base64 } });
++ return { role: m.role, content };
++ });
++ if (temperature != null) body.temperature = temperature;
++ if (json) {
++ body.tools = [{ name: json.name || 'result', description: json.description || 'Return the result.', input_schema: json.schema }];
++ body.tool_choice = { type: 'tool', name: json.name || 'result' };
++ }
++ return body;
++}
++
++// ── HTTP ─────────────────────────────────────────────────────
++class LlmHttpError extends Error { constructor(status, message, body) { super(message); this.status = status; this.body = body; } }
++
++async function _post(url, headers, body, timeoutMs) {
++ const res = await fetch(url, { method: 'POST', headers: { 'Content-Type': 'application/json', ...headers }, body: JSON.stringify(body), signal: AbortSignal.timeout(timeoutMs) });
++ const text = await res.text();
++ let j = {}; try { j = text ? JSON.parse(text) : {}; } catch { j = { raw: text }; }
++ if (!res.ok) throw new LlmHttpError(res.status, (j.error && (j.error.message || j.error)) || `HTTP ${res.status}: ${text.slice(0, 200)}`, j);
++ return j;
++}
++
++async function _callOpenAI(p, args, timeoutMs) {
++ const headers = p.apiKey ? { Authorization: `Bearer ${p.apiKey}` } : {};
++ let jsonMode = args.json ? 'schema' : null;
++ for (let attempt = 0; attempt < 3; attempt++) {
++ const body = _openaiBody({ ...args, p, jsonMode });
++ try {
++ const j = await _post(`${p.baseUrl}/chat/completions`, headers, body, timeoutMs);
++ const msg = j.choices && j.choices[0] && j.choices[0].message;
++ const text = (msg && typeof msg.content === 'string' ? msg.content : '').trim();
++ const u = j.usage || {};
++ return { text, usage: { input: u.prompt_tokens || 0, output: u.completion_tokens || 0, cached: (u.prompt_tokens_details && u.prompt_tokens_details.cached_tokens) || 0 } };
++ } catch (e) {
++ // Gateways without structured-output support: step down schema → json_object → prose.
++ if (e instanceof LlmHttpError && e.status === 400 && jsonMode && /response_format|json_schema|schema|strict/i.test(e.message)) {
++ jsonMode = jsonMode === 'schema' ? 'object' : null;
++ continue;
++ }
++ throw e;
++ }
++ }
++ throw new Error('structured output unsupported');
++}
++async function _callAnthropic(p, args, timeoutMs) {
++ const body = _anthropicBody({ ...args, p });
++ const j = await _post(`${p.baseUrl}/messages`, { 'x-api-key': p.apiKey, 'anthropic-version': '2023-06-01' }, body, timeoutMs);
++ const content = j.content || [];
++ let text = content.filter(c => c.type === 'text').map(c => c.text).join('').trim();
++ let jsonOut = null;
++ const tool = content.find(c => c.type === 'tool_use');
++ if (tool && tool.input) { jsonOut = tool.input; if (!text) text = JSON.stringify(tool.input); }
++ const u = j.usage || {};
++ const cached = u.cache_read_input_tokens || 0;
++ return { text, json: jsonOut, usage: { input: (u.input_tokens || 0) + cached + (u.cache_creation_input_tokens || 0), output: u.output_tokens || 0, cached } };
++}
++
++function _retryable(e) {
++ if (!e) return false;
++ if (e.name === 'TimeoutError' || e.name === 'AbortError') return true;
++ if (e instanceof LlmHttpError) return e.status === 429 || e.status >= 500;
++ return /fetch failed|ECONNRESET|ETIMEDOUT|EAI_AGAIN|socket/i.test(e.message || '');
++}
++
++function parseJsonLoose(text) {
++ if (!text) return null;
++ try { return JSON.parse(text); } catch { /* */ }
++ const m = String(text).match(/\{[\s\S]*\}/);
++ if (!m) return null;
++ let raw = m[0];
++ try { return JSON.parse(raw); } catch { /* repair */ }
++ try {
++ let t = raw.replace(/[“”]/g, '"').replace(/[‘’]/g, "'").replace(/,\s*([}\]])/g, '$1');
++ const osq = (t.match(/\[/g) || []).length, csq = (t.match(/\]/g) || []).length;
++ if (osq > csq) { t = t.replace(/\}\s*$/, ''); t += ']'.repeat(osq - csq); }
++ const ocb = (t.match(/\{/g) || []).length, ccb = (t.match(/\}/g) || []).length;
++ if (ocb > ccb) t += '}'.repeat(ocb - ccb);
++ return JSON.parse(t);
++ } catch { return null; }
++}
++
++/**
++ * The call.
++ * @param {object} o
++ * role, system (string | [{text, cache}]), messages ([{role, content}]), user (shortcut: appended user msg),
++ * image (data URL | http url | path | Buffer), imageMaxWidth (default 1024), json ({name, schema, strict?}),
++ * maxTokens, temperature, timeoutMs, retries (default 1), cacheKey,
++ * kind (ai_usage.kind), source, ownerUserId, provider (BYO override), skipGate (internal)
++ * @returns {Promise<null|{text:string, json:object|null, usage:{input,output,cached}, model:string, provider:string, latencyMs:number}>}
++ */
++async function complete(o = {}) {
++ const role = ROLES.includes(o.role) ? o.role : 'legacy';
++ const p = resolveProvider(role, o.provider || null);
++ if (p.shared) { if (!isEnabled() || !withinBudget()) return null; }
++ else if (!p.apiKey && !/localhost|127\.0\.0\.1|:\d+$/.test(p.baseUrl) && /api\.openai\.com|anthropic\.com/i.test(p.baseUrl)) return null;
++
++ let img = null;
++ if (o.image) {
++ const dataUrl = await toVisionJpeg(o.image, { maxWidth: o.imageMaxWidth || 1024, quality: o.imageQuality || 78 });
++ img = dataUrl ? _splitDataUrl(dataUrl) : null;
++ if (!img) return null;
++ }
++ const args = {
++ system: _systemBlocks(o.system),
++ messages: _normMessages(o.messages, o.user),
++ img,
++ json: o.json || null,
++ maxTokens: Math.max(1, Math.round(o.maxTokens || 400)),
++ temperature: o.temperature == null ? null : o.temperature,
++ cacheKey: o.cacheKey || null,
++ };
++ const timeoutMs = o.timeoutMs || DEFAULT_TIMEOUT_MS[role] || 25000;
++ const retries = o.retries == null ? 1 : Math.max(0, o.retries);
++ const started = Date.now();
++ let r = null, lastErr = null;
++ for (let attempt = 0; attempt <= retries; attempt++) {
++ try {
++ r = p.kind === 'anthropic' ? await _callAnthropic(p, args, timeoutMs) : await _callOpenAI(p, args, timeoutMs);
++ break;
++ } catch (e) {
++ lastErr = e;
++ if (attempt < retries && _retryable(e)) { await new Promise(res => setTimeout(res, 800 + Math.random() * 700)); continue; }
++ break;
++ }
++ }
++ if (!r) { console.warn(`[AI] ${role}/${p.model} failed:`, lastErr && lastErr.message); return null; }
++ const latencyMs = Date.now() - started;
++ const usage = r.usage || { input: 0, output: 0, cached: 0 };
++ if (!usage.input && !usage.output) {
++ // Self-hosted servers often omit usage — estimate so budgets still mean something.
++ const approx = (t) => Math.ceil(String(t || '').length / 4);
++ usage.input = approx(args.system.map(x => x.text).join('')) + args.messages.reduce((n, m) => n + approx(typeof m.content === 'string' ? m.content : JSON.stringify(m.content)), 0) + (img ? 800 : 0);
++ usage.output = approx(r.text);
++ usage.estimated = true;
++ }
++ const cost = estimateCost(p.model, usage);
++ try {
++ db.recordAiUsage({
++ kind: o.kind || role, model: p.model, input_tokens: usage.input, output_tokens: usage.output, cached_tokens: usage.cached || 0,
++ cost_usd: cost, owner_user_id: o.ownerUserId || null, source: o.source || null, role, provider: p.shared ? 'shared' : 'byo', latency_ms: latencyMs,
++ });
++ } catch { /* metering is best-effort */ }
++ let json = r.json || null;
++ if (o.json && !json) json = parseJsonLoose(r.text);
++ return { text: r.text || '', json, usage, model: p.model, provider: p.kind, shared: p.shared, latencyMs, cost };
++}
++
++/** 1-token connectivity probe for a provider override (BYO test) or the shared key. */
++async function testProvider(override = null) {
++ try {
++ const r = await complete({ role: 'chat', user: 'Reply with the single word: ok', maxTokens: 5, temperature: 0, timeoutMs: 15000, retries: 0, kind: 'status_check', provider: override, skipGate: true });
++ return r ? { ok: true, model: r.model, latencyMs: r.latencyMs } : { ok: false, error: 'AI disabled, over budget, or the provider rejected the request' };
++ } catch (e) { return { ok: false, error: e.message }; }
++}
++
++module.exports = { complete, testProvider, resolveProvider, modelForRole, priceFor, estimateCost, toVisionJpeg, parseJsonLoose, isEnabled, withinBudget, defaultModel, ROLES };
ddiff --git a/server/ai/viewers/budget.js b/server/ai/viewers/budget.js
iindex bad50b46..67ce22ac 100644
---- a/server/ai/viewers/budget.js
++++ b/server/ai/viewers/budget.js
@@@ -64,37 +64,32 @@ async function generate(userId, { system = '', user = '', image = null, maxToken
return ai.viewerComplete({ system, user, image, maxTokens, temperature, ownerUserId: userId });
}
-- // BYO key (OpenAI-compatible) — unmetered by the provider; record an estimate.
-- const cfg = st.cfg;
-- const messages = [];
-- if (system) messages.push({ role: 'system', content: system });
-- const userContent = image
-- ? [{ type: 'text', text: user }, { type: 'image_url', image_url: { url: image } }]
-- : user;
-- messages.push({ role: 'user', content: userContent });
-- let text;
-- try {
-- text = await aiProvider.chatCompletion({
-- baseUrl: cfg.byo_base_url || aiProvider.DEFAULT_BASE_URL,
-- apiKey: cfg.byo_key,
-- model: cfg.byo_model || 'gpt-4o-mini',
-- messages, temperature, maxTokens,
-- });
-- } catch (e) {
-- console.warn('[AI-Viewers] BYO completion failed:', e.message);
-- return null;
++ // BYO key — same llm.js path as the shared key (real system role, caching, timeouts),
++ // metered from the provider's usage report (estimated only when the server omits it).
++ const r = await ai.llm.complete({
++ role: 'chat', system, user, image, imageMaxWidth: 768, maxTokens, temperature,
++ kind: 'ai_viewers', source: SOURCE, ownerUserId: userId,
++ provider: byoProvider(st.cfg),
++ });
++ return r && r.text ? r.text.trim() || null : null;
++}
++
++/** llm.js provider override for a streamer's BYO settings (column fields + settings_json.byo). */
++function byoProvider(cfg) {
++ let extra = {};
++ try { extra = (JSON.parse(cfg.settings_json || '{}') || {}).byo || {}; } catch { extra = {}; }
++ const models = {};
++ for (const role of ['chat', 'vision', 'director', 'summary']) {
++ const m = extra[`model_${role}`] || (extra.models && extra.models[role]);
++ if (m) models[role] = String(m);
}
-- try {
-- const inTok = approxTokens(system) + approxTokens(user) + (image ? 800 : 0);
-- const outTok = approxTokens(text);
-- db.recordAiUsage({
-- kind: 'ai_viewers', model: cfg.byo_model || 'byo',
-- input_tokens: inTok, output_tokens: outTok,
-- cost_usd: ai.estimateCost(inTok, outTok),
-- owner_user_id: userId, source: SOURCE,
-- });
-- } catch { /* metering is best-effort */ }
-- return (text || '').trim() || null;
++ return {
++ baseUrl: cfg.byo_base_url || extra.base_url || aiProvider.DEFAULT_BASE_URL,
++ apiKey: cfg.byo_key || '',
++ model: cfg.byo_model || extra.model || 'gpt-4o-mini',
++ models,
++ kind: extra.provider === 'anthropic' ? 'anthropic' : undefined,
++ };
}
--module.exports = { generate, budgetStatus, byoUsable, SOURCE };
++module.exports = { generate, budgetStatus, byoUsable, byoProvider, SOURCE };
ddiff --git a/server/db/database.js b/server/db/database.js
iindex 57d411f1..18b73d4d 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -1070,6 +1070,20 @@ function initDb() {
['ai_max_cost_usd_per_day', '0', 'Daily AI spend cap in USD (0 = no cap)', 'number'],
['ai_input_cost_per_mtok', '3.0', 'Estimated input cost per million tokens (for cost breakdown)', 'number'],
['ai_output_cost_per_mtok', '15.0', 'Estimated output cost per million tokens (for cost breakdown)', 'number'],
++ // Per-role model routing (blank = use ai_model). chat = short bot lines, vision = frame/image
++ // analysis (must be vision-capable), director = the AI-viewers planner (structured JSON),
++ // summary = memories/overviews/insight folds.
++ ['ai_model_chat', '', 'Model for short chat-style generations (AI viewers lines). Blank = ai_model', 'string'],
++ ['ai_model_vision', '', 'Model for image/frame analysis (must support vision). Blank = ai_model', 'string'],
++ ['ai_model_director', '', 'Model for the AI-viewers director (plans several lines per call, JSON output). Blank = ai_model', 'string'],
++ ['ai_model_summary', '', 'Model for summaries/overviews/memory folds. Blank = ai_model', 'string'],
++ ['ai_pricing_json', '{"gpt-5-nano":{"in":0.05,"out":0.4,"cached":0.005},"gpt-5-mini":{"in":0.25,"out":2,"cached":0.025},"gpt-5":{"in":1.25,"out":10,"cached":0.125},"gpt-4o-mini":{"in":0.15,"out":0.6,"cached":0.075},"gpt-4.1-mini":{"in":0.4,"out":1.6,"cached":0.1},"claude-haiku":{"in":0.8,"out":4,"cached":0.08},"claude-sonnet":{"in":3,"out":15,"cached":0.3}}', 'USD per 1M tokens by model-id prefix: {"<model>":{"in","out","cached"}}. Longest prefix wins; missing → ai_input/output_cost_per_mtok', 'json'],
++ ['ai_viewers_enabled', 'true', 'Kill switch for the AI chat viewers feature (all channels)', 'boolean'],
++ ['ai_viewers_engine', 'v2', 'AI viewers engine: v2 (legacy per-line) | v3 (director, unified context)', 'string'],
++ ['ai_viewers_max_roster', '12', 'Max AI viewers per channel', 'number'],
++ ['ai_viewers_max_lines_per_min', '12', 'Hard ceiling on bot lines per minute per channel', 'number'],
++ ['ai_viewers_global_cap_usd_per_day', '0', 'Daily USD cap for ALL AI-viewer spend on the shared key (0 = none)', 'number'],
++ ['ai_viewers_default_settings_json', '{}', 'Admin defaults for per-channel AI viewer settings (overrides built-in defaults)', 'json'],
];
const seedAi = database.prepare("INSERT OR IGNORE INTO site_settings (key, value, description, type) VALUES (?, ?, ?, ?)");
for (const [k, v, d, t] of aiSeeds) seedAi.run(k, v, d, t);
@@@ -1241,6 +1255,16 @@ function initDb() {
generated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)`);
++ // Per-call metering detail for the llm.js layer: cached prompt tokens (prompt caching),
++ // role (chat/vision/director/summary), provider (shared|byo), latency.
++ try {
++ const cols = database.prepare('PRAGMA table_info(ai_usage)').all().map(c => c.name);
++ if (!cols.includes('cached_tokens')) database.exec('ALTER TABLE ai_usage ADD COLUMN cached_tokens INTEGER DEFAULT 0');
++ if (!cols.includes('role')) database.exec('ALTER TABLE ai_usage ADD COLUMN role TEXT');
++ if (!cols.includes('provider')) database.exec('ALTER TABLE ai_usage ADD COLUMN provider TEXT');
++ if (!cols.includes('latency_ms')) database.exec('ALTER TABLE ai_usage ADD COLUMN latency_ms INTEGER');
++ database.exec('CREATE INDEX IF NOT EXISTS idx_ai_usage_source_day ON ai_usage(source, created_at)');
++ } catch (e) { console.warn('[DB] ai_usage metering columns:', e.message); }
// Assembled AI Timeline payload per streamer (streamer overview + every session's
// AI overview + memory moments), cached JSON so the channel tab is cheap to serve and
@@@ -3016,9 +3040,9 @@ function cleanupMalformedAiText() {
if (fixed) console.log(`[AI] Cleaned ${fixed} malformed JSON AI text value(s)`);
return fixed;
}
--function recordAiUsage({ kind, model, input_tokens = 0, output_tokens = 0, cost_usd = 0, owner_user_id = null, source = null }) {
-- return run('INSERT INTO ai_usage (kind, model, input_tokens, output_tokens, cost_usd, owner_user_id, source) VALUES (?, ?, ?, ?, ?, ?, ?)',
-- [kind || null, model || null, input_tokens || 0, output_tokens || 0, cost_usd || 0, owner_user_id || null, source || null]);
++function recordAiUsage({ kind, model, input_tokens = 0, output_tokens = 0, cached_tokens = 0, cost_usd = 0, owner_user_id = null, source = null, role = null, provider = null, latency_ms = null }) {
++ return run('INSERT INTO ai_usage (kind, model, input_tokens, output_tokens, cached_tokens, cost_usd, owner_user_id, source, role, provider, latency_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)',
++ [kind || null, model || null, input_tokens || 0, output_tokens || 0, cached_tokens || 0, cost_usd || 0, owner_user_id || null, source || null, role || null, provider || null, latency_ms == null ? null : Math.round(latency_ms)]);
}
function getAiCostToday() {
const r = get("SELECT COALESCE(SUM(cost_usd),0) AS c FROM ai_usage WHERE created_at >= date('now')");
@@@ -3040,9 +3064,25 @@ function getAiUsageSummary(days = 30) {
const byKind = all(`SELECT kind, COUNT(*) AS calls, SUM(cost_usd) AS cost_usd
FROM ai_usage WHERE created_at >= date('now', ?) GROUP BY kind ORDER BY cost_usd DESC`, [`-${days} days`]);
const totals = get(`SELECT COUNT(*) AS calls, COALESCE(SUM(input_tokens),0) AS input_tokens,
-- COALESCE(SUM(output_tokens),0) AS output_tokens, COALESCE(SUM(cost_usd),0) AS cost_usd
++ COALESCE(SUM(output_tokens),0) AS output_tokens, COALESCE(SUM(cached_tokens),0) AS cached_tokens,
++ COALESCE(SUM(cost_usd),0) AS cost_usd
FROM ai_usage WHERE created_at >= date('now', ?)`, [`-${days} days`]);
-- return { byDay, byKind, totals, today: getAiCostToday() };
++ const byRole = all(`SELECT COALESCE(role,'legacy') AS role, COUNT(*) AS calls, SUM(input_tokens) AS input_tokens, SUM(cached_tokens) AS cached_tokens,
++ SUM(output_tokens) AS output_tokens, SUM(cost_usd) AS cost_usd, AVG(latency_ms) AS avg_latency_ms
++ FROM ai_usage WHERE created_at >= date('now', ?) GROUP BY role ORDER BY cost_usd DESC`, [`-${days} days`]);
++ const bySource = all(`SELECT COALESCE(source,'platform') AS source, COALESCE(provider,'shared') AS provider, COUNT(*) AS calls, SUM(cost_usd) AS cost_usd, SUM(input_tokens) AS input_tokens, SUM(cached_tokens) AS cached_tokens
++ FROM ai_usage WHERE created_at >= date('now', ?) GROUP BY source, provider ORDER BY cost_usd DESC`, [`-${days} days`]);
++ const byOwner = all(`SELECT a.owner_user_id AS user_id, u.username, COUNT(*) AS calls, SUM(a.cost_usd) AS cost_usd,
++ SUM(a.input_tokens) AS input_tokens, SUM(a.cached_tokens) AS cached_tokens, SUM(a.output_tokens) AS output_tokens,
++ SUM(CASE WHEN a.created_at >= date('now') THEN a.cost_usd ELSE 0 END) AS cost_today,
++ SUM(CASE WHEN a.provider = 'byo' THEN a.cost_usd ELSE 0 END) AS cost_byo
++ FROM ai_usage a LEFT JOIN users u ON u.id = a.owner_user_id
++ WHERE a.created_at >= date('now', ?) AND a.owner_user_id IS NOT NULL
++ GROUP BY a.owner_user_id ORDER BY cost_usd DESC LIMIT 100`, [`-${days} days`]);
++ const byModel = all(`SELECT model, COUNT(*) AS calls, SUM(cost_usd) AS cost_usd, SUM(input_tokens) AS input_tokens, SUM(cached_tokens) AS cached_tokens, SUM(output_tokens) AS output_tokens
++ FROM ai_usage WHERE created_at >= date('now', ?) GROUP BY model ORDER BY cost_usd DESC`, [`-${days} days`]);
++ const cachedShare = totals.input_tokens ? totals.cached_tokens / totals.input_tokens : 0;
++ return { byDay, byKind, byRole, bySource, byOwner, byModel, totals, cachedShare, today: getAiCostToday() };
}
// Memories across ALL of a streamer's streams (for the per-streamer AI overview + explorer).