openvibers/openvibe.live Commits
Live: LIVE_MEDIA_ANALYSIS=ai sends finished VODs and clips to OpenVibe.AI's media.analyze (no local ffmpeg+whisper for them) (#111)
Changes
@@@ -21,10 +21,14 @@ ADMIN_PASSWORD=changeme123# service, with a service token for audience openvibe.ai (grants ai.run.create, ai.run.read and# ai.credential.manage for client `live`). Provider keys, a streamer's own included, live in the AI# service; Live calls a provider only to test a key a streamer typed and has not saved yet. The admin--# ai_enabled switch and the local daily cap still apply; local whisper.cpp transcription stays here.++# ai_enabled switch and the local daily cap still apply to other AI workflows. Live stream transcription stays here.# AI_SERVICE=off turns AI off entirely.#AI_SERVICE=off#OV_AI_INTERNAL_URL=http://127.0.0.1:4700++# Finished VOD/clip analysis: local (current frame/whisper path) or ai (one media.analyze run).++#LIVE_MEDIA_ANALYSIS=local++# How long one call waits for media.analyze before re-joining the run 15 minutes later (ms).++#LIVE_MEDIA_ANALYSIS_WAIT_MS=60000# ServerPORT=3000@@@ -1,5 +1,20 @@# VODs & Clips++## Finished media analysis++++`LIVE_MEDIA_ANALYSIS=local` is the default: Live's existing frame and transcript jobs produce++VOD and clip overviews and timed transcripts. Set `LIVE_MEDIA_ANALYSIS=ai` to send each finished++VOD or clip's signed Media URL to OpenVibe.AI's `media.analyze` workflow (AI runs one analysis at a time at nice 15).++Live fills a missing overview (an overview it already has is kept), derives the short overview, and stores timed speech++segments in `vod_ai_state` or `clip_ai_state`. Live waits up to a minute for the run++(`LIVE_MEDIA_ANALYSIS_WAIT_MS`, 60000); a longer analysis is re-joined 15 minutes later by its idempotency key++(`live.media.analyze:<kind>:<id>:<attempt>`), so the same recording is never analysed twice at once and no loop waits++for it.++AI's signals, scenes, and highlights have no Live state columns and are not stored. Failed or++unavailable analyses retry with backoff; Live does not run its VOD/clip decoder in this mode.++Live stream captions and sound events continue using their real-time paths. The++`live_media_analysis_retry_total` metric counts deferred analyses by media kind.++> Written before the media split. Storage, cutting, thumbnails and files now live in OpenVibe.Media;> Live proxies to it (`server/media-proxy/`) and starts/stops recordings (`server/streaming/recorder.js`).> Pastes live in OpenVibe.Community. Live's old local `vods`/`clips`/`pastes` tables are frozen: nothing@@@ -160,6 +160,7 @@ async function _mediaSource(row, kind = 'vod') {const _overviewInFlight = new Set();async function generateVodOverview(vod) {if (!vod) return null;++ if (require('./ai-media-analyze').enabled()) return require('./ai-media-analyze').analyze('vod', vod);if (!await isEnabled() || !await withinBudget()) return null;// Guard against the on-finalize trigger and the backfill poller processing the same// VOD at once (that would double-extract frames + duplicate timeline memories).@@@ -218,6 +219,7 @@ async function _generateVodOverviewInner(vod) {*/async function ensureVodTimeline(vod) {if (!vod || !vod.stream_id) return;++ if (require('./ai-media-analyze').enabled()) return;if (!await isEnabled() || !await withinBudget()) return;const ma = require('./media-analysis');const src = await _mediaSource(vod);@@@ -265,6 +267,7 @@ async function _applyTxLowPower() {* a clean silent run is marked terminal — so an interrupted run is never lost.*/async function generateVodTranscript(vod) {++ if (require('./ai-media-analyze').enabled()) return require('./ai-media-analyze').analyze('vod', vod);if (!vod || !await transcriptionEnabled()) return null;return _txRun(async () => {await _applyTxLowPower();@@@ -302,6 +305,7 @@ async function generateVodTranscript(vod) {/** Transcript-only pass for a clip — FREE local whisper (see generateVodTranscript). */async function generateClipTranscript(clip) {++ if (require('./ai-media-analyze').enabled()) return require('./ai-media-analyze').analyze('clip', clip);if (!clip || !await transcriptionEnabled()) return null;return _txRun(async () => {await _applyTxLowPower();@@@ -335,6 +339,7 @@ async function generateClipTranscript(clip) {*/async function generateClipOverview(clip) {if (!clip) return null;++ if (require('./ai-media-analyze').enabled()) return require('./ai-media-analyze').analyze('clip', clip);if (!await isEnabled() || !await withinBudget()) return null;const src = await _mediaSource(clip, 'clip');if (!src) { try { await db.setClipAiOverview(clip.id, { overview: ' ', transcript: null }); } catch { /* */ } return null; }nnew file mode 100644@@@ -0,0 +1,100 @@++'use strict';++++// Finished Media recordings are analysed as one OpenVibe.AI workflow run. This++// module never decodes media on Live; the signed URL also works for private items.++const db = require('../db/database');++const media = require('../media-client');++const aiService = require('./ai-service');++++const inFlight = new Set();++let failures = 0;++let failureCounter = null;++++function enabled() { return String(process.env.LIVE_MEDIA_ANALYSIS || 'local').toLowerCase() === 'ai'; }++function registerMetrics(registry) {++ if (registry && typeof registry.counter === 'function') {++ failureCounter = registry.counter({ name: 'live_media_analysis_retry_total', help: 'Finished media analyses deferred after an OpenVibe.AI failure', labelNames: ['kind'] });++ }++}++function retryMinutes(attempt) { return Math.min(360, 2 ** Math.min(8, Math.max(0, attempt - 1))); }++++async function source(kind, item) {++ const signed = await media.signedMediaUrl(kind, item.id, 6 * 3600).catch(() => null);++ if (signed && signed.url) return signed.url;++ const meta = kind === 'vod' ? await media.getVod(item.id).catch(() => null) : await media.getClip(item.id).catch(() => null);++ if (meta && meta.playback_url) return media.publicUrl(meta.playback_url);++ return kind === 'vod' ? media.vodPlaybackUrl(item.id) : media.clipUrl(item.id);++}++++async function analyze(kind, item) {++ if (!item || !item.id || !['vod', 'clip'].includes(kind)) return null;++ const key = `${kind}:${item.id}`;++ if (inFlight.has(key)) return null;++ inFlight.add(key);++ const bump = kind === 'vod' ? db.bumpVodTranscriptAttempt : db.bumpClipTranscriptAttempt;++ const status = kind === 'vod' ? db.setVodTranscriptStatus : db.setClipTranscriptStatus;++ try {++ const state = kind === 'vod' ? await db.getVodAiState(item.id) : await db.getClipAiState(item.id);++ const mediaUrl = await source(kind, item);++ if (!mediaUrl) throw new Error('Media did not provide a readable URL');++ // One AI run per attempt: a retry after a failure starts a new run, while a run still going when the poll++ // budget ends is re-joined later by the same key (never a second analysis of the same recording at once).++ const attemptNo = Number(state && state.transcript_attempts) || 0;++ const run = await aiService.run('media.analyze', { media_url: mediaUrl, language: 'en', allow_paid: false }, {++ target: { service: 'live', type: kind, id: String(item.id) },++ attribution: aiService.ownerRef(item.user_id),++ idempotencyKey: `live.media.analyze:${kind}:${item.id}:${attemptNo}`,++ // Up to a minute (a clip is usually done by then), never a poll loop: a longer analysis is re-joined++ // in 15 minutes by its key, so the backfill loop is never held for the length of a recording.++ waitMs: Math.max(0, Number(process.env.LIVE_MEDIA_ANALYSIS_WAIT_MS ?? 60000) || 0), pollMs: 0,++ });++ if (run && (run.status === 'queued' || run.status === 'running')) {++ await status(item.id, 'retry', 'OpenVibe.AI is still analysing it', 15);++ return null;++ }++ const out = aiService.usable(run);++ if (!out || !out.overview || typeof out.overview.text !== 'string' || !out.overview.text.trim() || !out.transcript || !Array.isArray(out.transcript.segments)) {++ throw new Error((run && run.error && (run.error.code || run.error.detail)) || (run && run.status) || 'no usable media.analyze result');++ }++ // The state setters derive ai_overview_short and store segments in the++ // existing [{start,end,text}] JSON format. AI signals/highlights have no++ // corresponding state columns; keep all other Live state untouched.++ const overview = out.overview.text.trim();++ const segments = out.transcript.segments.map(g => ({ start: g.start, end: g.end, text: g.text }));++ // An overview Live already has (a model's, or a person's edit) is kept: AI's fills only a missing one.++ if (!state || !String(state.ai_overview || '').trim()) {++ if (kind === 'vod') await db.setVodAiOverview(item.id, overview);++ else await db.setClipAiOverview(item.id, { overview });++ }++ const partialSpeech = Array.isArray(out.gaps) && out.gaps.some(g => /^Speech-to-text failed/i.test(String(g)));++ if (out.transcript.available && !partialSpeech) {++ const hasSpeech = !!(segments.length || String(out.transcript.text || '').trim());++ if (hasSpeech || !state || !state.ai_transcript_json) {++ if (kind === 'vod') await db.setVodTranscript(item.id, out.transcript.text || ' ', segments);++ else await db.setClipTranscript(item.id, out.transcript.text || ' ', segments);++ }++ await status(item.id, hasSpeech || (state && state.ai_transcript_json) ? 'done' : 'empty');++ } else if (out.streams && out.streams.audio === false) {++ if (!state || !state.ai_transcript_json) {++ if (kind === 'vod') await db.setVodTranscript(item.id, ' ', []);++ else await db.setClipTranscript(item.id, ' ', []);++ }++ await status(item.id, state && state.ai_transcript_json ? 'done' : 'empty', 'no audio stream');++ } else if (!state || !state.ai_transcript_json) {++ const attempt = await bump(item.id);++ await status(item.id, 'retry', partialSpeech ? 'AI speech-to-text incomplete' : 'AI speech-to-text unavailable', retryMinutes(attempt));++ }++ return out;++ } catch (e) {++ const attempt = await bump(item.id);++ await status(item.id, 'retry', e.message, retryMinutes(attempt));++ failures++;++ if (failureCounter) failureCounter.inc({ kind });++ console.warn(`[AI media] ${kind} ${item.id}: ${e.message}; retry in ${retryMinutes(attempt)} min`);++ return null;++ } finally {++ inFlight.delete(key);++ }++}++++module.exports = { enabled, analyze, registerMetrics, stats: () => ({ failures }) };@@@ -13,6 +13,7 @@const db = require('../db/database');const ai = require('./ai-analysis');const media = require('../media-client');++const remote = require('./ai-media-analyze');let _timer = null;let _busy = false;@@@ -40,54 +41,68 @@ async function tick() {if (_busy) return;_busy = true;try {-- // Transcripts — FREE local whisper (no API/cost), so they run independent of-- // the AI master switch. We DON'T stop while streams are live — instead we drop-- // to low-power (fewer whisper threads) and a smaller batch so we never starve-- // the live encoders.let anyLive = false;try { anyLive = ((db.getLiveStreams && await db.getLiveStreams()) || []).length > 0; } catch { /* */ }-- // When the continuous timeline is running it is already decoding live audio-- // constantly, and VOD backfill does NOT share its serialisation — so both would-- // 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.-- // 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 && await 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 {-- let done = 0;-- for (const row of await db.getVodsNeedingTranscript(batch + 6)) {-- if (done >= batch) break;-- const vod = await _vodMeta(row);-- if (!vod) continue;-- await ai.generateVodTranscript(vod);-- done++;-- }-- } catch (e) { console.warn('[AI backfill] vod transcript:', e.message); }-- try {-- let done = 0;-- for (const row of await db.getClipsNeedingTranscript(batch + 6)) {-- if (done >= batch) break;-- const clip = await _clipMeta(row);-- if (!clip) continue;-- await ai.generateClipTranscript(clip);-- done++;-- }-- } catch (e) { console.warn('[AI backfill] clip transcript:', e.message); }++ if (remote.enabled()) {++ for (const [kind, metadata] of [['vod', _vodMeta], ['clip', _clipMeta]]) {++ try {++ for (const row of await db.getMediaNeedingAiAnalysis(kind, 6)) {++ const item = await metadata(row);++ if (!item) continue;++ await remote.analyze(kind, item);++ break;++ }++ } catch (e) { console.warn(`[AI backfill] ${kind} media.analyze:`, e.message); }++ }}++ if (!remote.enabled()) {++ // Transcripts — FREE local whisper (no API/cost), so they run independent of++ // the AI master switch. We DON'T stop while streams are live — instead we drop++ // to low-power (fewer whisper threads) and a smaller batch so we never starve++ // the live encoders.++ // When the continuous timeline is running it is already decoding live audio++ // constantly, and VOD backfill does NOT share its serialisation — so both would++ // 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.++ // 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 && await 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 {++ let done = 0;++ for (const row of await db.getVodsNeedingTranscript(batch + 6)) {++ if (done >= batch) break;++ const vod = await _vodMeta(row);++ if (!vod) continue;++ await ai.generateVodTranscript(vod);++ done++;++ }++ } catch (e) { console.warn('[AI backfill] vod transcript:', e.message); }++ try {++ let done = 0;++ for (const row of await db.getClipsNeedingTranscript(batch + 6)) {++ if (done >= batch) break;++ const clip = await _clipMeta(row);++ if (!clip) continue;++ await ai.generateClipTranscript(clip);++ done++;++ }++ } catch (e) { console.warn('[AI backfill] clip transcript:', e.message); }++ }++ }if (!await ai.isEnabled()) return;// Paste summaries. Pastes moved to Media at the migration, but the analysis job@@@ -119,6 +134,8 @@ async function tick() {} catch (e) { console.warn('[AI backfill] paste:', e.message); }}++ if (remote.enabled()) return;++// VOD overviews (frames + audio pulled from the Media playback URL; self-marks).try {// Take a few candidates and process the first usable one: the newest row is@@@ -627,6 +627,18 @@ async function getClipsNeedingTranscript(limit = 2) {AND (transcript_next_at IS NULL OR transcript_next_at <= ov_now())ORDER BY (transcript_status='retry'), clip_id DESC LIMIT ?`, [limit]);}++// In AI mode one workflow fills both fields. The transcript retry schedule also++// throttles failed overview attempts, including rows with an older transcript.++async function getMediaNeedingAiAnalysis(kind, limit = 6) {++ const clip = kind === 'clip';++ const table = clip ? 'clip_ai_state' : 'vod_ai_state';++ const id = clip ? 'clip_id' : 'vod_id';++ return await all(`SELECT ${id} AS id, s.* FROM ${table} s++ WHERE (ai_overview IS NULL OR ai_overview = '' OR ai_transcript_json IS NULL)++ AND (transcript_next_at IS NULL OR transcript_next_at <= ov_now())++ AND (transcript_status IS NULL OR transcript_status <> 'processing')++ ORDER BY ${id} DESC LIMIT ?`, [limit]);++}// status setter. On a 'retry', pass retryDelayMin to schedule the next eligible attempt// (exponential backoff); any other status clears the schedule.async function setVodTranscriptStatus(id, status, error = null, retryDelayMin = 0) {@@@ -3874,7 +3886,7 @@ module.exports = {createStream, endStream, onStreamLifecycle, endOtherLiveStreamsForSlot, updateViewerCount,addStreamMemory, getStreamMemories, getLatestStreamMemory, updateStreamAiOverview,setVodAiOverview, setClipAiOverview, setVodTranscript, setClipTranscript, getStreamMemoriesInRange,-- getVodsNeedingOverview, getClipsNeedingOverview, getVodsNeedingTimeline, getVodsNeedingTranscript, getClipsNeedingTranscript,++ getVodsNeedingOverview, getClipsNeedingOverview, getVodsNeedingTimeline, getVodsNeedingTranscript, getClipsNeedingTranscript, getMediaNeedingAiAnalysis,setVodTranscriptStatus, setClipTranscriptStatus, bumpVodTranscriptAttempt, bumpClipTranscriptAttempt,cleanupMalformedAiText, recordAiUsage, getAiCostToday, getAiCostTodayForUser, getAiUsageSummary,getStreamMemoriesByUser, countStreamMemoriesByUser, getStreamTranscriptSegments,@@@ -244,6 +244,7 @@ function releaseEnv() {// answers direct loopback callers only; through nginx it is a 404 (server/web/observability.js).const observability = require('./web/observability');const { registry: metricsRegistry } = observability.mountMetrics(app, { release });++if (!drill.enabled) require('./ai/ai-media-analyze').registerMetrics(metricsRegistry);const vibeCodingPublishServer = new VibeCodingPublishServer(db);function normalizeOrigin(origin) {@@@ -112,8 +112,12 @@ async function apply(event, data) {try {const ai = require('../ai/ai-analysis');const vodMeta = { id: vodId, ...data };-- if (ai.transcriptionEnabled && await ai.transcriptionEnabled()) ai.generateVodTranscript(vodMeta).catch(() => {});-- if (ai.isEnabled && await ai.isEnabled() && ai.withinBudget && await ai.withinBudget()) ai.generateVodOverview(vodMeta).catch(() => {});++ const remote = require('../ai/ai-media-analyze');++ if (remote.enabled()) remote.analyze('vod', vodMeta).catch(() => {});++ else {++ if (ai.transcriptionEnabled && await ai.transcriptionEnabled()) ai.generateVodTranscript(vodMeta).catch(() => {});++ if (ai.isEnabled && await ai.isEnabled() && ai.withinBudget && await ai.withinBudget()) ai.generateVodOverview(vodMeta).catch(() => {});++ }} catch { /* backfill poller will pick it up */ }console.log(`[MediaOutcome] VOD ${vodId} ready (${data.duration || data.duration_seconds || '?'}s)`);},@@@ -141,7 +145,9 @@ async function apply(event, data) {after: async () => {try {const ai = require('../ai/ai-analysis');-- if (ai.isEnabled && await ai.isEnabled() && ai.withinBudget && await ai.withinBudget()) {++ const remote = require('../ai/ai-media-analyze');++ if (remote.enabled()) remote.analyze('clip', { id: data.id, ...data }).catch(() => {});++ else if (ai.isEnabled && await ai.isEnabled() && ai.withinBudget && await ai.withinBudget()) {ai.generateClipOverview({ id: data.id, ...data }).catch(() => {});}} catch { /* */ }nnew file mode 100644@@@ -0,0 +1,158 @@++'use strict';++const assert = require('assert');++const http = require('http');++const db = require('../server/db/database');++const media = require('../server/media-client');++const principal = require('../server/net/network-principal');++const local = require('../server/ai/ai-analysis');++const remote = require('../server/ai/ai-media-analyze');++const backfill = require('../server/ai/backfill-job');++++process.env.LIVE_MEDIA_ANALYSIS = 'ai';++let calls = [];++let refuse = false;++let stillRunning = false;++function answer(body) {++ calls.push(body);++ if (refuse) return { status: 503, body: { code: 'media.no_space', detail: 'disk is full' } };++ if (stillRunning) return { status: 202, body: { run: { id: `run_${calls.length}`, status: 'running' } } };++ return { status: 201, body: { run: {++ id: `run_${calls.length}`, status: 'succeeded', synthetic: false,++ output: {++ overview: { text: 'A creator explains the project. Then the stream ends.', source: 'local_model' },++ transcript: { available: true, language: 'en', text: 'Welcome everyone.', segments: [++ { start: 1.25, end: 2.8, text: 'Welcome' },++ { start: 2.9, end: 4.1, text: 'everyone.' },++ ] },++ streams: { video: true, audio: true },++ highlights: [{ start: 1, end: 5, score: 2, reasons: ['speech'], excerpt: 'Welcome everyone.' }],++ scenes: [], signals: {}, gaps: [], duration_seconds: 30, analyzed_seconds: 30, speech_ratio: 0.1,++ },++ } } };++}++const server = http.createServer((req, res) => {++ let raw = '';++ req.on('data', chunk => { raw += chunk; });++ req.on('end', () => {++ const result = answer(raw ? JSON.parse(raw) : { poll: req.url });++ res.setHeader('Content-Type', 'application/json');++ res.statusCode = result.status;++ res.end(JSON.stringify(result.body));++ });++});++++(async () => {++ await db.initDb();++ const originalFetch = global.fetch;++ let listening = false;++ try {++ await new Promise((resolve, reject) => {++ server.once('error', reject);++ server.listen(0, '127.0.0.1', resolve);++ });++ listening = true;++ } catch (e) {++ if (e.code !== 'EPERM') throw e;++ // Some sandboxes forbid listening sockets. Exercise the same fake AI++ // response through fetch while preserving the normal HTTP server path.++ global.fetch = async (url, opts) => {++ const result = answer(JSON.parse(opts.body));++ return new Response(JSON.stringify(result.body), { status: result.status, headers: { 'Content-Type': 'application/json' } });++ };++ }++ process.env.OV_AI_INTERNAL_URL = listening ? `http://127.0.0.1:${server.address().port}` : 'http://127.0.0.1:4700';++ principal.serviceHeaders = async () => ({ Authorization: 'Bearer test-service-token' });++ media.signedMediaUrl = async (kind, id) => ({ url: `https://openvibe.media/${kind}/${id}?signature=test` });++ media.getVod = async id => ({ id, status: 'ready', user_id: 7, duration_seconds: 30 });++ media.getClip = async id => ({ id, status: 'ready', user_id: 7, duration_seconds: 30 });++ let localCalls = 0;++ local.transcriptionEnabled = async () => { localCalls++; return true; };++ local.generateVodTranscript = async () => { localCalls++; throw new Error('local VOD transcription started'); };++ local.generateClipTranscript = async () => { localCalls++; throw new Error('local clip transcription started'); };++ local.generateVodOverview = async () => { localCalls++; throw new Error('local VOD overview started'); };++ local.generateClipOverview = async () => { localCalls++; throw new Error('local clip overview started'); };++ local.isEnabled = async () => true;++ local.pasteAnalysisEnabled = async () => false;++ try {++ await db.run('INSERT INTO vod_ai_state (vod_id) VALUES (801)');++ await db.run('INSERT INTO clip_ai_state (clip_id, clip_notified) VALUES (901, 1)');++ await backfill.tick();++ assert.deepStrictEqual(calls.map(c => c.workflow), ['media.analyze', 'media.analyze']);++ assert.deepStrictEqual(calls.map(c => c.input.media_url), [++ 'https://openvibe.media/vod/801?signature=test', 'https://openvibe.media/clip/901?signature=test',++ ]);++ assert.ok(calls.every(c => c.input.allow_paid === false && c.input.language === 'en'));++ const vod = await db.getVodAiState(801);++ const clip = await db.getClipAiState(901);++ assert.strictEqual(clip.clip_notified, 1, 'unmapped clip state survives analysis');++ for (const state of [vod, clip]) {++ assert.strictEqual(state.ai_overview, 'A creator explains the project. Then the stream ends.');++ assert.ok(state.ai_overview_short);++ assert.deepStrictEqual(JSON.parse(state.ai_transcript_json), [++ { start: 1.25, end: 2.8, text: 'Welcome' }, { start: 2.9, end: 4.1, text: 'everyone.' },++ ]);++ assert.strictEqual(state.transcript_status, 'done');++ }++ assert.strictEqual(localCalls, 0);++++ await db.run("INSERT INTO vod_ai_state (vod_id, ai_overview, ai_overview_short) VALUES (802, 'Existing', 'Existing')");++ refuse = true;++ const before = remote.stats().failures;++ await backfill.tick();++ const failed = await db.getVodAiState(802);++ assert.strictEqual(failed.ai_overview, 'Existing');++ assert.strictEqual(failed.ai_overview_short, 'Existing');++ assert.strictEqual(failed.ai_transcript_json, null);++ assert.strictEqual(failed.transcript_status, 'retry');++ assert.ok(failed.transcript_next_at);++ assert.strictEqual(remote.stats().failures, before + 1);++ const attempted = calls.length;++ await backfill.tick();++ assert.strictEqual(calls.length, attempted, 'backoff prevents an immediate retry');++ assert.strictEqual(localCalls, 0, 'AI failure never starts the local path');++++ // An overview Live already has is kept: AI's fills only a missing one (the transcript still lands).++ refuse = false;++ await db.run("INSERT INTO vod_ai_state (vod_id, ai_overview, ai_overview_short) VALUES (803, 'A model overview', 'A model overview')");++ await backfill.tick();++ const kept = await db.getVodAiState(803);++ assert.strictEqual(kept.ai_overview, 'A model overview', 'never replaced by AI\'s');++ assert.ok(kept.ai_transcript_json && JSON.parse(kept.ai_transcript_json).length === 2, 'the transcript landed');++ assert.strictEqual(calls[calls.length - 1].idempotency_key, 'live.media.analyze:vod:803:0', 'one AI run per attempt');++++ // A run still going when the poll budget ends is re-joined later by the same key, not counted as a failure.++ stillRunning = true;++ process.env.LIVE_MEDIA_ANALYSIS_WAIT_MS = '0';++ await db.run('INSERT INTO vod_ai_state (vod_id) VALUES (804)');++ const failuresBefore = remote.stats().failures;++ const vod804 = { id: 804, status: 'ready', user_id: 7, duration_seconds: 30 };++ await remote.analyze('vod', vod804);++ const pending = await db.getVodAiState(804);++ assert.strictEqual(pending.transcript_status, 'retry');++ assert.strictEqual(Number(pending.transcript_attempts) || 0, 0, 'no attempt counted while AI is still working');++ assert.strictEqual(remote.stats().failures, failuresBefore, 'not a failure');++ stillRunning = false;++ await remote.analyze('vod', vod804);++ assert.strictEqual(calls[calls.length - 1].idempotency_key, calls[calls.length - 2].idempotency_key, 'the same key re-joins the run');++ assert.strictEqual((await db.getVodAiState(804)).transcript_status, 'done');++ delete process.env.LIVE_MEDIA_ANALYSIS_WAIT_MS;++++ await db.run("INSERT INTO clip_ai_state (clip_id, ai_overview, ai_overview_short) VALUES (902, 'Kept clip', 'Kept clip')");++ global.fetch = async () => { throw new Error('AI connection refused'); };++ await backfill.tick();++ const down = await db.getClipAiState(902);++ assert.strictEqual(down.ai_overview, 'Kept clip');++ assert.strictEqual(down.ai_overview_short, 'Kept clip');++ assert.strictEqual(down.transcript_status, 'retry');++ assert.ok(down.transcript_next_at);++ assert.strictEqual(remote.stats().failures, before + 2);++ assert.strictEqual(localCalls, 0);++++ delete process.env.LIVE_MEDIA_ANALYSIS;++ assert.strictEqual(remote.enabled(), false, 'default remains local');++ console.log('AI media analyze: VOD, clip, retry, and default mode passed');++ } finally {++ global.fetch = originalFetch;++ if (listening) server.close();++ }++})().catch(e => { console.error(e); process.exit(1); });
@@ -21,10 +21,14 @@ ADMIN_PASSWORD=changeme123
@@ -21,10 +21,14 @@ ADMIN_PASSWORD=changeme123
# service, with a service token for audience openvibe.ai (grants ai.run.create, ai.run.read and
# service, with a service token for audience openvibe.ai (grants ai.run.create, ai.run.read and
# ai.credential.manage for client `live`). Provider keys, a streamer's own included, live in the AI
# ai.credential.manage for client `live`). Provider keys, a streamer's own included, live in the AI
# service; Live calls a provider only to test a key a streamer typed and has not saved yet. The admin
# service; Live calls a provider only to test a key a streamer typed and has not saved yet. The admin
-# ai_enabled switch and the local daily cap still apply; local whisper.cpp transcription stays here.
+# ai_enabled switch and the local daily cap still apply to other AI workflows. Live stream transcription stays here.
# AI_SERVICE=off turns AI off entirely.
# AI_SERVICE=off turns AI off entirely.
#AI_SERVICE=off
#AI_SERVICE=off
#OV_AI_INTERNAL_URL=http://127.0.0.1:4700
#OV_AI_INTERNAL_URL=http://127.0.0.1:4700
+# Finished VOD/clip analysis: local (current frame/whisper path) or ai (one media.analyze run).
+#LIVE_MEDIA_ANALYSIS=local
+# How long one call waits for media.analyze before re-joining the run 15 minutes later (ms).
+#LIVE_MEDIA_ANALYSIS_WAIT_MS=60000
# Server
# Server
PORT=3000
PORT=3000
@@ -1,5 +1,20 @@
@@ -1,5 +1,20 @@
# VODs & Clips
# VODs & Clips
+## Finished media analysis
+
+`LIVE_MEDIA_ANALYSIS=local` is the default: Live's existing frame and transcript jobs produce
+VOD and clip overviews and timed transcripts. Set `LIVE_MEDIA_ANALYSIS=ai` to send each finished
+VOD or clip's signed Media URL to OpenVibe.AI's `media.analyze` workflow (AI runs one analysis at a time at nice 15).
+Live fills a missing overview (an overview it already has is kept), derives the short overview, and stores timed speech
+segments in `vod_ai_state` or `clip_ai_state`. Live waits up to a minute for the run
+(`LIVE_MEDIA_ANALYSIS_WAIT_MS`, 60000); a longer analysis is re-joined 15 minutes later by its idempotency key
+(`live.media.analyze:<kind>:<id>:<attempt>`), so the same recording is never analysed twice at once and no loop waits
+for it.
+AI's signals, scenes, and highlights have no Live state columns and are not stored. Failed or
+unavailable analyses retry with backoff; Live does not run its VOD/clip decoder in this mode.
+Live stream captions and sound events continue using their real-time paths. The
+`live_media_analysis_retry_total` metric counts deferred analyses by media kind.
+
> Written before the media split. Storage, cutting, thumbnails and files now live in OpenVibe.Media;
> Written before the media split. Storage, cutting, thumbnails and files now live in OpenVibe.Media;
> Live proxies to it (`server/media-proxy/`) and starts/stops recordings (`server/streaming/recorder.js`).
> Live proxies to it (`server/media-proxy/`) and starts/stops recordings (`server/streaming/recorder.js`).
> Pastes live in OpenVibe.Community. Live's old local `vods`/`clips`/`pastes` tables are frozen: nothing
> Pastes live in OpenVibe.Community. Live's old local `vods`/`clips`/`pastes` tables are frozen: nothing
@@ -160,6 +160,7 @@ async function _mediaSource(row, kind = 'vod') {
@@ -160,6 +160,7 @@ async function _mediaSource(row, kind = 'vod') {
const _overviewInFlight = new Set();
const _overviewInFlight = new Set();
async function generateVodOverview(vod) {
async function generateVodOverview(vod) {
if (!vod) return null;
if (!vod) return null;
+ if (require('./ai-media-analyze').enabled()) return require('./ai-media-analyze').analyze('vod', vod);
if (!await isEnabled() || !await withinBudget()) return null;
if (!await isEnabled() || !await withinBudget()) return null;
// Guard against the on-finalize trigger and the backfill poller processing the same
// Guard against the on-finalize trigger and the backfill poller processing the same
// VOD at once (that would double-extract frames + duplicate timeline memories).
// VOD at once (that would double-extract frames + duplicate timeline memories).
@@ -218,6 +219,7 @@ async function _generateVodOverviewInner(vod) {
@@ -218,6 +219,7 @@ async function _generateVodOverviewInner(vod) {
*/
*/
async function ensureVodTimeline(vod) {
async function ensureVodTimeline(vod) {
if (!vod || !vod.stream_id) return;
if (!vod || !vod.stream_id) return;
+ if (require('./ai-media-analyze').enabled()) return;
if (!await isEnabled() || !await withinBudget()) return;
if (!await isEnabled() || !await withinBudget()) return;
const ma = require('./media-analysis');
const ma = require('./media-analysis');
const src = await _mediaSource(vod);
const src = await _mediaSource(vod);
@@ -265,6 +267,7 @@ async function _applyTxLowPower() {
@@ -265,6 +267,7 @@ async function _applyTxLowPower() {
* a clean silent run is marked terminal — so an interrupted run is never lost.
* a clean silent run is marked terminal — so an interrupted run is never lost.
*/
*/
async function generateVodTranscript(vod) {
async function generateVodTranscript(vod) {
+ if (require('./ai-media-analyze').enabled()) return require('./ai-media-analyze').analyze('vod', vod);
if (!vod || !await transcriptionEnabled()) return null;
if (!vod || !await transcriptionEnabled()) return null;
return _txRun(async () => {
return _txRun(async () => {
await _applyTxLowPower();
await _applyTxLowPower();
@@ -302,6 +305,7 @@ async function generateVodTranscript(vod) {
@@ -302,6 +305,7 @@ async function generateVodTranscript(vod) {
/** Transcript-only pass for a clip — FREE local whisper (see generateVodTranscript). */
/** Transcript-only pass for a clip — FREE local whisper (see generateVodTranscript). */
async function generateClipTranscript(clip) {
async function generateClipTranscript(clip) {
+ if (require('./ai-media-analyze').enabled()) return require('./ai-media-analyze').analyze('clip', clip);
if (!clip || !await transcriptionEnabled()) return null;
if (!clip || !await transcriptionEnabled()) return null;
return _txRun(async () => {
return _txRun(async () => {
await _applyTxLowPower();
await _applyTxLowPower();
@@ -335,6 +339,7 @@ async function generateClipTranscript(clip) {
@@ -335,6 +339,7 @@ async function generateClipTranscript(clip) {
*/
*/
async function generateClipOverview(clip) {
async function generateClipOverview(clip) {
if (!clip) return null;
if (!clip) return null;
+ if (require('./ai-media-analyze').enabled()) return require('./ai-media-analyze').analyze('clip', clip);
if (!await isEnabled() || !await withinBudget()) return null;
if (!await isEnabled() || !await withinBudget()) return null;
const src = await _mediaSource(clip, 'clip');
const src = await _mediaSource(clip, 'clip');
if (!src) { try { await db.setClipAiOverview(clip.id, { overview: ' ', transcript: null }); } catch { /* */ } return null; }
if (!src) { try { await db.setClipAiOverview(clip.id, { overview: ' ', transcript: null }); } catch { /* */ } return null; }
new file mode 100644
new file mode 100644
@@ -0,0 +1,100 @@
@@ -0,0 +1,100 @@
+'use strict';
+
+// Finished Media recordings are analysed as one OpenVibe.AI workflow run. This
+// module never decodes media on Live; the signed URL also works for private items.
+const db = require('../db/database');
+const media = require('../media-client');
+const aiService = require('./ai-service');
+
+const inFlight = new Set();
+let failures = 0;
+let failureCounter = null;
+
+function enabled() { return String(process.env.LIVE_MEDIA_ANALYSIS || 'local').toLowerCase() === 'ai'; }
+function registerMetrics(registry) {
+ if (registry && typeof registry.counter === 'function') {
+ failureCounter = registry.counter({ name: 'live_media_analysis_retry_total', help: 'Finished media analyses deferred after an OpenVibe.AI failure', labelNames: ['kind'] });
+ }
+}
+function retryMinutes(attempt) { return Math.min(360, 2 ** Math.min(8, Math.max(0, attempt - 1))); }
+
+async function source(kind, item) {
+ const signed = await media.signedMediaUrl(kind, item.id, 6 * 3600).catch(() => null);
+ if (signed && signed.url) return signed.url;
+ const meta = kind === 'vod' ? await media.getVod(item.id).catch(() => null) : await media.getClip(item.id).catch(() => null);
+ if (meta && meta.playback_url) return media.publicUrl(meta.playback_url);
+ return kind === 'vod' ? media.vodPlaybackUrl(item.id) : media.clipUrl(item.id);
+}
+
+async function analyze(kind, item) {
+ if (!item || !item.id || !['vod', 'clip'].includes(kind)) return null;
+ const key = `${kind}:${item.id}`;
+ if (inFlight.has(key)) return null;
+ inFlight.add(key);
+ const bump = kind === 'vod' ? db.bumpVodTranscriptAttempt : db.bumpClipTranscriptAttempt;
+ const status = kind === 'vod' ? db.setVodTranscriptStatus : db.setClipTranscriptStatus;
+ try {
+ const state = kind === 'vod' ? await db.getVodAiState(item.id) : await db.getClipAiState(item.id);
+ const mediaUrl = await source(kind, item);
+ if (!mediaUrl) throw new Error('Media did not provide a readable URL');
+ // One AI run per attempt: a retry after a failure starts a new run, while a run still going when the poll
+ // budget ends is re-joined later by the same key (never a second analysis of the same recording at once).
+ const attemptNo = Number(state && state.transcript_attempts) || 0;
+ const run = await aiService.run('media.analyze', { media_url: mediaUrl, language: 'en', allow_paid: false }, {
+ target: { service: 'live', type: kind, id: String(item.id) },
+ attribution: aiService.ownerRef(item.user_id),
+ idempotencyKey: `live.media.analyze:${kind}:${item.id}:${attemptNo}`,
+ // Up to a minute (a clip is usually done by then), never a poll loop: a longer analysis is re-joined
+ // in 15 minutes by its key, so the backfill loop is never held for the length of a recording.
+ waitMs: Math.max(0, Number(process.env.LIVE_MEDIA_ANALYSIS_WAIT_MS ?? 60000) || 0), pollMs: 0,
+ });
+ if (run && (run.status === 'queued' || run.status === 'running')) {
+ await status(item.id, 'retry', 'OpenVibe.AI is still analysing it', 15);
+ return null;
+ }
+ const out = aiService.usable(run);
+ if (!out || !out.overview || typeof out.overview.text !== 'string' || !out.overview.text.trim() || !out.transcript || !Array.isArray(out.transcript.segments)) {
+ throw new Error((run && run.error && (run.error.code || run.error.detail)) || (run && run.status) || 'no usable media.analyze result');
+ }
+ // The state setters derive ai_overview_short and store segments in the
+ // existing [{start,end,text}] JSON format. AI signals/highlights have no
+ // corresponding state columns; keep all other Live state untouched.
+ const overview = out.overview.text.trim();
+ const segments = out.transcript.segments.map(g => ({ start: g.start, end: g.end, text: g.text }));
+ // An overview Live already has (a model's, or a person's edit) is kept: AI's fills only a missing one.
+ if (!state || !String(state.ai_overview || '').trim()) {
+ if (kind === 'vod') await db.setVodAiOverview(item.id, overview);
+ else await db.setClipAiOverview(item.id, { overview });
+ }
+ const partialSpeech = Array.isArray(out.gaps) && out.gaps.some(g => /^Speech-to-text failed/i.test(String(g)));
+ if (out.transcript.available && !partialSpeech) {
+ const hasSpeech = !!(segments.length || String(out.transcript.text || '').trim());
+ if (hasSpeech || !state || !state.ai_transcript_json) {
+ if (kind === 'vod') await db.setVodTranscript(item.id, out.transcript.text || ' ', segments);
+ else await db.setClipTranscript(item.id, out.transcript.text || ' ', segments);
+ }
+ await status(item.id, hasSpeech || (state && state.ai_transcript_json) ? 'done' : 'empty');
+ } else if (out.streams && out.streams.audio === false) {
+ if (!state || !state.ai_transcript_json) {
+ if (kind === 'vod') await db.setVodTranscript(item.id, ' ', []);
+ else await db.setClipTranscript(item.id, ' ', []);
+ }
+ await status(item.id, state && state.ai_transcript_json ? 'done' : 'empty', 'no audio stream');
+ } else if (!state || !state.ai_transcript_json) {
+ const attempt = await bump(item.id);
+ await status(item.id, 'retry', partialSpeech ? 'AI speech-to-text incomplete' : 'AI speech-to-text unavailable', retryMinutes(attempt));
+ }
+ return out;
+ } catch (e) {
+ const attempt = await bump(item.id);
+ await status(item.id, 'retry', e.message, retryMinutes(attempt));
+ failures++;
+ if (failureCounter) failureCounter.inc({ kind });
+ console.warn(`[AI media] ${kind} ${item.id}: ${e.message}; retry in ${retryMinutes(attempt)} min`);
+ return null;
+ } finally {
+ inFlight.delete(key);
+ }
+}
+
+module.exports = { enabled, analyze, registerMetrics, stats: () => ({ failures }) };
@@ -13,6 +13,7 @@
@@ -13,6 +13,7 @@
const db = require('../db/database');
const db = require('../db/database');
const ai = require('./ai-analysis');
const ai = require('./ai-analysis');
const media = require('../media-client');
const media = require('../media-client');
+const remote = require('./ai-media-analyze');
let _timer = null;
let _timer = null;
let _busy = false;
let _busy = false;
@@ -40,54 +41,68 @@ async function tick() {
@@ -40,54 +41,68 @@ async function tick() {
if (_busy) return;
if (_busy) return;
_busy = true;
_busy = true;
try {
try {
- // Transcripts — FREE local whisper (no API/cost), so they run independent of
- // the AI master switch. We DON'T stop while streams are live — instead we drop
- // to low-power (fewer whisper threads) and a smaller batch so we never starve
- // the live encoders.
let anyLive = false;
let anyLive = false;
try { anyLive = ((db.getLiveStreams && await db.getLiveStreams()) || []).length > 0; } catch { /* */ }
try { anyLive = ((db.getLiveStreams && await db.getLiveStreams()) || []).length > 0; } catch { /* */ }
- // When the continuous timeline is running it is already decoding live audio
- // constantly, and VOD backfill does NOT share its serialisation — so both would
- // 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.
- // 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 && await 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 {
- let done = 0;
- for (const row of await db.getVodsNeedingTranscript(batch + 6)) {
- if (done >= batch) break;
- const vod = await _vodMeta(row);
- if (!vod) continue;
- await ai.generateVodTranscript(vod);
- done++;
- }
- } catch (e) { console.warn('[AI backfill] vod transcript:', e.message); }
- try {
- let done = 0;
- for (const row of await db.getClipsNeedingTranscript(batch + 6)) {
- if (done >= batch) break;
- const clip = await _clipMeta(row);
- if (!clip) continue;
- await ai.generateClipTranscript(clip);
- done++;
- }
- } catch (e) { console.warn('[AI backfill] clip transcript:', e.message); }
+ if (remote.enabled()) {
+ for (const [kind, metadata] of [['vod', _vodMeta], ['clip', _clipMeta]]) {
+ try {
+ for (const row of await db.getMediaNeedingAiAnalysis(kind, 6)) {
+ const item = await metadata(row);
+ if (!item) continue;
+ await remote.analyze(kind, item);
+ break;
+ }
+ } catch (e) { console.warn(`[AI backfill] ${kind} media.analyze:`, e.message); }
+ }
}
}
+ if (!remote.enabled()) {
+ // Transcripts — FREE local whisper (no API/cost), so they run independent of
+ // the AI master switch. We DON'T stop while streams are live — instead we drop
+ // to low-power (fewer whisper threads) and a smaller batch so we never starve
+ // the live encoders.
+ // When the continuous timeline is running it is already decoding live audio
+ // constantly, and VOD backfill does NOT share its serialisation — so both would
+ // 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.
+ // 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 && await 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 {
+ let done = 0;
+ for (const row of await db.getVodsNeedingTranscript(batch + 6)) {
+ if (done >= batch) break;
+ const vod = await _vodMeta(row);
+ if (!vod) continue;
+ await ai.generateVodTranscript(vod);
+ done++;
+ }
+ } catch (e) { console.warn('[AI backfill] vod transcript:', e.message); }
+ try {
+ let done = 0;
+ for (const row of await db.getClipsNeedingTranscript(batch + 6)) {
+ if (done >= batch) break;
+ const clip = await _clipMeta(row);
+ if (!clip) continue;
+ await ai.generateClipTranscript(clip);
+ done++;
+ }
+ } catch (e) { console.warn('[AI backfill] clip transcript:', e.message); }
+ }
+ }
if (!await ai.isEnabled()) return;
if (!await ai.isEnabled()) return;
// Paste summaries. Pastes moved to Media at the migration, but the analysis job
// Paste summaries. Pastes moved to Media at the migration, but the analysis job
@@ -119,6 +134,8 @@ async function tick() {
@@ -119,6 +134,8 @@ async function tick() {
} catch (e) { console.warn('[AI backfill] paste:', e.message); }
} catch (e) { console.warn('[AI backfill] paste:', e.message); }
}
}
+ if (remote.enabled()) return;
+
// VOD overviews (frames + audio pulled from the Media playback URL; self-marks).
// VOD overviews (frames + audio pulled from the Media playback URL; self-marks).
try {
try {
// Take a few candidates and process the first usable one: the newest row is
// Take a few candidates and process the first usable one: the newest row is
@@ -627,6 +627,18 @@ async function getClipsNeedingTranscript(limit = 2) {
@@ -627,6 +627,18 @@ async function getClipsNeedingTranscript(limit = 2) {
AND (transcript_next_at IS NULL OR transcript_next_at <= ov_now())
AND (transcript_next_at IS NULL OR transcript_next_at <= ov_now())
ORDER BY (transcript_status='retry'), clip_id DESC LIMIT ?`, [limit]);
ORDER BY (transcript_status='retry'), clip_id DESC LIMIT ?`, [limit]);
}
}
+// In AI mode one workflow fills both fields. The transcript retry schedule also
+// throttles failed overview attempts, including rows with an older transcript.
+async function getMediaNeedingAiAnalysis(kind, limit = 6) {
+ const clip = kind === 'clip';
+ const table = clip ? 'clip_ai_state' : 'vod_ai_state';
+ const id = clip ? 'clip_id' : 'vod_id';
+ return await all(`SELECT ${id} AS id, s.* FROM ${table} s
+ WHERE (ai_overview IS NULL OR ai_overview = '' OR ai_transcript_json IS NULL)
+ AND (transcript_next_at IS NULL OR transcript_next_at <= ov_now())
+ AND (transcript_status IS NULL OR transcript_status <> 'processing')
+ ORDER BY ${id} DESC LIMIT ?`, [limit]);
+}
// status setter. On a 'retry', pass retryDelayMin to schedule the next eligible attempt
// status setter. On a 'retry', pass retryDelayMin to schedule the next eligible attempt
// (exponential backoff); any other status clears the schedule.
// (exponential backoff); any other status clears the schedule.
async function setVodTranscriptStatus(id, status, error = null, retryDelayMin = 0) {
async function setVodTranscriptStatus(id, status, error = null, retryDelayMin = 0) {
@@ -3874,7 +3886,7 @@ module.exports = {
@@ -3874,7 +3886,7 @@ module.exports = {
createStream, endStream, onStreamLifecycle, endOtherLiveStreamsForSlot, updateViewerCount,
createStream, endStream, onStreamLifecycle, endOtherLiveStreamsForSlot, updateViewerCount,
addStreamMemory, getStreamMemories, getLatestStreamMemory, updateStreamAiOverview,
addStreamMemory, getStreamMemories, getLatestStreamMemory, updateStreamAiOverview,
setVodAiOverview, setClipAiOverview, setVodTranscript, setClipTranscript, getStreamMemoriesInRange,
setVodAiOverview, setClipAiOverview, setVodTranscript, setClipTranscript, getStreamMemoriesInRange,
- getVodsNeedingOverview, getClipsNeedingOverview, getVodsNeedingTimeline, getVodsNeedingTranscript, getClipsNeedingTranscript,
+ getVodsNeedingOverview, getClipsNeedingOverview, getVodsNeedingTimeline, getVodsNeedingTranscript, getClipsNeedingTranscript, getMediaNeedingAiAnalysis,
setVodTranscriptStatus, setClipTranscriptStatus, bumpVodTranscriptAttempt, bumpClipTranscriptAttempt,
setVodTranscriptStatus, setClipTranscriptStatus, bumpVodTranscriptAttempt, bumpClipTranscriptAttempt,
cleanupMalformedAiText, recordAiUsage, getAiCostToday, getAiCostTodayForUser, getAiUsageSummary,
cleanupMalformedAiText, recordAiUsage, getAiCostToday, getAiCostTodayForUser, getAiUsageSummary,
getStreamMemoriesByUser, countStreamMemoriesByUser, getStreamTranscriptSegments,
getStreamMemoriesByUser, countStreamMemoriesByUser, getStreamTranscriptSegments,
@@ -244,6 +244,7 @@ function releaseEnv() {
@@ -244,6 +244,7 @@ function releaseEnv() {
// answers direct loopback callers only; through nginx it is a 404 (server/web/observability.js).
// answers direct loopback callers only; through nginx it is a 404 (server/web/observability.js).
const observability = require('./web/observability');
const observability = require('./web/observability');
const { registry: metricsRegistry } = observability.mountMetrics(app, { release });
const { registry: metricsRegistry } = observability.mountMetrics(app, { release });
+if (!drill.enabled) require('./ai/ai-media-analyze').registerMetrics(metricsRegistry);
const vibeCodingPublishServer = new VibeCodingPublishServer(db);
const vibeCodingPublishServer = new VibeCodingPublishServer(db);
function normalizeOrigin(origin) {
function normalizeOrigin(origin) {
@@ -112,8 +112,12 @@ async function apply(event, data) {
@@ -112,8 +112,12 @@ async function apply(event, data) {
try {
try {
const ai = require('../ai/ai-analysis');
const ai = require('../ai/ai-analysis');
const vodMeta = { id: vodId, ...data };
const vodMeta = { id: vodId, ...data };
- if (ai.transcriptionEnabled && await ai.transcriptionEnabled()) ai.generateVodTranscript(vodMeta).catch(() => {});
- if (ai.isEnabled && await ai.isEnabled() && ai.withinBudget && await ai.withinBudget()) ai.generateVodOverview(vodMeta).catch(() => {});
+ const remote = require('../ai/ai-media-analyze');
+ if (remote.enabled()) remote.analyze('vod', vodMeta).catch(() => {});
+ else {
+ if (ai.transcriptionEnabled && await ai.transcriptionEnabled()) ai.generateVodTranscript(vodMeta).catch(() => {});
+ if (ai.isEnabled && await ai.isEnabled() && ai.withinBudget && await ai.withinBudget()) ai.generateVodOverview(vodMeta).catch(() => {});
+ }
} catch { /* backfill poller will pick it up */ }
} catch { /* backfill poller will pick it up */ }
console.log(`[MediaOutcome] VOD ${vodId} ready (${data.duration || data.duration_seconds || '?'}s)`);
console.log(`[MediaOutcome] VOD ${vodId} ready (${data.duration || data.duration_seconds || '?'}s)`);
},
},
@@ -141,7 +145,9 @@ async function apply(event, data) {
@@ -141,7 +145,9 @@ async function apply(event, data) {
after: async () => {
after: async () => {
try {
try {
const ai = require('../ai/ai-analysis');
const ai = require('../ai/ai-analysis');
- if (ai.isEnabled && await ai.isEnabled() && ai.withinBudget && await ai.withinBudget()) {
+ const remote = require('../ai/ai-media-analyze');
+ if (remote.enabled()) remote.analyze('clip', { id: data.id, ...data }).catch(() => {});
+ else if (ai.isEnabled && await ai.isEnabled() && ai.withinBudget && await ai.withinBudget()) {
ai.generateClipOverview({ id: data.id, ...data }).catch(() => {});
ai.generateClipOverview({ id: data.id, ...data }).catch(() => {});
}
}
} catch { /* */ }
} catch { /* */ }
new file mode 100644
new file mode 100644
@@ -0,0 +1,158 @@
@@ -0,0 +1,158 @@
+'use strict';
+const assert = require('assert');
+const http = require('http');
+const db = require('../server/db/database');
+const media = require('../server/media-client');
+const principal = require('../server/net/network-principal');
+const local = require('../server/ai/ai-analysis');
+const remote = require('../server/ai/ai-media-analyze');
+const backfill = require('../server/ai/backfill-job');
+
+process.env.LIVE_MEDIA_ANALYSIS = 'ai';
+let calls = [];
+let refuse = false;
+let stillRunning = false;
+function answer(body) {
+ calls.push(body);
+ if (refuse) return { status: 503, body: { code: 'media.no_space', detail: 'disk is full' } };
+ if (stillRunning) return { status: 202, body: { run: { id: `run_${calls.length}`, status: 'running' } } };
+ return { status: 201, body: { run: {
+ id: `run_${calls.length}`, status: 'succeeded', synthetic: false,
+ output: {
+ overview: { text: 'A creator explains the project. Then the stream ends.', source: 'local_model' },
+ transcript: { available: true, language: 'en', text: 'Welcome everyone.', segments: [
+ { start: 1.25, end: 2.8, text: 'Welcome' },
+ { start: 2.9, end: 4.1, text: 'everyone.' },
+ ] },
+ streams: { video: true, audio: true },
+ highlights: [{ start: 1, end: 5, score: 2, reasons: ['speech'], excerpt: 'Welcome everyone.' }],
+ scenes: [], signals: {}, gaps: [], duration_seconds: 30, analyzed_seconds: 30, speech_ratio: 0.1,
+ },
+ } } };
+}
+const server = http.createServer((req, res) => {
+ let raw = '';
+ req.on('data', chunk => { raw += chunk; });
+ req.on('end', () => {
+ const result = answer(raw ? JSON.parse(raw) : { poll: req.url });
+ res.setHeader('Content-Type', 'application/json');
+ res.statusCode = result.status;
+ res.end(JSON.stringify(result.body));
+ });
+});
+
+(async () => {
+ await db.initDb();
+ const originalFetch = global.fetch;
+ let listening = false;
+ try {
+ await new Promise((resolve, reject) => {
+ server.once('error', reject);
+ server.listen(0, '127.0.0.1', resolve);
+ });
+ listening = true;
+ } catch (e) {
+ if (e.code !== 'EPERM') throw e;
+ // Some sandboxes forbid listening sockets. Exercise the same fake AI
+ // response through fetch while preserving the normal HTTP server path.
+ global.fetch = async (url, opts) => {
+ const result = answer(JSON.parse(opts.body));
+ return new Response(JSON.stringify(result.body), { status: result.status, headers: { 'Content-Type': 'application/json' } });
+ };
+ }
+ process.env.OV_AI_INTERNAL_URL = listening ? `http://127.0.0.1:${server.address().port}` : 'http://127.0.0.1:4700';
+ principal.serviceHeaders = async () => ({ Authorization: 'Bearer test-service-token' });
+ media.signedMediaUrl = async (kind, id) => ({ url: `https://openvibe.media/${kind}/${id}?signature=test` });
+ media.getVod = async id => ({ id, status: 'ready', user_id: 7, duration_seconds: 30 });
+ media.getClip = async id => ({ id, status: 'ready', user_id: 7, duration_seconds: 30 });
+ let localCalls = 0;
+ local.transcriptionEnabled = async () => { localCalls++; return true; };
+ local.generateVodTranscript = async () => { localCalls++; throw new Error('local VOD transcription started'); };
+ local.generateClipTranscript = async () => { localCalls++; throw new Error('local clip transcription started'); };
+ local.generateVodOverview = async () => { localCalls++; throw new Error('local VOD overview started'); };
+ local.generateClipOverview = async () => { localCalls++; throw new Error('local clip overview started'); };
+ local.isEnabled = async () => true;
+ local.pasteAnalysisEnabled = async () => false;
+ try {
+ await db.run('INSERT INTO vod_ai_state (vod_id) VALUES (801)');
+ await db.run('INSERT INTO clip_ai_state (clip_id, clip_notified) VALUES (901, 1)');
+ await backfill.tick();
+ assert.deepStrictEqual(calls.map(c => c.workflow), ['media.analyze', 'media.analyze']);
+ assert.deepStrictEqual(calls.map(c => c.input.media_url), [
+ 'https://openvibe.media/vod/801?signature=test', 'https://openvibe.media/clip/901?signature=test',
+ ]);
+ assert.ok(calls.every(c => c.input.allow_paid === false && c.input.language === 'en'));
+ const vod = await db.getVodAiState(801);
+ const clip = await db.getClipAiState(901);
+ assert.strictEqual(clip.clip_notified, 1, 'unmapped clip state survives analysis');
+ for (const state of [vod, clip]) {
+ assert.strictEqual(state.ai_overview, 'A creator explains the project. Then the stream ends.');
+ assert.ok(state.ai_overview_short);
+ assert.deepStrictEqual(JSON.parse(state.ai_transcript_json), [
+ { start: 1.25, end: 2.8, text: 'Welcome' }, { start: 2.9, end: 4.1, text: 'everyone.' },
+ ]);
+ assert.strictEqual(state.transcript_status, 'done');
+ }
+ assert.strictEqual(localCalls, 0);
+
+ await db.run("INSERT INTO vod_ai_state (vod_id, ai_overview, ai_overview_short) VALUES (802, 'Existing', 'Existing')");
+ refuse = true;
+ const before = remote.stats().failures;
+ await backfill.tick();
+ const failed = await db.getVodAiState(802);
+ assert.strictEqual(failed.ai_overview, 'Existing');
+ assert.strictEqual(failed.ai_overview_short, 'Existing');
+ assert.strictEqual(failed.ai_transcript_json, null);
+ assert.strictEqual(failed.transcript_status, 'retry');
+ assert.ok(failed.transcript_next_at);
+ assert.strictEqual(remote.stats().failures, before + 1);
+ const attempted = calls.length;
+ await backfill.tick();
+ assert.strictEqual(calls.length, attempted, 'backoff prevents an immediate retry');
+ assert.strictEqual(localCalls, 0, 'AI failure never starts the local path');
+
+ // An overview Live already has is kept: AI's fills only a missing one (the transcript still lands).
+ refuse = false;
+ await db.run("INSERT INTO vod_ai_state (vod_id, ai_overview, ai_overview_short) VALUES (803, 'A model overview', 'A model overview')");
+ await backfill.tick();
+ const kept = await db.getVodAiState(803);
+ assert.strictEqual(kept.ai_overview, 'A model overview', 'never replaced by AI\'s');
+ assert.ok(kept.ai_transcript_json && JSON.parse(kept.ai_transcript_json).length === 2, 'the transcript landed');
+ assert.strictEqual(calls[calls.length - 1].idempotency_key, 'live.media.analyze:vod:803:0', 'one AI run per attempt');
+
+ // A run still going when the poll budget ends is re-joined later by the same key, not counted as a failure.
+ stillRunning = true;
+ process.env.LIVE_MEDIA_ANALYSIS_WAIT_MS = '0';
+ await db.run('INSERT INTO vod_ai_state (vod_id) VALUES (804)');
+ const failuresBefore = remote.stats().failures;
+ const vod804 = { id: 804, status: 'ready', user_id: 7, duration_seconds: 30 };
+ await remote.analyze('vod', vod804);
+ const pending = await db.getVodAiState(804);
+ assert.strictEqual(pending.transcript_status, 'retry');
+ assert.strictEqual(Number(pending.transcript_attempts) || 0, 0, 'no attempt counted while AI is still working');
+ assert.strictEqual(remote.stats().failures, failuresBefore, 'not a failure');
+ stillRunning = false;
+ await remote.analyze('vod', vod804);
+ assert.strictEqual(calls[calls.length - 1].idempotency_key, calls[calls.length - 2].idempotency_key, 'the same key re-joins the run');
+ assert.strictEqual((await db.getVodAiState(804)).transcript_status, 'done');
+ delete process.env.LIVE_MEDIA_ANALYSIS_WAIT_MS;
+
+ await db.run("INSERT INTO clip_ai_state (clip_id, ai_overview, ai_overview_short) VALUES (902, 'Kept clip', 'Kept clip')");
+ global.fetch = async () => { throw new Error('AI connection refused'); };
+ await backfill.tick();
+ const down = await db.getClipAiState(902);
+ assert.strictEqual(down.ai_overview, 'Kept clip');
+ assert.strictEqual(down.ai_overview_short, 'Kept clip');
+ assert.strictEqual(down.transcript_status, 'retry');
+ assert.ok(down.transcript_next_at);
+ assert.strictEqual(remote.stats().failures, before + 2);
+ assert.strictEqual(localCalls, 0);
+
+ delete process.env.LIVE_MEDIA_ANALYSIS;
+ assert.strictEqual(remote.enabled(), false, 'default remains local');
+ console.log('AI media analyze: VOD, clip, retry, and default mode passed');
+ } finally {
+ global.fetch = originalFetch;
+ if (listening) server.close();
+ }
+})().catch(e => { console.error(e); process.exit(1); });