openvibers/openvibe.ai Commits
AI's user modules (Contracts 0.41.0, WS-B task 9): a run on a person's behalf follows their ai.preferences (style, length and perspective reach the system prompt; history: false keeps neither the input nor a cache entry; shaped output is never cached for reuse); ai.usage_summary (30-day runs, tokens, services) written as the owner by a 5-minute scan and a daily refresh, only when it changed
Changes
@@@ -13,7 +13,7 @@"better-sqlite3": "^11.10.0","dotenv": "^16.4.7","express": "^4.21.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.33.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.41.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0"},@@@ -1395,9 +1395,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.33.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.33.0",-- "integrity": "sha512-7q8NUhcPxgb8lxrVdC3pFbiPdjzliJM1mMP/1G5a2PmOr2tJVAXwYa8+a8YgCJvYeMSeMHWijLkIEiZ9IhllOg==",++ "version": "0.41.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.41.0",++ "integrity": "sha512-uT+Prd6NOO7iT2zn2cstxpPa6qmybc/O5KH0AQ8R5alzN/+WyUufd+h/yE0YxAw1+H/mJ4I2Q4xT+Go+CiPW2w==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -19,7 +19,7 @@"better-sqlite3": "^11.10.0","dotenv": "^16.4.7","express": "^4.21.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.33.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.41.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0"},@@@ -33,7 +33,10 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = glconst engine = createEngine({ registry, pool, fetcher });// ai.run.* to OpenVibe.Events through the outbox (server/events.js); before recovery, so its failures are announced.require('./events').init(db, { log });-- const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log });++ // ai.preferences (read) and ai.usage_summary (written): Network user modules (server/user-modules.js).++ const userModules = require('./user-modules').createUserModules({ db, config, env, fetchImpl, clock, log });++ userModules.ensureSchema();++ const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules });seed({ registry, quotas, config, env, db });const interrupted = runs.recoverInterrupted();if (interrupted) log.warn(`[ai] marked ${interrupted} interrupted run(s) failed (run.interrupted); callers can retry them`);@@@ -53,6 +56,7 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = glhousekeeping.unref?.();let server = null;++ if (listen) userModules.start();if (listen) {server = await new Promise((resolve, reject) => {const s = app.listen(config.port, config.host, () => resolve(s));@@@ -68,6 +72,7 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = glasync function close() {clearInterval(housekeeping);++ userModules.stop();keys.stop();if (server) {server.closeAllConnections?.();@@@ -18,6 +18,7 @@*/const { ids } = require('openvibe-contracts');const { AiError, sha256, stableStringify, parseJson, iso } = require('./util');++const { preferenceLines } = require('./user-modules');const schemas = require('./schemas');const events = require('./events');@@@ -57,7 +58,7 @@ function retainedInput(input, wf, debugRaw) {return { value, full: !redacted };}--function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console }) {++function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null }) {const inflight = new Map(); // run id -> { controller, promise, release, caller }const queue = []; // run ids waiting for the workerlet active = 0;@@@ -274,7 +275,13 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now});};try {-- const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash });++ // A run on a person's behalf follows their ai.preferences (Network user module): style, length and++ // perspective shape the prompt; history: false keeps neither their input nor a cache entry.++ const preferences = userModules && row.on_behalf_of ? await userModules.preferencesFor(row.on_behalf_of) : {};++ if (preferences.history === false) {++ db.prepare("UPDATE runs SET input = 'null', cache_key = NULL, options = json_set(options, '$.input_retained', json('false'), '$.history', json('false')) WHERE id = ?").run(id);++ }++ const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash, preferences });if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');// The status, its citations and the ai.run.succeeded event commit together.const done = db.transaction(() => {@@@ -290,7 +297,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowif (!done.changes) return; // cancelled while finishingif (r.provider) quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost });if (r.fallbackUsed) registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });-- if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none') {++ // Output shaped by someone's preferences, or kept for nobody (history off), is never cached for reuse.++ if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none' && preferences.history !== false && !preferenceLines(preferences)) {cache.put({ key: row.cache_key, scope, privacy: wf.cache_mode, workflowKey: wf.key, workflowVersion: wf.version, templateVersion: row.template_version, routeKey: row.route_key, routeVersion: row.route_version, model: r.model, inputHash: row.input_hash, output: r.output, runId: id, ttlSec: wf.cache_ttl_sec || config.runs.defaultCacheTtlSec });}} catch (err) {nnew file mode 100644@@@ -0,0 +1,155 @@++'use strict';++/**++ * AI's user modules on OpenVibe.Network (openvibe-contracts 0.41.0, roadmap WS-B task 9):++ *++ * ai.preferences read (AI is a listed reader of all five fields): style, length and perspective shape++ * the prompts of runs made on a person's behalf (engine.js withPreferences);++ * history: false keeps neither their input nor a cache entry (runs.js). Cached++ * 5 minutes per person; Network down or no record = no preferences.++ * ai.usage_summary written as the owner: runs_30d, tokens_30d, by_service, last_run_at, computed_at,++ * for runs on the person's behalf. scan() every 5 minutes (runs finished since the++ * last one), refresh() daily (the window moves); only when it changed (ai_module_pushes).++ *++ * Off without OV_OAUTH_CLIENT_SECRET. Only usr_ subjects.++ */++const crypto = require('crypto');++++const USR = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const PREFS_TTL_MS = 5 * 60 * 1000;++const DAY_MS = 86400000;++++function subjectOf(onBehalfOf) {++ let ref = onBehalfOf;++ if (typeof ref === 'string') { try { ref = JSON.parse(ref); } catch { return null; } }++ return ref && ref.type === 'user' && USR.test(String(ref.id || '')) ? ref.id : null;++}++++function createUserModules({ db, config, env = process.env, fetchImpl = globalThis.fetch, clock = { now: () => Date.now() }, log = console } = {}) {++ const secret = env.OV_OAUTH_CLIENT_SECRET || '';++ const enabled = !!secret && String(env.AI_USER_MODULES || '').toLowerCase() !== 'off';++ const base = String(config.networkInternalUrl || 'http://127.0.0.1:4000').replace(/\/+$/, '');++ const prefsCache = new Map();++ const stats = { written: 0, unchanged: 0, failed: 0, lastError: null };++ let tokens = null, lastScan = null, timers = [];++++ function tokenClient() {++ if (!tokens) {++ const { createServiceTokenClient } = require('openvibe-sdk/auth');++ tokens = createServiceTokenClient({ tokenUrl: `${base}/oauth/token`, clientId: env.OV_OAUTH_CLIENT_ID || 'ai', clientSecret: secret, fetch: fetchImpl });++ }++ return tokens;++ }++ async function call(method, ns, subject, data) {++ const t = await tokenClient().getToken({ audience: 'openvibe.network' });++ const res = await fetchImpl(`${base}/internal/modules/${ns}/${subject}`, {++ method, signal: AbortSignal.timeout(5000),++ headers: { Authorization: `Bearer ${t}`, Accept: 'application/json', ...(data ? { 'Content-Type': 'application/json' } : {}) },++ body: data ? JSON.stringify({ data }) : undefined,++ });++ return { status: res.status, body: await res.json().catch(() => null) };++ }++++ /** The person's ai.preferences for a run's on_behalf_of ({} when none, not a person, or unavailable). */++ async function preferencesFor(onBehalfOf) {++ const subject = subjectOf(onBehalfOf);++ if (!enabled || !subject) return {};++ const hit = prefsCache.get(subject);++ if (hit && clock.now() - hit.at < PREFS_TTL_MS) return hit.data;++ let data = {};++ try {++ const r = await call('GET', 'ai.preferences', subject);++ if (r.status === 200 && r.body && r.body.data && typeof r.body.data === 'object') data = r.body.data;++ else if (r.status !== 404) return hit ? hit.data : {};++ } catch { return hit ? hit.data : {}; }++ if (prefsCache.size > 5000) prefsCache.clear();++ prefsCache.set(subject, { data, at: clock.now() });++ return data;++ }++++ function ensureSchema() {++ db.exec(`CREATE TABLE IF NOT EXISTS ai_module_pushes (subject_id TEXT PRIMARY KEY, hash TEXT NOT NULL, pushed_at TEXT NOT NULL)`);++ }++++ /** ai.usage_summary for one person over the 30 days before now, or null when there were no runs. */++ function summarize(subject, now = clock.now()) {++ const since = new Date(now - 30 * DAY_MS).toISOString();++ const rows = db.prepare(`SELECT COALESCE(source_service, requester_id) AS svc, COUNT(*) AS n, SUM(tokens_in + tokens_out) AS tokens, MAX(created_at) AS last++ FROM runs WHERE json_extract(on_behalf_of, '$.type') = 'user' AND json_extract(on_behalf_of, '$.id') = ? AND created_at >= ?++ GROUP BY svc`).all(subject, since);++ const lastEver = db.prepare("SELECT MAX(created_at) AS last FROM runs WHERE json_extract(on_behalf_of, '$.type') = 'user' AND json_extract(on_behalf_of, '$.id') = ?").get(subject).last;++ if (!lastEver) return null;++ const by = {};++ for (const r of rows.slice(0, 30)) if (/^[a-z][a-z0-9-]{1,39}$/.test(String(r.svc || ''))) by[r.svc] = r.n;++ return {++ runs_30d: rows.reduce((a, r) => a + r.n, 0),++ tokens_30d: rows.reduce((a, r) => a + (Number(r.tokens) || 0), 0),++ by_service: by,++ last_run_at: lastEver,++ };++ }++++ async function push(subject, now = clock.now()) {++ if (!enabled || !USR.test(String(subject || ''))) return false;++ const data = summarize(subject, now);++ if (!data) return false;++ const hash = crypto.createHash('sha256').update(JSON.stringify(data)).digest('hex').slice(0, 32);++ const last = db.prepare('SELECT hash FROM ai_module_pushes WHERE subject_id = ?').get(subject);++ if (last && last.hash === hash) { stats.unchanged++; return false; }++ try {++ const r = await call('PUT', 'ai.usage_summary', subject, { ...data, computed_at: new Date(now).toISOString() });++ if (r.status >= 300) throw new Error(`Network answered ${r.status} ${(r.body && r.body.code) || ''}`);++ } catch (err) { stats.failed++; stats.lastError = err.message; return false; }++ db.prepare(`INSERT INTO ai_module_pushes (subject_id, hash, pushed_at) VALUES (?, ?, ?)++ ON CONFLICT(subject_id) DO UPDATE SET hash = excluded.hash, pushed_at = excluded.pushed_at`).run(subject, hash, new Date(now).toISOString());++ stats.written++;++ return true;++ }++++ const people = (sql, ...args) => db.prepare(sql).all(...args).map((r) => r.s).filter((s) => USR.test(String(s || '')));++++ /** People with a run created since the previous scan. */++ async function scan(now = clock.now()) {++ if (!enabled) return 0;++ const from = lastScan || new Date(now - 10 * 60 * 1000).toISOString();++ const to = new Date(now).toISOString();++ const subjects = people("SELECT DISTINCT json_extract(on_behalf_of, '$.id') AS s FROM runs WHERE on_behalf_of IS NOT NULL AND created_at >= ? AND created_at < ?", from, to);++ for (const s of subjects) await push(s, now);++ lastScan = to;++ return subjects.length;++ }++++ /** Everyone with a run in the last 31 days or a record already written (their window moves). */++ async function refresh(now = clock.now()) {++ if (!enabled) return 0;++ const since = new Date(now - 31 * DAY_MS).toISOString();++ const subjects = people(`SELECT DISTINCT json_extract(on_behalf_of, '$.id') AS s FROM runs WHERE on_behalf_of IS NOT NULL AND created_at >= ?++ UNION SELECT subject_id AS s FROM ai_module_pushes`, since);++ for (const s of subjects) await push(s, now);++ return subjects.length;++ }++++ function start() {++ if (!enabled || timers.length) return false;++ ensureSchema();++ const safe = (fn) => () => { fn().catch((err) => { stats.lastError = err.message; }); };++ timers = [setInterval(safe(() => scan()), 5 * 60 * 1000), setInterval(safe(() => refresh()), DAY_MS), setTimeout(safe(() => refresh()), 3 * 60 * 1000)];++ for (const t of timers) t.unref?.();++ return true;++ }++ function stop() { for (const t of timers) { clearInterval(t); clearTimeout(t); } timers = []; }++++ return { enabled, preferencesFor, ensureSchema, summarize, push, scan, refresh, start, stop, stats: () => ({ enabled, ...stats }) };++}++++/** The system-prompt lines a person's ai.preferences add ('' when none). */++function preferenceLines(prefs) {++ if (!prefs || typeof prefs !== 'object') return '';++ const lines = [];++ if (['neutral', 'casual', 'formal'].includes(prefs.style)) lines.push(`Write in a ${prefs.style} style.`);++ if (prefs.length === 'short') lines.push('Keep it short.');++ else if (prefs.length === 'long') lines.push('Go into detail: a longer answer is welcome.');++ if (typeof prefs.perspective === 'string' && prefs.perspective.trim()) lines.push(`Write from this perspective: ${prefs.perspective.trim().slice(0, 80)}.`);++ return lines.length ? `The person this is for asked for the following. Follow it unless the task's own format rules say otherwise.\n${lines.join('\n')}` : '';++}++++module.exports = { createUserModules, preferenceLines, subjectOf };@@@ -18,6 +18,7 @@const fs = require('fs');const { AiError, sha256, parseJsonLoose } = require('../util');const { render } = require('../templates');++const { preferenceLines } = require('../user-modules');const schemas = require('../schemas');const { PREPARE, POSTPROCESS } = require('./hooks');@@@ -86,7 +87,9 @@ function createEngine({ registry, pool, fetcher }) {return { output: prep.output, noProvider: true, template };}const vars = prep.vars || {};-- const system = render(template.system_prompt, vars).trim();++ // The person's ai.preferences (runs on their behalf; server/user-modules.js), after the template's own rules.++ const asked = preferenceLines(ctx && ctx.preferences);++ const system = [render(template.system_prompt, vars).trim(), asked].filter(Boolean).join('\n\n');const user = render(template.user_prompt, vars);const routeKey = step.route || template.default_route || wf.default_route || 'default.chat';const route = routeFor(routeKey);nnew file mode 100644@@@ -0,0 +1,95 @@++'use strict';++/**++ * AI's Network user modules (server/user-modules.js, Contracts 0.41.0, WS-B task 9):++ * - a run on a person's behalf follows their ai.preferences: style/length/perspective reach the system++ * prompt, history: false keeps neither the input nor a cache entry; other runs are untouched++ * - ai.usage_summary: runs, tokens and services over 30 days, written as the owner, only when changed++ */++const assert = require('assert');++const http = require('http');++const { boot, request, token, suite, seamServer, ALL } = require('./helpers');++const { preferenceLines } = require('../server/user-modules');++++const t = suite('user modules');++const live = token('live', ALL);++const A = { type: 'user', id: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPA' };++const B = { type: 'user', id: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPB' };++let h, seam, network;++const prefs = { [A.id]: { style: 'casual', length: 'short', perspective: 'a night-shift worker', history: false } };++const puts = [];++const systemOf = () => (seam.last.system || []).map((s) => s.text || s).join('\n');++const summarize = (extra = {}) => request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: live, body: { workflow: 'ai.summarize', input: { text: 'A long day at the plant, told briefly.' }, ...extra } });++++t.test('boot with a stub Network and a seam provider', async () => {++ network = http.createServer((req, res) => {++ let raw = '';++ req.on('data', (c) => { raw += c; });++ req.on('end', () => {++ res.setHeader('content-type', 'application/json');++ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'svc-ai', token_type: 'Bearer', expires_in: 300 }));++ assert.strictEqual(req.headers.authorization, 'Bearer svc-ai');++ const m = req.url.match(/^\/internal\/modules\/(ai\.[a-z_]+)\/(usr_[0-9A-Z]+)$/);++ if (m && req.method === 'GET' && m[1] === 'ai.preferences') {++ if (!prefs[m[2]]) { res.statusCode = 404; return res.end(JSON.stringify({ code: 'modules.not_found' })); }++ return res.end(JSON.stringify({ namespace: 'ai.preferences', revision: 1, data: prefs[m[2]] }));++ }++ if (m && req.method === 'PUT' && m[1] === 'ai.usage_summary') { puts.push({ subject: m[2], data: JSON.parse(raw).data }); res.statusCode = 201; return res.end('{}'); }++ res.statusCode = 404; res.end('{}');++ });++ });++ await new Promise((r) => network.listen(0, '127.0.0.1', r));++ seam = await seamServer(() => ({ text: `summary #${seam.calls}`, usage: { input: 30, output: 10 } }));++ h = await boot({ env: { AI_STUB_FALLBACK: 'false', OV_OAUTH_CLIENT_SECRET: 'x'.repeat(40), OV_NETWORK_INTERNAL_URL: `http://127.0.0.1:${network.address().port}` } });++ await request(h.base, 'POST', '/api/v1/providers', { tok: live, body: { key: 'seam', kind: 'http', base_url: seam.url, auth_mode: 'none', capabilities: ['summarize', 'generate', 'chat', 'json'] } });++ await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: live, body: { primary: { provider: 'seam' }, fallbacks: [] } });++});++++t.test('preference lines', () => {++ assert.strictEqual(preferenceLines({}), '');++ assert.strictEqual(preferenceLines({ personalization: false, history: true }), '', 'only style, length and perspective change a prompt');++ const l = preferenceLines({ style: 'formal', length: 'long', perspective: 'x'.repeat(200) });++ assert.ok(l.includes('formal style') && l.includes('longer answer') && l.includes('x'.repeat(80) + '.') && !l.includes('x'.repeat(81)));++});++++t.test("a run on A's behalf follows A's preferences; history off keeps no input and no cache entry", async () => {++ const r = await summarize({ on_behalf_of: A });++ assert.strictEqual(r.body.run.status, 'succeeded', r.text);++ const sys = systemOf();++ assert.ok(sys.includes('casual style') && sys.includes('Keep it short.') && sys.includes('a night-shift worker'), sys);++ const row = h.db.prepare('SELECT input, cache_key, options FROM runs WHERE id = ?').get(r.body.run.id);++ assert.strictEqual(row.input, 'null'); assert.strictEqual(row.cache_key, null);++ assert.strictEqual(JSON.parse(row.options).history, false);++ const calls = seam.calls;++ const again = await summarize({ on_behalf_of: A });++ assert.strictEqual(again.body.run.status, 'succeeded', 'computed again: nothing was cached');++ assert.strictEqual(seam.calls, calls + 1);++});++++t.test('runs for someone without preferences, or for nobody, are untouched', async () => {++ await summarize({ on_behalf_of: B });++ assert.ok(!systemOf().includes('The person this is for asked'));++ const r = await summarize();++ assert.ok(!systemOf().includes('The person this is for asked'));++ assert.notStrictEqual(h.db.prepare('SELECT input FROM runs WHERE id = ?').get(r.body.run.id).input, 'null');++});++++t.test('ai.usage_summary: 30-day runs and tokens per person, written only when changed', async () => {++ const mods = require('../server/user-modules').createUserModules({ db: h.db, config: h.config || { networkInternalUrl: h.env.OV_NETWORK_INTERNAL_URL }, env: h.env, log: {} });++ mods.ensureSchema();++ const s = mods.summarize(A.id);++ assert.strictEqual(s.runs_30d, 2); assert.strictEqual(s.tokens_30d, 80);++ assert.deepStrictEqual(s.by_service, { live: 2 });++ assert.ok(await mods.push(A.id));++ assert.strictEqual(puts.at(-1).subject, A.id);++ assert.ok(puts.at(-1).data.computed_at);++ const { modules } = require('openvibe-contracts');++ if (modules.get('ai.usage_summary')) assert.ok(modules.validateData('ai.usage_summary', puts.at(-1).data).valid);++ assert.strictEqual(await mods.push(A.id), false, 'unchanged: not written again');++ assert.strictEqual(await mods.scan(Date.now() + 1000), 2, 'the scan sees A and B (their runs were just made)');++ assert.deepStrictEqual(puts.map((p) => p.subject).sort(), [A.id, B.id].sort());++ assert.strictEqual(mods.summarize('usr_01JAB2C3D4E5F6G7H8J9K0MNPC'), null);++});++++t.test('shutdown', async () => { await h.stop(); await seam.close(); network.close(); });++++t.run();
@@ -13,7 +13,7 @@
@@ -13,7 +13,7 @@
"better-sqlite3": "^11.10.0",
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.33.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.41.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0"
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0"
},
},
@@ -1395,9 +1395,9 @@
@@ -1395,9 +1395,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.33.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.33.0",
- "integrity": "sha512-7q8NUhcPxgb8lxrVdC3pFbiPdjzliJM1mMP/1G5a2PmOr2tJVAXwYa8+a8YgCJvYeMSeMHWijLkIEiZ9IhllOg==",
+ "version": "0.41.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.41.0",
+ "integrity": "sha512-uT+Prd6NOO7iT2zn2cstxpPa6qmybc/O5KH0AQ8R5alzN/+WyUufd+h/yE0YxAw1+H/mJ4I2Q4xT+Go+CiPW2w==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -19,7 +19,7 @@
@@ -19,7 +19,7 @@
"better-sqlite3": "^11.10.0",
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.33.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.41.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0"
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0"
},
},
@@ -33,7 +33,10 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
@@ -33,7 +33,10 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
const engine = createEngine({ registry, pool, fetcher });
const engine = createEngine({ registry, pool, fetcher });
// ai.run.* to OpenVibe.Events through the outbox (server/events.js); before recovery, so its failures are announced.
// ai.run.* to OpenVibe.Events through the outbox (server/events.js); before recovery, so its failures are announced.
require('./events').init(db, { log });
require('./events').init(db, { log });
- const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log });
+ // ai.preferences (read) and ai.usage_summary (written): Network user modules (server/user-modules.js).
+ const userModules = require('./user-modules').createUserModules({ db, config, env, fetchImpl, clock, log });
+ userModules.ensureSchema();
+ const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules });
seed({ registry, quotas, config, env, db });
seed({ registry, quotas, config, env, db });
const interrupted = runs.recoverInterrupted();
const interrupted = runs.recoverInterrupted();
if (interrupted) log.warn(`[ai] marked ${interrupted} interrupted run(s) failed (run.interrupted); callers can retry them`);
if (interrupted) log.warn(`[ai] marked ${interrupted} interrupted run(s) failed (run.interrupted); callers can retry them`);
@@ -53,6 +56,7 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
@@ -53,6 +56,7 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
housekeeping.unref?.();
housekeeping.unref?.();
let server = null;
let server = null;
+ if (listen) userModules.start();
if (listen) {
if (listen) {
server = await new Promise((resolve, reject) => {
server = await new Promise((resolve, reject) => {
const s = app.listen(config.port, config.host, () => resolve(s));
const s = app.listen(config.port, config.host, () => resolve(s));
@@ -68,6 +72,7 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
@@ -68,6 +72,7 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
async function close() {
async function close() {
clearInterval(housekeeping);
clearInterval(housekeeping);
+ userModules.stop();
keys.stop();
keys.stop();
if (server) {
if (server) {
server.closeAllConnections?.();
server.closeAllConnections?.();
@@ -18,6 +18,7 @@
@@ -18,6 +18,7 @@
*/
*/
const { ids } = require('openvibe-contracts');
const { ids } = require('openvibe-contracts');
const { AiError, sha256, stableStringify, parseJson, iso } = require('./util');
const { AiError, sha256, stableStringify, parseJson, iso } = require('./util');
+const { preferenceLines } = require('./user-modules');
const schemas = require('./schemas');
const schemas = require('./schemas');
const events = require('./events');
const events = require('./events');
@@ -57,7 +58,7 @@ function retainedInput(input, wf, debugRaw) {
@@ -57,7 +58,7 @@ function retainedInput(input, wf, debugRaw) {
return { value, full: !redacted };
return { value, full: !redacted };
}
}
-function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console }) {
+function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null }) {
const inflight = new Map(); // run id -> { controller, promise, release, caller }
const inflight = new Map(); // run id -> { controller, promise, release, caller }
const queue = []; // run ids waiting for the worker
const queue = []; // run ids waiting for the worker
let active = 0;
let active = 0;
@@ -274,7 +275,13 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -274,7 +275,13 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
});
});
};
};
try {
try {
- const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash });
+ // A run on a person's behalf follows their ai.preferences (Network user module): style, length and
+ // perspective shape the prompt; history: false keeps neither their input nor a cache entry.
+ const preferences = userModules && row.on_behalf_of ? await userModules.preferencesFor(row.on_behalf_of) : {};
+ if (preferences.history === false) {
+ db.prepare("UPDATE runs SET input = 'null', cache_key = NULL, options = json_set(options, '$.input_retained', json('false'), '$.history', json('false')) WHERE id = ?").run(id);
+ }
+ const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash, preferences });
if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');
if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');
// The status, its citations and the ai.run.succeeded event commit together.
// The status, its citations and the ai.run.succeeded event commit together.
const done = db.transaction(() => {
const done = db.transaction(() => {
@@ -290,7 +297,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -290,7 +297,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
if (!done.changes) return; // cancelled while finishing
if (!done.changes) return; // cancelled while finishing
if (r.provider) quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost });
if (r.provider) quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost });
if (r.fallbackUsed) registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });
if (r.fallbackUsed) registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });
- if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none') {
+ // Output shaped by someone's preferences, or kept for nobody (history off), is never cached for reuse.
+ if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none' && preferences.history !== false && !preferenceLines(preferences)) {
cache.put({ key: row.cache_key, scope, privacy: wf.cache_mode, workflowKey: wf.key, workflowVersion: wf.version, templateVersion: row.template_version, routeKey: row.route_key, routeVersion: row.route_version, model: r.model, inputHash: row.input_hash, output: r.output, runId: id, ttlSec: wf.cache_ttl_sec || config.runs.defaultCacheTtlSec });
cache.put({ key: row.cache_key, scope, privacy: wf.cache_mode, workflowKey: wf.key, workflowVersion: wf.version, templateVersion: row.template_version, routeKey: row.route_key, routeVersion: row.route_version, model: r.model, inputHash: row.input_hash, output: r.output, runId: id, ttlSec: wf.cache_ttl_sec || config.runs.defaultCacheTtlSec });
}
}
} catch (err) {
} catch (err) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,155 @@
@@ -0,0 +1,155 @@
+'use strict';
+/**
+ * AI's user modules on OpenVibe.Network (openvibe-contracts 0.41.0, roadmap WS-B task 9):
+ *
+ * ai.preferences read (AI is a listed reader of all five fields): style, length and perspective shape
+ * the prompts of runs made on a person's behalf (engine.js withPreferences);
+ * history: false keeps neither their input nor a cache entry (runs.js). Cached
+ * 5 minutes per person; Network down or no record = no preferences.
+ * ai.usage_summary written as the owner: runs_30d, tokens_30d, by_service, last_run_at, computed_at,
+ * for runs on the person's behalf. scan() every 5 minutes (runs finished since the
+ * last one), refresh() daily (the window moves); only when it changed (ai_module_pushes).
+ *
+ * Off without OV_OAUTH_CLIENT_SECRET. Only usr_ subjects.
+ */
+const crypto = require('crypto');
+
+const USR = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const PREFS_TTL_MS = 5 * 60 * 1000;
+const DAY_MS = 86400000;
+
+function subjectOf(onBehalfOf) {
+ let ref = onBehalfOf;
+ if (typeof ref === 'string') { try { ref = JSON.parse(ref); } catch { return null; } }
+ return ref && ref.type === 'user' && USR.test(String(ref.id || '')) ? ref.id : null;
+}
+
+function createUserModules({ db, config, env = process.env, fetchImpl = globalThis.fetch, clock = { now: () => Date.now() }, log = console } = {}) {
+ const secret = env.OV_OAUTH_CLIENT_SECRET || '';
+ const enabled = !!secret && String(env.AI_USER_MODULES || '').toLowerCase() !== 'off';
+ const base = String(config.networkInternalUrl || 'http://127.0.0.1:4000').replace(/\/+$/, '');
+ const prefsCache = new Map();
+ const stats = { written: 0, unchanged: 0, failed: 0, lastError: null };
+ let tokens = null, lastScan = null, timers = [];
+
+ function tokenClient() {
+ if (!tokens) {
+ const { createServiceTokenClient } = require('openvibe-sdk/auth');
+ tokens = createServiceTokenClient({ tokenUrl: `${base}/oauth/token`, clientId: env.OV_OAUTH_CLIENT_ID || 'ai', clientSecret: secret, fetch: fetchImpl });
+ }
+ return tokens;
+ }
+ async function call(method, ns, subject, data) {
+ const t = await tokenClient().getToken({ audience: 'openvibe.network' });
+ const res = await fetchImpl(`${base}/internal/modules/${ns}/${subject}`, {
+ method, signal: AbortSignal.timeout(5000),
+ headers: { Authorization: `Bearer ${t}`, Accept: 'application/json', ...(data ? { 'Content-Type': 'application/json' } : {}) },
+ body: data ? JSON.stringify({ data }) : undefined,
+ });
+ return { status: res.status, body: await res.json().catch(() => null) };
+ }
+
+ /** The person's ai.preferences for a run's on_behalf_of ({} when none, not a person, or unavailable). */
+ async function preferencesFor(onBehalfOf) {
+ const subject = subjectOf(onBehalfOf);
+ if (!enabled || !subject) return {};
+ const hit = prefsCache.get(subject);
+ if (hit && clock.now() - hit.at < PREFS_TTL_MS) return hit.data;
+ let data = {};
+ try {
+ const r = await call('GET', 'ai.preferences', subject);
+ if (r.status === 200 && r.body && r.body.data && typeof r.body.data === 'object') data = r.body.data;
+ else if (r.status !== 404) return hit ? hit.data : {};
+ } catch { return hit ? hit.data : {}; }
+ if (prefsCache.size > 5000) prefsCache.clear();
+ prefsCache.set(subject, { data, at: clock.now() });
+ return data;
+ }
+
+ function ensureSchema() {
+ db.exec(`CREATE TABLE IF NOT EXISTS ai_module_pushes (subject_id TEXT PRIMARY KEY, hash TEXT NOT NULL, pushed_at TEXT NOT NULL)`);
+ }
+
+ /** ai.usage_summary for one person over the 30 days before now, or null when there were no runs. */
+ function summarize(subject, now = clock.now()) {
+ const since = new Date(now - 30 * DAY_MS).toISOString();
+ const rows = db.prepare(`SELECT COALESCE(source_service, requester_id) AS svc, COUNT(*) AS n, SUM(tokens_in + tokens_out) AS tokens, MAX(created_at) AS last
+ FROM runs WHERE json_extract(on_behalf_of, '$.type') = 'user' AND json_extract(on_behalf_of, '$.id') = ? AND created_at >= ?
+ GROUP BY svc`).all(subject, since);
+ const lastEver = db.prepare("SELECT MAX(created_at) AS last FROM runs WHERE json_extract(on_behalf_of, '$.type') = 'user' AND json_extract(on_behalf_of, '$.id') = ?").get(subject).last;
+ if (!lastEver) return null;
+ const by = {};
+ for (const r of rows.slice(0, 30)) if (/^[a-z][a-z0-9-]{1,39}$/.test(String(r.svc || ''))) by[r.svc] = r.n;
+ return {
+ runs_30d: rows.reduce((a, r) => a + r.n, 0),
+ tokens_30d: rows.reduce((a, r) => a + (Number(r.tokens) || 0), 0),
+ by_service: by,
+ last_run_at: lastEver,
+ };
+ }
+
+ async function push(subject, now = clock.now()) {
+ if (!enabled || !USR.test(String(subject || ''))) return false;
+ const data = summarize(subject, now);
+ if (!data) return false;
+ const hash = crypto.createHash('sha256').update(JSON.stringify(data)).digest('hex').slice(0, 32);
+ const last = db.prepare('SELECT hash FROM ai_module_pushes WHERE subject_id = ?').get(subject);
+ if (last && last.hash === hash) { stats.unchanged++; return false; }
+ try {
+ const r = await call('PUT', 'ai.usage_summary', subject, { ...data, computed_at: new Date(now).toISOString() });
+ if (r.status >= 300) throw new Error(`Network answered ${r.status} ${(r.body && r.body.code) || ''}`);
+ } catch (err) { stats.failed++; stats.lastError = err.message; return false; }
+ db.prepare(`INSERT INTO ai_module_pushes (subject_id, hash, pushed_at) VALUES (?, ?, ?)
+ ON CONFLICT(subject_id) DO UPDATE SET hash = excluded.hash, pushed_at = excluded.pushed_at`).run(subject, hash, new Date(now).toISOString());
+ stats.written++;
+ return true;
+ }
+
+ const people = (sql, ...args) => db.prepare(sql).all(...args).map((r) => r.s).filter((s) => USR.test(String(s || '')));
+
+ /** People with a run created since the previous scan. */
+ async function scan(now = clock.now()) {
+ if (!enabled) return 0;
+ const from = lastScan || new Date(now - 10 * 60 * 1000).toISOString();
+ const to = new Date(now).toISOString();
+ const subjects = people("SELECT DISTINCT json_extract(on_behalf_of, '$.id') AS s FROM runs WHERE on_behalf_of IS NOT NULL AND created_at >= ? AND created_at < ?", from, to);
+ for (const s of subjects) await push(s, now);
+ lastScan = to;
+ return subjects.length;
+ }
+
+ /** Everyone with a run in the last 31 days or a record already written (their window moves). */
+ async function refresh(now = clock.now()) {
+ if (!enabled) return 0;
+ const since = new Date(now - 31 * DAY_MS).toISOString();
+ const subjects = people(`SELECT DISTINCT json_extract(on_behalf_of, '$.id') AS s FROM runs WHERE on_behalf_of IS NOT NULL AND created_at >= ?
+ UNION SELECT subject_id AS s FROM ai_module_pushes`, since);
+ for (const s of subjects) await push(s, now);
+ return subjects.length;
+ }
+
+ function start() {
+ if (!enabled || timers.length) return false;
+ ensureSchema();
+ const safe = (fn) => () => { fn().catch((err) => { stats.lastError = err.message; }); };
+ timers = [setInterval(safe(() => scan()), 5 * 60 * 1000), setInterval(safe(() => refresh()), DAY_MS), setTimeout(safe(() => refresh()), 3 * 60 * 1000)];
+ for (const t of timers) t.unref?.();
+ return true;
+ }
+ function stop() { for (const t of timers) { clearInterval(t); clearTimeout(t); } timers = []; }
+
+ return { enabled, preferencesFor, ensureSchema, summarize, push, scan, refresh, start, stop, stats: () => ({ enabled, ...stats }) };
+}
+
+/** The system-prompt lines a person's ai.preferences add ('' when none). */
+function preferenceLines(prefs) {
+ if (!prefs || typeof prefs !== 'object') return '';
+ const lines = [];
+ if (['neutral', 'casual', 'formal'].includes(prefs.style)) lines.push(`Write in a ${prefs.style} style.`);
+ if (prefs.length === 'short') lines.push('Keep it short.');
+ else if (prefs.length === 'long') lines.push('Go into detail: a longer answer is welcome.');
+ if (typeof prefs.perspective === 'string' && prefs.perspective.trim()) lines.push(`Write from this perspective: ${prefs.perspective.trim().slice(0, 80)}.`);
+ return lines.length ? `The person this is for asked for the following. Follow it unless the task's own format rules say otherwise.\n${lines.join('\n')}` : '';
+}
+
+module.exports = { createUserModules, preferenceLines, subjectOf };
@@ -18,6 +18,7 @@
@@ -18,6 +18,7 @@
const fs = require('fs');
const fs = require('fs');
const { AiError, sha256, parseJsonLoose } = require('../util');
const { AiError, sha256, parseJsonLoose } = require('../util');
const { render } = require('../templates');
const { render } = require('../templates');
+const { preferenceLines } = require('../user-modules');
const schemas = require('../schemas');
const schemas = require('../schemas');
const { PREPARE, POSTPROCESS } = require('./hooks');
const { PREPARE, POSTPROCESS } = require('./hooks');
@@ -86,7 +87,9 @@ function createEngine({ registry, pool, fetcher }) {
@@ -86,7 +87,9 @@ function createEngine({ registry, pool, fetcher }) {
return { output: prep.output, noProvider: true, template };
return { output: prep.output, noProvider: true, template };
}
}
const vars = prep.vars || {};
const vars = prep.vars || {};
- const system = render(template.system_prompt, vars).trim();
+ // The person's ai.preferences (runs on their behalf; server/user-modules.js), after the template's own rules.
+ const asked = preferenceLines(ctx && ctx.preferences);
+ const system = [render(template.system_prompt, vars).trim(), asked].filter(Boolean).join('\n\n');
const user = render(template.user_prompt, vars);
const user = render(template.user_prompt, vars);
const routeKey = step.route || template.default_route || wf.default_route || 'default.chat';
const routeKey = step.route || template.default_route || wf.default_route || 'default.chat';
const route = routeFor(routeKey);
const route = routeFor(routeKey);
new file mode 100644
new file mode 100644
@@ -0,0 +1,95 @@
@@ -0,0 +1,95 @@
+'use strict';
+/**
+ * AI's Network user modules (server/user-modules.js, Contracts 0.41.0, WS-B task 9):
+ * - a run on a person's behalf follows their ai.preferences: style/length/perspective reach the system
+ * prompt, history: false keeps neither the input nor a cache entry; other runs are untouched
+ * - ai.usage_summary: runs, tokens and services over 30 days, written as the owner, only when changed
+ */
+const assert = require('assert');
+const http = require('http');
+const { boot, request, token, suite, seamServer, ALL } = require('./helpers');
+const { preferenceLines } = require('../server/user-modules');
+
+const t = suite('user modules');
+const live = token('live', ALL);
+const A = { type: 'user', id: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPA' };
+const B = { type: 'user', id: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPB' };
+let h, seam, network;
+const prefs = { [A.id]: { style: 'casual', length: 'short', perspective: 'a night-shift worker', history: false } };
+const puts = [];
+const systemOf = () => (seam.last.system || []).map((s) => s.text || s).join('\n');
+const summarize = (extra = {}) => request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: live, body: { workflow: 'ai.summarize', input: { text: 'A long day at the plant, told briefly.' }, ...extra } });
+
+t.test('boot with a stub Network and a seam provider', async () => {
+ network = http.createServer((req, res) => {
+ let raw = '';
+ req.on('data', (c) => { raw += c; });
+ req.on('end', () => {
+ res.setHeader('content-type', 'application/json');
+ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'svc-ai', token_type: 'Bearer', expires_in: 300 }));
+ assert.strictEqual(req.headers.authorization, 'Bearer svc-ai');
+ const m = req.url.match(/^\/internal\/modules\/(ai\.[a-z_]+)\/(usr_[0-9A-Z]+)$/);
+ if (m && req.method === 'GET' && m[1] === 'ai.preferences') {
+ if (!prefs[m[2]]) { res.statusCode = 404; return res.end(JSON.stringify({ code: 'modules.not_found' })); }
+ return res.end(JSON.stringify({ namespace: 'ai.preferences', revision: 1, data: prefs[m[2]] }));
+ }
+ if (m && req.method === 'PUT' && m[1] === 'ai.usage_summary') { puts.push({ subject: m[2], data: JSON.parse(raw).data }); res.statusCode = 201; return res.end('{}'); }
+ res.statusCode = 404; res.end('{}');
+ });
+ });
+ await new Promise((r) => network.listen(0, '127.0.0.1', r));
+ seam = await seamServer(() => ({ text: `summary #${seam.calls}`, usage: { input: 30, output: 10 } }));
+ h = await boot({ env: { AI_STUB_FALLBACK: 'false', OV_OAUTH_CLIENT_SECRET: 'x'.repeat(40), OV_NETWORK_INTERNAL_URL: `http://127.0.0.1:${network.address().port}` } });
+ await request(h.base, 'POST', '/api/v1/providers', { tok: live, body: { key: 'seam', kind: 'http', base_url: seam.url, auth_mode: 'none', capabilities: ['summarize', 'generate', 'chat', 'json'] } });
+ await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: live, body: { primary: { provider: 'seam' }, fallbacks: [] } });
+});
+
+t.test('preference lines', () => {
+ assert.strictEqual(preferenceLines({}), '');
+ assert.strictEqual(preferenceLines({ personalization: false, history: true }), '', 'only style, length and perspective change a prompt');
+ const l = preferenceLines({ style: 'formal', length: 'long', perspective: 'x'.repeat(200) });
+ assert.ok(l.includes('formal style') && l.includes('longer answer') && l.includes('x'.repeat(80) + '.') && !l.includes('x'.repeat(81)));
+});
+
+t.test("a run on A's behalf follows A's preferences; history off keeps no input and no cache entry", async () => {
+ const r = await summarize({ on_behalf_of: A });
+ assert.strictEqual(r.body.run.status, 'succeeded', r.text);
+ const sys = systemOf();
+ assert.ok(sys.includes('casual style') && sys.includes('Keep it short.') && sys.includes('a night-shift worker'), sys);
+ const row = h.db.prepare('SELECT input, cache_key, options FROM runs WHERE id = ?').get(r.body.run.id);
+ assert.strictEqual(row.input, 'null'); assert.strictEqual(row.cache_key, null);
+ assert.strictEqual(JSON.parse(row.options).history, false);
+ const calls = seam.calls;
+ const again = await summarize({ on_behalf_of: A });
+ assert.strictEqual(again.body.run.status, 'succeeded', 'computed again: nothing was cached');
+ assert.strictEqual(seam.calls, calls + 1);
+});
+
+t.test('runs for someone without preferences, or for nobody, are untouched', async () => {
+ await summarize({ on_behalf_of: B });
+ assert.ok(!systemOf().includes('The person this is for asked'));
+ const r = await summarize();
+ assert.ok(!systemOf().includes('The person this is for asked'));
+ assert.notStrictEqual(h.db.prepare('SELECT input FROM runs WHERE id = ?').get(r.body.run.id).input, 'null');
+});
+
+t.test('ai.usage_summary: 30-day runs and tokens per person, written only when changed', async () => {
+ const mods = require('../server/user-modules').createUserModules({ db: h.db, config: h.config || { networkInternalUrl: h.env.OV_NETWORK_INTERNAL_URL }, env: h.env, log: {} });
+ mods.ensureSchema();
+ const s = mods.summarize(A.id);
+ assert.strictEqual(s.runs_30d, 2); assert.strictEqual(s.tokens_30d, 80);
+ assert.deepStrictEqual(s.by_service, { live: 2 });
+ assert.ok(await mods.push(A.id));
+ assert.strictEqual(puts.at(-1).subject, A.id);
+ assert.ok(puts.at(-1).data.computed_at);
+ const { modules } = require('openvibe-contracts');
+ if (modules.get('ai.usage_summary')) assert.ok(modules.validateData('ai.usage_summary', puts.at(-1).data).valid);
+ assert.strictEqual(await mods.push(A.id), false, 'unchanged: not written again');
+ assert.strictEqual(await mods.scan(Date.now() + 1000), 2, 'the scan sees A and B (their runs were just made)');
+ assert.deepStrictEqual(puts.map((p) => p.subject).sort(), [A.id, B.id].sort());
+ assert.strictEqual(mods.summarize('usr_01JAB2C3D4E5F6G7H8J9K0MNPC'), null);
+});
+
+t.test('shutdown', async () => { await h.stop(); await seam.close(); network.close(); });
+
+t.run();