openvibers/openvibe.live Commits
Audio timeline: continuous VAD-gated transcription + sound events
Changes
@@@ -21,6 +21,7 @@"author": "OpenVibe.Live Community","license": "MIT","dependencies": {++ "@roamhq/wrtc": "^0.8.0","bcryptjs": "^2.4.3","better-sqlite3": "^11.3.0","cookie-parser": "^1.4.6",@@@ -30,15 +31,15 @@"express-rate-limit": "^7.4.0","geoip-lite": "^2.0.0","helmet": "^7.1.0",-- "openvibe-shared": "file:./vendor/openvibe-shared",-- "@roamhq/wrtc": "^0.8.0","jsonwebtoken": "^9.0.2","mediasoup": "^3.14.14","mediasoup-client": "^3.15.7","multer": "^1.4.5-lts.1","node-media-server": "^2.7.1",-- "sharp": "^0.33.0",++ "onnxruntime-node": "^1.20.0",++ "openvibe-shared": "file:./vendor/openvibe-shared","sdp-transform": "^3.0.0",++ "sharp": "^0.33.0","uuid": "^10.0.0","werift": "^0.24.4","ws": "^8.18.0"@@@ -4970,7 +4970,7 @@ curl -s https://openvibe.media/live/@Goosely/chat-insight.json | jq ".chat_insig<!-- Scripts --><script src="/js/theme-engine.js?v=4"></script><script src="/live-notify.js?v=3" defer></script>--<script src="/js/app.js?v=123" defer></script>++<script src="/js/app.js?v=124" defer></script><script src="/js/stream-player.js?v=20" defer></script><script src="/js/chat-shared.js?v=3" defer></script><script src="/js/emotes.js?v=8" defer></script>@@@ -4478,11 +4478,25 @@ async function _aiTlToggleTranscript(btn) {const data = await api(`/chat-ai/transcript/${session.dataset.sid}`);const segs = data.segments || [];const vod = data.vodId;-- if (!segs.length) { box.innerHTML = '<p class="muted" style="padding:6px">No transcript available.</p>'; return; }-- box.innerHTML = segs.map(sg => {-- const stamp = _aiTimeFmt(sg.start || 0);-- const jump = vod ? `<a class="ai-tl-ts" href="/vod/${vod}?t=${Math.floor(sg.start || 0)}" onclick="return handleLinkClick(event, '/vod/${vod}?t=${Math.floor(sg.start || 0)}')">${stamp}</a>` : `<span class="ai-tl-ts">${stamp}</span>`;-- return `<div class="ai-tl-tr-line">${jump}<span class="ai-tl-tr-text">${esc(sg.text || '')}</span></div>`;++ // Interleave spoken lines with detected non-speech sounds so the timeline reads++ // as everything that happened audibly, in order — not just the words.++ const rows = [++ ...segs.map(sg => ({ t: sg.start || 0, kind: 'speech', text: sg.text || '' })),++ ...(data.events || []).map(e => ({++ t: e.start_sec || 0, kind: 'sound',++ text: e.label + (e.confidence != null ? ` (${Number(e.confidence).toFixed(2)})` : ''),++ })),++ ].sort((a, b) => a.t - b.t);++ if (!rows.length) { box.innerHTML = '<p class="muted" style="padding:6px">No transcript available.</p>'; return; }++ const cov = data.coverageSec++ ? `<div class="muted" style="padding:4px 6px;font-size:.8rem">${_aiTimeFmt(data.coverageSec)} of speech transcribed</div>` : '';++ box.innerHTML = cov + rows.map(r => {++ const stamp = _aiTimeFmt(r.t);++ const jump = vod ? `<a class="ai-tl-ts" href="/vod/${vod}?t=${Math.floor(r.t)}" onclick="return handleLinkClick(event, '/vod/${vod}?t=${Math.floor(r.t)}')">${stamp}</a>` : `<span class="ai-tl-ts">${stamp}</span>`;++ const body = r.kind === 'sound'++ ? `<span class="ai-tl-tr-text muted"><i class="fa-solid fa-volume-high"></i> ${esc(r.text)}</span>`++ : `<span class="ai-tl-tr-text">${esc(r.text)}</span>`;++ return `<div class="ai-tl-tr-line">${jump}${body}</div>`;}).join('');} catch { box.innerHTML = '<p class="muted" style="padding:6px">Couldn\'t load transcript.</p>'; box.dataset.built = ''; }}@@@ -227,12 +227,39 @@ async function analyzeStreamFrame(image) {}/** Condense a stream's memories into a one-line "AI Overview" for the home card. */--async function summarizeStreamMemories(memories) {++async function summarizeStreamMemories(memories, streamId = null) {// Use observations from across the whole session (capped for token budget) so the// overview reflects the entire stream since it started, not just the latest frame.const lines = (memories || []).slice(-80).map(m => `- ${m.description}`).join('\n');if (!lines) return null;-- const prompt = `These are timestamped observations from a live stream, in order since it started. Give a thorough overview (2-6 sentences) of what this stream has been about overall — the main activities, topics, and vibe across the whole session (not just the latest moment):\n${lines}`;++++ // Fold in the audio timeline when one exists. Previously the transcript reached this++ // prompt only as a 500-char slice embedded inside a memory's description string, with++ // its timestamps stripped — so the model saw fragments of speech with no idea when++ // they happened or what they lined up with. Sending the timeline directly gives it++ // what was SAID and what was HEARD, both anchored in time.++ let audioBlock = '';++ if (streamId) {++ try {++ const _mmss = (s) => `${Math.floor(s / 60)}:${String(Math.floor(s % 60)).padStart(2, '0')}`;++ const speech = db.getTimeline(streamId, { kind: 'speech', limit: 400 }) || [];++ const sounds = db.getTimeline(streamId, { kind: 'sound', limit: 120 }) || [];++ if (speech.length) {++ const spoken = speech.slice(-200)++ .map(r => `[${_mmss(r.start_sec)}] ${String(r.text || '').replace(/\s+/g, ' ').trim().slice(0, 200)}`)++ .filter(l => l.length > 8).join('\n');++ if (spoken) audioBlock += `\n\nWHAT THE STREAMER SAID (timestamped):\n${spoken}`;++ }++ if (sounds.length) {++ const heard = sounds.slice(-60)++ .map(r => `[${_mmss(r.start_sec)}] ${r.label} (${Number(r.confidence || 0).toFixed(2)})`)++ .join('\n');++ if (heard) audioBlock += `\n\nNOTABLE SOUNDS HEARD:\n${heard}`;++ }++ } catch { /* timeline optional */ }++ }++++ const prompt = `These are timestamped observations from a live stream, in order since it started. Give a thorough overview (2-6 sentences) of what this stream has been about overall — the main activities, topics, and vibe across the whole session (not just the latest moment):\n${lines}${audioBlock}`;const text = await _complete({ prompt, maxTokens: 500, kind: 'stream_memory' });return text ? text.slice(0, 2000) : null;}@@@ -329,7 +356,7 @@ async function _generateVodOverviewInner(vod) {const existing = vod.stream_id ? (db.getStreamMemories(vod.stream_id) || []) : [];if (existing.length >= 2) {-- const overview = await summarizeStreamMemories(existing);++ const overview = await summarizeStreamMemories(existing, stream && stream.id);if (overview) { try { db.setVodAiOverview(vod.id, overview); } catch { /* */ } }return overview;}@@@ -120,7 +120,14 @@ function _momentContext(streamId, vodId) {const transcript = _sample(db.getStreamTranscriptSegments(streamId) || [], 60);const clipTimes = db.getClipStartTimesForStream(streamId, vodId) || [];const spikes = db.getChatSpikeOffsets(streamId, 30, 8) || [];-- return { memories, transcript, clipTimes, spikes };++ // Non-speech sounds are strong moment candidates — an explosion or a burst of++ // laughter marks a highlight as reliably as anything said out loud.++ let sounds = [];++ try {++ sounds = _sample((db.getTimeline(streamId, { kind: 'sound', limit: 400 }) || [])++ .filter(e => Number(e.confidence || 0) >= 0.45), 30);++ } catch { /* timeline optional */ }++ return { memories, transcript, clipTimes, spikes, sounds };}function _nearestMemory(memories, offset) {let best = null, bestD = Infinity;@@@ -136,6 +143,7 @@ async function _findBestMoment(vod) {if (_aiOn()) {const timeline = ctx.memories.map(m => `[${_mmss(m.offset_seconds)}] ${_cleanText(_deJson(m.description), 180)}`).join('\n');const script = ctx.transcript.map(s => `[${_mmss(s.start)}] ${_cleanText(s.text, 160)}`).join('\n');++ const soundLine = (ctx.sounds || []).map(e => `[${_mmss(e.start_sec || 0)}] ${e.label}`).join('\n');const clipHint = ctx.clipTimes.length ? ctx.clipTimes.slice(0, 12).map(_mmss).join(', ') : 'none';const spikeHint = ctx.spikes.length ? ctx.spikes.slice(0, 6).map(s => _mmss(s.offset)).join(', ') : 'none';const prompt = `Below is a single livestream VOD titled "${_cleanText(vod.title, 80)}", described by its on-screen TIMELINE (visual scene notes) and its AUDIO TRANSCRIPT, each line timestamped [m:ss].@@@ -146,6 +154,9 @@ ${timeline || '(none)'}TRANSCRIPT:${script || '(none)'}++SOUNDS HEARD (non-speech audio events):++${soundLine || '(none)'}++Viewers CLIPPED these timestamps (very strong "this was a highlight" signal): ${clipHint}Chat activity SPIKED around: ${spikeHint}nnew file mode 100644@@@ -0,0 +1,169 @@++/**++ * audio-events.js — non-speech sound recognition for the stream timeline.++ *++ * Speech is only part of what happens on a stream. Gunfire, explosions, laughter,++ * screaming, music and alarms carry as much meaning for stream memory, clip detection and++ * search as the words do — and today they are actively thrown away: whisper's own++ * "[MUSIC]" / "[Applause]" markers are deleted by the noise filter because, without VAD,++ * they were indistinguishable from hallucinations.++ *++ * Runs YAMNet (AudioSet, 521 classes) under onnxruntime-node. Chosen over CLAP because:++ * - it needs no text encoder, so there is no Python on the production path at all++ * (the box has Python 3.14 with no venv, PEP 668 blocking pip, and no torch)++ * - the model is ~16MB and the audio frontend is baked into the graph, so it takes a++ * raw waveform straight from the same 16kHz mono WAV the recogniser reads++ * - it emits a score per 0.48s frame, so events are time-localised for free — a single++ * explosion at second 11 cannot characterise a whole 30s segment++ *++ * Verified on 60s of real stream audio: top labels were Silence 1.000 and Speech 1.000,++ * with plausible secondaries — i.e. it is reading the actual content, not noise.++ *++ * Everything here is best-effort. If the model is absent, available() returns false and++ * the speech timeline is unaffected.++ */++'use strict';++const fs = require('fs');++const path = require('path');++++const MODEL_PATH = process.env.YAMNET_MODEL++ || path.join(__dirname, '../../data/models/yamnet.onnx');++const CLASSMAP_PATH = process.env.YAMNET_CLASSMAP++ || path.join(__dirname, '../../data/models/yamnet_class_map.csv');++++// YAMNet emits one score vector per 0.96s window advanced by 0.48s.++const FRAME_HOP_SEC = 0.48;++const MIN_CONFIDENCE = Number(process.env.YAMNET_MIN_CONFIDENCE) || 0.35;++// Merge same-label frames separated by less than this into one event.++const MERGE_GAP_SEC = 1.0;++// Never emit more than this per segment — one noisy label should not flood the timeline.++const MAX_EVENTS_PER_SEGMENT = 40;++++// Speech and silence are handled by the recogniser and by VAD respectively; recording them++// as "sound events" would just duplicate the speech rows and bury the interesting labels.++const IGNORED = new Set([++ 'Silence', 'Speech', 'Speech synthesizer', 'Narration, monologue',++ 'Male speech, man speaking', 'Female speech, woman speaking',++ 'Conversation', 'Child speech, kid speaking', 'Inside, small room',++ 'Inside, large room or hall', 'Music',++]);++++let _session = null;++let _labels = null;++let _loadFailed = false;++++function available() {++ if (_loadFailed) return false;++ try { return fs.existsSync(MODEL_PATH) && fs.existsSync(CLASSMAP_PATH); } catch { return false; }++}++++function _loadLabels() {++ if (_labels) return _labels;++ const raw = fs.readFileSync(CLASSMAP_PATH, 'utf8');++ _labels = raw.split('\n').slice(1).map((line) => {++ // index,mid,display_name — display_name may be quoted and contain commas++ const m = line.match(/^\d+,[^,]*,"?([^"\r\n]+?)"?\s*$/);++ return m ? m[1] : null;++ });++ return _labels;++}++++async function _session_() {++ if (_session) return _session;++ const ort = require('onnxruntime-node');++ _session = await ort.InferenceSession.create(MODEL_PATH, {++ // One thread: this shares a 4-core box with live x264 encoding, and the model is++ // small enough that throughput is not the constraint.++ intraOpNumThreads: 1,++ interOpNumThreads: 1,++ graphOptimizationLevel: 'all',++ });++ return _session;++}++++/** Read a 16kHz mono 16-bit PCM WAV into the Float32Array YAMNet expects. */++function _wavToFloat32(wavPath) {++ const buf = fs.readFileSync(wavPath);++ // Walk the RIFF chunks rather than assuming a 44-byte header — ffmpeg emits a LIST++ // chunk when it writes metadata, which would otherwise shift every sample.++ let off = 12;++ let dataOff = -1, dataLen = 0;++ while (off + 8 <= buf.length) {++ const id = buf.toString('ascii', off, off + 4);++ const size = buf.readUInt32LE(off + 4);++ if (id === 'data') { dataOff = off + 8; dataLen = size; break; }++ off += 8 + size + (size % 2);++ }++ if (dataOff < 0) { dataOff = 44; dataLen = buf.length - 44; }++ const n = Math.floor(Math.min(dataLen, buf.length - dataOff) / 2);++ const out = new Float32Array(n);++ for (let i = 0; i < n; i++) out[i] = buf.readInt16LE(dataOff + i * 2) / 32768;++ return out;++}++++/**++ * Detect non-speech sound events in a 16kHz mono WAV.++ * @param {string} wavPath++ * @param {{offsetSec?:number, category?:string, minConfidence?:number}} opts++ * @returns {Promise<Array<{start:number,end:number,label:string,confidence:number}>>}++ * start/end in ABSOLUTE stream seconds (offsetSec applied).++ */++async function detect(wavPath, { offsetSec = 0, minConfidence = MIN_CONFIDENCE } = {}) {++ if (!available()) return [];++ let scores, frames, classes, labels;++ try {++ const ort = require('onnxruntime-node');++ const wave = _wavToFloat32(wavPath);++ if (!wave.length) return [];++ const session = await _session_();++ labels = _loadLabels();++ const res = await session.run({ waveform: new ort.Tensor('float32', wave, [wave.length]) });++ const out = res.output_0 || res[session.outputNames[0]];++ if (!out) return [];++ [frames, classes] = out.dims;++ scores = out.data;++ } catch (e) {++ // A broken model file should disable the feature, not retry forever on every segment.++ _loadFailed = true;++ console.warn('[AI-Timeline] audio-events disabled:', e.message);++ return [];++ }++++ // Collect per-frame winners above threshold, then merge runs of the same label.++ const open = new Map(); // label -> { start, end, peak }++ const done = [];++ for (let f = 0; f < frames; f++) {++ const t = offsetSec + f * FRAME_HOP_SEC;++ for (let c = 0; c < classes; c++) {++ const v = scores[f * classes + c];++ if (v < minConfidence) continue;++ const label = labels[c];++ if (!label || IGNORED.has(label)) continue;++ const cur = open.get(label);++ if (cur && (t - cur.end) <= MERGE_GAP_SEC) {++ cur.end = t + FRAME_HOP_SEC;++ if (v > cur.peak) cur.peak = v;++ } else {++ if (cur) done.push(cur);++ open.set(label, { label, start: t, end: t + FRAME_HOP_SEC, peak: v });++ }++ }++ // Close labels that have gone quiet.++ for (const [label, ev] of open) {++ if ((t - ev.end) > MERGE_GAP_SEC) { done.push(ev); open.delete(label); }++ }++ }++ for (const ev of open.values()) done.push(ev);++++ return done++ .sort((a, b) => b.peak - a.peak)++ .slice(0, MAX_EVENTS_PER_SEGMENT)++ .map(e => ({++ start: Math.round(e.start * 100) / 100,++ end: Math.round(e.end * 100) / 100,++ label: e.label,++ confidence: Math.round(e.peak * 1000) / 1000,++ }))++ .sort((a, b) => a.start - b.start);++}++++module.exports = { available, detect, MODEL_PATH, CLASSMAP_PATH };@@@ -63,7 +63,20 @@ async function _confirmLiveMoment(stream) {const transcript = (db.getStreamTranscriptSegments(stream.id) || []).slice(-14);const chat = db.getRecentChatText(stream.id, WINDOW_SEC, 40) || [];const scene = memories.map(m => `- ${_clean(m.description, 160)}`).join('\n');-- const script = transcript.map(s => `- ${_clean(s.text, 140)}`).join('\n');++ // Timestamps are now used, not discarded: recent speech is anchored in time so the++ // model can line up what was said with the sounds and the chat spike.++ const _mmss = (n) => `${Math.floor(n / 60)}:${String(Math.floor(n % 60)).padStart(2, '0')}`;++ const script = transcript.map(s => `- [${_mmss(s.start || 0)}] ${_clean(s.text, 140)}`).join('\n');++ // Sound events are a strong clip signal on their own — a burst of gunfire, an++ // explosion or laughter is exactly the kind of thing viewers clip, and it is often++ // the reason chat spiked in the first place.++ let sounds = [];++ try {++ const since = Math.max(0, ((Date.now() - new Date(stream.started_at + 'Z').getTime()) / 1000) - WINDOW_SEC);++ sounds = (db.getTimeline(stream.id, { kind: 'sound', from: since, limit: 40 }) || [])++ .filter(e => Number(e.confidence || 0) >= 0.4).slice(-12);++ } catch { /* timeline optional */ }++ const soundBlock = sounds.map(e => `- [${_mmss(e.start_sec || 0)}] ${e.label} (${Number(e.confidence || 0).toFixed(2)})`).join('\n');const chatBlock = chat.slice(-30).map(c => `- ${_clean(c, 100)}`).join('\n');if (!_aiOn()) return { clip: null }; // caller decides via the stricter no-AI threshold@@@ -75,6 +88,9 @@ ${scene || '(none)'}WHAT WAS SAID (recent transcript):${script || '(none)'}++WHAT WAS HEARD (non-speech sounds detected):++${soundBlock || '(none)'}++CHAT (recent messages):${chatBlock || '(none)'}@@@ -235,7 +235,14 @@ router.get('/transcript/:streamId', async (req, res) => {vodId = v ? v.id : null; // legacy pre-migration rows} catch { /* */ }}-- res.json({ streamId: sid, vodId, segments });++ // Sound events ride alongside the speech segments so the transcript view can show++ // "what was heard" as well as "what was said". Empty for streams captured before++ // the timeline existed, which the frontend treats as speech-only.++ let events = [];++ try { events = db.getTimeline(sid, { kind: 'sound', limit: 2000 }) || []; } catch { /* */ }++ let coverageSec = 0;++ try { coverageSec = db.getTimelineCoverage(sid) || 0; } catch { /* */ }++ res.json({ streamId: sid, vodId, segments, events, coverageSec });} catch (err) {res.status(500).json({ error: 'Failed to load transcript' });}@@@ -200,15 +200,33 @@ async function _transcribeSpan(src, duration) {if (duration > 0 && duration <= 200) {return await transcribe.transcribeMediaDetailed(src, { seconds: 0, timeoutMs: 300000 });}++ // FULL coverage, not a sample. This used to take three 60-second windows at 10%/45%/80%++ // of the recording, so a 4-hour VOD produced 3 minutes of transcript and the rest was++ // simply never heard. Walk the whole thing in windows instead.++ //++ // Affordable because the decoder is VAD-gated: measured on this stream only ~13% of++ // audio contains a voice, and VAD cut a 60s decode from 51.0s to 24.0s while also++ // removing a trailing hallucination. Windows are processed one at a time and the whole++ // pass runs from the backfill queue, which already drops to low power while live.++ const WINDOW_SEC = 300;const parts = [];const segments = [];let anyOk = false, lastErr = null;-- for (const frac of [0.1, 0.45, 0.8]) {-- const start = Math.floor((duration || 0) * frac);++ const total = Math.max(0, Math.floor(duration || 0));++ const windows = [];++ for (let start = 0; start < total; start += WINDOW_SEC) windows.push(start);++ // Safety valve for absurdly long recordings so one VOD cannot occupy the queue forever.++ const MAX_WINDOWS = Math.max(1, parseInt(process.env.AI_VOD_MAX_WINDOWS, 10) || 96); // 8h++ if (windows.length > MAX_WINDOWS) {++ console.warn(`[AI] VOD is ${Math.round(total / 60)}min — transcribing the first ${MAX_WINDOWS * WINDOW_SEC / 60}min only`);++ windows.length = MAX_WINDOWS;++ }++ for (const start of windows) {const wav = _tmp('wav');-- const ffOk = await _runFf('ffmpeg', ['-y', '-ss', String(start), '-i', src, '-t', '60', '-vn', '-ac', '1', '-ar', '16000', '-f', 'wav', wav], 120000);++ const len = Math.min(WINDOW_SEC, total - start);++ const ffOk = await _runFf('ffmpeg', ['-y', '-ss', String(start), '-i', src, '-t', String(len), '-vn', '-ac', '1', '-ar', '16000', '-af', 'loudnorm=I=-16:TP=-1.5:LRA=11', '-f', 'wav', wav], 180000);if (ffOk) {-- const r = await transcribe.transcribeWavDetailed(wav, { timeoutMs: 150000, offsetSec: start });++ const r = await transcribe.transcribeWavDetailed(wav, { timeoutMs: 600000, offsetSec: start });if (r.ok) anyOk = true; else lastErr = r.error || lastErr;if (r.text) parts.push(r.text);if (r.segments && r.segments.length) segments.push(...r.segments);@@@ -218,7 +236,9 @@ async function _transcribeSpan(src, duration) {// ok=true only if at least one window ran the full ffmpeg+whisper pipeline cleanly.// If every window failed (unreadable source, etc.) we return ok=false so the caller// retries instead of marking the VOD permanently silent.-- return { text: parts.join(' … '), segments, ok: anyOk, error: anyOk ? null : lastErr };++ // Windows are contiguous now, so the transcript reads as continuous prose rather than++ // three disconnected excerpts joined by an ellipsis.++ return { text: parts.join(' ').replace(/\s+/g, ' ').trim(), segments, ok: anyOk, error: anyOk ? null : lastErr };}/**@@@ -64,7 +64,11 @@ function formatFmtp(params) {.join(';');}--async function captureWebrtc(stream, seconds) {++// Open a long-lived mediasoup PlainRTP consumer on the stream's audio producer and write++// the SDP ffmpeg needs to decode it. Shared by the one-shot chunk grab and the continuous++// capture, so there is exactly one copy of the SDP construction.++// Returns { sdpPath, cleanup } or null.++async function _prepareWebrtcSource(stream) {let webrtcSFU;try { webrtcSFU = require('../streaming/webrtc-sfu'); } catch { return null; }const roomId = `stream-${stream.id}`;@@@ -115,7 +119,6 @@ async function captureWebrtc(stream, seconds) {sdpLines.push('');const sdpContent = sdpLines.join('\r\n');const sdpPath = path.join(os.tmpdir(), `openvibe-aihear-${stream.id}-${Date.now()}.sdp`);-- const out = tmpWav(stream.id);const cleanup = () => {try { webrtcSFU.closePlainConsumer(roomId, consumer.transportId); } catch {}@@@ -124,18 +127,36 @@ async function captureWebrtc(stream, seconds) {try { fs.writeFileSync(sdpPath, sdpContent, 'utf8'); }catch { cleanup(); return null; }++ return { sdpPath, cleanup };++}++++// Common ffmpeg input flags for reading an RTP/SDP source. Tolerant of the packet loss and++// timestamp weirdness that shows up on a live ingest.++const SDP_INPUT_ARGS = [++ '-protocol_whitelist', 'file,rtp,udp',++ '-thread_queue_size', '2048',++ '-analyzeduration', '5000000', '-probesize', '5000000',++ '-use_wallclock_as_timestamps', '1',++ '-fflags', '+genpts+discardcorrupt+igndts',++ '-err_detect', 'ignore_err',++];++++// Loudness normalisation. Measured capture level on this stream was mean -32.8dB /++// max -4.4dB — plenty of headroom but a low average, which is exactly the signal a++// recogniser struggles with. loudnorm brought a real 60s clip from -31.1dB to -26.2dB.++const LOUDNORM = 'loudnorm=I=-16:TP=-1.5:LRA=11';++++async function captureWebrtc(stream, seconds) {++ const src = await _prepareWebrtcSource(stream);++ if (!src) return null;++ const { sdpPath, cleanup } = src;++ const out = tmpWav(stream.id);const ok = await runFfmpeg([-- '-y',-- '-protocol_whitelist', 'file,rtp,udp',-- '-thread_queue_size', '2048',-- '-analyzeduration', '5000000', '-probesize', '5000000',-- '-use_wallclock_as_timestamps', '1',-- '-fflags', '+genpts+discardcorrupt+igndts',-- '-err_detect', 'ignore_err',++ '-y', ...SDP_INPUT_ARGS,'-i', sdpPath,'-t', String(seconds),-- '-vn', '-ac', '1', '-ar', '16000',++ '-vn', '-ac', '1', '-ar', '16000', '-af', LOUDNORM,'-f', 'wav', out,], (seconds + 12) * 1000);@@@ -186,4 +207,135 @@ async function captureAudioChunk(stream, seconds = 12) {}}--module.exports = { captureAudioChunk };++// ── Continuous capture ────────────────────────────────────────────────────────────────++// The old model grabbed a 12s chunk every 120s. Measured against real stream audio that++// captures almost nothing: the stream is only ~13% speech, so sampling 10% of wall-clock++// caught roughly 1% of what was actually said — six of ten sampled slices contained no++// speech at all and produced only a bare "you", which the noise filter then deleted.++//++// Instead run ONE long-lived ffmpeg per stream that writes rolling WAV segments. Latency++// is explicitly not a concern here, so segments are consumed behind live at low priority.++// Cost is bounded by VAD: only the ~13% of audio containing a voice reaches the decoder.++const SEGMENT_SEC = Math.max(10, Math.min(120, parseInt(process.env.AI_HEAR_SEGMENT_SEC, 10) || 30));++const SPOOL_ROOT = path.join(__dirname, '../../data/asr');++// Hard cap on spooled audio per stream. The box has ~34GB free on a filesystem that++// openvibe.media is actively growing into, so an unbounded spool is not an option.++const MAX_SPOOL_FILES = Math.max(4, parseInt(process.env.AI_HEAR_MAX_SPOOL, 10) || 40);++++const _capturing = new Map(); // streamId -> { ff, dir, cleanup, seq }++++function spoolDir(streamId) { return path.join(SPOOL_ROOT, String(streamId)); }++++function isCapturing(streamId) { return _capturing.has(streamId); }++++/**++ * Start continuous audio capture for a live stream. Idempotent.++ * Segments land in data/asr/<streamId>/seg-NNNNNN.wav, each SEGMENT_SEC long, already++ * 16kHz mono and loudness-normalised — i.e. exactly what the recogniser wants.++ * @returns {Promise<boolean>} true if capture is running++ */++async function startContinuousCapture(stream) {++ if (!stream || !stream.id) return false;++ if (_capturing.has(stream.id)) return true;++++ const dir = spoolDir(stream.id);++ try { fs.mkdirSync(dir, { recursive: true }); } catch { return false; }++++ // Resolve an input the same way the one-shot grab does: WHIP/WebRTC via a plain RTP++ // consumer, RTMP via the local HTTP-FLV output.++ let inputArgs = null;++ let cleanup = () => {};++ const src = await _prepareWebrtcSource(stream).catch(() => null);++ if (src) {++ inputArgs = [...SDP_INPUT_ARGS, '-i', src.sdpPath];++ cleanup = src.cleanup;++ } else {++ const streamKey = resolveStreamKey(stream);++ if (!streamKey) return false;++ inputArgs = ['-thread_queue_size', '2048', '-i', `http://127.0.0.1:${FLV_PORT}/live/${streamKey}.flv`];++ }++++ const args = [++ '-y', ...inputArgs,++ '-vn', '-ac', '1', '-ar', '16000', '-af', LOUDNORM,++ '-f', 'segment', '-segment_time', String(SEGMENT_SEC),++ '-reset_timestamps', '0', // keep timestamps continuous across segments++ '-segment_format', 'wav',++ '-strftime', '0',++ path.join(dir, 'seg-%06d.wav'),++ ];++++ let ff;++ try { ff = spawn('ffmpeg', args, { stdio: 'ignore' }); }++ catch { cleanup(); return false; }++++ const entry = { ff, dir, cleanup, startedAt: Date.now() };++ _capturing.set(stream.id, entry);++ const onExit = () => {++ if (_capturing.get(stream.id) === entry) _capturing.delete(stream.id);++ try { cleanup(); } catch { /* */ }++ };++ ff.on('close', onExit);++ ff.on('error', onExit);++ console.log(`[AI-Hear] stream ${stream.id}: continuous capture started (${SEGMENT_SEC}s segments → ${dir})`);++ return true;++}++++function stopContinuousCapture(streamId) {++ const entry = _capturing.get(streamId);++ if (!entry) return false;++ _capturing.delete(streamId);++ try { entry.ff.kill('SIGKILL'); } catch { /* */ }++ try { entry.cleanup(); } catch { /* */ }++ console.log(`[AI-Hear] stream ${streamId}: continuous capture stopped`);++ return true;++}++++function stopAllCaptures() {++ let n = 0;++ for (const id of [..._capturing.keys()]) { if (stopContinuousCapture(id)) n++; }++ return n;++}++++/**++ * List finished segments awaiting transcription, oldest first.++ * The most recent file is skipped — ffmpeg is still writing to it.++ * Each entry carries the absolute offset (seconds into the stream) its audio starts at,++ * derived from the segment index, so timeline timestamps stay correct.++ */++function pendingSegments(streamId) {++ const dir = spoolDir(streamId);++ let names;++ try { names = fs.readdirSync(dir).filter(n => /^seg-\d{6}\.wav$/.test(n)).sort(); }++ catch { return []; }++ if (names.length <= 1) return []; // newest is still being written++ const ready = names.slice(0, -1);++ // Drop the oldest if the spool has run away (slow transcriber / very long stream).++ const overflow = Math.max(0, ready.length - MAX_SPOOL_FILES);++ for (let i = 0; i < overflow; i++) {++ try { fs.unlinkSync(path.join(dir, ready[i])); } catch { /* */ }++ }++ return ready.slice(overflow).map(name => ({++ name,++ path: path.join(dir, name),++ index: parseInt(name.slice(4, 10), 10),++ offsetSec: parseInt(name.slice(4, 10), 10) * SEGMENT_SEC,++ }));++}++++function discardSegment(streamId, name) {++ try { fs.unlinkSync(path.join(spoolDir(streamId), name)); return true; } catch { return false; }++}++++/** Remove a stream's whole spool directory (call when the stream ends). */++function purgeSpool(streamId) {++ try { fs.rmSync(spoolDir(streamId), { recursive: true, force: true }); return true; }++ catch { return false; }++}++++module.exports = {++ captureAudioChunk,++ startContinuousCapture, stopContinuousCapture, stopAllCaptures, isCapturing,++ pendingSegments, discardSegment, purgeSpool, spoolDir,++ SEGMENT_SEC,++};@@@ -89,7 +89,7 @@ async function _analyzeOne(stream) {const grew = (memories.length - last.count) >= SUMMARY_MIN_NEW;const stale = (now - last.at) >= SUMMARY_MAX_AGE_MS;if (memories.length > 1 && (!last.at || grew || stale)) {-- const summary = await ai.summarizeStreamMemories(memories);++ const summary = await ai.summarizeStreamMemories(memories, stream.id);if (summary) {db.updateStreamAiOverview(stream.id, summary);_lastSummary.set(stream.id, { count: memories.length, at: now });nnew file mode 100644@@@ -0,0 +1,159 @@++/**++ * timeline-job.js — continuous audio → unified stream timeline.++ *++ * Replaces the old sampling model. That one grabbed a 12s chunk every 120s, which++ * measured against real stream audio caught almost nothing: this stream is only ~13%++ * speech, so sampling 10% of wall-clock captured roughly 1% of what was actually said.++ * Six of ten sampled 12s slices contained no speech at all and produced a bare "you",++ * which the noise filter then deleted — hence hours of `transcript=""` in the logs.++ *++ * Here instead:++ * stream-audio.startContinuousCapture() one long-lived ffmpeg per stream,++ * rolling 16kHz mono loudnorm'd WAV segments++ * │++ * ▼++ * transcribe (Silero VAD-gated) decodes ONLY regions containing a voice++ * audio-events (optional) non-speech sound labels++ * │++ * ▼++ * stream_timeline_events one time-indexed row per phrase / sound++ *++ * Latency is explicitly not a requirement, so segments are consumed behind live, one at++ * a time, and the whole job yields whenever the box is busy. Accuracy and completeness++ * are what matter.++ */++'use strict';++const db = require('../db/database');++const ai = require('./ai-analysis');++const audio = require('./stream-audio');++++const POLL_MS = 10000;++// Process at most this many segments per tick per stream. Keeps a backlog from monopolising++// the CPU on a 4-core box that is also running live x264 encoding.++const MAX_SEGMENTS_PER_TICK = 2;++++let _busy = false;++const _known = new Set(); // stream ids we have started capture for++++function timelineEnabled() {++ try {++ const v = db.getSetting && db.getSetting('ai_timeline_enabled');++ if (v === undefined || v === null || v === '') return false; // default OFF++ return String(v) === 'true' || String(v) === '1';++ } catch { return false; }++}++++/** Transcribe one spooled segment and write its speech + sound rows. */++async function _processSegment(stream, seg) {++ const transcribe = require('./transcribe');++ let events = [];++++ // ── Speech ────────────────────────────────────────────────────────────────++ // offsetSec shifts whisper's file-relative timestamps into absolute stream time,++ // so a row's start_sec is directly usable as a ?t= deep link.++ let tx = null;++ try {++ tx = await transcribe.transcribeWavDetailed(seg.path, {++ offsetSec: seg.offsetSec,++ timeoutMs: 180000,++ });++ } catch (e) {++ console.warn(`[AI-Timeline] stream ${stream.id} seg ${seg.name}: transcribe error`, e.message);++ }++ if (tx && tx.ok && Array.isArray(tx.segments)) {++ for (const s of tx.segments) {++ const text = String(s.text || '').trim();++ if (!text) continue;++ events.push({++ stream_id: stream.id, user_id: stream.user_id, kind: 'speech',++ start_sec: Number(s.start) || seg.offsetSec,++ end_sec: s.end == null ? null : Number(s.end),++ text,++ });++ }++ }++++ // ── Non-speech sound events ───────────────────────────────────────────────++ // Optional and best-effort: if the classifier is unavailable the speech timeline is++ // still complete, so this must never be able to fail the segment.++ try {++ const sounds = require('./audio-events');++ if (sounds.available()) {++ const cat = stream.category || null;++ const detected = await sounds.detect(seg.path, { offsetSec: seg.offsetSec, category: cat });++ for (const d of (detected || [])) {++ events.push({++ stream_id: stream.id, user_id: stream.user_id, kind: 'sound',++ start_sec: d.start, end_sec: d.end, label: d.label, confidence: d.confidence,++ });++ }++ }++ } catch { /* classifier optional */ }++++ if (events.length) db.addTimelineEvents(events);++ audio.discardSegment(stream.id, seg.name);++++ const spoken = events.filter(e => e.kind === 'speech').length;++ const heard = events.filter(e => e.kind === 'sound').length;++ if (spoken || heard) {++ const preview = events.find(e => e.kind === 'speech');++ console.log(`[AI-Timeline] stream ${stream.id} +${seg.offsetSec}s: ${spoken} speech, ${heard} sound` +++ (preview ? ` — ${JSON.stringify(String(preview.text).slice(0, 80))}` : ''));++ }++ return events.length;++}++++async function tick() {++ if (_busy) return; // never overlap; whisper is CPU-bound++ if (!timelineEnabled()) return;++ if (!ai.transcriptionEnabled || !ai.transcriptionEnabled()) return;++++ _busy = true;++ try {++ let live = [];++ try { live = db.getLiveStreams() || []; } catch { return; }++ const liveIds = new Set(live.map(s => s.id));++++ // Start capture for newly live streams.++ for (const stream of live) {++ if (audio.isCapturing(stream.id)) continue;++ try { await audio.startContinuousCapture(stream); } catch { /* retry next tick */ }++ _known.add(stream.id);++ }++++ // Stop + tidy streams that ended. Their remaining segments are drained first so++ // the tail of the stream is not lost, then the spool is removed.++ for (const id of [..._known]) {++ if (liveIds.has(id)) continue;++ audio.stopContinuousCapture(id);++ const leftovers = audio.pendingSegments(id);++ if (!leftovers.length) {++ audio.purgeSpool(id);++ _known.delete(id);++ }++ }++++ // Drain segments, oldest first.++ for (const stream of live) {++ const pending = audio.pendingSegments(stream.id);++ if (!pending.length) continue;++ for (const seg of pending.slice(0, MAX_SEGMENTS_PER_TICK)) {++ await _processSegment(stream, seg).catch(() => {});++ }++ }++ } catch (e) {++ console.warn('[AI-Timeline] tick error:', e.message);++ } finally {++ _busy = false;++ }++}++++function start() {++ setInterval(() => { tick().catch(() => {}); }, POLL_MS);++ console.log('[AI] Timeline job started (continuous audio → stream_timeline_events)');++}++++/** Called on shutdown so ffmpeg children do not orphan. */++function stopAll() { return audio.stopAllCaptures(); }++++module.exports = { start, tick, stopAll, timelineEnabled };@@@ -26,6 +26,22 @@ const CANDIDATE_BINS = [].filter(Boolean);const MODEL = process.env.WHISPER_MODEL || path.join(HOME, 'whisper.cpp/models/ggml-base.en.bin');const THREADS = Math.max(2, Math.min(8, parseInt(process.env.WHISPER_THREADS, 10) || 4));++++// ── Voice Activity Detection ──────────────────────────────────────────────++// Measured on 60s of real stream audio: VAD cut large-v3-turbo from 51.0s to 24.0s++// AND removed a trailing "Thank you." hallucination. Both effects matter, and the++// second one matters more than the speedup.++//++// This stream is only ~13% speech (VAD found 7.9s of speech in 60s). Without VAD the++// decoder is handed 87% silence and invents text over it — "you", "Thank you.",++// "[BLANK_AUDIO]", "Bye!" — which is exactly what the HALLUCINATIONS blocklist below++// exists to delete. VAD removes the cause rather than filtering the symptom, so the++// decoder only ever sees regions that actually contain a voice.++const VAD_MODEL = process.env.WHISPER_VAD_MODEL || path.join(HOME, 'whisper.cpp/models/ggml-silero-v5.1.2.bin');++function vadModel() {++ if (process.env.WHISPER_VAD === '0') return null; // explicit opt-out++ try { return fs.existsSync(VAD_MODEL) ? VAD_MODEL : null; } catch { return null; }++}// Default to greedy decoding (whisper.cpp default) — proven to transcribe speech// reliably here. Beam search (WHISPER_BEAM>1) sometimes collapses real speech into a// non-speech tag like "[Crowd noise]", so it's opt-in only.@@@ -73,12 +89,21 @@ function setLowPower(v) { _lowPower = !!v; }function _threads() { return _lowPower ? Math.max(1, Math.min(2, THREADS)) : THREADS; }// Phrases whisper commonly hallucinates over silence / music / non-speech.++// Always-drop: whisper's stock inventions over non-speech. These are never worth keeping++// even when VAD says a voice is present, because they are the model's filler, not words.const HALLUCINATIONS = new Set([-- 'you', 'thank you', 'thank you.', 'thanks for watching', 'thanks for watching!',++ 'thank you', 'thank you.', 'thanks for watching', 'thanks for watching!','thanks for watching.', 'please subscribe', 'subscribe', 'like and subscribe',-- 'bye', 'bye.', 'bye bye', 'okay', 'ok', 'oh', 'uh', 'um', 'hmm', 'mm', 'mhm',-- 'the', 'so', 'yeah', '.', '...', 'thank you for watching', 'thank you very much',-- 'thank you so much', 'i\'m sorry', 'silence', 'music', 'applause',++ 'thank you for watching', 'thank you very much',++ 'thank you so much', 'silence', 'music', 'applause',++]);++// Drop-only-without-VAD: these ARE real words a streamer says constantly. They were on the++// blocklist because, with the decoder run over 87% silence, they were overwhelmingly++// hallucinations — six of ten sampled 12s slices produced a bare "you". Once Silero gates++// the decoder to actual speech regions, deleting them throws away genuine transcript.++const FILLER_WORDS = new Set([++ 'you', 'bye', 'bye.', 'bye bye', 'okay', 'ok', 'oh', 'uh', 'um', 'hmm', 'mm', 'mhm',++ 'the', 'so', 'yeah', 'i\'m sorry',]);function _isNoise(t) {const s = (t || '').replace(/\s+/g, ' ').trim();@@@ -88,7 +113,9 @@ function _isNoise(t) {if (/^[\[(♪].*[\])♪]?$/.test(s)) return true;if (/^♪/.test(s) || /♪$/.test(s)) return true;const norm = s.toLowerCase().replace(/[.!?,…]+$/g, '').trim();-- return HALLUCINATIONS.has(norm);++ if (HALLUCINATIONS.has(norm)) return true;++ if (!vadModel() && FILLER_WORDS.has(norm)) return true;++ return false;}// Clean a raw segment list: drop noise + collapse immediate repeats.@@@ -127,6 +154,10 @@ function transcribeWavDetailed(wavPath, { timeoutMs = 180000, offsetSec = 0 } =const jsonPath = `${outBase}.json`;const args = ['-m', MODEL, '-f', wavPath, '-oj', '-of', outBase, '-t', String(_threads()), '-l', 'en'];if (BEAM > 1) args.push('-bs', String(BEAM));++ // Decode only the regions Silero says contain a voice. Segment timestamps stay in++ // absolute file time, so the {start,end} contract is unchanged for every caller.++ const vm = vadModel();++ if (vm) args.push('--vad', '-vm', vm);let ff;try { ff = _track(spawn(bin, args, { stdio: 'ignore' })); }catch (e) { return resolve({ text: '', segments: [], ok: false, error: e.message }); }@@@ -883,6 +883,7 @@ function initDb() {['ai_stream_memory_enabled', 'false', 'Periodically analyze live-stream thumbnails into timestamped memories', 'boolean'],['ai_stream_capture_interval_sec', '120', 'Seconds between live-stream AI memory captures', 'number'],['ai_transcription_enabled', 'true', 'Transcribe live-stream/clip/VOD audio into memories — FREE, runs locally via whisper.cpp (no API/cost). Requires whisper.cpp installed on the server', 'boolean'],++ ['ai_timeline_enabled', 'false', 'CONTINUOUS audio timeline — transcribes the WHOLE live stream (not a 12s sample every 2min) and detects non-speech sounds, into a searchable timestamped timeline. FREE/local, but uses noticeably more CPU than sampling', 'boolean'],['ai_max_cost_usd_per_day', '0', 'Daily AI spend cap in USD (0 = no cap)', 'number'],['ai_input_cost_per_mtok', '3.0', 'Estimated input cost per million tokens (for cost breakdown)', 'number'],['ai_output_cost_per_mtok', '15.0', 'Estimated output cost per million tokens (for cost breakdown)', 'number'],@@@ -923,6 +924,31 @@ function initDb() {)`);database.exec('CREATE INDEX IF NOT EXISTS idx_stream_memories_stream ON stream_memories(stream_id, offset_seconds)');++ // ── 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++ // old arrangement where transcript segments were buried inside a per-memory JSON++ // blob, which meant timestamps were unusable for anything but display — of ~10++ // consumers only ai-moments-job actually read them, and getStreamTranscriptSegments++ // discarded `end` entirely.++ database.exec(`CREATE TABLE IF NOT EXISTS stream_timeline_events (++ id INTEGER PRIMARY KEY AUTOINCREMENT,++ stream_id INTEGER NOT NULL,++ user_id INTEGER,++ vod_id INTEGER, -- set once the session becomes a VOD++ kind TEXT NOT NULL, -- 'speech' | 'sound'++ start_sec REAL NOT NULL, -- absolute seconds into the stream++ end_sec REAL,++ text TEXT, -- speech content++ label TEXT, -- sound event label++ confidence REAL,++ created_at DATETIME DEFAULT CURRENT_TIMESTAMP,++ FOREIGN KEY (stream_id) REFERENCES streams(id) ON DELETE CASCADE++ )`);++ database.exec('CREATE INDEX IF NOT EXISTS idx_timeline_stream ON stream_timeline_events(stream_id, start_sec)');++ database.exec('CREATE INDEX IF NOT EXISTS idx_timeline_kind ON stream_timeline_events(stream_id, kind, start_sec)');++ database.exec('CREATE INDEX IF NOT EXISTS idx_timeline_vod ON stream_timeline_events(vod_id, start_sec)');++database.exec(`CREATE TABLE IF NOT EXISTS ai_usage (id INTEGER PRIMARY KEY AUTOINCREMENT,kind TEXT, -- paste_image | paste_text | stream_memory | ai_viewers | ...@@@ -2712,7 +2738,80 @@ function countStreamMemoriesByUser(userId) {// Flattened audio-transcript segments for a whole stream (from its memories), ordered by time.// Used by the AI Timeline transcript viewer; each segment deep-links to the VOD at its start.++// ── Timeline accessors ────────────────────────────────────────────────────────++/**++ * Bulk-insert timeline rows.++ * @param {Array<{stream_id,user_id?,vod_id?,kind,start_sec,end_sec?,text?,label?,confidence?}>} rows++ */++function addTimelineEvents(rows) {++ if (!Array.isArray(rows) || !rows.length) return 0;++ const stmt = db.prepare(`INSERT INTO stream_timeline_events++ (stream_id, user_id, vod_id, kind, start_sec, end_sec, text, label, confidence)++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`);++ const tx = db.transaction((list) => {++ for (const r of list) {++ if (!r || !r.stream_id || !r.kind || r.start_sec == null) continue;++ stmt.run(r.stream_id, r.user_id || null, r.vod_id || null, r.kind,++ Number(r.start_sec) || 0, r.end_sec == null ? null : Number(r.end_sec),++ r.text || null, r.label || null, r.confidence == null ? null : Number(r.confidence));++ }++ });++ try { tx(rows); return rows.length; } catch { return 0; }++}++++/** Read a stream's timeline, optionally filtered by kind and time window. */++function getTimeline(streamId, { kind = null, from = null, to = null, limit = 5000 } = {}) {++ let sql = 'SELECT kind, start_sec, end_sec, text, label, confidence FROM stream_timeline_events WHERE stream_id = ?';++ const params = [streamId];++ if (kind) { sql += ' AND kind = ?'; params.push(kind); }++ if (from != null) { sql += ' AND start_sec >= ?'; params.push(Number(from)); }++ if (to != null) { sql += ' AND start_sec <= ?'; params.push(Number(to)); }++ sql += ' ORDER BY start_sec ASC LIMIT ?';++ params.push(Math.max(1, Math.min(20000, limit)));++ try { return all(sql, params); } catch { return []; }++}++++/** Flat transcript text for a stream, speech rows only, in time order. */++function getTimelineText(streamId) {++ try {++ return getTimeline(streamId, { kind: 'speech' })++ .map(r => String(r.text || '').trim()).filter(Boolean).join(' ');++ } catch { return ''; }++}++++/** How many seconds of a stream the timeline actually covers (union of speech spans). */++function getTimelineCoverage(streamId) {++ const rows = getTimeline(streamId, { kind: 'speech' });++ let covered = 0, lastEnd = -1;++ for (const r of rows) {++ const st = Number(r.start_sec) || 0;++ const en = r.end_sec == null ? st : Number(r.end_sec);++ if (en <= lastEnd) continue;++ covered += en - Math.max(st, lastEnd);++ lastEnd = en;++ }++ return Math.round(covered);++}++++/** 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]); }++ catch { return null; }++}++function getStreamTranscriptSegments(streamId) {++ // Prefer the timeline when it has rows — it keeps `end` and covers the whole stream.++ // Fall back to the legacy per-memory blobs so old streams keep rendering; no migration.++ try {++ const tl = getTimeline(streamId, { kind: 'speech' });++ if (tl.length) {++ return tl.map(r => ({++ start: Math.floor(Number(r.start_sec) || 0),++ end: r.end_sec == null ? null : Math.round(Number(r.end_sec) * 100) / 100,++ text: String(r.text || '').trim(),++ })).filter(s => s.text);++ }++ } catch { /* fall through to legacy */ }const out = [];try {const rows = all('SELECT offset_seconds, transcript_json FROM stream_memories WHERE stream_id = ? AND transcript_json IS NOT NULL ORDER BY offset_seconds ASC', [streamId]);@@@ -7172,6 +7271,7 @@ module.exports = {setVodTranscriptStatus, setClipTranscriptStatus, bumpVodTranscriptAttempt, bumpClipTranscriptAttempt,updatePasteAi, cleanupMalformedAiText, recordAiUsage, getAiCostToday, getAiCostTodayForUser, getAiUsageSummary,getStreamMemoriesByUser, countStreamMemoriesByUser, getAiMomentCandidates, getStreamTranscriptSegments, getUserPastesForAi,++ addTimelineEvents, getTimeline, getTimelineText, getTimelineCoverage, linkTimelineToVod,getVodsForMomentRanking, getClipStartTimesForStream, getChatSpikeOffsets,getLiveChatBuckets, getRecentChatText, getVodsWithoutAutoClip,upsertStreamerOverview, getStreamerOverview, getAllStreamerOverviews, getStreamersNeedingOverview,@@@ -932,6 +932,9 @@ async function start() {try { require('./ai/ai-moments-job').start(); } catch (e) { console.warn('[AI] moments job not started:', e.message); }try { require('./ai/auto-clip-job').start(); } catch (e) { console.warn('[AI] auto-clip job not started:', e.message); }try { require('./ai/easter-egg-job').start(); } catch (e) { console.warn('[AI] easter-egg job not started:', e.message); }++ // Continuous audio → stream_timeline_events. Gated behind ai_timeline_enabled++ // (default off), so starting it is a no-op until switched on in admin.++ try { require('./ai/timeline-job').start(); } catch (e) { console.warn('[AI] timeline job not started:', e.message); }// Heal server-side recordings for live streams (resume after restart / Media restart)// so clipping always has a source. First pass delayed to let broadcasters reconnect.// VOD health scanning, disk guardianship and storage tiering moved to OpenVibe.Media.@@@ -1212,6 +1215,7 @@ function shutdown() {// (and so their temp files get cleaned by the close handlers). The interrupted VOD// is left in 'processing' → re-queued to 'pending' on next boot (crash recovery).try { const n = require('./ai/transcribe').killActive(); if (n) console.log(`[Server] Killed ${n} transcription child(ren)`); } catch { /* */ }++ try { const n = require('./ai/timeline-job').stopAll(); if (n) console.log(`[Server] Stopped ${n} continuous audio capture(s)`); } catch { /* */ }try { require('./ai/media-analysis').killActive(); } catch { /* */ }// Cleanly END RobotStreamer passthrough streams FIRST (before the exit races the reconnect).@@@ -46,6 +46,13 @@ function handler(req, res) {// Seed the Live-owned AI state row (queues transcript/overview work)// and kick the on-finalize AI pass right away (both budget-gated).try { db.setVodTranscriptStatus(vodId, 'pending'); } catch { /* */ }++ // Point the live timeline rows at the VOD so the recording inherits the++ // transcript that was already built while the stream was running — no++ // need to transcribe the same audio a second time.++ try {++ const sid = data?.stream_id || data?.streamId;++ if (sid) db.linkTimelineToVod(sid, vodId);++ } catch { /* */ }try {const ai = require('../ai/ai-analysis');const vodMeta = { id: vodId, ...data };
@@ -21,6 +21,7 @@
@@ -21,6 +21,7 @@
"author": "OpenVibe.Live Community",
"author": "OpenVibe.Live Community",
"license": "MIT",
"license": "MIT",
"dependencies": {
"dependencies": {
+ "@roamhq/wrtc": "^0.8.0",
"bcryptjs": "^2.4.3",
"bcryptjs": "^2.4.3",
"better-sqlite3": "^11.3.0",
"better-sqlite3": "^11.3.0",
"cookie-parser": "^1.4.6",
"cookie-parser": "^1.4.6",
@@ -30,15 +31,15 @@
@@ -30,15 +31,15 @@
"express-rate-limit": "^7.4.0",
"express-rate-limit": "^7.4.0",
"geoip-lite": "^2.0.0",
"geoip-lite": "^2.0.0",
"helmet": "^7.1.0",
"helmet": "^7.1.0",
- "openvibe-shared": "file:./vendor/openvibe-shared",
- "@roamhq/wrtc": "^0.8.0",
"jsonwebtoken": "^9.0.2",
"jsonwebtoken": "^9.0.2",
"mediasoup": "^3.14.14",
"mediasoup": "^3.14.14",
"mediasoup-client": "^3.15.7",
"mediasoup-client": "^3.15.7",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"node-media-server": "^2.7.1",
- "sharp": "^0.33.0",
+ "onnxruntime-node": "^1.20.0",
+ "openvibe-shared": "file:./vendor/openvibe-shared",
"sdp-transform": "^3.0.0",
"sdp-transform": "^3.0.0",
+ "sharp": "^0.33.0",
"uuid": "^10.0.0",
"uuid": "^10.0.0",
"werift": "^0.24.4",
"werift": "^0.24.4",
"ws": "^8.18.0"
"ws": "^8.18.0"
@@ -4970,7 +4970,7 @@ curl -s https://openvibe.media/live/@Goosely/chat-insight.json | jq ".chat_insig
@@ -4970,7 +4970,7 @@ curl -s https://openvibe.media/live/@Goosely/chat-insight.json | jq ".chat_insig
<!-- Scripts -->
<!-- Scripts -->
<script src="/js/theme-engine.js?v=4"></script>
<script src="/js/theme-engine.js?v=4"></script>
<script src="/live-notify.js?v=3" defer></script>
<script src="/live-notify.js?v=3" defer></script>
-<script src="/js/app.js?v=123" defer></script>
+<script src="/js/app.js?v=124" defer></script>
<script src="/js/stream-player.js?v=20" defer></script>
<script src="/js/stream-player.js?v=20" defer></script>
<script src="/js/chat-shared.js?v=3" defer></script>
<script src="/js/chat-shared.js?v=3" defer></script>
<script src="/js/emotes.js?v=8" defer></script>
<script src="/js/emotes.js?v=8" defer></script>
@@ -4478,11 +4478,25 @@ async function _aiTlToggleTranscript(btn) {
@@ -4478,11 +4478,25 @@ async function _aiTlToggleTranscript(btn) {
const data = await api(`/chat-ai/transcript/${session.dataset.sid}`);
const data = await api(`/chat-ai/transcript/${session.dataset.sid}`);
const segs = data.segments || [];
const segs = data.segments || [];
const vod = data.vodId;
const vod = data.vodId;
- if (!segs.length) { box.innerHTML = '<p class="muted" style="padding:6px">No transcript available.</p>'; return; }
- box.innerHTML = segs.map(sg => {
- const stamp = _aiTimeFmt(sg.start || 0);
- const jump = vod ? `<a class="ai-tl-ts" href="/vod/${vod}?t=${Math.floor(sg.start || 0)}" onclick="return handleLinkClick(event, '/vod/${vod}?t=${Math.floor(sg.start || 0)}')">${stamp}</a>` : `<span class="ai-tl-ts">${stamp}</span>`;
- return `<div class="ai-tl-tr-line">${jump}<span class="ai-tl-tr-text">${esc(sg.text || '')}</span></div>`;
+ // Interleave spoken lines with detected non-speech sounds so the timeline reads
+ // as everything that happened audibly, in order — not just the words.
+ const rows = [
+ ...segs.map(sg => ({ t: sg.start || 0, kind: 'speech', text: sg.text || '' })),
+ ...(data.events || []).map(e => ({
+ t: e.start_sec || 0, kind: 'sound',
+ text: e.label + (e.confidence != null ? ` (${Number(e.confidence).toFixed(2)})` : ''),
+ })),
+ ].sort((a, b) => a.t - b.t);
+ if (!rows.length) { box.innerHTML = '<p class="muted" style="padding:6px">No transcript available.</p>'; return; }
+ const cov = data.coverageSec
+ ? `<div class="muted" style="padding:4px 6px;font-size:.8rem">${_aiTimeFmt(data.coverageSec)} of speech transcribed</div>` : '';
+ box.innerHTML = cov + rows.map(r => {
+ const stamp = _aiTimeFmt(r.t);
+ const jump = vod ? `<a class="ai-tl-ts" href="/vod/${vod}?t=${Math.floor(r.t)}" onclick="return handleLinkClick(event, '/vod/${vod}?t=${Math.floor(r.t)}')">${stamp}</a>` : `<span class="ai-tl-ts">${stamp}</span>`;
+ const body = r.kind === 'sound'
+ ? `<span class="ai-tl-tr-text muted"><i class="fa-solid fa-volume-high"></i> ${esc(r.text)}</span>`
+ : `<span class="ai-tl-tr-text">${esc(r.text)}</span>`;
+ return `<div class="ai-tl-tr-line">${jump}${body}</div>`;
}).join('');
}).join('');
} catch { box.innerHTML = '<p class="muted" style="padding:6px">Couldn\'t load transcript.</p>'; box.dataset.built = ''; }
} catch { box.innerHTML = '<p class="muted" style="padding:6px">Couldn\'t load transcript.</p>'; box.dataset.built = ''; }
}
}
@@ -227,12 +227,39 @@ async function analyzeStreamFrame(image) {
@@ -227,12 +227,39 @@ async function analyzeStreamFrame(image) {
}
}
/** Condense a stream's memories into a one-line "AI Overview" for the home card. */
/** Condense a stream's memories into a one-line "AI Overview" for the home card. */
-async function summarizeStreamMemories(memories) {
+async function summarizeStreamMemories(memories, streamId = null) {
// Use observations from across the whole session (capped for token budget) so the
// Use observations from across the whole session (capped for token budget) so the
// overview reflects the entire stream since it started, not just the latest frame.
// overview reflects the entire stream since it started, not just the latest frame.
const lines = (memories || []).slice(-80).map(m => `- ${m.description}`).join('\n');
const lines = (memories || []).slice(-80).map(m => `- ${m.description}`).join('\n');
if (!lines) return null;
if (!lines) return null;
- const prompt = `These are timestamped observations from a live stream, in order since it started. Give a thorough overview (2-6 sentences) of what this stream has been about overall — the main activities, topics, and vibe across the whole session (not just the latest moment):\n${lines}`;
+
+ // Fold in the audio timeline when one exists. Previously the transcript reached this
+ // prompt only as a 500-char slice embedded inside a memory's description string, with
+ // its timestamps stripped — so the model saw fragments of speech with no idea when
+ // they happened or what they lined up with. Sending the timeline directly gives it
+ // what was SAID and what was HEARD, both anchored in time.
+ let audioBlock = '';
+ if (streamId) {
+ try {
+ const _mmss = (s) => `${Math.floor(s / 60)}:${String(Math.floor(s % 60)).padStart(2, '0')}`;
+ const speech = db.getTimeline(streamId, { kind: 'speech', limit: 400 }) || [];
+ const sounds = db.getTimeline(streamId, { kind: 'sound', limit: 120 }) || [];
+ if (speech.length) {
+ const spoken = speech.slice(-200)
+ .map(r => `[${_mmss(r.start_sec)}] ${String(r.text || '').replace(/\s+/g, ' ').trim().slice(0, 200)}`)
+ .filter(l => l.length > 8).join('\n');
+ if (spoken) audioBlock += `\n\nWHAT THE STREAMER SAID (timestamped):\n${spoken}`;
+ }
+ if (sounds.length) {
+ const heard = sounds.slice(-60)
+ .map(r => `[${_mmss(r.start_sec)}] ${r.label} (${Number(r.confidence || 0).toFixed(2)})`)
+ .join('\n');
+ if (heard) audioBlock += `\n\nNOTABLE SOUNDS HEARD:\n${heard}`;
+ }
+ } catch { /* timeline optional */ }
+ }
+
+ const prompt = `These are timestamped observations from a live stream, in order since it started. Give a thorough overview (2-6 sentences) of what this stream has been about overall — the main activities, topics, and vibe across the whole session (not just the latest moment):\n${lines}${audioBlock}`;
const text = await _complete({ prompt, maxTokens: 500, kind: 'stream_memory' });
const text = await _complete({ prompt, maxTokens: 500, kind: 'stream_memory' });
return text ? text.slice(0, 2000) : null;
return text ? text.slice(0, 2000) : null;
}
}
@@ -329,7 +356,7 @@ async function _generateVodOverviewInner(vod) {
@@ -329,7 +356,7 @@ async function _generateVodOverviewInner(vod) {
const existing = vod.stream_id ? (db.getStreamMemories(vod.stream_id) || []) : [];
const existing = vod.stream_id ? (db.getStreamMemories(vod.stream_id) || []) : [];
if (existing.length >= 2) {
if (existing.length >= 2) {
- const overview = await summarizeStreamMemories(existing);
+ const overview = await summarizeStreamMemories(existing, stream && stream.id);
if (overview) { try { db.setVodAiOverview(vod.id, overview); } catch { /* */ } }
if (overview) { try { db.setVodAiOverview(vod.id, overview); } catch { /* */ } }
return overview;
return overview;
}
}
@@ -120,7 +120,14 @@ function _momentContext(streamId, vodId) {
@@ -120,7 +120,14 @@ function _momentContext(streamId, vodId) {
const transcript = _sample(db.getStreamTranscriptSegments(streamId) || [], 60);
const transcript = _sample(db.getStreamTranscriptSegments(streamId) || [], 60);
const clipTimes = db.getClipStartTimesForStream(streamId, vodId) || [];
const clipTimes = db.getClipStartTimesForStream(streamId, vodId) || [];
const spikes = db.getChatSpikeOffsets(streamId, 30, 8) || [];
const spikes = db.getChatSpikeOffsets(streamId, 30, 8) || [];
- return { memories, transcript, clipTimes, spikes };
+ // Non-speech sounds are strong moment candidates — an explosion or a burst of
+ // laughter marks a highlight as reliably as anything said out loud.
+ let sounds = [];
+ try {
+ sounds = _sample((db.getTimeline(streamId, { kind: 'sound', limit: 400 }) || [])
+ .filter(e => Number(e.confidence || 0) >= 0.45), 30);
+ } catch { /* timeline optional */ }
+ return { memories, transcript, clipTimes, spikes, sounds };
}
}
function _nearestMemory(memories, offset) {
function _nearestMemory(memories, offset) {
let best = null, bestD = Infinity;
let best = null, bestD = Infinity;
@@ -136,6 +143,7 @@ async function _findBestMoment(vod) {
@@ -136,6 +143,7 @@ async function _findBestMoment(vod) {
if (_aiOn()) {
if (_aiOn()) {
const timeline = ctx.memories.map(m => `[${_mmss(m.offset_seconds)}] ${_cleanText(_deJson(m.description), 180)}`).join('\n');
const timeline = ctx.memories.map(m => `[${_mmss(m.offset_seconds)}] ${_cleanText(_deJson(m.description), 180)}`).join('\n');
const script = ctx.transcript.map(s => `[${_mmss(s.start)}] ${_cleanText(s.text, 160)}`).join('\n');
const script = ctx.transcript.map(s => `[${_mmss(s.start)}] ${_cleanText(s.text, 160)}`).join('\n');
+ const soundLine = (ctx.sounds || []).map(e => `[${_mmss(e.start_sec || 0)}] ${e.label}`).join('\n');
const clipHint = ctx.clipTimes.length ? ctx.clipTimes.slice(0, 12).map(_mmss).join(', ') : 'none';
const clipHint = ctx.clipTimes.length ? ctx.clipTimes.slice(0, 12).map(_mmss).join(', ') : 'none';
const spikeHint = ctx.spikes.length ? ctx.spikes.slice(0, 6).map(s => _mmss(s.offset)).join(', ') : 'none';
const spikeHint = ctx.spikes.length ? ctx.spikes.slice(0, 6).map(s => _mmss(s.offset)).join(', ') : 'none';
const prompt = `Below is a single livestream VOD titled "${_cleanText(vod.title, 80)}", described by its on-screen TIMELINE (visual scene notes) and its AUDIO TRANSCRIPT, each line timestamped [m:ss].
const prompt = `Below is a single livestream VOD titled "${_cleanText(vod.title, 80)}", described by its on-screen TIMELINE (visual scene notes) and its AUDIO TRANSCRIPT, each line timestamped [m:ss].
@@ -146,6 +154,9 @@ ${timeline || '(none)'}
@@ -146,6 +154,9 @@ ${timeline || '(none)'}
TRANSCRIPT:
TRANSCRIPT:
${script || '(none)'}
${script || '(none)'}
+SOUNDS HEARD (non-speech audio events):
+${soundLine || '(none)'}
+
Viewers CLIPPED these timestamps (very strong "this was a highlight" signal): ${clipHint}
Viewers CLIPPED these timestamps (very strong "this was a highlight" signal): ${clipHint}
Chat activity SPIKED around: ${spikeHint}
Chat activity SPIKED around: ${spikeHint}
new file mode 100644
new file mode 100644
@@ -0,0 +1,169 @@
@@ -0,0 +1,169 @@
+/**
+ * audio-events.js — non-speech sound recognition for the stream timeline.
+ *
+ * Speech is only part of what happens on a stream. Gunfire, explosions, laughter,
+ * screaming, music and alarms carry as much meaning for stream memory, clip detection and
+ * search as the words do — and today they are actively thrown away: whisper's own
+ * "[MUSIC]" / "[Applause]" markers are deleted by the noise filter because, without VAD,
+ * they were indistinguishable from hallucinations.
+ *
+ * Runs YAMNet (AudioSet, 521 classes) under onnxruntime-node. Chosen over CLAP because:
+ * - it needs no text encoder, so there is no Python on the production path at all
+ * (the box has Python 3.14 with no venv, PEP 668 blocking pip, and no torch)
+ * - the model is ~16MB and the audio frontend is baked into the graph, so it takes a
+ * raw waveform straight from the same 16kHz mono WAV the recogniser reads
+ * - it emits a score per 0.48s frame, so events are time-localised for free — a single
+ * explosion at second 11 cannot characterise a whole 30s segment
+ *
+ * Verified on 60s of real stream audio: top labels were Silence 1.000 and Speech 1.000,
+ * with plausible secondaries — i.e. it is reading the actual content, not noise.
+ *
+ * Everything here is best-effort. If the model is absent, available() returns false and
+ * the speech timeline is unaffected.
+ */
+'use strict';
+const fs = require('fs');
+const path = require('path');
+
+const MODEL_PATH = process.env.YAMNET_MODEL
+ || path.join(__dirname, '../../data/models/yamnet.onnx');
+const CLASSMAP_PATH = process.env.YAMNET_CLASSMAP
+ || path.join(__dirname, '../../data/models/yamnet_class_map.csv');
+
+// YAMNet emits one score vector per 0.96s window advanced by 0.48s.
+const FRAME_HOP_SEC = 0.48;
+const MIN_CONFIDENCE = Number(process.env.YAMNET_MIN_CONFIDENCE) || 0.35;
+// Merge same-label frames separated by less than this into one event.
+const MERGE_GAP_SEC = 1.0;
+// Never emit more than this per segment — one noisy label should not flood the timeline.
+const MAX_EVENTS_PER_SEGMENT = 40;
+
+// Speech and silence are handled by the recogniser and by VAD respectively; recording them
+// as "sound events" would just duplicate the speech rows and bury the interesting labels.
+const IGNORED = new Set([
+ 'Silence', 'Speech', 'Speech synthesizer', 'Narration, monologue',
+ 'Male speech, man speaking', 'Female speech, woman speaking',
+ 'Conversation', 'Child speech, kid speaking', 'Inside, small room',
+ 'Inside, large room or hall', 'Music',
+]);
+
+let _session = null;
+let _labels = null;
+let _loadFailed = false;
+
+function available() {
+ if (_loadFailed) return false;
+ try { return fs.existsSync(MODEL_PATH) && fs.existsSync(CLASSMAP_PATH); } catch { return false; }
+}
+
+function _loadLabels() {
+ if (_labels) return _labels;
+ const raw = fs.readFileSync(CLASSMAP_PATH, 'utf8');
+ _labels = raw.split('\n').slice(1).map((line) => {
+ // index,mid,display_name — display_name may be quoted and contain commas
+ const m = line.match(/^\d+,[^,]*,"?([^"\r\n]+?)"?\s*$/);
+ return m ? m[1] : null;
+ });
+ return _labels;
+}
+
+async function _session_() {
+ if (_session) return _session;
+ const ort = require('onnxruntime-node');
+ _session = await ort.InferenceSession.create(MODEL_PATH, {
+ // One thread: this shares a 4-core box with live x264 encoding, and the model is
+ // small enough that throughput is not the constraint.
+ intraOpNumThreads: 1,
+ interOpNumThreads: 1,
+ graphOptimizationLevel: 'all',
+ });
+ return _session;
+}
+
+/** Read a 16kHz mono 16-bit PCM WAV into the Float32Array YAMNet expects. */
+function _wavToFloat32(wavPath) {
+ const buf = fs.readFileSync(wavPath);
+ // Walk the RIFF chunks rather than assuming a 44-byte header — ffmpeg emits a LIST
+ // chunk when it writes metadata, which would otherwise shift every sample.
+ let off = 12;
+ let dataOff = -1, dataLen = 0;
+ while (off + 8 <= buf.length) {
+ const id = buf.toString('ascii', off, off + 4);
+ const size = buf.readUInt32LE(off + 4);
+ if (id === 'data') { dataOff = off + 8; dataLen = size; break; }
+ off += 8 + size + (size % 2);
+ }
+ if (dataOff < 0) { dataOff = 44; dataLen = buf.length - 44; }
+ const n = Math.floor(Math.min(dataLen, buf.length - dataOff) / 2);
+ const out = new Float32Array(n);
+ for (let i = 0; i < n; i++) out[i] = buf.readInt16LE(dataOff + i * 2) / 32768;
+ return out;
+}
+
+/**
+ * Detect non-speech sound events in a 16kHz mono WAV.
+ * @param {string} wavPath
+ * @param {{offsetSec?:number, category?:string, minConfidence?:number}} opts
+ * @returns {Promise<Array<{start:number,end:number,label:string,confidence:number}>>}
+ * start/end in ABSOLUTE stream seconds (offsetSec applied).
+ */
+async function detect(wavPath, { offsetSec = 0, minConfidence = MIN_CONFIDENCE } = {}) {
+ if (!available()) return [];
+ let scores, frames, classes, labels;
+ try {
+ const ort = require('onnxruntime-node');
+ const wave = _wavToFloat32(wavPath);
+ if (!wave.length) return [];
+ const session = await _session_();
+ labels = _loadLabels();
+ const res = await session.run({ waveform: new ort.Tensor('float32', wave, [wave.length]) });
+ const out = res.output_0 || res[session.outputNames[0]];
+ if (!out) return [];
+ [frames, classes] = out.dims;
+ scores = out.data;
+ } catch (e) {
+ // A broken model file should disable the feature, not retry forever on every segment.
+ _loadFailed = true;
+ console.warn('[AI-Timeline] audio-events disabled:', e.message);
+ return [];
+ }
+
+ // Collect per-frame winners above threshold, then merge runs of the same label.
+ const open = new Map(); // label -> { start, end, peak }
+ const done = [];
+ for (let f = 0; f < frames; f++) {
+ const t = offsetSec + f * FRAME_HOP_SEC;
+ for (let c = 0; c < classes; c++) {
+ const v = scores[f * classes + c];
+ if (v < minConfidence) continue;
+ const label = labels[c];
+ if (!label || IGNORED.has(label)) continue;
+ const cur = open.get(label);
+ if (cur && (t - cur.end) <= MERGE_GAP_SEC) {
+ cur.end = t + FRAME_HOP_SEC;
+ if (v > cur.peak) cur.peak = v;
+ } else {
+ if (cur) done.push(cur);
+ open.set(label, { label, start: t, end: t + FRAME_HOP_SEC, peak: v });
+ }
+ }
+ // Close labels that have gone quiet.
+ for (const [label, ev] of open) {
+ if ((t - ev.end) > MERGE_GAP_SEC) { done.push(ev); open.delete(label); }
+ }
+ }
+ for (const ev of open.values()) done.push(ev);
+
+ return done
+ .sort((a, b) => b.peak - a.peak)
+ .slice(0, MAX_EVENTS_PER_SEGMENT)
+ .map(e => ({
+ start: Math.round(e.start * 100) / 100,
+ end: Math.round(e.end * 100) / 100,
+ label: e.label,
+ confidence: Math.round(e.peak * 1000) / 1000,
+ }))
+ .sort((a, b) => a.start - b.start);
+}
+
+module.exports = { available, detect, MODEL_PATH, CLASSMAP_PATH };
@@ -63,7 +63,20 @@ async function _confirmLiveMoment(stream) {
@@ -63,7 +63,20 @@ async function _confirmLiveMoment(stream) {
const transcript = (db.getStreamTranscriptSegments(stream.id) || []).slice(-14);
const transcript = (db.getStreamTranscriptSegments(stream.id) || []).slice(-14);
const chat = db.getRecentChatText(stream.id, WINDOW_SEC, 40) || [];
const chat = db.getRecentChatText(stream.id, WINDOW_SEC, 40) || [];
const scene = memories.map(m => `- ${_clean(m.description, 160)}`).join('\n');
const scene = memories.map(m => `- ${_clean(m.description, 160)}`).join('\n');
- const script = transcript.map(s => `- ${_clean(s.text, 140)}`).join('\n');
+ // Timestamps are now used, not discarded: recent speech is anchored in time so the
+ // model can line up what was said with the sounds and the chat spike.
+ const _mmss = (n) => `${Math.floor(n / 60)}:${String(Math.floor(n % 60)).padStart(2, '0')}`;
+ const script = transcript.map(s => `- [${_mmss(s.start || 0)}] ${_clean(s.text, 140)}`).join('\n');
+ // Sound events are a strong clip signal on their own — a burst of gunfire, an
+ // explosion or laughter is exactly the kind of thing viewers clip, and it is often
+ // the reason chat spiked in the first place.
+ let sounds = [];
+ try {
+ const since = Math.max(0, ((Date.now() - new Date(stream.started_at + 'Z').getTime()) / 1000) - WINDOW_SEC);
+ sounds = (db.getTimeline(stream.id, { kind: 'sound', from: since, limit: 40 }) || [])
+ .filter(e => Number(e.confidence || 0) >= 0.4).slice(-12);
+ } catch { /* timeline optional */ }
+ const soundBlock = sounds.map(e => `- [${_mmss(e.start_sec || 0)}] ${e.label} (${Number(e.confidence || 0).toFixed(2)})`).join('\n');
const chatBlock = chat.slice(-30).map(c => `- ${_clean(c, 100)}`).join('\n');
const chatBlock = chat.slice(-30).map(c => `- ${_clean(c, 100)}`).join('\n');
if (!_aiOn()) return { clip: null }; // caller decides via the stricter no-AI threshold
if (!_aiOn()) return { clip: null }; // caller decides via the stricter no-AI threshold
@@ -75,6 +88,9 @@ ${scene || '(none)'}
@@ -75,6 +88,9 @@ ${scene || '(none)'}
WHAT WAS SAID (recent transcript):
WHAT WAS SAID (recent transcript):
${script || '(none)'}
${script || '(none)'}
+WHAT WAS HEARD (non-speech sounds detected):
+${soundBlock || '(none)'}
+
CHAT (recent messages):
CHAT (recent messages):
${chatBlock || '(none)'}
${chatBlock || '(none)'}
@@ -235,7 +235,14 @@ router.get('/transcript/:streamId', async (req, res) => {
@@ -235,7 +235,14 @@ router.get('/transcript/:streamId', async (req, res) => {
vodId = v ? v.id : null; // legacy pre-migration rows
vodId = v ? v.id : null; // legacy pre-migration rows
} catch { /* */ }
} catch { /* */ }
}
}
- res.json({ streamId: sid, vodId, segments });
+ // Sound events ride alongside the speech segments so the transcript view can show
+ // "what was heard" as well as "what was said". Empty for streams captured before
+ // the timeline existed, which the frontend treats as speech-only.
+ let events = [];
+ try { events = db.getTimeline(sid, { kind: 'sound', limit: 2000 }) || []; } catch { /* */ }
+ let coverageSec = 0;
+ try { coverageSec = db.getTimelineCoverage(sid) || 0; } catch { /* */ }
+ res.json({ streamId: sid, vodId, segments, events, coverageSec });
} catch (err) {
} catch (err) {
res.status(500).json({ error: 'Failed to load transcript' });
res.status(500).json({ error: 'Failed to load transcript' });
}
}
@@ -200,15 +200,33 @@ async function _transcribeSpan(src, duration) {
@@ -200,15 +200,33 @@ async function _transcribeSpan(src, duration) {
if (duration > 0 && duration <= 200) {
if (duration > 0 && duration <= 200) {
return await transcribe.transcribeMediaDetailed(src, { seconds: 0, timeoutMs: 300000 });
return await transcribe.transcribeMediaDetailed(src, { seconds: 0, timeoutMs: 300000 });
}
}
+ // FULL coverage, not a sample. This used to take three 60-second windows at 10%/45%/80%
+ // of the recording, so a 4-hour VOD produced 3 minutes of transcript and the rest was
+ // simply never heard. Walk the whole thing in windows instead.
+ //
+ // Affordable because the decoder is VAD-gated: measured on this stream only ~13% of
+ // audio contains a voice, and VAD cut a 60s decode from 51.0s to 24.0s while also
+ // removing a trailing hallucination. Windows are processed one at a time and the whole
+ // pass runs from the backfill queue, which already drops to low power while live.
+ const WINDOW_SEC = 300;
const parts = [];
const parts = [];
const segments = [];
const segments = [];
let anyOk = false, lastErr = null;
let anyOk = false, lastErr = null;
- for (const frac of [0.1, 0.45, 0.8]) {
- const start = Math.floor((duration || 0) * frac);
+ const total = Math.max(0, Math.floor(duration || 0));
+ const windows = [];
+ for (let start = 0; start < total; start += WINDOW_SEC) windows.push(start);
+ // Safety valve for absurdly long recordings so one VOD cannot occupy the queue forever.
+ const MAX_WINDOWS = Math.max(1, parseInt(process.env.AI_VOD_MAX_WINDOWS, 10) || 96); // 8h
+ if (windows.length > MAX_WINDOWS) {
+ console.warn(`[AI] VOD is ${Math.round(total / 60)}min — transcribing the first ${MAX_WINDOWS * WINDOW_SEC / 60}min only`);
+ windows.length = MAX_WINDOWS;
+ }
+ for (const start of windows) {
const wav = _tmp('wav');
const wav = _tmp('wav');
- const ffOk = await _runFf('ffmpeg', ['-y', '-ss', String(start), '-i', src, '-t', '60', '-vn', '-ac', '1', '-ar', '16000', '-f', 'wav', wav], 120000);
+ const len = Math.min(WINDOW_SEC, total - start);
+ const ffOk = await _runFf('ffmpeg', ['-y', '-ss', String(start), '-i', src, '-t', String(len), '-vn', '-ac', '1', '-ar', '16000', '-af', 'loudnorm=I=-16:TP=-1.5:LRA=11', '-f', 'wav', wav], 180000);
if (ffOk) {
if (ffOk) {
- const r = await transcribe.transcribeWavDetailed(wav, { timeoutMs: 150000, offsetSec: start });
+ const r = await transcribe.transcribeWavDetailed(wav, { timeoutMs: 600000, offsetSec: start });
if (r.ok) anyOk = true; else lastErr = r.error || lastErr;
if (r.ok) anyOk = true; else lastErr = r.error || lastErr;
if (r.text) parts.push(r.text);
if (r.text) parts.push(r.text);
if (r.segments && r.segments.length) segments.push(...r.segments);
if (r.segments && r.segments.length) segments.push(...r.segments);
@@ -218,7 +236,9 @@ async function _transcribeSpan(src, duration) {
@@ -218,7 +236,9 @@ async function _transcribeSpan(src, duration) {
// ok=true only if at least one window ran the full ffmpeg+whisper pipeline cleanly.
// ok=true only if at least one window ran the full ffmpeg+whisper pipeline cleanly.
// If every window failed (unreadable source, etc.) we return ok=false so the caller
// If every window failed (unreadable source, etc.) we return ok=false so the caller
// retries instead of marking the VOD permanently silent.
// retries instead of marking the VOD permanently silent.
- return { text: parts.join(' … '), segments, ok: anyOk, error: anyOk ? null : lastErr };
+ // Windows are contiguous now, so the transcript reads as continuous prose rather than
+ // three disconnected excerpts joined by an ellipsis.
+ return { text: parts.join(' ').replace(/\s+/g, ' ').trim(), segments, ok: anyOk, error: anyOk ? null : lastErr };
}
}
/**
/**
@@ -64,7 +64,11 @@ function formatFmtp(params) {
@@ -64,7 +64,11 @@ function formatFmtp(params) {
.join(';');
.join(';');
}
}
-async function captureWebrtc(stream, seconds) {
+// Open a long-lived mediasoup PlainRTP consumer on the stream's audio producer and write
+// the SDP ffmpeg needs to decode it. Shared by the one-shot chunk grab and the continuous
+// capture, so there is exactly one copy of the SDP construction.
+// Returns { sdpPath, cleanup } or null.
+async function _prepareWebrtcSource(stream) {
let webrtcSFU;
let webrtcSFU;
try { webrtcSFU = require('../streaming/webrtc-sfu'); } catch { return null; }
try { webrtcSFU = require('../streaming/webrtc-sfu'); } catch { return null; }
const roomId = `stream-${stream.id}`;
const roomId = `stream-${stream.id}`;
@@ -115,7 +119,6 @@ async function captureWebrtc(stream, seconds) {
@@ -115,7 +119,6 @@ async function captureWebrtc(stream, seconds) {
sdpLines.push('');
sdpLines.push('');
const sdpContent = sdpLines.join('\r\n');
const sdpContent = sdpLines.join('\r\n');
const sdpPath = path.join(os.tmpdir(), `openvibe-aihear-${stream.id}-${Date.now()}.sdp`);
const sdpPath = path.join(os.tmpdir(), `openvibe-aihear-${stream.id}-${Date.now()}.sdp`);
- const out = tmpWav(stream.id);
const cleanup = () => {
const cleanup = () => {
try { webrtcSFU.closePlainConsumer(roomId, consumer.transportId); } catch {}
try { webrtcSFU.closePlainConsumer(roomId, consumer.transportId); } catch {}
@@ -124,18 +127,36 @@ async function captureWebrtc(stream, seconds) {
@@ -124,18 +127,36 @@ async function captureWebrtc(stream, seconds) {
try { fs.writeFileSync(sdpPath, sdpContent, 'utf8'); }
try { fs.writeFileSync(sdpPath, sdpContent, 'utf8'); }
catch { cleanup(); return null; }
catch { cleanup(); return null; }
+ return { sdpPath, cleanup };
+}
+
+// Common ffmpeg input flags for reading an RTP/SDP source. Tolerant of the packet loss and
+// timestamp weirdness that shows up on a live ingest.
+const SDP_INPUT_ARGS = [
+ '-protocol_whitelist', 'file,rtp,udp',
+ '-thread_queue_size', '2048',
+ '-analyzeduration', '5000000', '-probesize', '5000000',
+ '-use_wallclock_as_timestamps', '1',
+ '-fflags', '+genpts+discardcorrupt+igndts',
+ '-err_detect', 'ignore_err',
+];
+
+// Loudness normalisation. Measured capture level on this stream was mean -32.8dB /
+// max -4.4dB — plenty of headroom but a low average, which is exactly the signal a
+// recogniser struggles with. loudnorm brought a real 60s clip from -31.1dB to -26.2dB.
+const LOUDNORM = 'loudnorm=I=-16:TP=-1.5:LRA=11';
+
+async function captureWebrtc(stream, seconds) {
+ const src = await _prepareWebrtcSource(stream);
+ if (!src) return null;
+ const { sdpPath, cleanup } = src;
+ const out = tmpWav(stream.id);
const ok = await runFfmpeg([
const ok = await runFfmpeg([
- '-y',
- '-protocol_whitelist', 'file,rtp,udp',
- '-thread_queue_size', '2048',
- '-analyzeduration', '5000000', '-probesize', '5000000',
- '-use_wallclock_as_timestamps', '1',
- '-fflags', '+genpts+discardcorrupt+igndts',
- '-err_detect', 'ignore_err',
+ '-y', ...SDP_INPUT_ARGS,
'-i', sdpPath,
'-i', sdpPath,
'-t', String(seconds),
'-t', String(seconds),
- '-vn', '-ac', '1', '-ar', '16000',
+ '-vn', '-ac', '1', '-ar', '16000', '-af', LOUDNORM,
'-f', 'wav', out,
'-f', 'wav', out,
], (seconds + 12) * 1000);
], (seconds + 12) * 1000);
@@ -186,4 +207,135 @@ async function captureAudioChunk(stream, seconds = 12) {
@@ -186,4 +207,135 @@ async function captureAudioChunk(stream, seconds = 12) {
}
}
}
}
-module.exports = { captureAudioChunk };
+// ── Continuous capture ────────────────────────────────────────────────────────────────
+// The old model grabbed a 12s chunk every 120s. Measured against real stream audio that
+// captures almost nothing: the stream is only ~13% speech, so sampling 10% of wall-clock
+// caught roughly 1% of what was actually said — six of ten sampled slices contained no
+// speech at all and produced only a bare "you", which the noise filter then deleted.
+//
+// Instead run ONE long-lived ffmpeg per stream that writes rolling WAV segments. Latency
+// is explicitly not a concern here, so segments are consumed behind live at low priority.
+// Cost is bounded by VAD: only the ~13% of audio containing a voice reaches the decoder.
+const SEGMENT_SEC = Math.max(10, Math.min(120, parseInt(process.env.AI_HEAR_SEGMENT_SEC, 10) || 30));
+const SPOOL_ROOT = path.join(__dirname, '../../data/asr');
+// Hard cap on spooled audio per stream. The box has ~34GB free on a filesystem that
+// openvibe.media is actively growing into, so an unbounded spool is not an option.
+const MAX_SPOOL_FILES = Math.max(4, parseInt(process.env.AI_HEAR_MAX_SPOOL, 10) || 40);
+
+const _capturing = new Map(); // streamId -> { ff, dir, cleanup, seq }
+
+function spoolDir(streamId) { return path.join(SPOOL_ROOT, String(streamId)); }
+
+function isCapturing(streamId) { return _capturing.has(streamId); }
+
+/**
+ * Start continuous audio capture for a live stream. Idempotent.
+ * Segments land in data/asr/<streamId>/seg-NNNNNN.wav, each SEGMENT_SEC long, already
+ * 16kHz mono and loudness-normalised — i.e. exactly what the recogniser wants.
+ * @returns {Promise<boolean>} true if capture is running
+ */
+async function startContinuousCapture(stream) {
+ if (!stream || !stream.id) return false;
+ if (_capturing.has(stream.id)) return true;
+
+ const dir = spoolDir(stream.id);
+ try { fs.mkdirSync(dir, { recursive: true }); } catch { return false; }
+
+ // Resolve an input the same way the one-shot grab does: WHIP/WebRTC via a plain RTP
+ // consumer, RTMP via the local HTTP-FLV output.
+ let inputArgs = null;
+ let cleanup = () => {};
+ const src = await _prepareWebrtcSource(stream).catch(() => null);
+ if (src) {
+ inputArgs = [...SDP_INPUT_ARGS, '-i', src.sdpPath];
+ cleanup = src.cleanup;
+ } else {
+ const streamKey = resolveStreamKey(stream);
+ if (!streamKey) return false;
+ inputArgs = ['-thread_queue_size', '2048', '-i', `http://127.0.0.1:${FLV_PORT}/live/${streamKey}.flv`];
+ }
+
+ const args = [
+ '-y', ...inputArgs,
+ '-vn', '-ac', '1', '-ar', '16000', '-af', LOUDNORM,
+ '-f', 'segment', '-segment_time', String(SEGMENT_SEC),
+ '-reset_timestamps', '0', // keep timestamps continuous across segments
+ '-segment_format', 'wav',
+ '-strftime', '0',
+ path.join(dir, 'seg-%06d.wav'),
+ ];
+
+ let ff;
+ try { ff = spawn('ffmpeg', args, { stdio: 'ignore' }); }
+ catch { cleanup(); return false; }
+
+ const entry = { ff, dir, cleanup, startedAt: Date.now() };
+ _capturing.set(stream.id, entry);
+ const onExit = () => {
+ if (_capturing.get(stream.id) === entry) _capturing.delete(stream.id);
+ try { cleanup(); } catch { /* */ }
+ };
+ ff.on('close', onExit);
+ ff.on('error', onExit);
+ console.log(`[AI-Hear] stream ${stream.id}: continuous capture started (${SEGMENT_SEC}s segments → ${dir})`);
+ return true;
+}
+
+function stopContinuousCapture(streamId) {
+ const entry = _capturing.get(streamId);
+ if (!entry) return false;
+ _capturing.delete(streamId);
+ try { entry.ff.kill('SIGKILL'); } catch { /* */ }
+ try { entry.cleanup(); } catch { /* */ }
+ console.log(`[AI-Hear] stream ${streamId}: continuous capture stopped`);
+ return true;
+}
+
+function stopAllCaptures() {
+ let n = 0;
+ for (const id of [..._capturing.keys()]) { if (stopContinuousCapture(id)) n++; }
+ return n;
+}
+
+/**
+ * List finished segments awaiting transcription, oldest first.
+ * The most recent file is skipped — ffmpeg is still writing to it.
+ * Each entry carries the absolute offset (seconds into the stream) its audio starts at,
+ * derived from the segment index, so timeline timestamps stay correct.
+ */
+function pendingSegments(streamId) {
+ const dir = spoolDir(streamId);
+ let names;
+ try { names = fs.readdirSync(dir).filter(n => /^seg-\d{6}\.wav$/.test(n)).sort(); }
+ catch { return []; }
+ if (names.length <= 1) return []; // newest is still being written
+ const ready = names.slice(0, -1);
+ // Drop the oldest if the spool has run away (slow transcriber / very long stream).
+ const overflow = Math.max(0, ready.length - MAX_SPOOL_FILES);
+ for (let i = 0; i < overflow; i++) {
+ try { fs.unlinkSync(path.join(dir, ready[i])); } catch { /* */ }
+ }
+ return ready.slice(overflow).map(name => ({
+ name,
+ path: path.join(dir, name),
+ index: parseInt(name.slice(4, 10), 10),
+ offsetSec: parseInt(name.slice(4, 10), 10) * SEGMENT_SEC,
+ }));
+}
+
+function discardSegment(streamId, name) {
+ try { fs.unlinkSync(path.join(spoolDir(streamId), name)); return true; } catch { return false; }
+}
+
+/** Remove a stream's whole spool directory (call when the stream ends). */
+function purgeSpool(streamId) {
+ try { fs.rmSync(spoolDir(streamId), { recursive: true, force: true }); return true; }
+ catch { return false; }
+}
+
+module.exports = {
+ captureAudioChunk,
+ startContinuousCapture, stopContinuousCapture, stopAllCaptures, isCapturing,
+ pendingSegments, discardSegment, purgeSpool, spoolDir,
+ SEGMENT_SEC,
+};
@@ -89,7 +89,7 @@ async function _analyzeOne(stream) {
@@ -89,7 +89,7 @@ async function _analyzeOne(stream) {
const grew = (memories.length - last.count) >= SUMMARY_MIN_NEW;
const grew = (memories.length - last.count) >= SUMMARY_MIN_NEW;
const stale = (now - last.at) >= SUMMARY_MAX_AGE_MS;
const stale = (now - last.at) >= SUMMARY_MAX_AGE_MS;
if (memories.length > 1 && (!last.at || grew || stale)) {
if (memories.length > 1 && (!last.at || grew || stale)) {
- const summary = await ai.summarizeStreamMemories(memories);
+ const summary = await ai.summarizeStreamMemories(memories, stream.id);
if (summary) {
if (summary) {
db.updateStreamAiOverview(stream.id, summary);
db.updateStreamAiOverview(stream.id, summary);
_lastSummary.set(stream.id, { count: memories.length, at: now });
_lastSummary.set(stream.id, { count: memories.length, at: now });
new file mode 100644
new file mode 100644
@@ -0,0 +1,159 @@
@@ -0,0 +1,159 @@
+/**
+ * timeline-job.js — continuous audio → unified stream timeline.
+ *
+ * Replaces the old sampling model. That one grabbed a 12s chunk every 120s, which
+ * measured against real stream audio caught almost nothing: this stream is only ~13%
+ * speech, so sampling 10% of wall-clock captured roughly 1% of what was actually said.
+ * Six of ten sampled 12s slices contained no speech at all and produced a bare "you",
+ * which the noise filter then deleted — hence hours of `transcript=""` in the logs.
+ *
+ * Here instead:
+ * stream-audio.startContinuousCapture() one long-lived ffmpeg per stream,
+ * rolling 16kHz mono loudnorm'd WAV segments
+ * │
+ * ▼
+ * transcribe (Silero VAD-gated) decodes ONLY regions containing a voice
+ * audio-events (optional) non-speech sound labels
+ * │
+ * ▼
+ * stream_timeline_events one time-indexed row per phrase / sound
+ *
+ * Latency is explicitly not a requirement, so segments are consumed behind live, one at
+ * a time, and the whole job yields whenever the box is busy. Accuracy and completeness
+ * are what matter.
+ */
+'use strict';
+const db = require('../db/database');
+const ai = require('./ai-analysis');
+const audio = require('./stream-audio');
+
+const POLL_MS = 10000;
+// Process at most this many segments per tick per stream. Keeps a backlog from monopolising
+// the CPU on a 4-core box that is also running live x264 encoding.
+const MAX_SEGMENTS_PER_TICK = 2;
+
+let _busy = false;
+const _known = new Set(); // stream ids we have started capture for
+
+function timelineEnabled() {
+ try {
+ const v = db.getSetting && db.getSetting('ai_timeline_enabled');
+ if (v === undefined || v === null || v === '') return false; // default OFF
+ return String(v) === 'true' || String(v) === '1';
+ } catch { return false; }
+}
+
+/** Transcribe one spooled segment and write its speech + sound rows. */
+async function _processSegment(stream, seg) {
+ const transcribe = require('./transcribe');
+ let events = [];
+
+ // ── Speech ────────────────────────────────────────────────────────────────
+ // offsetSec shifts whisper's file-relative timestamps into absolute stream time,
+ // so a row's start_sec is directly usable as a ?t= deep link.
+ let tx = null;
+ try {
+ tx = await transcribe.transcribeWavDetailed(seg.path, {
+ offsetSec: seg.offsetSec,
+ timeoutMs: 180000,
+ });
+ } catch (e) {
+ console.warn(`[AI-Timeline] stream ${stream.id} seg ${seg.name}: transcribe error`, e.message);
+ }
+ if (tx && tx.ok && Array.isArray(tx.segments)) {
+ for (const s of tx.segments) {
+ const text = String(s.text || '').trim();
+ if (!text) continue;
+ events.push({
+ stream_id: stream.id, user_id: stream.user_id, kind: 'speech',
+ start_sec: Number(s.start) || seg.offsetSec,
+ end_sec: s.end == null ? null : Number(s.end),
+ text,
+ });
+ }
+ }
+
+ // ── Non-speech sound events ───────────────────────────────────────────────
+ // Optional and best-effort: if the classifier is unavailable the speech timeline is
+ // still complete, so this must never be able to fail the segment.
+ try {
+ const sounds = require('./audio-events');
+ if (sounds.available()) {
+ const cat = stream.category || null;
+ const detected = await sounds.detect(seg.path, { offsetSec: seg.offsetSec, category: cat });
+ for (const d of (detected || [])) {
+ events.push({
+ stream_id: stream.id, user_id: stream.user_id, kind: 'sound',
+ start_sec: d.start, end_sec: d.end, label: d.label, confidence: d.confidence,
+ });
+ }
+ }
+ } catch { /* classifier optional */ }
+
+ if (events.length) db.addTimelineEvents(events);
+ audio.discardSegment(stream.id, seg.name);
+
+ const spoken = events.filter(e => e.kind === 'speech').length;
+ const heard = events.filter(e => e.kind === 'sound').length;
+ if (spoken || heard) {
+ const preview = events.find(e => e.kind === 'speech');
+ console.log(`[AI-Timeline] stream ${stream.id} +${seg.offsetSec}s: ${spoken} speech, ${heard} sound` +
+ (preview ? ` — ${JSON.stringify(String(preview.text).slice(0, 80))}` : ''));
+ }
+ return events.length;
+}
+
+async function tick() {
+ if (_busy) return; // never overlap; whisper is CPU-bound
+ if (!timelineEnabled()) return;
+ if (!ai.transcriptionEnabled || !ai.transcriptionEnabled()) return;
+
+ _busy = true;
+ try {
+ let live = [];
+ try { live = db.getLiveStreams() || []; } catch { return; }
+ const liveIds = new Set(live.map(s => s.id));
+
+ // Start capture for newly live streams.
+ for (const stream of live) {
+ if (audio.isCapturing(stream.id)) continue;
+ try { await audio.startContinuousCapture(stream); } catch { /* retry next tick */ }
+ _known.add(stream.id);
+ }
+
+ // Stop + tidy streams that ended. Their remaining segments are drained first so
+ // the tail of the stream is not lost, then the spool is removed.
+ for (const id of [..._known]) {
+ if (liveIds.has(id)) continue;
+ audio.stopContinuousCapture(id);
+ const leftovers = audio.pendingSegments(id);
+ if (!leftovers.length) {
+ audio.purgeSpool(id);
+ _known.delete(id);
+ }
+ }
+
+ // Drain segments, oldest first.
+ for (const stream of live) {
+ const pending = audio.pendingSegments(stream.id);
+ if (!pending.length) continue;
+ for (const seg of pending.slice(0, MAX_SEGMENTS_PER_TICK)) {
+ await _processSegment(stream, seg).catch(() => {});
+ }
+ }
+ } catch (e) {
+ console.warn('[AI-Timeline] tick error:', e.message);
+ } finally {
+ _busy = false;
+ }
+}
+
+function start() {
+ setInterval(() => { tick().catch(() => {}); }, POLL_MS);
+ console.log('[AI] Timeline job started (continuous audio → stream_timeline_events)');
+}
+
+/** Called on shutdown so ffmpeg children do not orphan. */
+function stopAll() { return audio.stopAllCaptures(); }
+
+module.exports = { start, tick, stopAll, timelineEnabled };
@@ -26,6 +26,22 @@ const CANDIDATE_BINS = [
@@ -26,6 +26,22 @@ const CANDIDATE_BINS = [
].filter(Boolean);
].filter(Boolean);
const MODEL = process.env.WHISPER_MODEL || path.join(HOME, 'whisper.cpp/models/ggml-base.en.bin');
const MODEL = process.env.WHISPER_MODEL || path.join(HOME, 'whisper.cpp/models/ggml-base.en.bin');
const THREADS = Math.max(2, Math.min(8, parseInt(process.env.WHISPER_THREADS, 10) || 4));
const THREADS = Math.max(2, Math.min(8, parseInt(process.env.WHISPER_THREADS, 10) || 4));
+
+// ── Voice Activity Detection ──────────────────────────────────────────────
+// Measured on 60s of real stream audio: VAD cut large-v3-turbo from 51.0s to 24.0s
+// AND removed a trailing "Thank you." hallucination. Both effects matter, and the
+// second one matters more than the speedup.
+//
+// This stream is only ~13% speech (VAD found 7.9s of speech in 60s). Without VAD the
+// decoder is handed 87% silence and invents text over it — "you", "Thank you.",
+// "[BLANK_AUDIO]", "Bye!" — which is exactly what the HALLUCINATIONS blocklist below
+// exists to delete. VAD removes the cause rather than filtering the symptom, so the
+// decoder only ever sees regions that actually contain a voice.
+const VAD_MODEL = process.env.WHISPER_VAD_MODEL || path.join(HOME, 'whisper.cpp/models/ggml-silero-v5.1.2.bin');
+function vadModel() {
+ if (process.env.WHISPER_VAD === '0') return null; // explicit opt-out
+ try { return fs.existsSync(VAD_MODEL) ? VAD_MODEL : null; } catch { return null; }
+}
// Default to greedy decoding (whisper.cpp default) — proven to transcribe speech
// Default to greedy decoding (whisper.cpp default) — proven to transcribe speech
// reliably here. Beam search (WHISPER_BEAM>1) sometimes collapses real speech into a
// reliably here. Beam search (WHISPER_BEAM>1) sometimes collapses real speech into a
// non-speech tag like "[Crowd noise]", so it's opt-in only.
// non-speech tag like "[Crowd noise]", so it's opt-in only.
@@ -73,12 +89,21 @@ function setLowPower(v) { _lowPower = !!v; }
@@ -73,12 +89,21 @@ function setLowPower(v) { _lowPower = !!v; }
function _threads() { return _lowPower ? Math.max(1, Math.min(2, THREADS)) : THREADS; }
function _threads() { return _lowPower ? Math.max(1, Math.min(2, THREADS)) : THREADS; }
// Phrases whisper commonly hallucinates over silence / music / non-speech.
// Phrases whisper commonly hallucinates over silence / music / non-speech.
+// Always-drop: whisper's stock inventions over non-speech. These are never worth keeping
+// even when VAD says a voice is present, because they are the model's filler, not words.
const HALLUCINATIONS = new Set([
const HALLUCINATIONS = new Set([
- 'you', 'thank you', 'thank you.', 'thanks for watching', 'thanks for watching!',
+ 'thank you', 'thank you.', 'thanks for watching', 'thanks for watching!',
'thanks for watching.', 'please subscribe', 'subscribe', 'like and subscribe',
'thanks for watching.', 'please subscribe', 'subscribe', 'like and subscribe',
- 'bye', 'bye.', 'bye bye', 'okay', 'ok', 'oh', 'uh', 'um', 'hmm', 'mm', 'mhm',
- 'the', 'so', 'yeah', '.', '...', 'thank you for watching', 'thank you very much',
- 'thank you so much', 'i\'m sorry', 'silence', 'music', 'applause',
+ 'thank you for watching', 'thank you very much',
+ 'thank you so much', 'silence', 'music', 'applause',
+]);
+// Drop-only-without-VAD: these ARE real words a streamer says constantly. They were on the
+// blocklist because, with the decoder run over 87% silence, they were overwhelmingly
+// hallucinations — six of ten sampled 12s slices produced a bare "you". Once Silero gates
+// the decoder to actual speech regions, deleting them throws away genuine transcript.
+const FILLER_WORDS = new Set([
+ 'you', 'bye', 'bye.', 'bye bye', 'okay', 'ok', 'oh', 'uh', 'um', 'hmm', 'mm', 'mhm',
+ 'the', 'so', 'yeah', 'i\'m sorry',
]);
]);
function _isNoise(t) {
function _isNoise(t) {
const s = (t || '').replace(/\s+/g, ' ').trim();
const s = (t || '').replace(/\s+/g, ' ').trim();
@@ -88,7 +113,9 @@ function _isNoise(t) {
@@ -88,7 +113,9 @@ function _isNoise(t) {
if (/^[\[(♪].*[\])♪]?$/.test(s)) return true;
if (/^[\[(♪].*[\])♪]?$/.test(s)) return true;
if (/^♪/.test(s) || /♪$/.test(s)) return true;
if (/^♪/.test(s) || /♪$/.test(s)) return true;
const norm = s.toLowerCase().replace(/[.!?,…]+$/g, '').trim();
const norm = s.toLowerCase().replace(/[.!?,…]+$/g, '').trim();
- return HALLUCINATIONS.has(norm);
+ if (HALLUCINATIONS.has(norm)) return true;
+ if (!vadModel() && FILLER_WORDS.has(norm)) return true;
+ return false;
}
}
// Clean a raw segment list: drop noise + collapse immediate repeats.
// Clean a raw segment list: drop noise + collapse immediate repeats.
@@ -127,6 +154,10 @@ function transcribeWavDetailed(wavPath, { timeoutMs = 180000, offsetSec = 0 } =
@@ -127,6 +154,10 @@ function transcribeWavDetailed(wavPath, { timeoutMs = 180000, offsetSec = 0 } =
const jsonPath = `${outBase}.json`;
const jsonPath = `${outBase}.json`;
const args = ['-m', MODEL, '-f', wavPath, '-oj', '-of', outBase, '-t', String(_threads()), '-l', 'en'];
const args = ['-m', MODEL, '-f', wavPath, '-oj', '-of', outBase, '-t', String(_threads()), '-l', 'en'];
if (BEAM > 1) args.push('-bs', String(BEAM));
if (BEAM > 1) args.push('-bs', String(BEAM));
+ // Decode only the regions Silero says contain a voice. Segment timestamps stay in
+ // absolute file time, so the {start,end} contract is unchanged for every caller.
+ const vm = vadModel();
+ if (vm) args.push('--vad', '-vm', vm);
let ff;
let ff;
try { ff = _track(spawn(bin, args, { stdio: 'ignore' })); }
try { ff = _track(spawn(bin, args, { stdio: 'ignore' })); }
catch (e) { return resolve({ text: '', segments: [], ok: false, error: e.message }); }
catch (e) { return resolve({ text: '', segments: [], ok: false, error: e.message }); }
@@ -883,6 +883,7 @@ function initDb() {
@@ -883,6 +883,7 @@ function initDb() {
['ai_stream_memory_enabled', 'false', 'Periodically analyze live-stream thumbnails into timestamped memories', 'boolean'],
['ai_stream_memory_enabled', 'false', 'Periodically analyze live-stream thumbnails into timestamped memories', 'boolean'],
['ai_stream_capture_interval_sec', '120', 'Seconds between live-stream AI memory captures', 'number'],
['ai_stream_capture_interval_sec', '120', 'Seconds between live-stream AI memory captures', 'number'],
['ai_transcription_enabled', 'true', 'Transcribe live-stream/clip/VOD audio into memories — FREE, runs locally via whisper.cpp (no API/cost). Requires whisper.cpp installed on the server', 'boolean'],
['ai_transcription_enabled', 'true', 'Transcribe live-stream/clip/VOD audio into memories — FREE, runs locally via whisper.cpp (no API/cost). Requires whisper.cpp installed on the server', 'boolean'],
+ ['ai_timeline_enabled', 'false', 'CONTINUOUS audio timeline — transcribes the WHOLE live stream (not a 12s sample every 2min) and detects non-speech sounds, into a searchable timestamped timeline. FREE/local, but uses noticeably more CPU than sampling', 'boolean'],
['ai_max_cost_usd_per_day', '0', 'Daily AI spend cap in USD (0 = no cap)', 'number'],
['ai_max_cost_usd_per_day', '0', 'Daily AI spend cap in USD (0 = no cap)', 'number'],
['ai_input_cost_per_mtok', '3.0', 'Estimated input cost per million tokens (for cost breakdown)', 'number'],
['ai_input_cost_per_mtok', '3.0', 'Estimated input cost per million tokens (for cost breakdown)', 'number'],
['ai_output_cost_per_mtok', '15.0', 'Estimated output cost per million tokens (for cost breakdown)', 'number'],
['ai_output_cost_per_mtok', '15.0', 'Estimated output cost per million tokens (for cost breakdown)', 'number'],
@@ -923,6 +924,31 @@ function initDb() {
@@ -923,6 +924,31 @@ 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)');
+ // ── 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
+ // old arrangement where transcript segments were buried inside a per-memory JSON
+ // blob, which meant timestamps were unusable for anything but display — of ~10
+ // consumers only ai-moments-job actually read them, and getStreamTranscriptSegments
+ // discarded `end` entirely.
+ database.exec(`CREATE TABLE IF NOT EXISTS stream_timeline_events (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ stream_id INTEGER NOT NULL,
+ user_id INTEGER,
+ vod_id INTEGER, -- set once the session becomes a VOD
+ kind TEXT NOT NULL, -- 'speech' | 'sound'
+ start_sec REAL NOT NULL, -- absolute seconds into the stream
+ end_sec REAL,
+ text TEXT, -- speech content
+ label TEXT, -- sound event label
+ confidence REAL,
+ created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+ FOREIGN KEY (stream_id) REFERENCES streams(id) ON DELETE CASCADE
+ )`);
+ database.exec('CREATE INDEX IF NOT EXISTS idx_timeline_stream ON stream_timeline_events(stream_id, start_sec)');
+ database.exec('CREATE INDEX IF NOT EXISTS idx_timeline_kind ON stream_timeline_events(stream_id, kind, start_sec)');
+ database.exec('CREATE INDEX IF NOT EXISTS idx_timeline_vod ON stream_timeline_events(vod_id, start_sec)');
+
database.exec(`CREATE TABLE IF NOT EXISTS ai_usage (
database.exec(`CREATE TABLE IF NOT EXISTS ai_usage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
id INTEGER PRIMARY KEY AUTOINCREMENT,
kind TEXT, -- paste_image | paste_text | stream_memory | ai_viewers | ...
kind TEXT, -- paste_image | paste_text | stream_memory | ai_viewers | ...
@@ -2712,7 +2738,80 @@ function countStreamMemoriesByUser(userId) {
@@ -2712,7 +2738,80 @@ function countStreamMemoriesByUser(userId) {
// Flattened audio-transcript segments for a whole stream (from its memories), ordered by time.
// Flattened audio-transcript segments for a whole stream (from its memories), ordered by time.
// Used by the AI Timeline transcript viewer; each segment deep-links to the VOD at its start.
// Used by the AI Timeline transcript viewer; each segment deep-links to the VOD at its start.
+// ── Timeline accessors ────────────────────────────────────────────────────────
+/**
+ * Bulk-insert timeline rows.
+ * @param {Array<{stream_id,user_id?,vod_id?,kind,start_sec,end_sec?,text?,label?,confidence?}>} rows
+ */
+function addTimelineEvents(rows) {
+ if (!Array.isArray(rows) || !rows.length) return 0;
+ const stmt = db.prepare(`INSERT INTO stream_timeline_events
+ (stream_id, user_id, vod_id, kind, start_sec, end_sec, text, label, confidence)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`);
+ const tx = db.transaction((list) => {
+ for (const r of list) {
+ if (!r || !r.stream_id || !r.kind || r.start_sec == null) continue;
+ stmt.run(r.stream_id, r.user_id || null, r.vod_id || null, r.kind,
+ Number(r.start_sec) || 0, r.end_sec == null ? null : Number(r.end_sec),
+ r.text || null, r.label || null, r.confidence == null ? null : Number(r.confidence));
+ }
+ });
+ try { tx(rows); return rows.length; } catch { return 0; }
+}
+
+/** Read a stream's timeline, optionally filtered by kind and time window. */
+function getTimeline(streamId, { kind = null, from = null, to = null, limit = 5000 } = {}) {
+ let sql = 'SELECT kind, start_sec, end_sec, text, label, confidence FROM stream_timeline_events WHERE stream_id = ?';
+ const params = [streamId];
+ if (kind) { sql += ' AND kind = ?'; params.push(kind); }
+ if (from != null) { sql += ' AND start_sec >= ?'; params.push(Number(from)); }
+ if (to != null) { sql += ' AND start_sec <= ?'; params.push(Number(to)); }
+ sql += ' ORDER BY start_sec ASC LIMIT ?';
+ params.push(Math.max(1, Math.min(20000, limit)));
+ try { return all(sql, params); } catch { return []; }
+}
+
+/** Flat transcript text for a stream, speech rows only, in time order. */
+function getTimelineText(streamId) {
+ try {
+ return getTimeline(streamId, { kind: 'speech' })
+ .map(r => String(r.text || '').trim()).filter(Boolean).join(' ');
+ } catch { return ''; }
+}
+
+/** How many seconds of a stream the timeline actually covers (union of speech spans). */
+function getTimelineCoverage(streamId) {
+ const rows = getTimeline(streamId, { kind: 'speech' });
+ let covered = 0, lastEnd = -1;
+ for (const r of rows) {
+ const st = Number(r.start_sec) || 0;
+ const en = r.end_sec == null ? st : Number(r.end_sec);
+ if (en <= lastEnd) continue;
+ covered += en - Math.max(st, lastEnd);
+ lastEnd = en;
+ }
+ return Math.round(covered);
+}
+
+/** 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]); }
+ catch { return null; }
+}
+
function getStreamTranscriptSegments(streamId) {
function getStreamTranscriptSegments(streamId) {
+ // Prefer the timeline when it has rows — it keeps `end` and covers the whole stream.
+ // Fall back to the legacy per-memory blobs so old streams keep rendering; no migration.
+ try {
+ const tl = getTimeline(streamId, { kind: 'speech' });
+ if (tl.length) {
+ return tl.map(r => ({
+ start: Math.floor(Number(r.start_sec) || 0),
+ end: r.end_sec == null ? null : Math.round(Number(r.end_sec) * 100) / 100,
+ text: String(r.text || '').trim(),
+ })).filter(s => s.text);
+ }
+ } catch { /* fall through to legacy */ }
const out = [];
const out = [];
try {
try {
const rows = all('SELECT offset_seconds, transcript_json FROM stream_memories WHERE stream_id = ? AND transcript_json IS NOT NULL ORDER BY offset_seconds ASC', [streamId]);
const rows = all('SELECT offset_seconds, transcript_json FROM stream_memories WHERE stream_id = ? AND transcript_json IS NOT NULL ORDER BY offset_seconds ASC', [streamId]);
@@ -7172,6 +7271,7 @@ module.exports = {
@@ -7172,6 +7271,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,
getVodsForMomentRanking, getClipStartTimesForStream, getChatSpikeOffsets,
getVodsForMomentRanking, getClipStartTimesForStream, getChatSpikeOffsets,
getLiveChatBuckets, getRecentChatText, getVodsWithoutAutoClip,
getLiveChatBuckets, getRecentChatText, getVodsWithoutAutoClip,
upsertStreamerOverview, getStreamerOverview, getAllStreamerOverviews, getStreamersNeedingOverview,
upsertStreamerOverview, getStreamerOverview, getAllStreamerOverviews, getStreamersNeedingOverview,
@@ -932,6 +932,9 @@ async function start() {
@@ -932,6 +932,9 @@ async function start() {
try { require('./ai/ai-moments-job').start(); } catch (e) { console.warn('[AI] moments job not started:', e.message); }
try { require('./ai/ai-moments-job').start(); } catch (e) { console.warn('[AI] moments job not started:', e.message); }
try { require('./ai/auto-clip-job').start(); } catch (e) { console.warn('[AI] auto-clip job not started:', e.message); }
try { require('./ai/auto-clip-job').start(); } catch (e) { console.warn('[AI] auto-clip job not started:', e.message); }
try { require('./ai/easter-egg-job').start(); } catch (e) { console.warn('[AI] easter-egg job not started:', e.message); }
try { require('./ai/easter-egg-job').start(); } catch (e) { console.warn('[AI] easter-egg job not started:', e.message); }
+ // Continuous audio → stream_timeline_events. Gated behind ai_timeline_enabled
+ // (default off), so starting it is a no-op until switched on in admin.
+ try { require('./ai/timeline-job').start(); } catch (e) { console.warn('[AI] timeline job not started:', e.message); }
// Heal server-side recordings for live streams (resume after restart / Media restart)
// Heal server-side recordings for live streams (resume after restart / Media restart)
// so clipping always has a source. First pass delayed to let broadcasters reconnect.
// so clipping always has a source. First pass delayed to let broadcasters reconnect.
// VOD health scanning, disk guardianship and storage tiering moved to OpenVibe.Media.
// VOD health scanning, disk guardianship and storage tiering moved to OpenVibe.Media.
@@ -1212,6 +1215,7 @@ function shutdown() {
@@ -1212,6 +1215,7 @@ function shutdown() {
// (and so their temp files get cleaned by the close handlers). The interrupted VOD
// (and so their temp files get cleaned by the close handlers). The interrupted VOD
// is left in 'processing' → re-queued to 'pending' on next boot (crash recovery).
// is left in 'processing' → re-queued to 'pending' on next boot (crash recovery).
try { const n = require('./ai/transcribe').killActive(); if (n) console.log(`[Server] Killed ${n} transcription child(ren)`); } catch { /* */ }
try { const n = require('./ai/transcribe').killActive(); if (n) console.log(`[Server] Killed ${n} transcription child(ren)`); } catch { /* */ }
+ try { const n = require('./ai/timeline-job').stopAll(); if (n) console.log(`[Server] Stopped ${n} continuous audio capture(s)`); } catch { /* */ }
try { require('./ai/media-analysis').killActive(); } catch { /* */ }
try { require('./ai/media-analysis').killActive(); } catch { /* */ }
// Cleanly END RobotStreamer passthrough streams FIRST (before the exit races the reconnect).
// Cleanly END RobotStreamer passthrough streams FIRST (before the exit races the reconnect).
@@ -46,6 +46,13 @@ function handler(req, res) {
@@ -46,6 +46,13 @@ function handler(req, res) {
// Seed the Live-owned AI state row (queues transcript/overview work)
// Seed the Live-owned AI state row (queues transcript/overview work)
// and kick the on-finalize AI pass right away (both budget-gated).
// and kick the on-finalize AI pass right away (both budget-gated).
try { db.setVodTranscriptStatus(vodId, 'pending'); } catch { /* */ }
try { db.setVodTranscriptStatus(vodId, 'pending'); } catch { /* */ }
+ // Point the live timeline rows at the VOD so the recording inherits the
+ // transcript that was already built while the stream was running — no
+ // need to transcribe the same audio a second time.
+ try {
+ const sid = data?.stream_id || data?.streamId;
+ if (sid) db.linkTimelineToVod(sid, vodId);
+ } catch { /* */ }
try {
try {
const ai = require('../ai/ai-analysis');
const ai = require('../ai/ai-analysis');
const vodMeta = { id: vodId, ...data };
const vodMeta = { id: vodId, ...data };