openvibers/openvibe.chat Commits

The chat-AI summaries run in Chat (plan T3 step 2): Live's job ported over Chat's own messages and projections (same cadence, selection, prompts, fields and limits), calling OpenVibe.AI with Chat's own token (workflows chat.global/chat.profile, namespace chat), with an extractive fallback when AI is unavailable, writing Chat's tables under the authority rule; /api/chat/ai/{global,timeline,user/:id,anon/:anonId,relay/:platform/:username,timeline/:username} with Live's shapes. Off until CHAT_AI_ENABLED (set at the cutover, when Live's job stops). test/chat-ai.test.js; 43/43 (verified).

OP OpenVibers committed 29 Sept 2026 · adb2ab3faa · 10 files, +1102 −0


Changes

ddiff --git a/README.md b/README.md
iindex f764bce..6cfbe0d 100644
---- a/README.md
++++ b/README.md
@@@ -78,6 +78,17 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
`chat.moderated-channels-result@1`, `chat.emote-count-result@1`). Live caches each answer 30 s.
- Alert sounds: the bridge op `playAlertSound [streamerId, streamId, kind]` makes Chat resolve the
channel's own settings row, read the clip and broadcast it, so Live only names the alert.
++ - **The chat-AI summaries** (plan T3 step 2, decision 5) — Live's `server/ai/chat-ai.js` job moved
++ here: a background poller (off unless `CHAT_AI_ENABLED=1`) that folds Chat's own messages into
++ rolling global/per-user/per-relay/per-anon insights and the append-only timeline, writing
++ `chat_ai_summaries`/`chat_timeline_events` through the staged writers. The model is OpenVibe.AI
++ called with Chat's own service token (`ai.run.create`/`ai.run.read`, audience `openvibe.ai`,
++ namespace `chat.*`, workflows `chat.global`/`chat.profile`); when AI does not answer the job
++ stores a deterministic extractive summary instead, so the routes always have something to serve.
++ Routes (Live's paths, parameters, shapes and public visibility, now under `/api/chat/ai/`):
++ `GET /global`, `/timeline` (`?before=&since=&q=&limit=`), `/user/:id`, `/anon/:anonId`,
++ `/relay/:platform/:username`, `/timeline/:username`. A write while Live still owns the two
++ tables is skipped with one log line, never forced.
- Every write keeps the `table_authority` gate: while Live still writes the table the route
answers `503 { ok: false, error: 'not yet' }` and never writes.
@@@ -91,6 +102,7 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
| Live's own chat pushes and writes (AI viewers, relays, donations, `/api/mod`, recaps, calls) | **Live → Chat** | `POST /internal/live/calls` (`chat.live_bridge.write`), presence `GET /internal/live/presence` (`chat.presence.read`) |
| The six staged tables (moderators, moderation settings, emotes, tags, AI summaries, timeline) | **Chat** (after the T3 flip; Live until then) | Chat's own SQLite; Live reads through `GET /internal/moderation/*` (`chat.moderation.read`), 30 s cached |
| Emote image bytes | **OpenVibe.Media** (namespace `chat`) | `server/media/client.js` (openvibe-sdk `createObjectsClient`), Chat's service token (`media.object.upload` / `.delete`); the row keeps `media_url` + `media_asset_id` |
++| Chat-AI summaries (global / per-chatter insight + timeline) | **OpenVibe.AI** (namespace `chat.*`) | `server/ai/client.js` (openvibe-sdk `createAiClient`), Chat's service token (`ai.run.create` / `ai.run.read`); workflows `chat.global` / `chat.profile`; extractive fallback when AI does not answer |
| Identity | **OpenVibe.Network** | user tokens are resolved by Live (its account links); service tokens from `/oauth/token` |
| Events | **OpenVibe.Events** | `events_outbox` → `POST /api/v1/events` when `EVENTS_URL` is set (`events.event.publish`); Chat's subscriptions deliver to `POST /internal/events` (`server/events/consumer.js`, `events.subscription.manage`) |
| A person's chat preferences | **OpenVibe.Network** user module `chat.preferences` (Chat owns the namespace) | `server/prefs/` → `GET/PUT/DELETE /internal/modules/chat.preferences/:subject` (`network.modules.read` / `.write`), cached per person |
@@@ -420,6 +432,7 @@ server/net/service-auth.js service tokens: client (Chat → others) and guard (o
server/prefs/ chat preferences in the Network user module chat.preferences (routes, cache, migration from Live)
server/calls/ moved from Live: the call server (/ws/call), its REST routes, Live's stream hooks (/internal/calls), the calls lifecycle
server/media/client.js OpenVibe.Media objects (namespace chat): emote image bytes, with Chat's own service token
++server/ai/ moved from Live: chat-ai.js (the rolling insight job, off unless CHAT_AI_ENABLED), client.js (OpenVibe.AI runs, namespace chat.*), extractive.js (the fallback summary)
server/db/ schema.sql, database.js (Live's chat functions, same names and arguments)
scripts/ import-from-live, parity-check, parity, mirror-flush, table-authority, migrate-chat-preferences, subscribe-events, n-1-record
docs/ cutover.md, calls-cutover.md, staged-tables-cutover.md, parity.md, live-patch.diff, capabilities-proposal/
ddiff --git a/server/ai/chat-ai.js b/server/ai/chat-ai.js
nnew file mode 100644
iindex 0000000..47a86b1
---- /dev/null
++++ b/server/ai/chat-ai.js
@@@ -0,0 +1,464 @@
++/**
++ * chat-ai.js — rolling AI insight for chat, moved into Chat (plan T3 step 2, decision 5).
++ *
++ * Live's server/ai/chat-ai.js, ported: same tunables, cadence, selection rules and output fields,
++ * but the messages and the summaries are Chat's own tables (server/db/database.js) and the model is
++ * OpenVibe.AI called with Chat's own service token (server/ai/client.js). Where Live's
++ * getUsersNeedingChatAi joined Live's tables, the reads join Chat's ctx_* projections.
++ *
++ * Two products, one engine:
++ * • GLOBAL — an overview + timeline of ALL chat across the whole site.
++ * • PER-SUBJECT — a "today (24h) vs all-time" read on a chatter (native user, relay user, anon).
++ *
++ * Lower cost + graceful scaling as Live: a DB poller with the high-water mark in
++ * chat_ai_summaries.last_message_id, an adaptive window, a few subjects per tick.
++ *
++ * Writes go through Chat's staged writers under the _assertChatWrites rule: while Live is still the
++ * table's authority a write is skipped with one log line, never forced. The job itself is off
++ * unless config.ai.enabled (CHAT_AI_ENABLED=1) — off in development and the test harness, like
++ * Chat's other background jobs.
++ */
++'use strict';
++
++const config = require('../config');
++const db = require('../db/database');
++const aiClient = require('./client');
++const extractive = require('./extractive');
++
++// ── Tunables (Live's) ────────────────────────────────────────────────────────
++const TICK_MS = config.ai.tickMs;
++
++const GLOBAL_MSG_THRESHOLD = 100; // refresh once this many new msgs pile up
++const GLOBAL_MIN_INTERVAL_MS = 5 * 60 * 1000; // never more often than this (flood guard)
++const GLOBAL_MAX_AGE_MS = 30 * 60 * 1000; // ...but refresh at least this often if any new
++
++const USER_MSG_THRESHOLD = 15; // per-user refresh trigger
++const USER_MAX_PER_TICK = 2; // cap LLM calls per tick
++const USER_PASS_EVERY_MS = 3 * 60 * 1000; // throttle the (heavier) user discovery scan
++const USER_STALE_MS = 24 * 60 * 60 * 1000; // refresh a lagging user at least daily
++const USER_DISCOVERY_LOOKBACK_DAYS = 14; // bound the discovery GROUP BY
++
++const WINDOW_TARGET_MESSAGES = 300; // adaptive-window sizing target
++const MAX_BATCH_MESSAGES = 300; // token cap per call
++const WINDOW_MIN_MS = 30 * 60 * 1000; // overview window is never shorter than 30 min
++const WINDOW_MAX_MS = 14 * 24 * 60 * 60 * 1000; // ...nor longer than 14 days
++const MEMORY_MAX_CHARS = 1600;
++const USER_MEMORY_MAX_CHARS = 1200;
++const TIMELINE_MAX = 40;
++const MSG_MAX_CHARS = 220;
++
++// OpenVibe.AI workflows for these summaries. Under the `chat` namespace, which is what Network
++// grants Chat (ai.run.create/ai.run.read, namespace chat.*); the equivalents in AI today are
++// live.chat.global / live.chat.profile, unreachable with Chat's token (see README, For Opus).
++const WORKFLOW_GLOBAL = 'chat.global';
++const WORKFLOW_PROFILE = 'chat.profile';
++
++let _running = false;
++let _timer = null;
++let _tickInFlight = false;
++let _lastUserPass = 0;
++let _writeSkipLogged = false;
++
++// ── Small utils ──────────────────────────────────────────────────────────────
++// DB timestamps are UTC 'YYYY-MM-DD HH:MM:SS' (CURRENT_TIMESTAMP). Match that format.
++function _sqlTime(d) { return new Date(d).toISOString().slice(0, 19).replace('T', ' '); }
++function _parseSqlTime(s) { return s ? new Date(String(s).replace(' ', 'T') + 'Z').getTime() : 0; }
++function _clip(str, n) { str = (str == null ? '' : String(str)).trim(); return str.length > n ? str.slice(0, n) : str; }
++
++/** Messages as data for OpenVibe.AI's chat templates (WS-O task 2). */
++function _msgData(rows, { includeChannel = false, now = 0 } = {}) {
++ return rows.slice(-MAX_BATCH_MESSAGES).map((r) => {
++ let where = null;
++ if (includeChannel) where = r.is_global ? 'global' : r.channel_username ? `#${r.channel_username}` : r.stream_id ? 'stream' : null;
++ const t = now ? _parseSqlTime(r.timestamp || r.created_at) : 0;
++ return {
++ mins_ago: t ? Math.max(0, Math.round((now - t) / 60000)) : null, where: where ? String(where).slice(0, 80) : null,
++ author: String(r.username || (r.user_id ? `user#${r.user_id}` : 'anon')).slice(0, 80),
++ kind: r.message_type && r.message_type !== 'chat' ? String(r.message_type).slice(0, 40) : null, text: _clip(r.message, MSG_MAX_CHARS),
++ };
++ });
++}
++
++function _windowLabel(ms) {
++ const min = Math.round(ms / 60000);
++ if (min < 90) return min <= 60 ? 'past hour' : `past ${min} min`;
++ const hrs = Math.round(ms / 3600000);
++ if (hrs < 36) return `past ${hrs} hours`;
++ const days = Math.max(1, Math.round(ms / 86400000));
++ return `past ${days} day${days === 1 ? '' : 's'}`;
++}
++
++// Stamp raw model additions into {ts,label,detail}, skipping placeholders.
++function _stampAdditions(additions, fallbackTs, nowMs) {
++ const now = nowMs || Date.now();
++ const out = [];
++ for (const a of (Array.isArray(additions) ? additions : [])) {
++ const label = _clip(a && (a.label || a.title), 80);
++ if (!label || _isPlaceholder(label)) continue;
++ let ts = fallbackTs;
++ const mins = Number(a && a.mins_ago);
++ if (Number.isFinite(mins) && mins >= 0 && mins <= 43200) ts = _sqlTime(now - mins * 60000);
++ out.push({ ts, label, detail: _clip(a.detail || a.description || '', 240) });
++ }
++ return out;
++}
++function _mergeTimeline(priorJson, additions, fallbackTs, nowMs) {
++ let prior = [];
++ try { prior = JSON.parse(priorJson || '[]'); if (!Array.isArray(prior)) prior = []; } catch { prior = []; }
++ prior.push(..._stampAdditions(additions, fallbackTs, nowMs));
++ prior.sort((x, y) => _parseSqlTime(x.ts) - _parseSqlTime(y.ts));
++ if (prior.length > TIMELINE_MAX) prior = prior.slice(prior.length - TIMELINE_MAX);
++ return JSON.stringify(prior);
++}
++function _isPlaceholder(s) { return /^(short title|one sentence|label|title|detail|\.\.\.)$/i.test(String(s || '').trim()); }
++function _cleanTimeline(arr) {
++ if (!Array.isArray(arr)) return [];
++ return arr
++ .filter(t => t && t.label && !_isPlaceholder(t.label))
++ .sort((a, b) => _parseSqlTime(a.ts) - _parseSqlTime(b.ts));
++}
++
++// ── The authority gate ───────────────────────────────────────────────────────
++// Live still writes the two summaries tables until the cutover: the job skips (one log line), it
++// never forces a write, and writes that would race the flip are caught the same way.
++function _tablesHold() {
++ return db.tableAuthority('chat_ai_summaries') === 'chat' && db.tableAuthority('chat_timeline_events') === 'chat';
++}
++function _logWriteSkip() {
++ if (_writeSkipLogged) return;
++ _writeSkipLogged = true;
++ console.log('[ChatAI] chat_ai_summaries/chat_timeline_events are still written by Live (table_authority live) — the job skips its writes');
++}
++function _write(fn, ...args) {
++ try { return fn(...args); }
++ catch (err) {
++ if (err && err.code === 'table.not_chat') { _logWriteSkip(); return null; }
++ throw err;
++ }
++}
++
++// ── GLOBAL ───────────────────────────────────────────────────────────────────
++async function _refreshGlobal() {
++ const prior = db.getChatAiSummary('global', 0, 'global');
++ const hw = prior ? (prior.last_message_id || 0) : 0;
++ const maxId = db.getMaxChatMessageId();
++ const newCount = maxId > hw ? db.countChatMessagesSince(hw) : 0;
++ if (newCount === 0) return false;
++
++ const ageMs = prior && prior.updated_at ? (Date.now() - _parseSqlTime(prior.updated_at)) : Infinity;
++ if (ageMs < GLOBAL_MIN_INTERVAL_MS) return false;
++ if (newCount < GLOBAL_MSG_THRESHOLD && ageMs < GLOBAL_MAX_AGE_MS) return false;
++
++ const now = Date.now();
++ const nthTs = db.getNthRecentChatTs(WINDOW_TARGET_MESSAGES);
++ let startMs = nthTs ? _parseSqlTime(nthTs) : (now - WINDOW_MAX_MS);
++ startMs = Math.min(startMs, now - WINDOW_MIN_MS);
++ startMs = Math.max(startMs, now - WINDOW_MAX_MS);
++ const windowStart = _sqlTime(startMs);
++ const windowLabel = _windowLabel(now - startMs);
++
++ const rows = db.getChatMessagesForAi({ sinceTs: windowStart, order: 'desc', limit: MAX_BATCH_MESSAGES });
++ if (!rows.length) return false;
++ const priorMemory = prior ? (prior.memory_json || '') : '';
++ let priorTl = [];
++ try { priorTl = JSON.parse(prior ? (prior.timeline_json || '[]') : '[]'); } catch { priorTl = []; }
++
++ let parsed = await aiClient.structured(WORKFLOW_GLOBAL, {
++ window_label: windowLabel, prior_memory: _clip(priorMemory, MEMORY_MAX_CHARS),
++ recent_labels: priorTl.slice(-8).map(t => _clip(t.label, 200)).filter(Boolean),
++ messages: _msgData(rows, { includeChannel: true, now }),
++ });
++ if (!parsed) parsed = extractive.globalFrom(rows, { windowLabel, priorMemory, now });
++ if (!parsed) return false;
++
++ const nowIso = _sqlTime(now);
++ _write(db.addChatTimelineEvents, 'global', 0, _stampAdditions(parsed.timeline, nowIso, now));
++ _write(db.upsertChatAiSummary, {
++ scope: 'global', subject_id: 0, window: 'global',
++ overview: _clip(parsed.recent_overview || '', 2000),
++ memory_json: _clip(parsed.memory || priorMemory, MEMORY_MAX_CHARS),
++ timeline_json: _mergeTimeline(prior ? prior.timeline_json : '[]', parsed.timeline, nowIso, now),
++ message_count: (prior ? (prior.message_count || 0) : 0) + newCount,
++ window_message_count: rows.length,
++ last_message_id: maxId,
++ window_label: windowLabel,
++ window_start: windowStart,
++ window_end: nowIso,
++ });
++ console.log(`[ChatAI] global refreshed (${newCount} new, window=${windowLabel}, ${rows.length} msgs)`);
++ return true;
++}
++
++// ── PER-USER ─────────────────────────────────────────────────────────────────
++async function _refreshUser(uid, maxId) {
++ const prior = db.getChatAiSummary('user', uid, 'rolling');
++ const now = Date.now();
++
++ const dayStart = _sqlTime(now - 24 * 60 * 60 * 1000);
++ let dayRows = db.getChatMessagesForAi({ sinceTs: dayStart, userId: uid, order: 'asc', limit: MAX_BATCH_MESSAGES });
++ let has24h = dayRows.length > 0;
++ if (!dayRows.length) dayRows = db.getChatMessagesForAi({ userId: uid, order: 'desc', limit: 80 });
++ if (!dayRows.length) return false;
++
++ const uname = dayRows[dayRows.length - 1].username || `user#${uid}`;
++ const priorMemory = prior ? (prior.memory_json || '') : '';
++ const totalSeen = (prior ? (prior.message_count || 0) : 0);
++
++ let parsed = await aiClient.structured(WORKFLOW_PROFILE, {
++ subject_kind: 'user', name: _clip(uname, 120), recent_24h: has24h, seen: totalSeen, prior_memory: _clip(priorMemory, USER_MEMORY_MAX_CHARS),
++ messages: _msgData(dayRows, { now }),
++ });
++ if (!parsed) parsed = extractive.profileFrom(dayRows, { name: uname, subjectKind: 'user', priorMemory, has24h, seen: totalSeen, now });
++ if (!parsed) return false;
++
++ const newCount = db.countChatMessagesSince(prior ? (prior.last_message_id || 0) : 0, uid);
++ const nowIso = _sqlTime(now);
++ const _lastRow = dayRows[dayRows.length - 1] || {};
++ const activityTs = _lastRow.timestamp || _lastRow.created_at || nowIso;
++ _write(db.upsertChatAiSummary, {
++ scope: 'user', subject_id: uid, window: 'rolling',
++ overview: JSON.stringify({
++ today: _clip(parsed.overview_24h || '', 1200),
++ alltime: _clip(parsed.overview_alltime || '', 1200),
++ has_24h: has24h,
++ }),
++ memory_json: _clip(parsed.memory || priorMemory, USER_MEMORY_MAX_CHARS),
++ timeline_json: _mergeTimeline(prior ? prior.timeline_json : '[]', parsed.timeline, activityTs, now),
++ message_count: totalSeen + newCount,
++ window_message_count: dayRows.length,
++ last_message_id: maxId || db.getMaxChatMessageId(),
++ window_label: has24h ? 'past 24h' : 'recent',
++ window_start: has24h ? dayStart : null,
++ window_end: nowIso,
++ });
++ console.log(`[ChatAI] user ${uid} (${uname}) refreshed (${newCount} new, 24h=${has24h})`);
++ return true;
++}
++
++async function _userPass() {
++ const staleCutoff = _sqlTime(Date.now() - USER_STALE_MS);
++ const sinceTs = _sqlTime(Date.now() - USER_DISCOVERY_LOOKBACK_DAYS * 86400000);
++ let candidates = [];
++ try {
++ candidates = db.getUsersNeedingChatAi({ threshold: USER_MSG_THRESHOLD, staleCutoffIso: staleCutoff, sinceTs, limit: USER_MAX_PER_TICK });
++ } catch (e) { console.warn('[ChatAI] user discovery failed:', e.message); return; }
++ for (const c of candidates) {
++ if (!c.uid) continue;
++ try { await _refreshUser(c.uid, c.max_id); }
++ catch (e) { console.warn(`[ChatAI] user ${c.uid} refresh failed:`, e.message); }
++ }
++}
++
++// ── PER RELAY-USER (external platform chatters bridged in) ────────────────────
++async function _refreshRelayUser(ru) {
++ const prior = db.getChatAiSummary('relay', ru.id, 'rolling');
++ const now = Date.now();
++ const dayStart = _sqlTime(now - 24 * 60 * 60 * 1000);
++
++ let dayRows = db.getRelayChatMessagesForAi({ platform: ru.platform, rawUsername: ru.username, sinceTs: dayStart, order: 'asc', limit: MAX_BATCH_MESSAGES });
++ let has24h = dayRows.length > 0;
++ if (!dayRows.length) dayRows = db.getRelayChatMessagesForAi({ platform: ru.platform, rawUsername: ru.username, order: 'desc', limit: 80 });
++ if (!dayRows.length) return false;
++
++ const uname = ru.display_name || ru.username;
++ const priorMemory = prior ? (prior.memory_json || '') : '';
++
++ let parsed = await aiClient.structured(WORKFLOW_PROFILE, {
++ subject_kind: 'relay', name: _clip(uname, 120), platform: _clip(ru.platform, 40), recent_24h: has24h, prior_memory: _clip(priorMemory, USER_MEMORY_MAX_CHARS),
++ messages: _msgData(dayRows, { now }),
++ });
++ if (!parsed) parsed = extractive.profileFrom(dayRows, { name: uname, subjectKind: 'relay', priorMemory, has24h, seen: ru.message_count || 0, now });
++ if (!parsed) return false;
++
++ const nowIso = _sqlTime(now);
++ _write(db.upsertChatAiSummary, {
++ scope: 'relay', subject_id: ru.id, window: 'rolling',
++ overview: JSON.stringify({
++ today: _clip(parsed.overview_24h || '', 1200),
++ alltime: _clip(parsed.overview_alltime || '', 1200),
++ has_24h: has24h,
++ }),
++ memory_json: _clip(parsed.memory || priorMemory, USER_MEMORY_MAX_CHARS),
++ timeline_json: _mergeTimeline(prior ? prior.timeline_json : '[]', parsed.timeline, nowIso, now),
++ message_count: ru.message_count || 0,
++ window_message_count: dayRows.length,
++ last_message_id: db.getMaxChatMessageId(),
++ window_label: has24h ? 'past 24h' : 'recent',
++ window_start: has24h ? dayStart : null,
++ window_end: nowIso,
++ });
++ console.log(`[ChatAI] relay ${ru.platform}:${ru.username} refreshed (24h=${has24h}, ${dayRows.length} msgs)`);
++ return true;
++}
++
++async function _relayPass() {
++ const lookbackIso = _sqlTime(Date.now() - USER_DISCOVERY_LOOKBACK_DAYS * 86400000);
++ let candidates = [];
++ try { candidates = db.getRelayUsersNeedingChatAi({ lookbackIso, threshold: 8, limit: USER_MAX_PER_TICK }); }
++ catch (e) { console.warn('[ChatAI] relay discovery failed:', e.message); return; }
++ for (const ru of candidates) {
++ if (!ru.id) continue;
++ try { await _refreshRelayUser(ru); }
++ catch (e) { console.warn(`[ChatAI] relay ${ru.id} refresh failed:`, e.message); }
++ }
++}
++
++// ── PER ANON (not-logged-in chatters, keyed by their stable anon_id) ──────────
++async function _refreshAnon(anonId) {
++ const subjectId = db.anonSubjectId(anonId);
++ if (!subjectId) return false;
++ const prior = db.getChatAiSummary('anon', subjectId, 'rolling');
++ const now = Date.now();
++ const dayStart = _sqlTime(now - 24 * 60 * 60 * 1000);
++
++ let dayRows = db.getAnonChatMessagesForAi({ anonId, sinceTs: dayStart, order: 'asc', limit: MAX_BATCH_MESSAGES });
++ let has24h = dayRows.length > 0;
++ if (!dayRows.length) dayRows = db.getAnonChatMessagesForAi({ anonId, order: 'desc', limit: 80 });
++ if (!dayRows.length) return false;
++
++ const priorMemory = prior ? (prior.memory_json || '') : '';
++ let parsed = await aiClient.structured(WORKFLOW_PROFILE, {
++ subject_kind: 'anon', name: _clip(anonId, 120), recent_24h: has24h, prior_memory: _clip(priorMemory, USER_MEMORY_MAX_CHARS),
++ messages: _msgData(dayRows, { now }),
++ });
++ if (!parsed) parsed = extractive.profileFrom(dayRows, { name: anonId, subjectKind: 'anon', priorMemory, has24h, seen: 0, now });
++ if (!parsed) return false;
++
++ const nowIso = _sqlTime(now);
++ _write(db.upsertChatAiSummary, {
++ scope: 'anon', subject_id: subjectId, window: 'rolling',
++ overview: JSON.stringify({
++ today: _clip(parsed.overview_24h || '', 1200),
++ alltime: _clip(parsed.overview_alltime || '', 1200),
++ has_24h: has24h,
++ }),
++ memory_json: _clip(parsed.memory || priorMemory, USER_MEMORY_MAX_CHARS),
++ timeline_json: _mergeTimeline(prior ? prior.timeline_json : '[]', parsed.timeline, nowIso, now),
++ message_count: dayRows.length,
++ window_message_count: dayRows.length,
++ last_message_id: db.getMaxChatMessageId(),
++ window_label: has24h ? 'past 24h' : 'recent',
++ window_start: has24h ? dayStart : null,
++ window_end: nowIso,
++ });
++ console.log(`[ChatAI] anon ${anonId} refreshed (24h=${has24h}, ${dayRows.length} msgs)`);
++ return true;
++}
++
++async function _anonPass() {
++ const staleCutoff = _sqlTime(Date.now() - USER_STALE_MS);
++ const sinceTs = _sqlTime(Date.now() - USER_DISCOVERY_LOOKBACK_DAYS * 86400000);
++ let candidates = [];
++ try {
++ candidates = db.getAnonsNeedingChatAi({ threshold: USER_MSG_THRESHOLD, staleCutoffIso: staleCutoff, sinceTs, limit: USER_MAX_PER_TICK });
++ } catch (e) { console.warn('[ChatAI] anon discovery failed:', e.message); return; }
++ for (const c of candidates) {
++ if (!c.anon_id) continue;
++ try { await _refreshAnon(c.anon_id); }
++ catch (e) { console.warn(`[ChatAI] anon ${c.anon_id} refresh failed:`, e.message); }
++ }
++}
++
++// ── Poller ───────────────────────────────────────────────────────────────────
++async function _tick() {
++ if (_tickInFlight) return;
++ _tickInFlight = true;
++ try {
++ if (!config.ai.enabled) return; // the job's switch (off in dev and tests)
++ if (!_tablesHold()) { _logWriteSkip(); return; }
++ try { await _refreshGlobal(); }
++ catch (e) { console.warn('[ChatAI] global refresh failed:', e.message); }
++
++ if (Date.now() - _lastUserPass >= USER_PASS_EVERY_MS) {
++ _lastUserPass = Date.now();
++ await _userPass();
++ await _relayPass();
++ await _anonPass();
++ }
++ } catch (e) {
++ console.warn('[ChatAI] tick error:', e.message);
++ } finally {
++ _tickInFlight = false;
++ }
++}
++
++// One-time: backfill the growing timeline log from the existing summary JSON.
++async function _seedTimelineEvents() {
++ try {
++ if ((db.getChatTimelineEvents({ scope: 'global', limit: 1 }) || []).length) return;
++ const row = db.getChatAiSummary('global', 0, 'global');
++ if (!row) return;
++ let tl = []; try { tl = JSON.parse(row.timeline_json || '[]'); } catch { tl = []; }
++ const cleaned = _cleanTimeline(tl);
++ if (cleaned.length) { _write(db.addChatTimelineEvents, 'global', 0, cleaned); console.log(`[ChatAI] Seeded ${cleaned.length} timeline event(s)`); }
++ } catch { /* */ }
++}
++
++function start() {
++ if (_running) return;
++ if (!config.ai.enabled) return; // off under the drill/test switches and in dev
++ _running = true;
++ _seedTimelineEvents().catch(() => {});
++ _timer = setInterval(() => { _tick().catch(() => {}); }, TICK_MS);
++ if (_timer.unref) _timer.unref();
++ console.log('[AI] Chat-AI job started (global overview/timeline + per-chatter insights)');
++}
++
++function stop() {
++ _running = false;
++ if (_timer) { clearInterval(_timer); _timer = null; }
++}
++
++// ── Public read helpers (used by the routes) ─────────────────────────────────
++function _parseOverviews(row) {
++ let overviews = { today: '', alltime: '', has_24h: false };
++ try { overviews = { ...overviews, ...JSON.parse(row.overview || '{}') }; } catch { /* */ }
++ let timeline = [];
++ try { timeline = _cleanTimeline(JSON.parse(row.timeline_json || '[]')); } catch { /* */ }
++ return {
++ overview_24h: overviews.today || '',
++ overview_alltime: overviews.alltime || '',
++ has_24h: !!overviews.has_24h,
++ memory: row.memory_json || '',
++ timeline,
++ message_count: row.message_count || 0,
++ updated_at: row.updated_at || null,
++ };
++}
++
++function getGlobalInsight() {
++ const row = db.getChatAiSummary('global', 0, 'global');
++ if (!row) return null;
++ let timeline = [];
++ try { timeline = _cleanTimeline(JSON.parse(row.timeline_json || '[]')); } catch { /* */ }
++ return {
++ overview: row.overview || '',
++ memory: row.memory_json || '',
++ timeline,
++ window_label: row.window_label || '',
++ message_count: row.message_count || 0,
++ window_message_count: row.window_message_count || 0,
++ updated_at: row.updated_at || null,
++ };
++}
++
++function getUserInsight(userId) {
++ const row = db.getChatAiSummary('user', userId, 'rolling');
++ return row ? _parseOverviews(row) : null;
++}
++
++function getRelayUserInsight(relayId) {
++ const row = db.getChatAiSummary('relay', relayId, 'rolling');
++ return row ? _parseOverviews(row) : null;
++}
++
++function getAnonInsight(anonId) {
++ const row = db.getChatAiSummary('anon', db.anonSubjectId(anonId), 'rolling');
++ return row ? _parseOverviews(row) : null;
++}
++
++module.exports = {
++ start, stop, _tick, _msgData,
++ getGlobalInsight, getUserInsight, getRelayUserInsight, getAnonInsight,
++ WORKFLOW_GLOBAL, WORKFLOW_PROFILE,
++};
ddiff --git a/server/ai/client.js b/server/ai/client.js
nnew file mode 100644
iindex 0000000..eaa6ad6
---- /dev/null
++++ b/server/ai/client.js
@@@ -0,0 +1,60 @@
++/**
++ * OpenVibe.Chat → OpenVibe.AI (run API v1, namespace `chat`).
++ *
++ * Chat's OWN service token (client `chat`; Network grants ai.run.create / ai.run.read on audience
++ * openvibe.ai, namespace chat.*) through openvibe-sdk/ai's createAiClient. The chat-AI job calls
++ * `structured(workflow, input)` and gets the run's `output`, or null when AI is off, refuses the
++ * namespace, or the run did not succeed — the job then falls back to its extractive summary.
++ */
++'use strict';
++
++const config = require('../config');
++
++let _ai = null;
++
++function ai() {
++ if (_ai) return _ai;
++ if (!config.oauth.clientSecret) throw new Error('OV_OAUTH_CLIENT_SECRET is not set');
++ const { createClient } = require('openvibe-sdk');
++ const { createServiceTokenClient } = require('openvibe-sdk/auth');
++ const { createAiClient } = require('openvibe-sdk/ai');
++ const tokenClient = createServiceTokenClient({
++ tokenUrl: `${config.networkInternalUrl}/oauth/token`,
++ clientId: config.oauth.clientId,
++ clientSecret: config.oauth.clientSecret,
++ audience: 'openvibe.ai',
++ scope: config.ai.scope,
++ });
++ const http = createClient({ tokenProvider: tokenClient, timeoutMs: config.ai.waitMs + 10000 });
++ _ai = createAiClient(http, { baseUrl: config.ai.internalUrl });
++ return _ai;
++}
++
++/** Is the job allowed to call AI at all (config gate; no key → no calls). */
++function enabled() {
++ return !!(config.ai.enabled && config.oauth.clientSecret);
++}
++
++/**
++ * Run a workflow and return its `output` object, or null: disabled, a refusal, a failed run, or a
++ * transport error all answer null — exactly Live's ai-service.structured() contract.
++ */
++async function structured(workflow, input, opts = {}) {
++ if (!enabled()) return null;
++ try {
++ const { runs } = ai();
++ const body = await runs.create(workflow, input, {
++ wait: config.ai.waitMs,
++ ...(opts.idempotencyKey ? { idempotencyKey: opts.idempotencyKey } : {}),
++ ...(opts.attribution !== undefined ? { attribution: opts.attribution } : {}),
++ });
++ const run = body && (body.run || body);
++ if (!run || run.status !== 'succeeded' || run.synthetic) return null;
++ return run.output || null;
++ } catch (err) {
++ console.warn(`[AI] ${workflow}: ${err && err.message}`);
++ return null;
++ }
++}
++
++module.exports = { structured, enabled, _setClient(c) { _ai = c; }, _reset() { _ai = null; } };
ddiff --git a/server/ai/extractive.js b/server/ai/extractive.js
nnew file mode 100644
iindex 0000000..396c38f
---- /dev/null
++++ b/server/ai/extractive.js
@@@ -0,0 +1,74 @@
++/**
++ * The chat-AI job's fallback: a deterministic, extractive summary built from the message batch
++ * itself when OpenVibe.AI does not answer (disabled, over quota, refuses the namespace, or the run
++ * fails). It produces the same output fields as the `chat.global` / `chat.profile` workflows, so
++ * the stored row and every route shape are identical either way.
++ *
++ * Live had no extractive summariser (its job simply skipped a failed call); Chat needs one because
++ * the `chat.*` workflows do not exist in OpenVibe.AI yet (see the report's "For Opus"), so this is
++ * the summary the feature serves until they do.
++ */
++'use strict';
++
++const MSG_MAX_CHARS = 220;
++
++function _clip(str, n) { str = (str == null ? '' : String(str)).trim(); return str.length > n ? str.slice(0, n) : str; }
++function _oneLine(str, n) { return _clip(String(str == null ? '' : str).replace(/\s+/g, ' '), n); }
++
++function _parseSqlTime(s) { return s ? new Date(String(s).replace(' ', 'T') + 'Z').getTime() : 0; }
++
++function _minsAgo(ts, now) {
++ const t = _parseSqlTime(ts);
++ return t ? Math.max(0, Math.round((now - t) / 60000)) : null;
++}
++
++// A placeholder the model itself refuses ("short title", "one sentence") never becomes a label.
++const _PLACEHOLDER = /^(short title|one sentence|label|title|detail|\.\.\.)$/i;
++
++/** Notable moments: non-chat activity (donations, clips, sound events), else the latest messages. */
++function _timeline(rows, now) {
++ const notable = rows.filter((r) => r.message_type && r.message_type !== 'chat' && r.message_type !== 'tts');
++ const picked = (notable.length ? notable : rows.slice(-5)).slice(-10);
++ return picked
++ .map((r) => {
++ const label = _oneLine(r.message, 80);
++ if (!label || _PLACEHOLDER.test(label)) return null;
++ return { label, detail: _oneLine(r.username || r.channel_username || '', 120), mins_ago: _minsAgo(r.timestamp || r.created_at, now) };
++ })
++ .filter(Boolean);
++}
++
++/** The global overview: how much traffic, who and where, with a few representative lines. */
++function globalFrom(rows, { windowLabel = 'recent window', priorMemory = '', now = Date.now() } = {}) {
++ const authors = new Set(rows.map((r) => r.username).filter(Boolean));
++ const rooms = new Set(rows.map((r) => r.channel_username || (r.is_global ? 'global' : null)).filter(Boolean));
++ const sample = rows.slice(-3).map((r) => _oneLine(r.message, 120)).filter(Boolean);
++ const recent_overview = _clip(
++ `${rows.length} messages from ${authors.size} chatter${authors.size === 1 ? '' : 's'} in ${rooms.size || 1} room${rooms.size === 1 ? '' : 's'} during the ${windowLabel}.` +
++ (sample.length ? ` Latest: ${sample.join(' · ')}` : ''),
++ 2000
++ );
++ return {
++ recent_overview,
++ memory: _clip(priorMemory || (sample[0] ? `Latest: ${sample[0]}` : ''), 1600),
++ timeline: _timeline(rows, now),
++ };
++}
++
++/** The per-subject (user / relay / anon) insight: today vs. all-time from the batch. */
++function profileFrom(rows, { name = '', subjectKind = 'user', priorMemory = '', has24h = false, seen = 0, now = Date.now() } = {}) {
++ const sample = rows.slice(-3).map((r) => _oneLine(r.message, 120)).filter(Boolean);
++ const who = name || `${subjectKind} user`;
++ const overview_24h = has24h
++ ? _clip(`${who}: ${rows.length} message${rows.length === 1 ? '' : 's'} in the last 24 hours.` + (sample.length ? ` Latest: ${sample.join(' · ')}` : ''), 1200)
++ : '';
++ const overview_alltime = _clip(`${who} has ${seen + rows.length} message${seen + rows.length === 1 ? '' : 's'} seen.` + (sample.length ? ` Latest: ${sample.join(' · ')}` : ''), 1200);
++ return {
++ overview_24h,
++ overview_alltime,
++ memory: _clip(priorMemory || (sample[0] ? `Latest: ${sample[0]}` : ''), 1200),
++ timeline: _timeline(rows, now),
++ };
++}
++
++module.exports = { globalFrom, profileFrom, _timeline, MSG_MAX_CHARS };
ddiff --git a/server/app.js b/server/app.js
iindex 93f780d..538ffae 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -226,6 +226,8 @@ function createApp({ chatServer, bridge, mirror, relay, events = null, callServe
app.use('/api/chat/channels', require('./chat/channel-mod-routes'));
// ICE servers for call rooms (STUN, and TURN with credentials when TURN_URL is set): server/net/turn.js.
app.use('/api/chat/ice-servers', require('./net/turn').createIceRoutes());
++ // Chat-AI insight (/api/chat/ai/…): before /api/chat, whose /:streamId routes would take /ai/….
++ app.use('/api/chat/ai', require('./chat/ai-routes'));
app.use('/api/chat', require('./chat/routes'));
app.use('/api/dm', require('./chat/dm-routes'));
app.use('/api/tts', require('./chat/tts-routes'));
ddiff --git a/server/chat/ai-routes.js b/server/chat/ai-routes.js
nnew file mode 100644
iindex 0000000..805a4fb
---- /dev/null
++++ b/server/chat/ai-routes.js
@@@ -0,0 +1,132 @@
++/**
++ * ai-routes.js — read API for chat AI insight, Chat's (/api/chat/ai/…, nginx sends /api/chat here).
++ *
++ * Live's server/ai/chat-ai-routes.js, moved (plan T3 step 2): the same six paths, query parameters,
++ * response shapes, error codes and caching headers, reading Chat's own chat_ai_summaries /
++ * chat_timeline_events (via the chat-ai job's read helpers) and Chat's ctx_* projections.
++ *
++ * GET /global → { insight } (or null)
++ * GET /timeline → { events, hasMore } ?before=ms&since=ms&q=&limit=
++ * GET /user/:id → { insight, streamer, user }
++ * GET /anon/:anonId → { insight, user } (anon id like "anon12345")
++ * GET /relay/:platform/:username → { insight, user }
++ * GET /timeline/:username → the channel timeline shell
++ *
++ * Visibility is Live's: these six are public reads (Live put no auth in front of them; the UI only
++ * offers the insight cards to staff). No Cache-Control, as Live.
++ *
++ * The captions and VOD-transcript routes in Live's file are Live's stream features (whisper, VOD
++ * moments) and stay there.
++ */
++'use strict';
++
++const express = require('express');
++const router = express.Router();
++const db = require('../db/database');
++const ctx = require('../live-context');
++const chatAi = require('../ai/chat-ai');
++
++router.get('/global', (req, res) => {
++ try {
++ const insight = chatAi.getGlobalInsight();
++ res.json({ insight: insight || null });
++ } catch (err) {
++ res.status(500).json({ error: 'Failed to load global chat insight' });
++ }
++});
++
++// Browsable/searchable global timeline. Query: before=<ms>, since=<ms>, q=<search>, limit=<n>.
++router.get('/timeline', (req, res) => {
++ try {
++ const num = (v) => { const n = parseInt(v, 10); return Number.isFinite(n) ? n : null; };
++ const limit = Math.min(60, Math.max(1, num(req.query.limit) || 25));
++ const events = db.getChatTimelineEvents({
++ scope: 'global', subjectId: 0,
++ before: num(req.query.before), since: num(req.query.since),
++ q: req.query.q || null, limit,
++ });
++ res.json({ events, hasMore: events.length >= limit });
++ } catch (err) {
++ res.status(500).json({ error: 'Failed to load timeline' });
++ }
++});
++
++router.get('/user/:id', (req, res) => {
++ try {
++ const uid = parseInt(req.params.id, 10);
++ if (!Number.isFinite(uid)) return res.status(400).json({ error: 'Invalid user id' });
++ const user = ctx.getUserById(uid);
++ const insight = chatAi.getUserInsight(uid);
++ // `streamer` (who they are as a broadcaster: overview + stream memories) is Live's stream
++ // feature, not chat; Chat answers null for it (see README, For Opus). The key is kept so
++ // the response shape matches Live's.
++ res.json({
++ insight: insight || null,
++ streamer: null,
++ user: user ? { id: user.id, username: user.username, display_name: user.display_name } : null,
++ });
++ } catch (err) {
++ res.status(500).json({ error: 'Failed to load user chat insight' });
++ }
++});
++
++// An anonymous chatter's insight, keyed by their stable anon_id ("anon<N>").
++router.get('/anon/:anonId', (req, res) => {
++ try {
++ const anonId = String(req.params.anonId || '');
++ if (!/^anon\d+$/i.test(anonId)) return res.status(400).json({ error: 'Invalid anon id' });
++ const insight = chatAi.getAnonInsight(anonId);
++ res.json({ insight: insight || null, user: { anon_id: anonId, username: anonId } });
++ } catch (err) {
++ res.status(500).json({ error: 'Failed to load anon chat insight' });
++ }
++});
++
++// A bridged external (relay) chatter's insight, keyed by platform + username.
++router.get('/relay/:platform/:username', (req, res) => {
++ try {
++ const ru = db.getRelayUser(req.params.platform, req.params.username);
++ if (!ru) return res.json({ insight: null, user: null });
++ const insight = chatAi.getRelayUserInsight(ru.id);
++ res.json({
++ insight: insight || null,
++ user: { platform: ru.platform, username: ru.display_name || ru.username, message_count: ru.message_count || 0, first_seen: ru.first_seen || null },
++ });
++ } catch (err) {
++ res.status(500).json({ error: 'Failed to load relay chat insight' });
++ }
++});
++
++// The channel page's AI timeline shell. Live's route fuses its streamer AI timeline (stream
++// overviews, VOD moments, combined overview, session titles) with the chatter insight; the streamer
++// half is Live's stream feature and Chat cannot build it (README, For Opus). Chat serves the chat
++// half with Live's exact keys so the endpoint and the simple consumer keep working.
++router.get('/timeline/:username', (req, res) => {
++ try {
++ const uname = String(req.params.username || '').trim();
++ const user = ctx.getUserByUsername(uname);
++ if (!user) return res.status(404).json({ error: 'Channel not found' });
++ const offset = Math.max(0, parseInt(req.query.offset, 10) || 0);
++ const limit = Math.min(30, Math.max(1, parseInt(req.query.limit, 10) || 12));
++ let chatInsight = null;
++ try { chatInsight = chatAi.getUserInsight(user.id) || null; } catch { chatInsight = null; }
++ res.json({
++ username: user.username,
++ display_name: user.display_name || user.username,
++ overview: undefined, // Live's streamer overview — Live's feature (For Opus)
++ chatInsight,
++ combinedOverview: null,
++ sessionCount: 0,
++ momentCount: 0,
++ generatedAt: null,
++ index: undefined,
++ sessions: [],
++ offset, limit,
++ hasMore: false,
++ });
++ } catch (err) {
++ res.status(500).json({ error: 'Failed to load AI timeline' });
++ }
++});
++
++module.exports = router;
ddiff --git a/server/config.js b/server/config.js
iindex ca1b94a..b229baa 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -160,6 +160,19 @@ module.exports = {
scope: 'media.object.upload media.object.delete',
},
++ // OpenVibe.AI — the chat-AI summaries (decision 5, plan T3 step 2). Chat runs Live's chat-ai job
++ // itself and calls AI with its OWN service token (client `chat`, audience openvibe.ai, network
++ // grants ai.run.create/ai.run.read on namespace chat.*). `enabled` is the job's switch: off in
++ // development and the test harness, on in production through CHAT_AI_ENABLED=1 (its other jobs
++ // are gated the same way — LIVE_MIRROR, EVENTS_URL).
++ ai: {
++ enabled: bool(process.env.CHAT_AI_ENABLED, false),
++ internalUrl: strip(process.env.OV_AI_INTERNAL_URL || 'http://127.0.0.1:4700'),
++ scope: 'ai.run.create ai.run.read',
++ waitMs: Math.max(0, int(process.env.CHAT_AI_WAIT_MS, 60000)),
++ tickMs: Math.max(1000, int(process.env.CHAT_AI_TICK_MS, 45000)),
++ },
++
// Channel sound clips live on disk. At cutover this is Live's sounds directory: stored rows
// hold absolute paths into it, and Live's donation alerts read the alert sounds from there.
sounds: {
ddiff --git a/server/db/database.js b/server/db/database.js
iindex 3323049..f6469b9 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -1314,6 +1314,153 @@ function addChatTimelineEvents(scope, subjectId, events) {
});
}
++// ── Chat AI reads (the job and its routes) ────────────────────
++// Live's server/db/database.js helpers, moved: same SQL, but Live's `users`/`streams` joins read
++// the ctx_* projections here (Live's rows are not in this database).
++const _CHAT_AI_WHERE = `cm.is_deleted = 0 AND cm.message_type != 'system'
++ AND COALESCE(cm.source_platform,'') != 'ai'
++ AND (cm.auto_delete_at IS NULL OR datetime(cm.auto_delete_at) > CURRENT_TIMESTAMP)`;
++
++function getChatAiSummary(scope, subjectId, window) {
++ return get('SELECT * FROM chat_ai_summaries WHERE scope = ? AND subject_id = ? AND window = ?',
++ [scope, subjectId || 0, window]) || null;
++}
++function getChatAiSummaries(scope, subjectId) {
++ return all('SELECT * FROM chat_ai_summaries WHERE scope = ? AND subject_id = ? ORDER BY window',
++ [scope, subjectId || 0]);
++}
++// Paginated + searchable timeline browse. `before` = epoch ms (exclusive upper bound); `q`
++// filters label/detail; `since` = epoch ms lower bound (for period jumps). Newest first.
++function getChatTimelineEvents({ scope = 'global', subjectId = 0, before = null, since = null, q = null, limit = 25 } = {}) {
++ const conds = ['scope = ?', 'subject_id = ?'];
++ const params = [scope, subjectId || 0];
++ if (before) { conds.push("ts < datetime(?, 'unixepoch')"); params.push(Math.floor(before / 1000)); }
++ if (since) { conds.push("ts >= datetime(?, 'unixepoch')"); params.push(Math.floor(since / 1000)); }
++ if (q && String(q).trim()) { const like = '%' + String(q).trim().slice(0, 60) + '%'; conds.push('(label LIKE ? OR detail LIKE ?)'); params.push(like, like); }
++ params.push(Math.min(60, Math.max(1, limit)));
++ try {
++ return all(`SELECT id, ts, label, detail FROM chat_timeline_events WHERE ${conds.join(' AND ')} ORDER BY ts DESC, id DESC LIMIT ?`, params) || [];
++ } catch { return []; }
++}
++
++function getMaxChatMessageId() {
++ return get('SELECT MAX(id) AS m FROM chat_messages')?.m || 0;
++}
++// Count analyzable messages newer than a high-water id (optionally for one user).
++function countChatMessagesSince(afterId, userId = null) {
++ let sql = `SELECT COUNT(*) AS c FROM chat_messages cm WHERE ${_CHAT_AI_WHERE} AND cm.id > ?`;
++ const params = [afterId || 0];
++ if (userId) { sql += ' AND cm.user_id = ?'; params.push(userId); }
++ return get(sql, params)?.c || 0;
++}
++// Fetch analyzable messages for AI batching (with the channel/broadcaster label).
++function getChatMessagesForAi({ afterId = null, sinceTs = null, userId = null, limit = 400, order = 'asc' } = {}) {
++ let sql = `SELECT cm.id, cm.user_id, cm.username, cm.message, cm.message_type, cm.timestamp,
++ cm.stream_id, cm.channel_user_id, cm.is_global,
++ ch.username AS channel_username, ch.display_name AS channel_display
++ FROM chat_messages cm
++ LEFT JOIN ctx_users ch ON cm.channel_user_id = ch.id
++ WHERE ${_CHAT_AI_WHERE}`;
++ const params = [];
++ if (afterId != null) { sql += ' AND cm.id > ?'; params.push(afterId); }
++ if (sinceTs != null) { sql += ' AND cm.timestamp >= ?'; params.push(sinceTs); }
++ if (userId) { sql += ' AND cm.user_id = ?'; params.push(userId); }
++ sql += ` ORDER BY cm.id ${order === 'desc' ? 'DESC' : 'ASC'} LIMIT ?`;
++ params.push(Math.max(1, Math.min(2000, limit)));
++ const rows = all(sql, params);
++ return order === 'desc' ? rows.reverse() : rows;
++}
++// Timestamp of the Nth-most-recent analyzable message — drives the adaptive overview window.
++function getNthRecentChatTs(n, userId = null) {
++ let sql = `SELECT cm.timestamp AS ts FROM chat_messages cm WHERE ${_CHAT_AI_WHERE}`;
++ const params = [];
++ if (userId) { sql += ' AND cm.user_id = ?'; params.push(userId); }
++ sql += ' ORDER BY cm.id DESC LIMIT 1 OFFSET ?';
++ params.push(Math.max(0, (n | 0) - 1));
++ return get(sql, params)?.ts || null;
++}
++
++// Users with enough new chat activity (or a stale summary) to warrant an AI refresh.
++function getUsersNeedingChatAi({ threshold = 15, staleCutoffIso, sinceTs, limit = 3 } = {}) {
++ const sql = `
++ SELECT cm.user_id AS uid,
++ MAX(cm.id) AS max_id,
++ SUM(CASE WHEN cm.id > COALESCE(cs.last_message_id, 0) THEN 1 ELSE 0 END) AS new_msgs,
++ COALESCE(cs.last_message_id, 0) AS hw,
++ cs.updated_at AS last_update
++ FROM chat_messages cm
++ LEFT JOIN chat_ai_summaries cs
++ ON cs.scope = 'user' AND cs.subject_id = cm.user_id AND cs.window = 'rolling'
++ WHERE ${_CHAT_AI_WHERE} AND cm.user_id IS NOT NULL AND cm.timestamp >= ?
++ GROUP BY cm.user_id
++ HAVING new_msgs > 0
++ AND ( new_msgs >= ? OR cs.last_message_id IS NULL OR cs.updated_at IS NULL OR cs.updated_at < ? )
++ ORDER BY (cs.updated_at IS NULL) DESC, new_msgs DESC
++ LIMIT ?`;
++ return all(sql, [sinceTs, threshold, staleCutoffIso, Math.max(1, limit)]);
++}
++
++// Relay messages for AI batching (mirrors getChatMessagesForAi).
++function getRelayChatMessagesForAi({ platform, rawUsername, sinceTs = null, limit = 300, order = 'asc' } = {}) {
++ let sql = `SELECT cm.id, cm.username, cm.message, cm.message_type, cm.timestamp, cm.source_platform
++ FROM chat_messages cm
++ WHERE ${_CHAT_AI_WHERE} AND ${_RELAY_MATCH}`;
++ const params = _relayMatchParams(platform, rawUsername);
++ if (sinceTs != null) { sql += ' AND cm.timestamp >= ?'; params.push(sinceTs); }
++ sql += ` ORDER BY cm.id ${order === 'desc' ? 'DESC' : 'ASC'} LIMIT ?`;
++ params.push(Math.max(1, Math.min(2000, limit)));
++ const rows = all(sql, params);
++ return order === 'desc' ? rows.reverse() : rows;
++}
++
++// Relay users with new activity since their last AI summary (or never summarised).
++function getRelayUsersNeedingChatAi({ lookbackIso, threshold = 8, limit = 2 } = {}) {
++ return all(`
++ SELECT r.rowid AS id, r.platform, r.username, r.display_name, r.message_count, r.last_seen,
++ cs.updated_at AS last_update
++ FROM relay_users r
++ LEFT JOIN chat_ai_summaries cs
++ ON cs.scope = 'relay' AND cs.window = 'rolling' AND cs.subject_id = r.rowid
++ WHERE r.last_seen >= ?
++ AND r.message_count >= ?
++ AND (cs.updated_at IS NULL OR cs.updated_at < r.last_seen)
++ ORDER BY (cs.updated_at IS NULL) DESC, r.last_seen DESC
++ LIMIT ?`, [lookbackIso, threshold, Math.max(1, limit)]);
++}
++
++// Anon messages for AI batching (mirrors getChatMessagesForAi / getRelayChatMessagesForAi).
++function getAnonChatMessagesForAi({ anonId, sinceTs = null, limit = 300, order = 'asc' } = {}) {
++ let sql = `SELECT cm.id, cm.username, cm.message, cm.message_type, cm.timestamp
++ FROM chat_messages cm
++ WHERE ${_CHAT_AI_WHERE} AND cm.user_id IS NULL AND cm.anon_id = ?`;
++ const params = [String(anonId)];
++ if (sinceTs != null) { sql += ' AND cm.timestamp >= ?'; params.push(sinceTs); }
++ sql += ` ORDER BY cm.id ${order === 'desc' ? 'DESC' : 'ASC'} LIMIT ?`;
++ params.push(Math.max(1, Math.min(2000, limit)));
++ const rows = all(sql, params);
++ return order === 'desc' ? rows.reverse() : rows;
++}
++
++// Anons with enough new chat activity (or a stale summary) to warrant an AI refresh.
++function getAnonsNeedingChatAi({ threshold = 12, staleCutoffIso, sinceTs, limit = 2 } = {}) {
++ const sql = `
++ SELECT cm.anon_id AS anon_id,
++ MAX(cm.id) AS max_id,
++ SUM(CASE WHEN cm.id > COALESCE(cs.last_message_id, 0) THEN 1 ELSE 0 END) AS new_msgs
++ FROM chat_messages cm
++ LEFT JOIN chat_ai_summaries cs
++ ON cs.scope = 'anon' AND cs.window = 'rolling'
++ AND cs.subject_id = CAST(SUBSTR(cm.anon_id, 5) AS INTEGER)
++ WHERE ${_CHAT_AI_WHERE} AND cm.user_id IS NULL AND cm.anon_id IS NOT NULL
++ AND cm.anon_id LIKE 'anon%' AND cm.timestamp >= ?
++ GROUP BY cm.anon_id
++ HAVING new_msgs > 0
++ AND ( new_msgs >= ? OR cs.last_message_id IS NULL OR cs.updated_at IS NULL OR cs.updated_at < ? )
++ ORDER BY (cs.updated_at IS NULL) DESC, new_msgs DESC
++ LIMIT ?`;
++ return all(sql, [sinceTs, threshold, staleCutoffIso, Math.max(1, limit)]);
++}
++
/**
* Live's changes to a staged table it still writes (its capture, relayed over the bridge): the rows
* as they are in Live now, or a delete. Applied only while the table is at 'live' — once Chat writes
@@@ -1478,4 +1625,17 @@ module.exports = {
revokeUserTag,
upsertChatAiSummary,
addChatTimelineEvents,
++ // chat AI (Live's server/ai/chat-ai.js, moved): the job's selection/read helpers and the routes' reads
++ getChatAiSummary,
++ getChatAiSummaries,
++ getChatTimelineEvents,
++ getMaxChatMessageId,
++ countChatMessagesSince,
++ getChatMessagesForAi,
++ getNthRecentChatTs,
++ getUsersNeedingChatAi,
++ getRelayChatMessagesForAi,
++ getRelayUsersNeedingChatAi,
++ getAnonChatMessagesForAi,
++ getAnonsNeedingChatAi,
};
ddiff --git a/server/index.js b/server/index.js
iindex 44648b8..bbd3695 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -22,6 +22,7 @@ const { createMirror } = require('./bridge/live-mirror');
const { createRelay } = require('./events/outbox');
const { createEventsConsumer } = require('./events/consumer');
const subscriptions = require('./events/subscriptions');
++const chatAi = require('./ai/chat-ai');
const { createApp } = require('./app');
async function start() {
@@@ -46,6 +47,9 @@ async function start() {
mirror.start();
relay.start();
events.start();
++ // The chat-AI job (Live's server/ai/chat-ai.js, now Chat's): off unless CHAT_AI_ENABLED=1, and
++ // its writes skip with one log line while Live is still the two summaries tables' authority.
++ chatAi.start();
await new Promise((resolve) => server.listen(config.port, config.host, resolve));
console.log(`[Chat] listening on http://${config.host}:${server.address().port} (ws /ws/chat${config.calls.enabled ? ', /ws/call' : ''})`);
@@@ -67,6 +71,7 @@ async function start() {
relay.stop();
mirror.stop();
events.stop();
++ chatAi.stop();
subs.stop();
try { await Promise.race([mirror.flush(), new Promise((r) => setTimeout(r, 3000))]); } catch { /* */ }
try { chatServer.close(); } catch { /* */ }
ddiff --git a/test/chat-ai.test.js b/test/chat-ai.test.js
nnew file mode 100644
iindex 0000000..e78c25a
---- /dev/null
++++ b/test/chat-ai.test.js
@@@ -0,0 +1,179 @@
++'use strict';
++/**
++ * Plan T3 step 2 — the chat-AI job and its routes, moved into Chat (decision 5):
++ * - the job's selection and write path over Chat's own messages (stub AI client)
++ * - the authority-not-flipped skip (one log line, never forced)
++ * - the extractive fallback when AI does not answer
++ * - each of the six routes' shape, visibility and error codes
++ * - the job's switch (off unless CHAT_AI_ENABLED)
++ */
++const assert = require('assert');
++const { boot, suite, sqliteNow } = require('./helpers');
++
++const t = suite('chat-ai');
++let h, db, ctx, chatAi, aiClient, streamer, alice, bob, channelId, streamId;
++
++function msg(o = {}) {
++ return db.run(
++ `INSERT INTO chat_messages (stream_id, user_id, username, message, message_type, is_global, channel_user_id, timestamp)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
++ [o.stream_id === undefined ? streamId : o.stream_id, o.user_id === undefined ? null : o.user_id,
++ o.username || null, o.message || 'a message', o.message_type || 'chat', o.is_global ? 1 : 0,
++ o.channel_user_id === undefined ? streamer.id : o.channel_user_id, o.timestamp || sqliteNow()]
++ ).lastInsertRowid;
++}
++const asChat = () => { db.setTableAuthority('chat_ai_summaries', 'chat'); db.setTableAuthority('chat_timeline_events', 'chat'); };
++
++t('boot', async () => {
++ h = await boot({ env: { CHAT_AI_ENABLED: '1' } });
++ db = require('../server/db/database');
++ ctx = require('../server/live-context');
++ chatAi = require('../server/ai/chat-ai');
++ aiClient = require('../server/ai/client');
++ streamer = h.addUser('streamer', { role: 'streamer' });
++ alice = h.addUser('alice');
++ bob = h.addUser('bob');
++ channelId = h.addChannel(streamer.id);
++ streamId = h.addStream(streamer.id, channelId);
++ await h.ctx.sync();
++ asChat();
++});
++
++t('the job selects Chat messages and writes the summary through the staged writers', async () => {
++ const calls = [];
++ aiClient._setClient({
++ runs: {
++ create: async (workflow, input) => {
++ calls.push({ workflow, input });
++ return { run: { status: 'succeeded', output: {
++ recent_overview: 'the room talked about games',
++ memory: 'memory line',
++ timeline: [{ label: 'a moment', detail: 'detail', mins_ago: 3 }],
++ } } };
++ },
++ },
++ });
++ msg({ user_id: alice.id, username: 'alice', message: 'hello world' });
++ msg({ user_id: bob.id, username: 'bob', message: 'a big donation', message_type: 'donation' });
++
++ await chatAi._tick();
++
++ const global = calls.find((c) => c.workflow === chatAi.WORKFLOW_GLOBAL);
++ assert.ok(global, 'the global workflow was called');
++ assert.strictEqual(global.input.messages.length, 2);
++ assert.strictEqual(global.input.messages[0].author, 'alice');
++
++ const row = db.getChatAiSummary('global', 0, 'global');
++ assert.ok(row, 'a global summary row was written');
++ assert.strictEqual(row.overview, 'the room talked about games');
++ assert.strictEqual(row.memory_json, 'memory line');
++ assert.ok(JSON.parse(row.timeline_json).some((x) => x.label === 'a moment'));
++ const events = db.getChatTimelineEvents({ scope: 'global', subjectId: 0, limit: 10 });
++ assert.ok(events.some((e) => e.label === 'a moment'), 'the timeline event was appended');
++});
++
++t('a write while Live is still the authority is skipped, never forced', async () => {
++ db.setTableAuthority('chat_ai_summaries', 'live');
++ db.setTableAuthority('chat_timeline_events', 'live');
++ const before = db.getChatAiSummary('global', 0, 'global');
++ msg({ user_id: alice.id, username: 'alice', message: 'while Live owns the table' });
++ await chatAi._tick(); // logs once, returns; no AI call, no write
++ const after = db.getChatAiSummary('global', 0, 'global');
++ assert.strictEqual(after.overview, before.overview);
++ assert.strictEqual(after.message_count, before.message_count);
++ asChat();
++});
++
++t('falls back to the extractive summary when AI does not answer', async () => {
++ aiClient._setClient({ runs: { create: async () => ({ run: { status: 'failed' } }) } });
++ db.run("UPDATE chat_ai_summaries SET updated_at = ? WHERE scope = 'global'", [sqliteNow(-40 * 60 * 1000)]);
++ msg({ user_id: alice.id, username: 'alice', message: 'the extractive line' });
++
++ await chatAi._tick();
++
++ const row = db.getChatAiSummary('global', 0, 'global');
++ assert.ok(/messages from \d+ chatter/.test(row.overview), row.overview);
++ assert.ok(row.overview.includes('the extractive line'), row.overview);
++ aiClient._reset();
++ aiClient._setClient(null);
++});
++
++t('the job is off unless CHAT_AI_ENABLED is set', async () => {
++ const cfg = require('../server/config');
++ assert.strictEqual(cfg.ai.enabled, true, 'this boot enabled it');
++ const before = db.getChatAiSummary('global', 0, 'global').message_count;
++ cfg.ai.enabled = false;
++ msg({ user_id: bob.id, username: 'bob', message: 'while the job is off' });
++ await chatAi._tick();
++ assert.strictEqual(db.getChatAiSummary('global', 0, 'global').message_count, before);
++ cfg.ai.enabled = true;
++});
++
++t('the six routes answer with Live\'s shapes, publicly', async () => {
++ db.upsertChatAiSummary({
++ scope: 'global', subject_id: 0, window: 'global', overview: 'everything happened', memory_json: 'mem',
++ timeline_json: JSON.stringify([{ ts: '2026-01-01 00:00:00', label: 'moment', detail: 'detail' }]),
++ message_count: 5, window_message_count: 4, window_label: 'past hour',
++ });
++ db.upsertChatAiSummary({
++ scope: 'user', subject_id: alice.id, window: 'rolling',
++ overview: JSON.stringify({ today: 'today read', alltime: 'all time read', has_24h: true }),
++ timeline_json: '[]', message_count: 9,
++ });
++ db.upsertChatAiSummary({
++ scope: 'anon', subject_id: 7, window: 'rolling',
++ overview: JSON.stringify({ today: 'anon today', alltime: 'anon all', has_24h: false }), timeline_json: '[]', message_count: 2,
++ });
++ db.recordRelayUser('twitch', 'SomeOne');
++ const ru = db.getRelayUser('twitch', 'someone');
++ db.upsertChatAiSummary({
++ scope: 'relay', subject_id: ru.id, window: 'rolling',
++ overview: JSON.stringify({ today: 'relay today', alltime: 'relay all', has_24h: true }), timeline_json: '[]', message_count: 3,
++ });
++
++ let r = await h.http('GET', '/api/chat/ai/global'); // no token: public, as Live
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.body.insight.overview, 'everything happened');
++ assert.strictEqual(r.body.insight.window_label, 'past hour');
++ assert.deepStrictEqual(r.body.insight.timeline.map((x) => x.label), ['moment']);
++
++ r = await h.http('GET', '/api/chat/ai/timeline?limit=10');
++ assert.strictEqual(r.status, 200, r.text);
++ assert.ok(Array.isArray(r.body.events));
++ assert.strictEqual(typeof r.body.hasMore, 'boolean');
++
++ r = await h.http('GET', `/api/chat/ai/user/${alice.id}`);
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.body.user.id, alice.id);
++ assert.strictEqual(r.body.user.username, 'alice');
++ assert.strictEqual(r.body.streamer, null, 'the streamer half is Live\'s (For Opus)');
++ assert.strictEqual(r.body.insight.overview_24h, 'today read');
++ assert.strictEqual(r.body.insight.overview_alltime, 'all time read');
++ assert.strictEqual(r.body.insight.has_24h, true);
++ r = await h.http('GET', '/api/chat/ai/user/not-a-number');
++ assert.strictEqual(r.status, 400);
++
++ r = await h.http('GET', '/api/chat/ai/anon/anon7');
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.body.user.anon_id, 'anon7');
++ assert.strictEqual(r.body.insight.overview_24h, 'anon today');
++ r = await h.http('GET', '/api/chat/ai/anon/nope');
++ assert.strictEqual(r.status, 400);
++
++ r = await h.http('GET', '/api/chat/ai/relay/twitch/someone');
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.body.user.platform, 'twitch');
++ assert.strictEqual(r.body.insight.overview_alltime, 'relay all');
++ r = await h.http('GET', '/api/chat/ai/relay/twitch/ghost');
++ assert.deepStrictEqual(r.body, { insight: null, user: null });
++
++ r = await h.http('GET', `/api/chat/ai/timeline/${alice.username}`);
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.body.username, 'alice');
++ assert.strictEqual(r.body.chatInsight.overview_24h, 'today read');
++ assert.deepStrictEqual(r.body.sessions, []);
++ r = await h.http('GET', '/api/chat/ai/timeline/nobody');
++ assert.strictEqual(r.status, 404);
++});
++
++t.run(async () => { try { chatAi.stop(); } catch { /* */ } });