openvibers/openvibe.live Commits
AI state: repair keyless tables, unjam backfill, transcribe the live session
Changes
@@@ -50,7 +50,11 @@ router.get('/user/:id', (req, res) => {try {const ov = db.getStreamerOverview(uid);if (ov && (ov.overview || ov.overview_short)) {-- const mems = (db.getStreamMemoriesByUser(uid, 8) || []).map(m => ({++ // 8 was hardcoded, and duplicates meant a viewer saw ~4 distinct moments —++ // the "memories don't include everything" complaint. Allow a caller to ask++ // for more, and default high enough to read as a real timeline.++ const memLimit = Math.min(200, Math.max(1, parseInt(req.query.memories, 10) || 40));++ const mems = (db.getStreamMemoriesByUser(uid, memLimit) || []).map(m => ({description: m.description || '',created_at: m.created_at,stream_id: m.stream_id,@@@ -243,18 +247,28 @@ router.get('/vod-transcripts', (req, res) => {else events.push({ start: r.start_sec, end: r.end_sec, label: r.label, confidence: r.confidence });}} catch { /* */ }-- // Fall back to the batch-transcribed blob for VODs with no live timeline.-- if (!segments.length) {-- try {-- const st = db.getVodAiState(vodId);-- if (st && st.ai_transcript_json) {-- const parsed = JSON.parse(st.ai_transcript_json);-- if (Array.isArray(parsed)) segments = parsed;-- }-- } catch { /* */ }-- }++ // The timeline and the batch-transcribed blob are two views of the same audio++ // and either can be the more complete one, so take whichever actually carries++ // more speech instead of only falling back when the timeline is empty. A++ // partially-linked timeline used to win by default and truncate the result:++ // vod 2163 served 426 characters from 7 linked rows while its blob held the++ // full 3548. Sound events only exist on the timeline, so they are kept either way.++ let blob = [];++ try {++ const st = db.getVodAiState(vodId);++ if (st && st.ai_transcript_json) {++ const parsed = JSON.parse(st.ai_transcript_json);++ if (Array.isArray(parsed)) blob = parsed;++ }++ } catch { /* */ }++ const spoken = (arr) => arr.reduce((n, x) => n + String(x.text || '').trim().length, 0);++ if (spoken(blob) > spoken(segments)) segments = blob;++ segments.sort((a, b) => (Number(a.start) || 0) - (Number(b.start) || 0));let overviewShort = null;try { overviewShort = (db.getVodAiState(vodId) || {}).ai_overview_short || null; } catch { /* */ }++ // `segments` may be [] for a VOD whose audio had no speech; that is a real++ // answer, not a missing one, so transcript stays null and segments stays [].++const text = segments.map(s => String(s.text || '').trim()).filter(Boolean).join(' ');out[vodId] = {transcript: text || null,@@@ -93,6 +93,20 @@ async function _processSegment(stream, seg) {}} catch { /* classifier optional */ }++ // Stamp the Media vod id NOW rather than waiting for the vod.ready webhook to++ // backfill it. getTimelineByVod() matches on vod_id, so while a stream is still++ // recording every row it writes is invisible to++ // openvibe.media/live/:sel/transcript.json — which is why `current` came back with++ // transcript: null, segments: [] after twenty minutes of live speech while the++ // finished sessions below it had transcripts. The recorder holds the vod id in++ // memory for the whole recording; after it stops, reuse whatever vod_id the webhook++ // already stamped on this stream so late-transcribed spool segments land on the VOD++ // too instead of being orphaned.++ let vodId = null;++ try { vodId = require('../streaming/recorder').getActiveRecording(stream.id)?.vodId || null; } catch { /* */ }++ if (!vodId) { try { vodId = db.getTimelineVodId(stream.id); } catch { /* */ } }++ if (vodId) for (const e of events) e.vod_id = vodId;++if (events.length) db.addTimelineEvents(events);audio.discardSegment(stream.id, seg.name);@@@ -111,6 +111,106 @@ function getDb() {return db;}++/**++ * Collapse duplicate rows in an AI-state table down to one row per id and put a UNIQUE++ * index on the key so `INSERT OR IGNORE` behaves as its callers assume.++ *++ * Values are merged per column rather than by keeping a single "best" row: duplicates++ * were created at different times, so the transcript may sit on one row and the overview++ * on another. Longest wins for text we accumulate; for transcript_status the most++ * settled state wins, so a stray 'pending' duplicate cannot resurrect finished work.++ */++function _dedupeKeyedTable(database, table, key) {++ try {++ const idx = `idx_${table}_${key}_unique`;++ const has = database.prepare(++ "SELECT 1 FROM sqlite_master WHERE type='index' AND name=?").get(idx);++ if (has) return; // already repaired++ // Column-driven, never a hardcoded list: clip_ai_state carries clip_notified /++ // clip_notify_at that vod_ai_state does not, and a fixed column list would drop++ // them on the floor during the rebuild.++ const cols = database.prepare(`PRAGMA table_info(${table})`).all().map(c => c.name);++ if (!cols.includes(key)) return;++ const others = cols.filter(c => c !== key);++ const dupes = database.prepare(++ `SELECT COUNT(*) - COUNT(DISTINCT ${key}) AS extra FROM ${table}`).get();++ if (dupes && dupes.extra > 0) {++ // Merge per column, preferring the richest value: duplicates were written at++ // different times, so a transcript can sit on one row and an overview on++ // another — keeping a single "best" row wholesale would discard the other.++ const pick = (c) => {++ if (c === 'transcript_status') {++ // Most-settled state wins, so a stray 'pending' duplicate cannot++ // resurrect work that already finished.++ return `(SELECT x.${c} FROM ${table} x WHERE x.${key} = k.${key} AND x.${c} IS NOT NULL++ ORDER BY CASE x.${c} WHEN 'done' THEN 0 WHEN 'empty' THEN 1 WHEN 'failed' THEN 2++ WHEN 'processing' THEN 3 WHEN 'retry' THEN 4 ELSE 5 END LIMIT 1)`;++ }++ if (c === 'transcript_next_at') {++ return `(SELECT MIN(x.${c}) FROM ${table} x WHERE x.${key} = k.${key} AND x.${c} IS NOT NULL)`;++ }++ if (c === 'transcript_attempts') {++ return `(SELECT MAX(COALESCE(x.${c},0)) FROM ${table} x WHERE x.${key} = k.${key})`;++ }++ // Everything else: any non-null value, longest first. For accumulated text++ // (transcripts, overviews) longest is the most complete; for flags and++ // timestamps it just means "a real value beats NULL".++ return `(SELECT x.${c} FROM ${table} x WHERE x.${key} = k.${key} AND x.${c} IS NOT NULL++ ORDER BY LENGTH(CAST(x.${c} AS TEXT)) DESC LIMIT 1)`;++ };++ const selects = [`k.${key} AS ${key}`, ...others.map(c => `${pick(c)} AS ${c}`)].join(',\n ');++ database.exec('BEGIN');++ try {++ database.exec(`CREATE TEMP TABLE _merge_${table} AS++ SELECT ${selects}++ FROM (SELECT DISTINCT ${key} FROM ${table}) k`);++ database.exec(`DELETE FROM ${table}`);++ database.exec(`INSERT INTO ${table} (${cols.join(', ')})++ SELECT ${cols.join(', ')} FROM _merge_${table}`);++ database.exec(`DROP TABLE _merge_${table}`);++ database.exec('COMMIT');++ console.log(`[DB] ${table}: merged ${dupes.extra} duplicate row(s) down to one per ${key}`);++ } catch (e) {++ try { database.exec('ROLLBACK'); } catch { /* */ }++ console.warn(`[DB] ${table} dedupe failed, leaving as-is:`, e.message);++ return; // never index over dirty data++ }++ }++ database.exec(`CREATE UNIQUE INDEX IF NOT EXISTS ${idx} ON ${table}(${key})`);++ } catch (e) {++ console.warn(`[DB] ${table} integrity repair skipped:`, e.message);++ }++}++++/**++ * One memory per (stream, offset). Re-analysing a stream used to append a second++ * description of the very same moment, so a viewer's memory list read as near-duplicate++ * pairs a minute apart. Keep the longest description (the richest capture, e.g. the one++ * that also carries the "heard:" transcript clause) and let the UNIQUE index make++ * addStreamMemory's INSERT OR IGNORE actually ignore.++ */++function _dedupeStreamMemories(database) {++ try {++ const idx = 'idx_stream_memories_moment_unique';++ if (database.prepare("SELECT 1 FROM sqlite_master WHERE type='index' AND name=?").get(idx)) return;++ const extra = database.prepare(`SELECT COUNT(*) - COUNT(DISTINCT stream_id || ':' || offset_seconds) AS extra++ FROM stream_memories`).get();++ if (extra && extra.extra > 0) {++ const res = database.prepare(`DELETE FROM stream_memories WHERE id NOT IN (++ SELECT id FROM (++ SELECT id, ROW_NUMBER() OVER (++ PARTITION BY stream_id, offset_seconds++ ORDER BY LENGTH(COALESCE(description,'')) DESC,++ (transcript_json IS NOT NULL) DESC, id DESC) AS rn++ FROM stream_memories) WHERE rn = 1)`).run();++ console.log(`[DB] stream_memories: removed ${res.changes} duplicate moment(s)`);++ }++ database.exec(`CREATE UNIQUE INDEX IF NOT EXISTS ${idx} ON stream_memories(stream_id, offset_seconds)`);++ } catch (e) {++ console.warn('[DB] stream_memories dedupe skipped:', e.message);++ }++}++function initDb() {const database = getDb();const schema = fs.readFileSync(path.join(__dirname, 'schema.sql'), 'utf8');@@@ -924,6 +1024,30 @@ function initDb() {)`);database.exec('CREATE INDEX IF NOT EXISTS idx_stream_memories_stream ON stream_memories(stream_id, offset_seconds)');++ // ── Integrity repair: one row per key ────────────────────────────────────++ // These three tables are all written through "insert once, update in place"++ // helpers whose no-op-on-duplicate behaviour depends on a uniqueness++ // constraint. Production drifted: vod_ai_state was created as `vod_id INT`++ // with NO primary key (database.js declares INTEGER PRIMARY KEY, but++ // CREATE TABLE IF NOT EXISTS never repairs an existing table). With nothing to++ // conflict against, `INSERT OR IGNORE INTO vod_ai_state (vod_id)` appended a++ // fresh row on EVERY call — 2818 rows for 487 VODs, one VOD holding 341.++ //++ // That is not just untidy, it silently broke the AI backfill: the work queues++ // are `SELECT ... WHERE ai_overview_short IS NULL ORDER BY vod_id DESC LIMIT 4`,++ // so a VOD with four empty duplicate rows fills the entire batch with itself and++ // no other VOD is ever processed. Every recent VOD had ai_overview null as a++ // result. stream_memories had the same shape of problem for a different reason:++ // no constraint at all, so re-analysing a stream stored the same moment again++ // (the /live/:sel/transcript.json memories list was ~50% duplicates).++ //++ // Merge duplicates field-by-field, preferring the richest value rather than an++ // arbitrary row — a transcript and an overview can live on different duplicates,++ // and picking one row wholesale would throw the other away.++ _dedupeKeyedTable(database, 'vod_ai_state', 'vod_id');++ _dedupeKeyedTable(database, 'clip_ai_state', 'clip_id');++ _dedupeStreamMemories(database);++// ── Unified audio timeline ───────────────────────────────────────────────// One time-indexed row per thing heard on a stream: a phrase that was spoken// ('speech') or a non-speech sound that was recognised ('sound'). Replaces the@@@ -2469,7 +2593,9 @@ function endOtherLiveStreamsForSlot(managedStreamId, keepStreamId) {// ── AI analysis helpers ──────────────────────────────────────function addStreamMemory({ stream_id, user_id = null, offset_seconds = 0, description, tags = null, thumbnail_url = null, transcript_json = null }) {-- return run(`INSERT INTO stream_memories (stream_id, user_id, offset_seconds, description, tags, thumbnail_url, transcript_json)++ // OR IGNORE against idx_stream_memories_moment_unique: re-analysing a stream must not++ // store a second description of a moment already captured.++ return run(`INSERT OR IGNORE INTO stream_memories (stream_id, user_id, offset_seconds, description, tags, thumbnail_url, transcript_json)VALUES (?, ?, ?, ?, ?, ?, ?)`,[stream_id, user_id, Math.max(0, Math.round(offset_seconds || 0)), description || '',tags ? (typeof tags === 'string' ? tags : JSON.stringify(tags)) : null, thumbnail_url,@@@ -2801,6 +2927,22 @@ function getTimelineByVod(vodId) {} catch { return []; }}++/**++ * The vod_id already stamped on this stream's timeline, if any.++ *++ * Transcription of spooled audio keeps running for a while after the vod.ready webhook++ * fires, and linkTimelineToVod() is a one-shot UPDATE — so those late rows used to stay++ * vod_id NULL forever and never appear in the VOD's transcript. (Stream 2128: 11 speech++ * rows orphaned against 2 linked; vod 2163 served 426 characters when the full++ * transcript was 3548.) Late writers call this to stamp themselves correctly.++ */++function getTimelineVodId(streamId) {++ try {++ const r = get('SELECT vod_id FROM stream_timeline_events WHERE stream_id = ? AND vod_id IS NOT NULL LIMIT 1', [streamId]);++ return r ? r.vod_id : null;++ } catch { return null; }++}++/** Attach a vod_id to a finished stream's rows so VOD views can reuse the timeline. */function linkTimelineToVod(streamId, vodId) {try { return run('UPDATE stream_timeline_events SET vod_id = ? WHERE stream_id = ? AND vod_id IS NULL', [vodId, streamId]); }@@@ -7279,7 +7421,7 @@ module.exports = {setVodTranscriptStatus, setClipTranscriptStatus, bumpVodTranscriptAttempt, bumpClipTranscriptAttempt,updatePasteAi, cleanupMalformedAiText, recordAiUsage, getAiCostToday, getAiCostTodayForUser, getAiUsageSummary,getStreamMemoriesByUser, countStreamMemoriesByUser, getAiMomentCandidates, getStreamTranscriptSegments, getUserPastesForAi,-- addTimelineEvents, getTimeline, getTimelineText, getTimelineCoverage, linkTimelineToVod, getTimelineByVod,++ addTimelineEvents, getTimeline, getTimelineText, getTimelineCoverage, linkTimelineToVod, getTimelineByVod, getTimelineVodId,getVodsForMomentRanking, getClipStartTimesForStream, getChatSpikeOffsets,getLiveChatBuckets, getRecentChatText, getVodsWithoutAutoClip,upsertStreamerOverview, getStreamerOverview, getAllStreamerOverviews, getStreamersNeedingOverview,nnew file mode 100644@@@ -0,0 +1,155 @@++/**++ * Regression: openvibe.media/live/:sel/transcript.json served broken data.++ *++ * Four independent faults, all reproduced here against a throwaway DB:++ *++ * 1. vod_ai_state / clip_ai_state drifted in production to a keyless schema++ * (`vod_id INT`, no PRIMARY KEY — CREATE TABLE IF NOT EXISTS never repairs an++ * existing table). Their writers use `INSERT OR IGNORE ... (vod_id) VALUES (?)`,++ * which needs a uniqueness constraint to be a no-op, so every call appended a row:++ * 2818 rows for 487 VODs, 2986 for 103 clips.++ *++ * 2. That silently jammed the AI backfill. The queues are++ * `WHERE ai_overview_short IS NULL ORDER BY vod_id DESC LIMIT 4`, so a VOD with four++ * empty duplicates filled the whole batch with itself and nothing else was ever++ * processed — every recent VOD had ai_overview null.++ *++ * 3. stream_memories had no constraint either, so re-analysing a stream stored the same++ * moment again; the memories list read as near-duplicate pairs a minute apart.++ *++ * 4. The transcript endpoint preferred the timeline over the batch blob whenever the++ * timeline was non-empty, so a partially-linked timeline truncated the result++ * (vod 2163: 426 characters served against a 3548-character transcript).++ *++ * The merge must be lossless: duplicates were written at different times, so a transcript++ * can sit on one row and an overview on another.++ */++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++++const tmp = path.join(os.tmpdir(), `ov-ai-state-${Date.now()}.db`);++process.env.DB_PATH = tmp;++++const Database = require('better-sqlite3');++++// ── Build a DB with the DRIFTED (keyless) production schema, then dirty it ──────────++{++ const raw = new Database(tmp);++ raw.exec(`CREATE TABLE vod_ai_state (++ vod_id INT, ai_overview_short TEXT, ai_transcript_json TEXT, transcript_status TEXT,++ transcript_attempts INT, transcript_error TEXT, transcript_next_at NUM)`);++ raw.exec(`CREATE TABLE clip_ai_state (++ clip_id INT, ai_overview_short TEXT, ai_transcript_json TEXT, transcript_status TEXT,++ transcript_attempts INT, transcript_error TEXT, transcript_next_at NUM,++ clip_notified INT, clip_notify_at NUM)`);++++ const v = raw.prepare(`INSERT INTO vod_ai_state++ (vod_id, ai_overview_short, ai_transcript_json, transcript_status, transcript_attempts)++ VALUES (?,?,?,?,?)`);++ // vod 100: transcript and overview live on DIFFERENT duplicate rows, and a later++ // duplicate carries a regressed 'pending' status. A lossless merge keeps all three.++ v.run(100, null, JSON.stringify([{ start: 0, end: 2, text: 'the full transcript' }]), 'done', 1);++ v.run(100, 'the overview', null, 'done', 2);++ v.run(100, null, null, 'pending', 0);++ // vod 101: four empty duplicates — the shape that jammed the overview queue.++ for (let i = 0; i < 4; i++) v.run(101, null, null, null, 0);++++ const c = raw.prepare(`INSERT INTO clip_ai_state++ (clip_id, ai_overview_short, clip_notified, clip_notify_at) VALUES (?,?,?,?)`);++ c.run(200, 'clip overview', 1, 1234); // extra columns must survive the rebuild++ c.run(200, null, null, null);++ raw.close();++}++++const db = require('../server/db/database');++db.initDb();++++// stream_memories and stream_timeline_events both carry a FK to streams(id), so the++// fixture needs real parent rows (user_id is the only NOT NULL column without a default).++{++ const w = new Database(tmp);++ w.pragma('foreign_keys = OFF'); // fixture only: streams.user_id -> users(id) is not under test++ const ins = w.prepare('INSERT OR IGNORE INTO streams (id, user_id) VALUES (?, 1)');++ for (const id of [900, 901, 902]) ins.run(id);++ w.close();++}++++// ── 1: one row per key, nothing lost ────────────────────────────────────────────────++const raw = new Database(tmp, { readonly: true });++const count = (sql) => raw.prepare(sql).get().n;++assert.strictEqual(count('SELECT COUNT(*) n FROM vod_ai_state'), 2, 'vod_ai_state must collapse to one row per vod');++assert.strictEqual(count('SELECT COUNT(*) n FROM clip_ai_state'), 1, 'clip_ai_state must collapse to one row per clip');++++const v100 = db.getVodAiState(100);++assert.ok(v100.ai_transcript_json && v100.ai_transcript_json.includes('the full transcript'),++ 'transcript must survive even though it lived on a different duplicate than the overview');++assert.strictEqual(v100.ai_overview_short, 'the overview',++ 'overview must survive even though it lived on a different duplicate than the transcript');++assert.strictEqual(v100.transcript_status, 'done',++ "a stray 'pending' duplicate must not resurrect finished work");++console.log('OK 1: duplicates merged losslessly (transcript + overview + settled status all kept)');++++const clip = raw.prepare('SELECT * FROM clip_ai_state WHERE clip_id = 200').get();++assert.strictEqual(clip.clip_notified, 1, 'clip_notified must survive the rebuild');++assert.strictEqual(clip.clip_notify_at, 1234, 'clip_notify_at must survive the rebuild');++assert.strictEqual(clip.ai_overview_short, 'clip overview');++console.log('OK 2: table-specific extra columns preserved (column-driven merge, not a fixed list)');++++// ── 2: the backfill queue is no longer jammed by one VOD ────────────────────────────++const queue = db.getVodsNeedingOverview(4).map(r => r.id);++assert.strictEqual(queue.length, new Set(queue).size, `overview queue still returns duplicates: ${queue}`);++assert.ok(queue.includes(101), 'the VOD lacking an overview must still be queued');++assert.ok(!queue.includes(100), 'a VOD that already has an overview must not be queued');++console.log(`OK 3: overview backfill queue returns distinct VODs (${JSON.stringify(queue)})`);++++// ── 3: re-seeding is genuinely idempotent now ───────────────────────────────────────++for (let i = 0; i < 5; i++) db.setVodTranscriptStatus(100, 'pending');++assert.strictEqual(count('SELECT COUNT(*) n FROM vod_ai_state WHERE vod_id = 100'), 1,++ 'INSERT OR IGNORE must no-op now that a UNIQUE index exists');++assert.ok((db.getVodAiState(100).ai_transcript_json || '').includes('the full transcript'),++ 're-seeding must not clobber an existing transcript');++console.log('OK 4: five re-seeds leave exactly one row, transcript intact');++++// ── 4: one memory per (stream, offset) ──────────────────────────────────────────────++db.addStreamMemory({ stream_id: 900, offset_seconds: 15, description: 'a red figure by a brick building' });++db.addStreamMemory({ stream_id: 900, offset_seconds: 15, description: 'a red horse statue by a brick building' });++db.addStreamMemory({ stream_id: 900, offset_seconds: 23, description: 'a different moment' });++assert.strictEqual(count('SELECT COUNT(*) n FROM stream_memories WHERE stream_id = 900'), 2,++ 're-describing the same moment must not create a second memory');++console.log('OK 5: duplicate stream memories rejected, distinct moments kept');++++// ── 5: transcript source selection prefers the more complete view ───────────────────++// Mirrors the endpoint's rule: whichever view carries more speech wins.++const spoken = (arr) => arr.reduce((n, x) => n + String(x.text || '').trim().length, 0);++const choose = (timeline, blob) => (spoken(blob) > spoken(timeline) ? blob : timeline);++const partialTimeline = [{ start: 0, text: 'x'.repeat(426) }];++const fullBlob = [{ start: 0, text: 'y'.repeat(3548) }];++assert.strictEqual(spoken(choose(partialTimeline, fullBlob)), 3548,++ 'a partially-linked timeline must not beat a fuller transcript blob');++assert.strictEqual(spoken(choose(fullBlob, partialTimeline)), 3548,++ 'and the richer source wins regardless of which side it is on');++assert.strictEqual(spoken(choose([], [])), 0, 'no speech anywhere is a real answer, not an error');++console.log('OK 6: transcript endpoint picks whichever source carries more speech');++++// ── 6: live rows are reachable — the headline complaint ─────────────────────────────++db.addTimelineEvents([++ { stream_id: 901, kind: 'speech', start_sec: 1, end_sec: 2, text: 'live speech', vod_id: 500 },++ { stream_id: 901, kind: 'sound', start_sec: 3, end_sec: 4, label: 'laughter', vod_id: 500 },++]);++const byVod = db.getTimelineByVod(500);++assert.strictEqual(byVod.length, 2, 'rows stamped with vod_id while live must be queryable by vod_id');++assert.strictEqual(db.getTimelineVodId(901), 500, 'a late writer must be able to recover the stream vod id');++// A row written before the vod id was known stays orphaned until relinked — that is the++// case linkTimelineToVod covers, and getTimelineVodId lets later writers self-stamp.++db.addTimelineEvents([{ stream_id: 902, kind: 'speech', start_sec: 1, text: 'orphan' }]);++assert.strictEqual(db.getTimelineByVod(501).length, 0, 'unlinked rows are not visible by vod_id');++db.linkTimelineToVod(902, 501);++assert.strictEqual(db.getTimelineByVod(501).length, 1, 'linkTimelineToVod must adopt orphaned rows');++console.log('OK 7: live timeline rows reachable by vod_id; orphans still adoptable');++++raw.close();++try { fs.unlinkSync(tmp); } catch { /* */ }++for (const ext of ['-wal', '-shm']) { try { fs.unlinkSync(tmp + ext); } catch { /* */ } }++console.log('✅ AI state integrity regression test passed');
@@ -50,7 +50,11 @@ router.get('/user/:id', (req, res) => {
@@ -50,7 +50,11 @@ router.get('/user/:id', (req, res) => {
try {
try {
const ov = db.getStreamerOverview(uid);
const ov = db.getStreamerOverview(uid);
if (ov && (ov.overview || ov.overview_short)) {
if (ov && (ov.overview || ov.overview_short)) {
- const mems = (db.getStreamMemoriesByUser(uid, 8) || []).map(m => ({
+ // 8 was hardcoded, and duplicates meant a viewer saw ~4 distinct moments —
+ // the "memories don't include everything" complaint. Allow a caller to ask
+ // for more, and default high enough to read as a real timeline.
+ const memLimit = Math.min(200, Math.max(1, parseInt(req.query.memories, 10) || 40));
+ const mems = (db.getStreamMemoriesByUser(uid, memLimit) || []).map(m => ({
description: m.description || '',
description: m.description || '',
created_at: m.created_at,
created_at: m.created_at,
stream_id: m.stream_id,
stream_id: m.stream_id,
@@ -243,18 +247,28 @@ router.get('/vod-transcripts', (req, res) => {
@@ -243,18 +247,28 @@ router.get('/vod-transcripts', (req, res) => {
else events.push({ start: r.start_sec, end: r.end_sec, label: r.label, confidence: r.confidence });
else events.push({ start: r.start_sec, end: r.end_sec, label: r.label, confidence: r.confidence });
}
}
} catch { /* */ }
} catch { /* */ }
- // Fall back to the batch-transcribed blob for VODs with no live timeline.
- if (!segments.length) {
- try {
- const st = db.getVodAiState(vodId);
- if (st && st.ai_transcript_json) {
- const parsed = JSON.parse(st.ai_transcript_json);
- if (Array.isArray(parsed)) segments = parsed;
- }
- } catch { /* */ }
- }
+ // The timeline and the batch-transcribed blob are two views of the same audio
+ // and either can be the more complete one, so take whichever actually carries
+ // more speech instead of only falling back when the timeline is empty. A
+ // partially-linked timeline used to win by default and truncate the result:
+ // vod 2163 served 426 characters from 7 linked rows while its blob held the
+ // full 3548. Sound events only exist on the timeline, so they are kept either way.
+ let blob = [];
+ try {
+ const st = db.getVodAiState(vodId);
+ if (st && st.ai_transcript_json) {
+ const parsed = JSON.parse(st.ai_transcript_json);
+ if (Array.isArray(parsed)) blob = parsed;
+ }
+ } catch { /* */ }
+ const spoken = (arr) => arr.reduce((n, x) => n + String(x.text || '').trim().length, 0);
+ if (spoken(blob) > spoken(segments)) segments = blob;
+ segments.sort((a, b) => (Number(a.start) || 0) - (Number(b.start) || 0));
let overviewShort = null;
let overviewShort = null;
try { overviewShort = (db.getVodAiState(vodId) || {}).ai_overview_short || null; } catch { /* */ }
try { overviewShort = (db.getVodAiState(vodId) || {}).ai_overview_short || null; } catch { /* */ }
+ // `segments` may be [] for a VOD whose audio had no speech; that is a real
+ // answer, not a missing one, so transcript stays null and segments stays [].
+
const text = segments.map(s => String(s.text || '').trim()).filter(Boolean).join(' ');
const text = segments.map(s => String(s.text || '').trim()).filter(Boolean).join(' ');
out[vodId] = {
out[vodId] = {
transcript: text || null,
transcript: text || null,
@@ -93,6 +93,20 @@ async function _processSegment(stream, seg) {
@@ -93,6 +93,20 @@ async function _processSegment(stream, seg) {
}
}
} catch { /* classifier optional */ }
} catch { /* classifier optional */ }
+ // Stamp the Media vod id NOW rather than waiting for the vod.ready webhook to
+ // backfill it. getTimelineByVod() matches on vod_id, so while a stream is still
+ // recording every row it writes is invisible to
+ // openvibe.media/live/:sel/transcript.json — which is why `current` came back with
+ // transcript: null, segments: [] after twenty minutes of live speech while the
+ // finished sessions below it had transcripts. The recorder holds the vod id in
+ // memory for the whole recording; after it stops, reuse whatever vod_id the webhook
+ // already stamped on this stream so late-transcribed spool segments land on the VOD
+ // too instead of being orphaned.
+ let vodId = null;
+ try { vodId = require('../streaming/recorder').getActiveRecording(stream.id)?.vodId || null; } catch { /* */ }
+ if (!vodId) { try { vodId = db.getTimelineVodId(stream.id); } catch { /* */ } }
+ if (vodId) for (const e of events) e.vod_id = vodId;
+
if (events.length) db.addTimelineEvents(events);
if (events.length) db.addTimelineEvents(events);
audio.discardSegment(stream.id, seg.name);
audio.discardSegment(stream.id, seg.name);
@@ -111,6 +111,106 @@ function getDb() {
@@ -111,6 +111,106 @@ function getDb() {
return db;
return db;
}
}
+/**
+ * Collapse duplicate rows in an AI-state table down to one row per id and put a UNIQUE
+ * index on the key so `INSERT OR IGNORE` behaves as its callers assume.
+ *
+ * Values are merged per column rather than by keeping a single "best" row: duplicates
+ * were created at different times, so the transcript may sit on one row and the overview
+ * on another. Longest wins for text we accumulate; for transcript_status the most
+ * settled state wins, so a stray 'pending' duplicate cannot resurrect finished work.
+ */
+function _dedupeKeyedTable(database, table, key) {
+ try {
+ const idx = `idx_${table}_${key}_unique`;
+ const has = database.prepare(
+ "SELECT 1 FROM sqlite_master WHERE type='index' AND name=?").get(idx);
+ if (has) return; // already repaired
+ // Column-driven, never a hardcoded list: clip_ai_state carries clip_notified /
+ // clip_notify_at that vod_ai_state does not, and a fixed column list would drop
+ // them on the floor during the rebuild.
+ const cols = database.prepare(`PRAGMA table_info(${table})`).all().map(c => c.name);
+ if (!cols.includes(key)) return;
+ const others = cols.filter(c => c !== key);
+ const dupes = database.prepare(
+ `SELECT COUNT(*) - COUNT(DISTINCT ${key}) AS extra FROM ${table}`).get();
+ if (dupes && dupes.extra > 0) {
+ // Merge per column, preferring the richest value: duplicates were written at
+ // different times, so a transcript can sit on one row and an overview on
+ // another — keeping a single "best" row wholesale would discard the other.
+ const pick = (c) => {
+ if (c === 'transcript_status') {
+ // Most-settled state wins, so a stray 'pending' duplicate cannot
+ // resurrect work that already finished.
+ return `(SELECT x.${c} FROM ${table} x WHERE x.${key} = k.${key} AND x.${c} IS NOT NULL
+ ORDER BY CASE x.${c} WHEN 'done' THEN 0 WHEN 'empty' THEN 1 WHEN 'failed' THEN 2
+ WHEN 'processing' THEN 3 WHEN 'retry' THEN 4 ELSE 5 END LIMIT 1)`;
+ }
+ if (c === 'transcript_next_at') {
+ return `(SELECT MIN(x.${c}) FROM ${table} x WHERE x.${key} = k.${key} AND x.${c} IS NOT NULL)`;
+ }
+ if (c === 'transcript_attempts') {
+ return `(SELECT MAX(COALESCE(x.${c},0)) FROM ${table} x WHERE x.${key} = k.${key})`;
+ }
+ // Everything else: any non-null value, longest first. For accumulated text
+ // (transcripts, overviews) longest is the most complete; for flags and
+ // timestamps it just means "a real value beats NULL".
+ return `(SELECT x.${c} FROM ${table} x WHERE x.${key} = k.${key} AND x.${c} IS NOT NULL
+ ORDER BY LENGTH(CAST(x.${c} AS TEXT)) DESC LIMIT 1)`;
+ };
+ const selects = [`k.${key} AS ${key}`, ...others.map(c => `${pick(c)} AS ${c}`)].join(',\n ');
+ database.exec('BEGIN');
+ try {
+ database.exec(`CREATE TEMP TABLE _merge_${table} AS
+ SELECT ${selects}
+ FROM (SELECT DISTINCT ${key} FROM ${table}) k`);
+ database.exec(`DELETE FROM ${table}`);
+ database.exec(`INSERT INTO ${table} (${cols.join(', ')})
+ SELECT ${cols.join(', ')} FROM _merge_${table}`);
+ database.exec(`DROP TABLE _merge_${table}`);
+ database.exec('COMMIT');
+ console.log(`[DB] ${table}: merged ${dupes.extra} duplicate row(s) down to one per ${key}`);
+ } catch (e) {
+ try { database.exec('ROLLBACK'); } catch { /* */ }
+ console.warn(`[DB] ${table} dedupe failed, leaving as-is:`, e.message);
+ return; // never index over dirty data
+ }
+ }
+ database.exec(`CREATE UNIQUE INDEX IF NOT EXISTS ${idx} ON ${table}(${key})`);
+ } catch (e) {
+ console.warn(`[DB] ${table} integrity repair skipped:`, e.message);
+ }
+}
+
+/**
+ * One memory per (stream, offset). Re-analysing a stream used to append a second
+ * description of the very same moment, so a viewer's memory list read as near-duplicate
+ * pairs a minute apart. Keep the longest description (the richest capture, e.g. the one
+ * that also carries the "heard:" transcript clause) and let the UNIQUE index make
+ * addStreamMemory's INSERT OR IGNORE actually ignore.
+ */
+function _dedupeStreamMemories(database) {
+ try {
+ const idx = 'idx_stream_memories_moment_unique';
+ if (database.prepare("SELECT 1 FROM sqlite_master WHERE type='index' AND name=?").get(idx)) return;
+ const extra = database.prepare(`SELECT COUNT(*) - COUNT(DISTINCT stream_id || ':' || offset_seconds) AS extra
+ FROM stream_memories`).get();
+ if (extra && extra.extra > 0) {
+ const res = database.prepare(`DELETE FROM stream_memories WHERE id NOT IN (
+ SELECT id FROM (
+ SELECT id, ROW_NUMBER() OVER (
+ PARTITION BY stream_id, offset_seconds
+ ORDER BY LENGTH(COALESCE(description,'')) DESC,
+ (transcript_json IS NOT NULL) DESC, id DESC) AS rn
+ FROM stream_memories) WHERE rn = 1)`).run();
+ console.log(`[DB] stream_memories: removed ${res.changes} duplicate moment(s)`);
+ }
+ database.exec(`CREATE UNIQUE INDEX IF NOT EXISTS ${idx} ON stream_memories(stream_id, offset_seconds)`);
+ } catch (e) {
+ console.warn('[DB] stream_memories dedupe skipped:', e.message);
+ }
+}
+
function initDb() {
function initDb() {
const database = getDb();
const database = getDb();
const schema = fs.readFileSync(path.join(__dirname, 'schema.sql'), 'utf8');
const schema = fs.readFileSync(path.join(__dirname, 'schema.sql'), 'utf8');
@@ -924,6 +1024,30 @@ function initDb() {
@@ -924,6 +1024,30 @@ function initDb() {
)`);
)`);
database.exec('CREATE INDEX IF NOT EXISTS idx_stream_memories_stream ON stream_memories(stream_id, offset_seconds)');
database.exec('CREATE INDEX IF NOT EXISTS idx_stream_memories_stream ON stream_memories(stream_id, offset_seconds)');
+ // ── Integrity repair: one row per key ────────────────────────────────────
+ // These three tables are all written through "insert once, update in place"
+ // helpers whose no-op-on-duplicate behaviour depends on a uniqueness
+ // constraint. Production drifted: vod_ai_state was created as `vod_id INT`
+ // with NO primary key (database.js declares INTEGER PRIMARY KEY, but
+ // CREATE TABLE IF NOT EXISTS never repairs an existing table). With nothing to
+ // conflict against, `INSERT OR IGNORE INTO vod_ai_state (vod_id)` appended a
+ // fresh row on EVERY call — 2818 rows for 487 VODs, one VOD holding 341.
+ //
+ // That is not just untidy, it silently broke the AI backfill: the work queues
+ // are `SELECT ... WHERE ai_overview_short IS NULL ORDER BY vod_id DESC LIMIT 4`,
+ // so a VOD with four empty duplicate rows fills the entire batch with itself and
+ // no other VOD is ever processed. Every recent VOD had ai_overview null as a
+ // result. stream_memories had the same shape of problem for a different reason:
+ // no constraint at all, so re-analysing a stream stored the same moment again
+ // (the /live/:sel/transcript.json memories list was ~50% duplicates).
+ //
+ // Merge duplicates field-by-field, preferring the richest value rather than an
+ // arbitrary row — a transcript and an overview can live on different duplicates,
+ // and picking one row wholesale would throw the other away.
+ _dedupeKeyedTable(database, 'vod_ai_state', 'vod_id');
+ _dedupeKeyedTable(database, 'clip_ai_state', 'clip_id');
+ _dedupeStreamMemories(database);
+
// ── Unified audio timeline ───────────────────────────────────────────────
// ── Unified audio timeline ───────────────────────────────────────────────
// One time-indexed row per thing heard on a stream: a phrase that was spoken
// One time-indexed row per thing heard on a stream: a phrase that was spoken
// ('speech') or a non-speech sound that was recognised ('sound'). Replaces the
// ('speech') or a non-speech sound that was recognised ('sound'). Replaces the
@@ -2469,7 +2593,9 @@ function endOtherLiveStreamsForSlot(managedStreamId, keepStreamId) {
@@ -2469,7 +2593,9 @@ function endOtherLiveStreamsForSlot(managedStreamId, keepStreamId) {
// ── AI analysis helpers ──────────────────────────────────────
// ── AI analysis helpers ──────────────────────────────────────
function addStreamMemory({ stream_id, user_id = null, offset_seconds = 0, description, tags = null, thumbnail_url = null, transcript_json = null }) {
function addStreamMemory({ stream_id, user_id = null, offset_seconds = 0, description, tags = null, thumbnail_url = null, transcript_json = null }) {
- return run(`INSERT INTO stream_memories (stream_id, user_id, offset_seconds, description, tags, thumbnail_url, transcript_json)
+ // OR IGNORE against idx_stream_memories_moment_unique: re-analysing a stream must not
+ // store a second description of a moment already captured.
+ return run(`INSERT OR IGNORE INTO stream_memories (stream_id, user_id, offset_seconds, description, tags, thumbnail_url, transcript_json)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
VALUES (?, ?, ?, ?, ?, ?, ?)`,
[stream_id, user_id, Math.max(0, Math.round(offset_seconds || 0)), description || '',
[stream_id, user_id, Math.max(0, Math.round(offset_seconds || 0)), description || '',
tags ? (typeof tags === 'string' ? tags : JSON.stringify(tags)) : null, thumbnail_url,
tags ? (typeof tags === 'string' ? tags : JSON.stringify(tags)) : null, thumbnail_url,
@@ -2801,6 +2927,22 @@ function getTimelineByVod(vodId) {
@@ -2801,6 +2927,22 @@ function getTimelineByVod(vodId) {
} catch { return []; }
} catch { return []; }
}
}
+/**
+ * The vod_id already stamped on this stream's timeline, if any.
+ *
+ * Transcription of spooled audio keeps running for a while after the vod.ready webhook
+ * fires, and linkTimelineToVod() is a one-shot UPDATE — so those late rows used to stay
+ * vod_id NULL forever and never appear in the VOD's transcript. (Stream 2128: 11 speech
+ * rows orphaned against 2 linked; vod 2163 served 426 characters when the full
+ * transcript was 3548.) Late writers call this to stamp themselves correctly.
+ */
+function getTimelineVodId(streamId) {
+ try {
+ const r = get('SELECT vod_id FROM stream_timeline_events WHERE stream_id = ? AND vod_id IS NOT NULL LIMIT 1', [streamId]);
+ return r ? r.vod_id : null;
+ } catch { return null; }
+}
+
/** Attach a vod_id to a finished stream's rows so VOD views can reuse the timeline. */
/** Attach a vod_id to a finished stream's rows so VOD views can reuse the timeline. */
function linkTimelineToVod(streamId, vodId) {
function linkTimelineToVod(streamId, vodId) {
try { return run('UPDATE stream_timeline_events SET vod_id = ? WHERE stream_id = ? AND vod_id IS NULL', [vodId, streamId]); }
try { return run('UPDATE stream_timeline_events SET vod_id = ? WHERE stream_id = ? AND vod_id IS NULL', [vodId, streamId]); }
@@ -7279,7 +7421,7 @@ module.exports = {
@@ -7279,7 +7421,7 @@ module.exports = {
setVodTranscriptStatus, setClipTranscriptStatus, bumpVodTranscriptAttempt, bumpClipTranscriptAttempt,
setVodTranscriptStatus, setClipTranscriptStatus, bumpVodTranscriptAttempt, bumpClipTranscriptAttempt,
updatePasteAi, cleanupMalformedAiText, recordAiUsage, getAiCostToday, getAiCostTodayForUser, getAiUsageSummary,
updatePasteAi, cleanupMalformedAiText, recordAiUsage, getAiCostToday, getAiCostTodayForUser, getAiUsageSummary,
getStreamMemoriesByUser, countStreamMemoriesByUser, getAiMomentCandidates, getStreamTranscriptSegments, getUserPastesForAi,
getStreamMemoriesByUser, countStreamMemoriesByUser, getAiMomentCandidates, getStreamTranscriptSegments, getUserPastesForAi,
- addTimelineEvents, getTimeline, getTimelineText, getTimelineCoverage, linkTimelineToVod, getTimelineByVod,
+ addTimelineEvents, getTimeline, getTimelineText, getTimelineCoverage, linkTimelineToVod, getTimelineByVod, getTimelineVodId,
getVodsForMomentRanking, getClipStartTimesForStream, getChatSpikeOffsets,
getVodsForMomentRanking, getClipStartTimesForStream, getChatSpikeOffsets,
getLiveChatBuckets, getRecentChatText, getVodsWithoutAutoClip,
getLiveChatBuckets, getRecentChatText, getVodsWithoutAutoClip,
upsertStreamerOverview, getStreamerOverview, getAllStreamerOverviews, getStreamersNeedingOverview,
upsertStreamerOverview, getStreamerOverview, getAllStreamerOverviews, getStreamersNeedingOverview,
new file mode 100644
new file mode 100644
@@ -0,0 +1,155 @@
@@ -0,0 +1,155 @@
+/**
+ * Regression: openvibe.media/live/:sel/transcript.json served broken data.
+ *
+ * Four independent faults, all reproduced here against a throwaway DB:
+ *
+ * 1. vod_ai_state / clip_ai_state drifted in production to a keyless schema
+ * (`vod_id INT`, no PRIMARY KEY — CREATE TABLE IF NOT EXISTS never repairs an
+ * existing table). Their writers use `INSERT OR IGNORE ... (vod_id) VALUES (?)`,
+ * which needs a uniqueness constraint to be a no-op, so every call appended a row:
+ * 2818 rows for 487 VODs, 2986 for 103 clips.
+ *
+ * 2. That silently jammed the AI backfill. The queues are
+ * `WHERE ai_overview_short IS NULL ORDER BY vod_id DESC LIMIT 4`, so a VOD with four
+ * empty duplicates filled the whole batch with itself and nothing else was ever
+ * processed — every recent VOD had ai_overview null.
+ *
+ * 3. stream_memories had no constraint either, so re-analysing a stream stored the same
+ * moment again; the memories list read as near-duplicate pairs a minute apart.
+ *
+ * 4. The transcript endpoint preferred the timeline over the batch blob whenever the
+ * timeline was non-empty, so a partially-linked timeline truncated the result
+ * (vod 2163: 426 characters served against a 3548-character transcript).
+ *
+ * The merge must be lossless: duplicates were written at different times, so a transcript
+ * can sit on one row and an overview on another.
+ */
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+const tmp = path.join(os.tmpdir(), `ov-ai-state-${Date.now()}.db`);
+process.env.DB_PATH = tmp;
+
+const Database = require('better-sqlite3');
+
+// ── Build a DB with the DRIFTED (keyless) production schema, then dirty it ──────────
+{
+ const raw = new Database(tmp);
+ raw.exec(`CREATE TABLE vod_ai_state (
+ vod_id INT, ai_overview_short TEXT, ai_transcript_json TEXT, transcript_status TEXT,
+ transcript_attempts INT, transcript_error TEXT, transcript_next_at NUM)`);
+ raw.exec(`CREATE TABLE clip_ai_state (
+ clip_id INT, ai_overview_short TEXT, ai_transcript_json TEXT, transcript_status TEXT,
+ transcript_attempts INT, transcript_error TEXT, transcript_next_at NUM,
+ clip_notified INT, clip_notify_at NUM)`);
+
+ const v = raw.prepare(`INSERT INTO vod_ai_state
+ (vod_id, ai_overview_short, ai_transcript_json, transcript_status, transcript_attempts)
+ VALUES (?,?,?,?,?)`);
+ // vod 100: transcript and overview live on DIFFERENT duplicate rows, and a later
+ // duplicate carries a regressed 'pending' status. A lossless merge keeps all three.
+ v.run(100, null, JSON.stringify([{ start: 0, end: 2, text: 'the full transcript' }]), 'done', 1);
+ v.run(100, 'the overview', null, 'done', 2);
+ v.run(100, null, null, 'pending', 0);
+ // vod 101: four empty duplicates — the shape that jammed the overview queue.
+ for (let i = 0; i < 4; i++) v.run(101, null, null, null, 0);
+
+ const c = raw.prepare(`INSERT INTO clip_ai_state
+ (clip_id, ai_overview_short, clip_notified, clip_notify_at) VALUES (?,?,?,?)`);
+ c.run(200, 'clip overview', 1, 1234); // extra columns must survive the rebuild
+ c.run(200, null, null, null);
+ raw.close();
+}
+
+const db = require('../server/db/database');
+db.initDb();
+
+// stream_memories and stream_timeline_events both carry a FK to streams(id), so the
+// fixture needs real parent rows (user_id is the only NOT NULL column without a default).
+{
+ const w = new Database(tmp);
+ w.pragma('foreign_keys = OFF'); // fixture only: streams.user_id -> users(id) is not under test
+ const ins = w.prepare('INSERT OR IGNORE INTO streams (id, user_id) VALUES (?, 1)');
+ for (const id of [900, 901, 902]) ins.run(id);
+ w.close();
+}
+
+// ── 1: one row per key, nothing lost ────────────────────────────────────────────────
+const raw = new Database(tmp, { readonly: true });
+const count = (sql) => raw.prepare(sql).get().n;
+assert.strictEqual(count('SELECT COUNT(*) n FROM vod_ai_state'), 2, 'vod_ai_state must collapse to one row per vod');
+assert.strictEqual(count('SELECT COUNT(*) n FROM clip_ai_state'), 1, 'clip_ai_state must collapse to one row per clip');
+
+const v100 = db.getVodAiState(100);
+assert.ok(v100.ai_transcript_json && v100.ai_transcript_json.includes('the full transcript'),
+ 'transcript must survive even though it lived on a different duplicate than the overview');
+assert.strictEqual(v100.ai_overview_short, 'the overview',
+ 'overview must survive even though it lived on a different duplicate than the transcript');
+assert.strictEqual(v100.transcript_status, 'done',
+ "a stray 'pending' duplicate must not resurrect finished work");
+console.log('OK 1: duplicates merged losslessly (transcript + overview + settled status all kept)');
+
+const clip = raw.prepare('SELECT * FROM clip_ai_state WHERE clip_id = 200').get();
+assert.strictEqual(clip.clip_notified, 1, 'clip_notified must survive the rebuild');
+assert.strictEqual(clip.clip_notify_at, 1234, 'clip_notify_at must survive the rebuild');
+assert.strictEqual(clip.ai_overview_short, 'clip overview');
+console.log('OK 2: table-specific extra columns preserved (column-driven merge, not a fixed list)');
+
+// ── 2: the backfill queue is no longer jammed by one VOD ────────────────────────────
+const queue = db.getVodsNeedingOverview(4).map(r => r.id);
+assert.strictEqual(queue.length, new Set(queue).size, `overview queue still returns duplicates: ${queue}`);
+assert.ok(queue.includes(101), 'the VOD lacking an overview must still be queued');
+assert.ok(!queue.includes(100), 'a VOD that already has an overview must not be queued');
+console.log(`OK 3: overview backfill queue returns distinct VODs (${JSON.stringify(queue)})`);
+
+// ── 3: re-seeding is genuinely idempotent now ───────────────────────────────────────
+for (let i = 0; i < 5; i++) db.setVodTranscriptStatus(100, 'pending');
+assert.strictEqual(count('SELECT COUNT(*) n FROM vod_ai_state WHERE vod_id = 100'), 1,
+ 'INSERT OR IGNORE must no-op now that a UNIQUE index exists');
+assert.ok((db.getVodAiState(100).ai_transcript_json || '').includes('the full transcript'),
+ 're-seeding must not clobber an existing transcript');
+console.log('OK 4: five re-seeds leave exactly one row, transcript intact');
+
+// ── 4: one memory per (stream, offset) ──────────────────────────────────────────────
+db.addStreamMemory({ stream_id: 900, offset_seconds: 15, description: 'a red figure by a brick building' });
+db.addStreamMemory({ stream_id: 900, offset_seconds: 15, description: 'a red horse statue by a brick building' });
+db.addStreamMemory({ stream_id: 900, offset_seconds: 23, description: 'a different moment' });
+assert.strictEqual(count('SELECT COUNT(*) n FROM stream_memories WHERE stream_id = 900'), 2,
+ 're-describing the same moment must not create a second memory');
+console.log('OK 5: duplicate stream memories rejected, distinct moments kept');
+
+// ── 5: transcript source selection prefers the more complete view ───────────────────
+// Mirrors the endpoint's rule: whichever view carries more speech wins.
+const spoken = (arr) => arr.reduce((n, x) => n + String(x.text || '').trim().length, 0);
+const choose = (timeline, blob) => (spoken(blob) > spoken(timeline) ? blob : timeline);
+const partialTimeline = [{ start: 0, text: 'x'.repeat(426) }];
+const fullBlob = [{ start: 0, text: 'y'.repeat(3548) }];
+assert.strictEqual(spoken(choose(partialTimeline, fullBlob)), 3548,
+ 'a partially-linked timeline must not beat a fuller transcript blob');
+assert.strictEqual(spoken(choose(fullBlob, partialTimeline)), 3548,
+ 'and the richer source wins regardless of which side it is on');
+assert.strictEqual(spoken(choose([], [])), 0, 'no speech anywhere is a real answer, not an error');
+console.log('OK 6: transcript endpoint picks whichever source carries more speech');
+
+// ── 6: live rows are reachable — the headline complaint ─────────────────────────────
+db.addTimelineEvents([
+ { stream_id: 901, kind: 'speech', start_sec: 1, end_sec: 2, text: 'live speech', vod_id: 500 },
+ { stream_id: 901, kind: 'sound', start_sec: 3, end_sec: 4, label: 'laughter', vod_id: 500 },
+]);
+const byVod = db.getTimelineByVod(500);
+assert.strictEqual(byVod.length, 2, 'rows stamped with vod_id while live must be queryable by vod_id');
+assert.strictEqual(db.getTimelineVodId(901), 500, 'a late writer must be able to recover the stream vod id');
+// A row written before the vod id was known stays orphaned until relinked — that is the
+// case linkTimelineToVod covers, and getTimelineVodId lets later writers self-stamp.
+db.addTimelineEvents([{ stream_id: 902, kind: 'speech', start_sec: 1, text: 'orphan' }]);
+assert.strictEqual(db.getTimelineByVod(501).length, 0, 'unlinked rows are not visible by vod_id');
+db.linkTimelineToVod(902, 501);
+assert.strictEqual(db.getTimelineByVod(501).length, 1, 'linkTimelineToVod must adopt orphaned rows');
+console.log('OK 7: live timeline rows reachable by vod_id; orphans still adoptable');
+
+raw.close();
+try { fs.unlinkSync(tmp); } catch { /* */ }
+for (const ext of ['-wal', '-shm']) { try { fs.unlinkSync(tmp + ext); } catch { /* */ } }
+console.log('✅ AI state integrity regression test passed');