openvibers/openvibe.live Commits
Transcription reliability: live/batch whisper lanes, boot recovery on the state tables, resumable VOD windows, no-audio detection, queue head-of-line fix, stale spool purge, status in VOD page
Changes
@@@ -4995,7 +4995,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=126" defer></script>++<script src="/js/app.js?v=127" defer></script><script src="/js/stream-player.js?v=21" defer></script><script src="/js/stream-pip.js?v=1" defer></script><script src="/js/chat-shared.js?v=3" defer></script>@@@ -201,7 +201,20 @@ function _renderMediaTranscript(descElId, kind, item) {let segments = null;try { segments = (item && item.ai_transcript_json) ? JSON.parse(item.ai_transcript_json) : null; } catch { segments = null; }const hasSegs = Array.isArray(segments) && segments.length > 0;-- if (!t && !hasSegs) return;++ if (!t && !hasSegs) {++ // No transcript (yet): say why, instead of silently showing nothing. 'empty' and++ // unknown states stay quiet — there is nothing useful to tell the viewer.++ const st = item && item.transcript_status;++ const msg = (st === 'pending' || st === 'retry' || st === 'processing' || st === null || st === undefined) && item && item.id++ ? (st === 'processing' ? 'Transcript in progress…' : (st ? 'Transcript queued — check back soon' : null))++ : (st === 'failed' ? 'Transcript unavailable for this recording' : null);++ if (!msg) return;++ const note = document.createElement('div');++ note.className = 'media-transcript media-transcript-pending';++ note.innerHTML = `<div class="media-transcript-head"><span class="media-transcript-toggle" style="cursor:default;opacity:.7"><i class="fa-solid ${st === 'failed' ? 'fa-file-circle-xmark' : 'fa-file-lines'}"></i> <span>${esc(msg)}</span></span></div>`;++ desc.parentNode.insertBefore(note, desc.nextSibling);++ return;++ }const videoId = kind === 'vod' ? 'vp-video' : 'clp-video';const words = (t || segments.map(s => s.text).join(' ')).split(/\s+/).filter(Boolean).length;@@@ -464,13 +464,22 @@ async function generateVodTranscript(vod) {}db.setVodTranscriptStatus(vod.id, 'processing');let r = { text: '', segments: [], ok: false, error: 'unknown' };-- try { r = await require('./media-analysis').transcribeOnly(src); } catch (e) { r = { text: '', segments: [], ok: false, error: e.message }; }++ try {++ // Continue from the last finished 5-minute window (persisted per window), so++ // the frequent deploy restarts no longer discard an hour of decoding.++ const prog = db.getVodTranscriptProgress ? db.getVodTranscriptProgress(vod.id) : { progressSec: 0, segments: [] };++ r = await require('./media-analysis').transcribeOnly(src, {++ resumeFromSec: prog.progressSec,++ priorSegments: prog.segments,++ onWindow: (sec, segs) => db.saveVodTranscriptProgress(vod.id, sec, segs),++ });++ } catch (e) { r = { text: '', segments: [], ok: false, error: e.message }; }if (r.text) { // got speech → store it (+segments)try { db.setVodTranscript(vod.id, r.text, r.segments || []); } catch { /* */ }db.setVodTranscriptStatus(vod.id, 'done');} else if (r.ok) { // ran clean, genuinely no speech → terminaltry { db.setVodTranscript(vod.id, ' ', []); } catch { /* */ }-- db.setVodTranscriptStatus(vod.id, 'empty');++ db.setVodTranscriptStatus(vod.id, 'empty', r.noAudio ? 'no audio stream' : null);} else { // failure → retry (bounded + backoff), never poisonconst n = db.bumpVodTranscriptAttempt(vod.id);db.setVodTranscriptStatus(vod.id, n >= MAX_TX_ATTEMPTS ? 'failed' : 'retry', r.error || 'transcription failed', _txBackoffMin(n));@@@ -498,7 +507,7 @@ async function generateClipTranscript(clip) {db.setClipTranscriptStatus(clip.id, 'done');} else if (r.ok) {try { db.setClipTranscript(clip.id, ' ', []); } catch { /* */ }-- db.setClipTranscriptStatus(clip.id, 'empty');++ db.setClipTranscriptStatus(clip.id, 'empty', r.noAudio ? 'no audio stream' : null);} else {const n = db.bumpClipTranscriptAttempt(clip.id);db.setClipTranscriptStatus(clip.id, n >= MAX_TX_ATTEMPTS ? 'failed' : 'retry', r.error || 'transcription failed', _txBackoffMin(n));@@@ -16,6 +16,7 @@ const media = require('../media-client');let _timer = null;let _busy = false;++let _loadWarnedAt = 0;// Resolve Media metadata for a state row ({id} = Media id). Returns a vod/clip-shaped// object the ai-analysis functions understand, or null (→ retry later).@@@ -50,23 +51,39 @@ async function tick() {// run whisper at once, on top of the live encoder. Measured effect on this 4-core// box: load average 5.44. VOD transcripts have no deadline whatsoever, so simply// stand them down until the stream ends; they catch up at full speed afterwards.-- let timelineBusy = false;-- try { timelineBusy = anyLive && require('./timeline-job').timelineEnabled(); } catch { /* */ }-- if (timelineBusy) {-- try { require('./transcribe').setLowPower(true); } catch { /* */ }++ // The live timeline has its OWN whisper lane now (transcribe.js), so VOD/clip work++ // no longer has to stand down for the whole stream — it runs niced, low-power, and++ // only when the box has headroom. (The old stand-down meant nothing was ever++ // transcribed while the owner streamed, which on this site is most of the day.)++ const cores = require('os').cpus().length || 4;++ const load1 = (require('os').loadavg()[0]) || 0;++ const headroom = load1 < cores * 1.25;++ if (!headroom) {++ if (!_loadWarnedAt || Date.now() - _loadWarnedAt > 600000) { _loadWarnedAt = Date.now(); console.log(`[AI backfill] load ${load1.toFixed(1)} on ${cores} cores — deferring batch transcription`); }} else if (ai.transcriptionEnabled && ai.transcriptionEnabled()) {try { require('./transcribe').setLowPower(anyLive); } catch { /* */ }const batch = anyLive ? 1 : 2; // throttle while live, catch up faster when idle++ // Take several candidates and process the first `batch` that resolve: a row whose++ // Media metadata won't resolve (still recording, Media down) used to occupy the++ // whole LIMIT every minute and block every row behind it.try {-- for (const row of db.getVodsNeedingTranscript(batch)) {++ let done = 0;++ for (const row of db.getVodsNeedingTranscript(batch + 6)) {++ if (done >= batch) break;const vod = await _vodMeta(row);-- if (vod) await ai.generateVodTranscript(vod);++ if (!vod) continue;++ await ai.generateVodTranscript(vod);++ done++;}} catch (e) { console.warn('[AI backfill] vod transcript:', e.message); }try {-- for (const row of db.getClipsNeedingTranscript(batch)) {++ let done = 0;++ for (const row of db.getClipsNeedingTranscript(batch + 6)) {++ if (done >= batch) break;const clip = await _clipMeta(row);-- if (clip) await ai.generateClipTranscript(clip);++ if (!clip) continue;++ await ai.generateClipTranscript(clip);++ done++;}} catch (e) { console.warn('[AI backfill] clip transcript:', e.message); }}@@@ -194,7 +194,24 @@ async function pickFrameTimes(src, duration, { existingOffsets = [], maxFrames =// Transcribe the audio: full for short media, else 3 spread 60s windows. Returns// { text, segments:[{start,end,text}] } with segment times relative to the recording.--async function _transcribeSpan(src, duration) {++// Does the source carry an audio stream at all? Video-only recordings (screen shares with++// no mic, some RS robots) used to fail ffmpeg's wav extraction, burn 8 retries and end up++// 'failed' — they are simply silent and should be terminal 'empty' on the first pass.++function _ffprobeHasAudio(src) {++ return new Promise((resolve) => {++ let out = '';++ let p;++ try {++ p = _track(spawn('ffprobe', ['-v', 'error', '-select_streams', 'a', '-show_entries', 'stream=codec_type', '-of', 'csv=p=0', src], { stdio: ['ignore', 'pipe', 'ignore'] }));++ } catch { return resolve(null); }++ const t = setTimeout(() => { try { p.kill('SIGKILL'); } catch { /* */ } resolve(null); }, 30000);++ p.stdout.on('data', (d) => { out += d; });++ p.on('close', (c) => { clearTimeout(t); resolve(c === 0 ? /audio/.test(out) : null); });++ p.on('error', () => { clearTimeout(t); resolve(null); });++ });++}++++async function _transcribeSpan(src, duration, { resumeFromSec = 0, priorSegments = null, onWindow = null } = {}) {const transcribe = require('./transcribe');if (!transcribe.available()) return { text: '', segments: [], ok: false, error: 'whisper unavailable' };if (duration > 0 && duration <= 200) {@@@ -210,11 +227,16 @@ async function _transcribeSpan(src, duration) {// 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;++ // Resume support: a deploy/restart used to throw away every finished window of an++ // hour-long VOD. Callers persist progress after each window (onWindow) and hand the++ // finished segments back (priorSegments/resumeFromSec) so we continue, not restart.++ const segments = Array.isArray(priorSegments) ? [...priorSegments] : [];++ let anyOk = segments.length > 0, lastErr = null;const total = Math.max(0, Math.floor(duration || 0));const windows = [];-- for (let start = 0; start < total; start += WINDOW_SEC) windows.push(start);++ const firstStart = Math.max(0, Math.floor((resumeFromSec || 0) / WINDOW_SEC) * WINDOW_SEC);++ for (let start = firstStart; start < total; start += WINDOW_SEC) windows.push(start);++ if (firstStart > 0) console.log(`[AI] transcript resuming at ${Math.round(firstStart / 60)}min of ${Math.round(total / 60)}min (${segments.length} segments kept)`);// 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); // 8hif (windows.length > MAX_WINDOWS) {@@@ -228,11 +250,16 @@ async function _transcribeSpan(src, duration) {if (ffOk) {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);++ if (r.ok && typeof onWindow === 'function') {++ try { await onWindow(Math.min(total, start + len), segments); } catch { /* progress is best-effort */ }++ }} else { lastErr = 'ffmpeg window extract failed'; }try { fs.existsSync(wav) && fs.unlinkSync(wav); } catch { /* */ }}++ segments.sort((a, b) => (a.start || 0) - (b.start || 0));++ parts.length = 0;++ for (const s of segments) if (s.text) parts.push(s.text);// 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.@@@ -280,10 +307,12 @@ async function analyzeMedia(src, { streamId = null, userId = null, numFrames = n* Transcript-only: probe duration then whisper-transcribe (FREE local, no vision).* @returns {Promise<{text:string, segments:Array}>}*/--async function transcribeOnly(src) {++async function transcribeOnly(src, opts = {}) {if (!src) return { text: '', segments: [], ok: false, error: 'no source' };++ const hasAudio = await _ffprobeHasAudio(src);++ if (hasAudio === false) return { text: '', segments: [], ok: true, noAudio: true, error: null };const duration = await _ffprobeDuration(src);-- try { return await _transcribeSpan(src, duration); } catch (e) { return { text: '', segments: [], ok: false, error: e.message }; }++ try { return await _transcribeSpan(src, duration, opts); } catch (e) { return { text: '', segments: [], ok: false, error: e.message }; }}module.exports = {@@@ -250,6 +250,15 @@ async function startContinuousCapture(stream) {const dir = spoolDir(stream.id);try { fs.mkdirSync(dir, { recursive: true }); } catch { return false; }++ // A previous process (deploy restart) may have left segments here. Their indexes++ // restart at 0 for the new ffmpeg, so old files would be mis-ordered against new ones++ // — the "newest file is still being written" rule then skipped a stale file and fed a++ // half-written one to whisper. Drop them; the VOD transcript covers that audio.++ try {++ const stale = fs.readdirSync(dir).filter(n => /^seg-\d{6}\.wav$/.test(n));++ for (const n of stale) { try { fs.unlinkSync(path.join(dir, n)); } catch { /* */ } }++ if (stale.length) console.log(`[AI-Hear] stream ${stream.id}: dropped ${stale.length} stale spool segment(s) from a previous run`);++ } catch { /* */ }// 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.@@@ -328,6 +337,7 @@ function pendingSegments(streamId, { includeLast = false } = {}) {for (let i = 0; i < overflow; i++) {try { fs.unlinkSync(path.join(dir, ready[i])); } catch { /* */ }}++ if (overflow) console.warn(`[AI-Hear] stream ${streamId}: transcriber ${overflow * SEGMENT_SEC}s behind — dropped ${overflow} oldest segment(s)`);const startedAt = (_capturing.get(streamId) || {}).startedAt || null;return ready.slice(overflow).map(name => {const index = parseInt(name.slice(4, 10), 10);@@@ -102,23 +102,40 @@ function killActive() {//// Put the limit HERE, where every caller must pass, so no future path can bypass it.const MAX_CONCURRENT = Math.max(1, parseInt(process.env.WHISPER_MAX_CONCURRENT, 10) || 1);--let _running = 0;--const _waiters = [];--function _acquire() {-- if (_running < MAX_CONCURRENT) { _running++; return Promise.resolve(); }-- return new Promise(resolve => _waiters.push(resolve));++// Two INDEPENDENT lanes. The live timeline decodes 10s segments with the small model and++// must keep up with real time; VOD/clip backfill decodes hour-long recordings with the++// large model. When they shared one slot, a single VOD job (minutes to hours, and killed++// + restarted from scratch on every deploy) held the slot and the live transcript simply++// stopped — segments piled up in the spool and were discarded. Live never waits on batch.++const _lanes = {++ live: { max: 1, running: 0, waiters: [] },++ batch: { max: MAX_CONCURRENT, running: 0, waiters: [] },++};++function _acquire(lane = 'batch') {++ const L = _lanes[lane] || _lanes.batch;++ if (L.running < L.max) { L.running++; return Promise.resolve(); }++ return new Promise(resolve => L.waiters.push(resolve));}--function _release() {-- const next = _waiters.shift();++function _release(lane = 'batch') {++ const L = _lanes[lane] || _lanes.batch;++ const next = L.waiters.shift();if (next) next(); // hand the slot straight over-- else _running = Math.max(0, _running - 1);++ else L.running = Math.max(0, L.running - 1);++}++function laneStatus() {++ return Object.fromEntries(Object.entries(_lanes).map(([k, L]) => [k, { running: L.running, queued: L.waiters.length }]));}// While a stream is live we lower the whisper thread count so VOD transcription can// still make progress without starving the live encoders.let _lowPower = false;function setLowPower(v) { _lowPower = !!v; }--function _threads() { return _lowPower ? Math.max(1, Math.min(2, THREADS)) : THREADS; }++function _threads(live = false) {++ // Live segments are short and use the small model: 2 threads keeps them well ahead of++ // real time while leaving cores for the encoders and the (niced) batch lane.++ if (live) return Math.max(1, Math.min(2, 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@@@ -194,7 +211,7 @@ function _transcribeWavInner(wavPath, { timeoutMs = 180000, offsetSec = 0, live}const outBase = `${wavPath}.out`;const jsonPath = `${outBase}.json`;-- const args = ['-m', _modelFor({ live }), '-f', wavPath, '-oj', '-of', outBase, '-t', String(_threads()), '-l', 'en'];++ const args = ['-m', _modelFor({ live }), '-f', wavPath, '-oj', '-of', outBase, '-t', String(_threads(live)), '-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.@@@ -278,9 +295,23 @@ async function transcribeMedia(mediaPath, opts = {}) {* The slot is held for the whole decode and always released, including on throw.*/async function transcribeWavDetailed(wavPath, opts = {}) {-- await _acquire();++ const lane = opts && opts.live ? 'live' : 'batch';++ await _acquire(lane);try { return await _transcribeWavInner(wavPath, opts); }-- finally { _release(); }++ finally { _release(lane); }++}++++/** Human-readable availability report (logged at boot so a broken install is visible). */++function describe() {++ const bin = whisperBin();++ return {++ available: available(),++ bin: bin || null,++ model: MODEL, modelExists: fs.existsSync(MODEL),++ modelLive: MODEL_LIVE, modelLiveExists: fs.existsSync(MODEL_LIVE),++ vadModel: vadModel() || null,++ threads: THREADS, beam: BEAM, maxConcurrent: MAX_CONCURRENT,++ };}--module.exports = { available, transcribeWav, transcribeWavDetailed, transcribeMedia, transcribeMediaDetailed, killActive, setLowPower };++module.exports = { available, describe, laneStatus, transcribeWav, transcribeWavDetailed, transcribeMedia, transcribeMediaDetailed, killActive, setLowPower };@@@ -278,6 +278,28 @@ function initDb() {}} catch (e) { console.warn('[DB] ai-overview column migration:', e.message); }++ // Transcript job-state recovery — on the tables that actually hold it. The older++ // recovery loop above targets the legacy local `vods`/`clips` tables, which moved to++ // OpenVibe.Media; it throws on the first PRAGMA and silently does nothing, which is++ // why rows killed mid-flight by a deploy stayed 'processing' forever and 'failed' was++ // permanent. Also adds resumable-progress columns (finished windows survive restarts).++ try {++ for (const t of ['vod_ai_state', 'clip_ai_state']) {++ const cols = database.prepare(`PRAGMA table_info(${t})`).all().map(c => c.name);++ if (!cols.includes('transcript_partial_json')) database.exec(`ALTER TABLE ${t} ADD COLUMN transcript_partial_json TEXT`);++ if (!cols.includes('transcript_progress_sec')) database.exec(`ALTER TABLE ${t} ADD COLUMN transcript_progress_sec INTEGER DEFAULT 0`);++ // Mid-flight at shutdown → back to the queue (progress is kept, attempts untouched).++ const p = database.prepare(`UPDATE ${t} SET transcript_status='retry', transcript_next_at=NULL WHERE transcript_status='processing'`).run();++ // Exhausted retries get a fresh ladder after a deploy — except sources that can++ // never work (Media reported the recording itself failed / no audio stream).++ const f = database.prepare(`UPDATE ${t} SET transcript_status='retry', transcript_attempts=0, transcript_next_at=NULL++ WHERE transcript_status='failed'++ AND COALESCE(transcript_error,'') NOT LIKE 'media reported%'++ AND COALESCE(transcript_error,'') NOT LIKE 'no audio stream%'`).run();++ if (p.changes || f.changes) console.log(`[DB] ${t}: re-queued ${p.changes} interrupted + ${f.changes} previously-failed transcript job(s)`);++ }++ } catch (e) { console.warn('[DB] transcript recovery:', e.message); }++// ── Migrations ────────────────────────────────────────────try {const cols = database.prepare("PRAGMA table_info('channels')").all().map(c => c.name);@@@ -1417,7 +1439,7 @@ function initDb() {WHERE transcript_status IS NULL AND ai_transcript = ' ' AND created_at >= datetime('now','-60 days')`);// Older poisoned items: assume genuinely silent (terminal) so we don't churn the whole archive.database.exec(`UPDATE ${t} SET transcript_status='empty' WHERE transcript_status IS NULL AND ai_transcript = ' '`);-- // Crash/deploy recovery: anything left mid-flight is re-queued on every boot.++ // (Legacy local tables — see the vod_ai_state/clip_ai_state recovery below.)database.exec(`UPDATE ${t} SET transcript_status=NULL, transcript_next_at=NULL WHERE transcript_status='processing'`);// Give previously-'failed' rows a fresh chance after a deploy — transient issues// (whisper temporarily missing, storage hiccup, a bad build) shouldn't be permanent.@@@ -2823,7 +2845,19 @@ function _segJson(segments) {}function setVodTranscript(vodId, transcript, segments) {_ensureVodAiState(vodId);-- return run('UPDATE vod_ai_state SET ai_transcript_json = ? WHERE vod_id = ?', [_segJson(segments) ?? (transcript ? JSON.stringify([]) : null), vodId]);++ return run('UPDATE vod_ai_state SET ai_transcript_json = ?, transcript_partial_json = NULL, transcript_progress_sec = 0 WHERE vod_id = ?', [_segJson(segments) ?? (transcript ? JSON.stringify([]) : null), vodId]);++}++// Resumable VOD transcription: persist finished windows so a restart continues from here.++function saveVodTranscriptProgress(vodId, progressSec, segments) {++ _ensureVodAiState(vodId);++ return run('UPDATE vod_ai_state SET transcript_partial_json = ?, transcript_progress_sec = ? WHERE vod_id = ?', [_segJson(segments), Math.max(0, Math.floor(progressSec || 0)), vodId]);++}++function getVodTranscriptProgress(vodId) {++ const row = get('SELECT transcript_partial_json, transcript_progress_sec FROM vod_ai_state WHERE vod_id = ?', [vodId]);++ if (!row) return { progressSec: 0, segments: [] };++ let segments = [];++ try { segments = row.transcript_partial_json ? JSON.parse(row.transcript_partial_json) : []; } catch { segments = []; }++ return { progressSec: row.transcript_progress_sec || 0, segments: Array.isArray(segments) ? segments : [] };}function setClipTranscript(clipId, transcript, segments) {_ensureClipAiState(clipId);@@@ -7743,6 +7777,7 @@ module.exports = {updateChannelAiBot, touchChannelAiBot, deleteChannelAiBot,// Site SettingsgetSetting, getSettingRow, getAllSettings, setSetting, deleteSetting,++ saveVodTranscriptProgress, getVodTranscriptProgress,getState, setState, deleteState,// Verification KeyscreateVerificationKey, getVerificationKeyByKey, getVerificationKeyByUsername,@@@ -976,7 +976,8 @@ async function start() {// AI stream-memory job (no-op until enabled in openvibe.network/admin → AI).try { require('./ai/stream-memory-job').start(); } catch (e) { console.warn('[AI] memory job not started:', e.message); }-- try { require('./ai/backfill-job').start(); } catch (e) { console.warn('[AI] backfill job not started:', e.message); }++ try { (() => { try { const d = require('./ai/transcribe').describe(); console.log(`[AI] whisper ${d.available ? 'available' : 'UNAVAILABLE'} — bin=${d.bin} model=${d.model}${d.modelExists ? '' : ' (MISSING)'} live=${d.modelLive}${d.modelLiveExists ? '' : ' (MISSING)'} vad=${d.vadModel || 'off'} threads=${d.threads}`); } catch (e) { console.warn('[AI] whisper probe failed:', e.message); } })();++ require('./ai/backfill-job').start(); } catch (e) { console.warn('[AI] backfill job not started:', e.message); }try { require('./ai/streamer-overview-job').start(); } catch (e) { console.warn('[AI] streamer-overview job not started:', e.message); }try { require('./ai/chat-ai').start(); } catch (e) { console.warn('[AI] chat-ai job not started:', e.message); }try { require('./ai/slogan-job').start(); } catch (e) { console.warn('[AI] slogan job not started:', e.message); }
@@ -4995,7 +4995,7 @@ curl -s https://openvibe.media/live/@Goosely/chat-insight.json | jq ".chat_insig
@@ -4995,7 +4995,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=126" defer></script>
+<script src="/js/app.js?v=127" defer></script>
<script src="/js/stream-player.js?v=21" defer></script>
<script src="/js/stream-player.js?v=21" defer></script>
<script src="/js/stream-pip.js?v=1" defer></script>
<script src="/js/stream-pip.js?v=1" defer></script>
<script src="/js/chat-shared.js?v=3" defer></script>
<script src="/js/chat-shared.js?v=3" defer></script>
@@ -201,7 +201,20 @@ function _renderMediaTranscript(descElId, kind, item) {
@@ -201,7 +201,20 @@ function _renderMediaTranscript(descElId, kind, item) {
let segments = null;
let segments = null;
try { segments = (item && item.ai_transcript_json) ? JSON.parse(item.ai_transcript_json) : null; } catch { segments = null; }
try { segments = (item && item.ai_transcript_json) ? JSON.parse(item.ai_transcript_json) : null; } catch { segments = null; }
const hasSegs = Array.isArray(segments) && segments.length > 0;
const hasSegs = Array.isArray(segments) && segments.length > 0;
- if (!t && !hasSegs) return;
+ if (!t && !hasSegs) {
+ // No transcript (yet): say why, instead of silently showing nothing. 'empty' and
+ // unknown states stay quiet — there is nothing useful to tell the viewer.
+ const st = item && item.transcript_status;
+ const msg = (st === 'pending' || st === 'retry' || st === 'processing' || st === null || st === undefined) && item && item.id
+ ? (st === 'processing' ? 'Transcript in progress…' : (st ? 'Transcript queued — check back soon' : null))
+ : (st === 'failed' ? 'Transcript unavailable for this recording' : null);
+ if (!msg) return;
+ const note = document.createElement('div');
+ note.className = 'media-transcript media-transcript-pending';
+ note.innerHTML = `<div class="media-transcript-head"><span class="media-transcript-toggle" style="cursor:default;opacity:.7"><i class="fa-solid ${st === 'failed' ? 'fa-file-circle-xmark' : 'fa-file-lines'}"></i> <span>${esc(msg)}</span></span></div>`;
+ desc.parentNode.insertBefore(note, desc.nextSibling);
+ return;
+ }
const videoId = kind === 'vod' ? 'vp-video' : 'clp-video';
const videoId = kind === 'vod' ? 'vp-video' : 'clp-video';
const words = (t || segments.map(s => s.text).join(' ')).split(/\s+/).filter(Boolean).length;
const words = (t || segments.map(s => s.text).join(' ')).split(/\s+/).filter(Boolean).length;
@@ -464,13 +464,22 @@ async function generateVodTranscript(vod) {
@@ -464,13 +464,22 @@ async function generateVodTranscript(vod) {
}
}
db.setVodTranscriptStatus(vod.id, 'processing');
db.setVodTranscriptStatus(vod.id, 'processing');
let r = { text: '', segments: [], ok: false, error: 'unknown' };
let r = { text: '', segments: [], ok: false, error: 'unknown' };
- try { r = await require('./media-analysis').transcribeOnly(src); } catch (e) { r = { text: '', segments: [], ok: false, error: e.message }; }
+ try {
+ // Continue from the last finished 5-minute window (persisted per window), so
+ // the frequent deploy restarts no longer discard an hour of decoding.
+ const prog = db.getVodTranscriptProgress ? db.getVodTranscriptProgress(vod.id) : { progressSec: 0, segments: [] };
+ r = await require('./media-analysis').transcribeOnly(src, {
+ resumeFromSec: prog.progressSec,
+ priorSegments: prog.segments,
+ onWindow: (sec, segs) => db.saveVodTranscriptProgress(vod.id, sec, segs),
+ });
+ } catch (e) { r = { text: '', segments: [], ok: false, error: e.message }; }
if (r.text) { // got speech → store it (+segments)
if (r.text) { // got speech → store it (+segments)
try { db.setVodTranscript(vod.id, r.text, r.segments || []); } catch { /* */ }
try { db.setVodTranscript(vod.id, r.text, r.segments || []); } catch { /* */ }
db.setVodTranscriptStatus(vod.id, 'done');
db.setVodTranscriptStatus(vod.id, 'done');
} else if (r.ok) { // ran clean, genuinely no speech → terminal
} else if (r.ok) { // ran clean, genuinely no speech → terminal
try { db.setVodTranscript(vod.id, ' ', []); } catch { /* */ }
try { db.setVodTranscript(vod.id, ' ', []); } catch { /* */ }
- db.setVodTranscriptStatus(vod.id, 'empty');
+ db.setVodTranscriptStatus(vod.id, 'empty', r.noAudio ? 'no audio stream' : null);
} else { // failure → retry (bounded + backoff), never poison
} else { // failure → retry (bounded + backoff), never poison
const n = db.bumpVodTranscriptAttempt(vod.id);
const n = db.bumpVodTranscriptAttempt(vod.id);
db.setVodTranscriptStatus(vod.id, n >= MAX_TX_ATTEMPTS ? 'failed' : 'retry', r.error || 'transcription failed', _txBackoffMin(n));
db.setVodTranscriptStatus(vod.id, n >= MAX_TX_ATTEMPTS ? 'failed' : 'retry', r.error || 'transcription failed', _txBackoffMin(n));
@@ -498,7 +507,7 @@ async function generateClipTranscript(clip) {
@@ -498,7 +507,7 @@ async function generateClipTranscript(clip) {
db.setClipTranscriptStatus(clip.id, 'done');
db.setClipTranscriptStatus(clip.id, 'done');
} else if (r.ok) {
} else if (r.ok) {
try { db.setClipTranscript(clip.id, ' ', []); } catch { /* */ }
try { db.setClipTranscript(clip.id, ' ', []); } catch { /* */ }
- db.setClipTranscriptStatus(clip.id, 'empty');
+ db.setClipTranscriptStatus(clip.id, 'empty', r.noAudio ? 'no audio stream' : null);
} else {
} else {
const n = db.bumpClipTranscriptAttempt(clip.id);
const n = db.bumpClipTranscriptAttempt(clip.id);
db.setClipTranscriptStatus(clip.id, n >= MAX_TX_ATTEMPTS ? 'failed' : 'retry', r.error || 'transcription failed', _txBackoffMin(n));
db.setClipTranscriptStatus(clip.id, n >= MAX_TX_ATTEMPTS ? 'failed' : 'retry', r.error || 'transcription failed', _txBackoffMin(n));
@@ -16,6 +16,7 @@ const media = require('../media-client');
@@ -16,6 +16,7 @@ const media = require('../media-client');
let _timer = null;
let _timer = null;
let _busy = false;
let _busy = false;
+let _loadWarnedAt = 0;
// Resolve Media metadata for a state row ({id} = Media id). Returns a vod/clip-shaped
// Resolve Media metadata for a state row ({id} = Media id). Returns a vod/clip-shaped
// object the ai-analysis functions understand, or null (→ retry later).
// object the ai-analysis functions understand, or null (→ retry later).
@@ -50,23 +51,39 @@ async function tick() {
@@ -50,23 +51,39 @@ async function tick() {
// run whisper at once, on top of the live encoder. Measured effect on this 4-core
// run whisper at once, on top of the live encoder. Measured effect on this 4-core
// box: load average 5.44. VOD transcripts have no deadline whatsoever, so simply
// box: load average 5.44. VOD transcripts have no deadline whatsoever, so simply
// stand them down until the stream ends; they catch up at full speed afterwards.
// stand them down until the stream ends; they catch up at full speed afterwards.
- let timelineBusy = false;
- try { timelineBusy = anyLive && require('./timeline-job').timelineEnabled(); } catch { /* */ }
- if (timelineBusy) {
- try { require('./transcribe').setLowPower(true); } catch { /* */ }
+ // The live timeline has its OWN whisper lane now (transcribe.js), so VOD/clip work
+ // no longer has to stand down for the whole stream — it runs niced, low-power, and
+ // only when the box has headroom. (The old stand-down meant nothing was ever
+ // transcribed while the owner streamed, which on this site is most of the day.)
+ const cores = require('os').cpus().length || 4;
+ const load1 = (require('os').loadavg()[0]) || 0;
+ const headroom = load1 < cores * 1.25;
+ if (!headroom) {
+ if (!_loadWarnedAt || Date.now() - _loadWarnedAt > 600000) { _loadWarnedAt = Date.now(); console.log(`[AI backfill] load ${load1.toFixed(1)} on ${cores} cores — deferring batch transcription`); }
} else if (ai.transcriptionEnabled && ai.transcriptionEnabled()) {
} else if (ai.transcriptionEnabled && ai.transcriptionEnabled()) {
try { require('./transcribe').setLowPower(anyLive); } catch { /* */ }
try { require('./transcribe').setLowPower(anyLive); } catch { /* */ }
const batch = anyLive ? 1 : 2; // throttle while live, catch up faster when idle
const batch = anyLive ? 1 : 2; // throttle while live, catch up faster when idle
+ // Take several candidates and process the first `batch` that resolve: a row whose
+ // Media metadata won't resolve (still recording, Media down) used to occupy the
+ // whole LIMIT every minute and block every row behind it.
try {
try {
- for (const row of db.getVodsNeedingTranscript(batch)) {
+ let done = 0;
+ for (const row of db.getVodsNeedingTranscript(batch + 6)) {
+ if (done >= batch) break;
const vod = await _vodMeta(row);
const vod = await _vodMeta(row);
- if (vod) await ai.generateVodTranscript(vod);
+ if (!vod) continue;
+ await ai.generateVodTranscript(vod);
+ done++;
}
}
} catch (e) { console.warn('[AI backfill] vod transcript:', e.message); }
} catch (e) { console.warn('[AI backfill] vod transcript:', e.message); }
try {
try {
- for (const row of db.getClipsNeedingTranscript(batch)) {
+ let done = 0;
+ for (const row of db.getClipsNeedingTranscript(batch + 6)) {
+ if (done >= batch) break;
const clip = await _clipMeta(row);
const clip = await _clipMeta(row);
- if (clip) await ai.generateClipTranscript(clip);
+ if (!clip) continue;
+ await ai.generateClipTranscript(clip);
+ done++;
}
}
} catch (e) { console.warn('[AI backfill] clip transcript:', e.message); }
} catch (e) { console.warn('[AI backfill] clip transcript:', e.message); }
}
}
@@ -194,7 +194,24 @@ async function pickFrameTimes(src, duration, { existingOffsets = [], maxFrames =
@@ -194,7 +194,24 @@ async function pickFrameTimes(src, duration, { existingOffsets = [], maxFrames =
// Transcribe the audio: full for short media, else 3 spread 60s windows. Returns
// Transcribe the audio: full for short media, else 3 spread 60s windows. Returns
// { text, segments:[{start,end,text}] } with segment times relative to the recording.
// { text, segments:[{start,end,text}] } with segment times relative to the recording.
-async function _transcribeSpan(src, duration) {
+// Does the source carry an audio stream at all? Video-only recordings (screen shares with
+// no mic, some RS robots) used to fail ffmpeg's wav extraction, burn 8 retries and end up
+// 'failed' — they are simply silent and should be terminal 'empty' on the first pass.
+function _ffprobeHasAudio(src) {
+ return new Promise((resolve) => {
+ let out = '';
+ let p;
+ try {
+ p = _track(spawn('ffprobe', ['-v', 'error', '-select_streams', 'a', '-show_entries', 'stream=codec_type', '-of', 'csv=p=0', src], { stdio: ['ignore', 'pipe', 'ignore'] }));
+ } catch { return resolve(null); }
+ const t = setTimeout(() => { try { p.kill('SIGKILL'); } catch { /* */ } resolve(null); }, 30000);
+ p.stdout.on('data', (d) => { out += d; });
+ p.on('close', (c) => { clearTimeout(t); resolve(c === 0 ? /audio/.test(out) : null); });
+ p.on('error', () => { clearTimeout(t); resolve(null); });
+ });
+}
+
+async function _transcribeSpan(src, duration, { resumeFromSec = 0, priorSegments = null, onWindow = null } = {}) {
const transcribe = require('./transcribe');
const transcribe = require('./transcribe');
if (!transcribe.available()) return { text: '', segments: [], ok: false, error: 'whisper unavailable' };
if (!transcribe.available()) return { text: '', segments: [], ok: false, error: 'whisper unavailable' };
if (duration > 0 && duration <= 200) {
if (duration > 0 && duration <= 200) {
@@ -210,11 +227,16 @@ async function _transcribeSpan(src, duration) {
@@ -210,11 +227,16 @@ async function _transcribeSpan(src, duration) {
// pass runs from the backfill queue, which already drops to low power while live.
// pass runs from the backfill queue, which already drops to low power while live.
const WINDOW_SEC = 300;
const WINDOW_SEC = 300;
const parts = [];
const parts = [];
- const segments = [];
- let anyOk = false, lastErr = null;
+ // Resume support: a deploy/restart used to throw away every finished window of an
+ // hour-long VOD. Callers persist progress after each window (onWindow) and hand the
+ // finished segments back (priorSegments/resumeFromSec) so we continue, not restart.
+ const segments = Array.isArray(priorSegments) ? [...priorSegments] : [];
+ let anyOk = segments.length > 0, lastErr = null;
const total = Math.max(0, Math.floor(duration || 0));
const total = Math.max(0, Math.floor(duration || 0));
const windows = [];
const windows = [];
- for (let start = 0; start < total; start += WINDOW_SEC) windows.push(start);
+ const firstStart = Math.max(0, Math.floor((resumeFromSec || 0) / WINDOW_SEC) * WINDOW_SEC);
+ for (let start = firstStart; start < total; start += WINDOW_SEC) windows.push(start);
+ if (firstStart > 0) console.log(`[AI] transcript resuming at ${Math.round(firstStart / 60)}min of ${Math.round(total / 60)}min (${segments.length} segments kept)`);
// Safety valve for absurdly long recordings so one VOD cannot occupy the queue forever.
// 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
const MAX_WINDOWS = Math.max(1, parseInt(process.env.AI_VOD_MAX_WINDOWS, 10) || 96); // 8h
if (windows.length > MAX_WINDOWS) {
if (windows.length > MAX_WINDOWS) {
@@ -228,11 +250,16 @@ async function _transcribeSpan(src, duration) {
@@ -228,11 +250,16 @@ async function _transcribeSpan(src, duration) {
if (ffOk) {
if (ffOk) {
const r = await transcribe.transcribeWavDetailed(wav, { timeoutMs: 600000, 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.segments && r.segments.length) segments.push(...r.segments);
if (r.segments && r.segments.length) segments.push(...r.segments);
+ if (r.ok && typeof onWindow === 'function') {
+ try { await onWindow(Math.min(total, start + len), segments); } catch { /* progress is best-effort */ }
+ }
} else { lastErr = 'ffmpeg window extract failed'; }
} else { lastErr = 'ffmpeg window extract failed'; }
try { fs.existsSync(wav) && fs.unlinkSync(wav); } catch { /* */ }
try { fs.existsSync(wav) && fs.unlinkSync(wav); } catch { /* */ }
}
}
+ segments.sort((a, b) => (a.start || 0) - (b.start || 0));
+ parts.length = 0;
+ for (const s of segments) if (s.text) parts.push(s.text);
// 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.
@@ -280,10 +307,12 @@ async function analyzeMedia(src, { streamId = null, userId = null, numFrames = n
@@ -280,10 +307,12 @@ async function analyzeMedia(src, { streamId = null, userId = null, numFrames = n
* Transcript-only: probe duration then whisper-transcribe (FREE local, no vision).
* Transcript-only: probe duration then whisper-transcribe (FREE local, no vision).
* @returns {Promise<{text:string, segments:Array}>}
* @returns {Promise<{text:string, segments:Array}>}
*/
*/
-async function transcribeOnly(src) {
+async function transcribeOnly(src, opts = {}) {
if (!src) return { text: '', segments: [], ok: false, error: 'no source' };
if (!src) return { text: '', segments: [], ok: false, error: 'no source' };
+ const hasAudio = await _ffprobeHasAudio(src);
+ if (hasAudio === false) return { text: '', segments: [], ok: true, noAudio: true, error: null };
const duration = await _ffprobeDuration(src);
const duration = await _ffprobeDuration(src);
- try { return await _transcribeSpan(src, duration); } catch (e) { return { text: '', segments: [], ok: false, error: e.message }; }
+ try { return await _transcribeSpan(src, duration, opts); } catch (e) { return { text: '', segments: [], ok: false, error: e.message }; }
}
}
module.exports = {
module.exports = {
@@ -250,6 +250,15 @@ async function startContinuousCapture(stream) {
@@ -250,6 +250,15 @@ async function startContinuousCapture(stream) {
const dir = spoolDir(stream.id);
const dir = spoolDir(stream.id);
try { fs.mkdirSync(dir, { recursive: true }); } catch { return false; }
try { fs.mkdirSync(dir, { recursive: true }); } catch { return false; }
+ // A previous process (deploy restart) may have left segments here. Their indexes
+ // restart at 0 for the new ffmpeg, so old files would be mis-ordered against new ones
+ // — the "newest file is still being written" rule then skipped a stale file and fed a
+ // half-written one to whisper. Drop them; the VOD transcript covers that audio.
+ try {
+ const stale = fs.readdirSync(dir).filter(n => /^seg-\d{6}\.wav$/.test(n));
+ for (const n of stale) { try { fs.unlinkSync(path.join(dir, n)); } catch { /* */ } }
+ if (stale.length) console.log(`[AI-Hear] stream ${stream.id}: dropped ${stale.length} stale spool segment(s) from a previous run`);
+ } catch { /* */ }
// Resolve an input the same way the one-shot grab does: WHIP/WebRTC via a plain RTP
// 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.
// consumer, RTMP via the local HTTP-FLV output.
@@ -328,6 +337,7 @@ function pendingSegments(streamId, { includeLast = false } = {}) {
@@ -328,6 +337,7 @@ function pendingSegments(streamId, { includeLast = false } = {}) {
for (let i = 0; i < overflow; i++) {
for (let i = 0; i < overflow; i++) {
try { fs.unlinkSync(path.join(dir, ready[i])); } catch { /* */ }
try { fs.unlinkSync(path.join(dir, ready[i])); } catch { /* */ }
}
}
+ if (overflow) console.warn(`[AI-Hear] stream ${streamId}: transcriber ${overflow * SEGMENT_SEC}s behind — dropped ${overflow} oldest segment(s)`);
const startedAt = (_capturing.get(streamId) || {}).startedAt || null;
const startedAt = (_capturing.get(streamId) || {}).startedAt || null;
return ready.slice(overflow).map(name => {
return ready.slice(overflow).map(name => {
const index = parseInt(name.slice(4, 10), 10);
const index = parseInt(name.slice(4, 10), 10);
@@ -102,23 +102,40 @@ function killActive() {
@@ -102,23 +102,40 @@ function killActive() {
//
//
// Put the limit HERE, where every caller must pass, so no future path can bypass it.
// Put the limit HERE, where every caller must pass, so no future path can bypass it.
const MAX_CONCURRENT = Math.max(1, parseInt(process.env.WHISPER_MAX_CONCURRENT, 10) || 1);
const MAX_CONCURRENT = Math.max(1, parseInt(process.env.WHISPER_MAX_CONCURRENT, 10) || 1);
-let _running = 0;
-const _waiters = [];
-function _acquire() {
- if (_running < MAX_CONCURRENT) { _running++; return Promise.resolve(); }
- return new Promise(resolve => _waiters.push(resolve));
+// Two INDEPENDENT lanes. The live timeline decodes 10s segments with the small model and
+// must keep up with real time; VOD/clip backfill decodes hour-long recordings with the
+// large model. When they shared one slot, a single VOD job (minutes to hours, and killed
+// + restarted from scratch on every deploy) held the slot and the live transcript simply
+// stopped — segments piled up in the spool and were discarded. Live never waits on batch.
+const _lanes = {
+ live: { max: 1, running: 0, waiters: [] },
+ batch: { max: MAX_CONCURRENT, running: 0, waiters: [] },
+};
+function _acquire(lane = 'batch') {
+ const L = _lanes[lane] || _lanes.batch;
+ if (L.running < L.max) { L.running++; return Promise.resolve(); }
+ return new Promise(resolve => L.waiters.push(resolve));
}
}
-function _release() {
- const next = _waiters.shift();
+function _release(lane = 'batch') {
+ const L = _lanes[lane] || _lanes.batch;
+ const next = L.waiters.shift();
if (next) next(); // hand the slot straight over
if (next) next(); // hand the slot straight over
- else _running = Math.max(0, _running - 1);
+ else L.running = Math.max(0, L.running - 1);
+}
+function laneStatus() {
+ return Object.fromEntries(Object.entries(_lanes).map(([k, L]) => [k, { running: L.running, queued: L.waiters.length }]));
}
}
// While a stream is live we lower the whisper thread count so VOD transcription can
// While a stream is live we lower the whisper thread count so VOD transcription can
// still make progress without starving the live encoders.
// still make progress without starving the live encoders.
let _lowPower = false;
let _lowPower = false;
function setLowPower(v) { _lowPower = !!v; }
function setLowPower(v) { _lowPower = !!v; }
-function _threads() { return _lowPower ? Math.max(1, Math.min(2, THREADS)) : THREADS; }
+function _threads(live = false) {
+ // Live segments are short and use the small model: 2 threads keeps them well ahead of
+ // real time while leaving cores for the encoders and the (niced) batch lane.
+ if (live) return Math.max(1, Math.min(2, 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
// Always-drop: whisper's stock inventions over non-speech. These are never worth keeping
@@ -194,7 +211,7 @@ function _transcribeWavInner(wavPath, { timeoutMs = 180000, offsetSec = 0, live
@@ -194,7 +211,7 @@ function _transcribeWavInner(wavPath, { timeoutMs = 180000, offsetSec = 0, live
}
}
const outBase = `${wavPath}.out`;
const outBase = `${wavPath}.out`;
const jsonPath = `${outBase}.json`;
const jsonPath = `${outBase}.json`;
- const args = ['-m', _modelFor({ live }), '-f', wavPath, '-oj', '-of', outBase, '-t', String(_threads()), '-l', 'en'];
+ const args = ['-m', _modelFor({ live }), '-f', wavPath, '-oj', '-of', outBase, '-t', String(_threads(live)), '-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
// 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.
// absolute file time, so the {start,end} contract is unchanged for every caller.
@@ -278,9 +295,23 @@ async function transcribeMedia(mediaPath, opts = {}) {
@@ -278,9 +295,23 @@ async function transcribeMedia(mediaPath, opts = {}) {
* The slot is held for the whole decode and always released, including on throw.
* The slot is held for the whole decode and always released, including on throw.
*/
*/
async function transcribeWavDetailed(wavPath, opts = {}) {
async function transcribeWavDetailed(wavPath, opts = {}) {
- await _acquire();
+ const lane = opts && opts.live ? 'live' : 'batch';
+ await _acquire(lane);
try { return await _transcribeWavInner(wavPath, opts); }
try { return await _transcribeWavInner(wavPath, opts); }
- finally { _release(); }
+ finally { _release(lane); }
+}
+
+/** Human-readable availability report (logged at boot so a broken install is visible). */
+function describe() {
+ const bin = whisperBin();
+ return {
+ available: available(),
+ bin: bin || null,
+ model: MODEL, modelExists: fs.existsSync(MODEL),
+ modelLive: MODEL_LIVE, modelLiveExists: fs.existsSync(MODEL_LIVE),
+ vadModel: vadModel() || null,
+ threads: THREADS, beam: BEAM, maxConcurrent: MAX_CONCURRENT,
+ };
}
}
-module.exports = { available, transcribeWav, transcribeWavDetailed, transcribeMedia, transcribeMediaDetailed, killActive, setLowPower };
+module.exports = { available, describe, laneStatus, transcribeWav, transcribeWavDetailed, transcribeMedia, transcribeMediaDetailed, killActive, setLowPower };
@@ -278,6 +278,28 @@ function initDb() {
@@ -278,6 +278,28 @@ function initDb() {
}
}
} catch (e) { console.warn('[DB] ai-overview column migration:', e.message); }
} catch (e) { console.warn('[DB] ai-overview column migration:', e.message); }
+ // Transcript job-state recovery — on the tables that actually hold it. The older
+ // recovery loop above targets the legacy local `vods`/`clips` tables, which moved to
+ // OpenVibe.Media; it throws on the first PRAGMA and silently does nothing, which is
+ // why rows killed mid-flight by a deploy stayed 'processing' forever and 'failed' was
+ // permanent. Also adds resumable-progress columns (finished windows survive restarts).
+ try {
+ for (const t of ['vod_ai_state', 'clip_ai_state']) {
+ const cols = database.prepare(`PRAGMA table_info(${t})`).all().map(c => c.name);
+ if (!cols.includes('transcript_partial_json')) database.exec(`ALTER TABLE ${t} ADD COLUMN transcript_partial_json TEXT`);
+ if (!cols.includes('transcript_progress_sec')) database.exec(`ALTER TABLE ${t} ADD COLUMN transcript_progress_sec INTEGER DEFAULT 0`);
+ // Mid-flight at shutdown → back to the queue (progress is kept, attempts untouched).
+ const p = database.prepare(`UPDATE ${t} SET transcript_status='retry', transcript_next_at=NULL WHERE transcript_status='processing'`).run();
+ // Exhausted retries get a fresh ladder after a deploy — except sources that can
+ // never work (Media reported the recording itself failed / no audio stream).
+ const f = database.prepare(`UPDATE ${t} SET transcript_status='retry', transcript_attempts=0, transcript_next_at=NULL
+ WHERE transcript_status='failed'
+ AND COALESCE(transcript_error,'') NOT LIKE 'media reported%'
+ AND COALESCE(transcript_error,'') NOT LIKE 'no audio stream%'`).run();
+ if (p.changes || f.changes) console.log(`[DB] ${t}: re-queued ${p.changes} interrupted + ${f.changes} previously-failed transcript job(s)`);
+ }
+ } catch (e) { console.warn('[DB] transcript recovery:', e.message); }
+
// ── Migrations ────────────────────────────────────────────
// ── Migrations ────────────────────────────────────────────
try {
try {
const cols = database.prepare("PRAGMA table_info('channels')").all().map(c => c.name);
const cols = database.prepare("PRAGMA table_info('channels')").all().map(c => c.name);
@@ -1417,7 +1439,7 @@ function initDb() {
@@ -1417,7 +1439,7 @@ function initDb() {
WHERE transcript_status IS NULL AND ai_transcript = ' ' AND created_at >= datetime('now','-60 days')`);
WHERE transcript_status IS NULL AND ai_transcript = ' ' AND created_at >= datetime('now','-60 days')`);
// Older poisoned items: assume genuinely silent (terminal) so we don't churn the whole archive.
// Older poisoned items: assume genuinely silent (terminal) so we don't churn the whole archive.
database.exec(`UPDATE ${t} SET transcript_status='empty' WHERE transcript_status IS NULL AND ai_transcript = ' '`);
database.exec(`UPDATE ${t} SET transcript_status='empty' WHERE transcript_status IS NULL AND ai_transcript = ' '`);
- // Crash/deploy recovery: anything left mid-flight is re-queued on every boot.
+ // (Legacy local tables — see the vod_ai_state/clip_ai_state recovery below.)
database.exec(`UPDATE ${t} SET transcript_status=NULL, transcript_next_at=NULL WHERE transcript_status='processing'`);
database.exec(`UPDATE ${t} SET transcript_status=NULL, transcript_next_at=NULL WHERE transcript_status='processing'`);
// Give previously-'failed' rows a fresh chance after a deploy — transient issues
// Give previously-'failed' rows a fresh chance after a deploy — transient issues
// (whisper temporarily missing, storage hiccup, a bad build) shouldn't be permanent.
// (whisper temporarily missing, storage hiccup, a bad build) shouldn't be permanent.
@@ -2823,7 +2845,19 @@ function _segJson(segments) {
@@ -2823,7 +2845,19 @@ function _segJson(segments) {
}
}
function setVodTranscript(vodId, transcript, segments) {
function setVodTranscript(vodId, transcript, segments) {
_ensureVodAiState(vodId);
_ensureVodAiState(vodId);
- return run('UPDATE vod_ai_state SET ai_transcript_json = ? WHERE vod_id = ?', [_segJson(segments) ?? (transcript ? JSON.stringify([]) : null), vodId]);
+ return run('UPDATE vod_ai_state SET ai_transcript_json = ?, transcript_partial_json = NULL, transcript_progress_sec = 0 WHERE vod_id = ?', [_segJson(segments) ?? (transcript ? JSON.stringify([]) : null), vodId]);
+}
+// Resumable VOD transcription: persist finished windows so a restart continues from here.
+function saveVodTranscriptProgress(vodId, progressSec, segments) {
+ _ensureVodAiState(vodId);
+ return run('UPDATE vod_ai_state SET transcript_partial_json = ?, transcript_progress_sec = ? WHERE vod_id = ?', [_segJson(segments), Math.max(0, Math.floor(progressSec || 0)), vodId]);
+}
+function getVodTranscriptProgress(vodId) {
+ const row = get('SELECT transcript_partial_json, transcript_progress_sec FROM vod_ai_state WHERE vod_id = ?', [vodId]);
+ if (!row) return { progressSec: 0, segments: [] };
+ let segments = [];
+ try { segments = row.transcript_partial_json ? JSON.parse(row.transcript_partial_json) : []; } catch { segments = []; }
+ return { progressSec: row.transcript_progress_sec || 0, segments: Array.isArray(segments) ? segments : [] };
}
}
function setClipTranscript(clipId, transcript, segments) {
function setClipTranscript(clipId, transcript, segments) {
_ensureClipAiState(clipId);
_ensureClipAiState(clipId);
@@ -7743,6 +7777,7 @@ module.exports = {
@@ -7743,6 +7777,7 @@ module.exports = {
updateChannelAiBot, touchChannelAiBot, deleteChannelAiBot,
updateChannelAiBot, touchChannelAiBot, deleteChannelAiBot,
// Site Settings
// Site Settings
getSetting, getSettingRow, getAllSettings, setSetting, deleteSetting,
getSetting, getSettingRow, getAllSettings, setSetting, deleteSetting,
+ saveVodTranscriptProgress, getVodTranscriptProgress,
getState, setState, deleteState,
getState, setState, deleteState,
// Verification Keys
// Verification Keys
createVerificationKey, getVerificationKeyByKey, getVerificationKeyByUsername,
createVerificationKey, getVerificationKeyByKey, getVerificationKeyByUsername,
@@ -976,7 +976,8 @@ async function start() {
@@ -976,7 +976,8 @@ async function start() {
// AI stream-memory job (no-op until enabled in openvibe.network/admin → AI).
// AI stream-memory job (no-op until enabled in openvibe.network/admin → AI).
try { require('./ai/stream-memory-job').start(); } catch (e) { console.warn('[AI] memory job not started:', e.message); }
try { require('./ai/stream-memory-job').start(); } catch (e) { console.warn('[AI] memory job not started:', e.message); }
- try { require('./ai/backfill-job').start(); } catch (e) { console.warn('[AI] backfill job not started:', e.message); }
+ try { (() => { try { const d = require('./ai/transcribe').describe(); console.log(`[AI] whisper ${d.available ? 'available' : 'UNAVAILABLE'} — bin=${d.bin} model=${d.model}${d.modelExists ? '' : ' (MISSING)'} live=${d.modelLive}${d.modelLiveExists ? '' : ' (MISSING)'} vad=${d.vadModel || 'off'} threads=${d.threads}`); } catch (e) { console.warn('[AI] whisper probe failed:', e.message); } })();
+ require('./ai/backfill-job').start(); } catch (e) { console.warn('[AI] backfill job not started:', e.message); }
try { require('./ai/streamer-overview-job').start(); } catch (e) { console.warn('[AI] streamer-overview job not started:', e.message); }
try { require('./ai/streamer-overview-job').start(); } catch (e) { console.warn('[AI] streamer-overview job not started:', e.message); }
try { require('./ai/chat-ai').start(); } catch (e) { console.warn('[AI] chat-ai job not started:', e.message); }
try { require('./ai/chat-ai').start(); } catch (e) { console.warn('[AI] chat-ai job not started:', e.message); }
try { require('./ai/slogan-job').start(); } catch (e) { console.warn('[AI] slogan job not started:', e.message); }
try { require('./ai/slogan-job').start(); } catch (e) { console.warn('[AI] slogan job not started:', e.message); }