openvibers/openvibe.live Commits

VOD and clip transcripts are made by OpenVibe.AI, not Live's own whisper (roadmap WS-O task 2; AI d46c217, be66fa4). A recording on Media is transcribed by live.media.transcribe one 5-minute window at a time, or whole when it is 200 seconds or less. AI reads it where it lies with range requests, loudness-normalises it and runs whisper.cpp with the same models. Each window sends media_url, start_sec and seconds; low_power follows the backfill's live-stream state (transcribe.isLowPower). Resume from the last finished window, progress after each window, and a failed window reported without failing the pass all work as before. ai-service.run takes pollMs, because a window takes minutes. Only a local file (legacy) is still decoded here, and live-stream captions stay local. test/transcript-ai.test.js makes a 10-minute recording and checks windows, low power, offsets, resume and a failed window against a stub AI. npm test passes.

OP OpenVibers committed 28 Sept 2026 · 860c944974 · 4 files, +132 −6


Changes

ddiff --git a/server/ai/ai-service.js b/server/ai/ai-service.js
iindex 7c5feaac..5800766e 100644
---- a/server/ai/ai-service.js
++++ b/server/ai/ai-service.js
@@@ -74,7 +74,7 @@ async function _fetchJson(method, path, body, timeoutMs, retried = false) {
* Run a workflow and wait for it. Returns the run object (any terminal status) or null.
* opts: { target, attribution, idempotencyKey, waitMs (default 60s), timeoutMs }
*/
--async function run(workflow, input, { target, attribution, idempotencyKey, waitMs = 60000, credentialSubject = null } = {}) {
++async function run(workflow, input, { target, attribution, idempotencyKey, waitMs = 60000, pollMs = 0, credentialSubject = null } = {}) {
if (!enabled()) return null;
const wait = Math.max(0, Math.min(60000, waitMs));
const created = await _fetchJson('POST', `/api/v1/runs?wait=${wait}`, {
@@@ -83,8 +83,9 @@ async function run(workflow, input, { target, attribution, idempotencyKey, waitM
}, wait + 15000);
let r = created && created.run;
if (!r) return null;
-- // Still queued/running after the wait: poll briefly (the service caps a wait at 60 s).
-- const deadline = Date.now() + wait;
++ // Still queued/running after the wait: poll (the service caps a wait at 60 s; pollMs keeps polling longer,
++ // for work like a transcript window that takes minutes).
++ const deadline = Date.now() + Math.max(wait, Math.min(30 * 60000, Number(pollMs) || 0));
while (!TERMINAL.has(r.status) && Date.now() < deadline) {
await new Promise(res => setTimeout(res, 1000));
const again = await _fetchJson('GET', `/api/v1/runs/${encodeURIComponent(r.id)}`, null, 15000);
ddiff --git a/server/ai/media-analysis.js b/server/ai/media-analysis.js
iindex a0546a39..54c0a12e 100644
---- a/server/ai/media-analysis.js
++++ b/server/ai/media-analysis.js
@@@ -211,9 +211,30 @@ function _ffprobeHasAudio(src) {
});
}
++/**
++ * One window of a recording, transcribed by OpenVibe.AI (live.media.transcribe; roadmap WS-O task 2): AI reads the
++ * recording where it lies (range requests, nothing downloaded), loudness-normalises it and runs whisper.cpp with the
++ * same models; low power while streams are live. → { ok, segments, error }
++ */
++async function _aiWindow(src, start, len) {
++ const aiService = require('./ai-service');
++ const r = await aiService.run('live.media.transcribe', {
++ media_url: src, language: 'en', start_sec: start, seconds: Math.max(0, Math.round(len)), low_power: require('./transcribe').isLowPower(),
++ }, { waitMs: 60000, pollMs: 20 * 60000 });
++ const out = aiService.usable(r);
++ if (!out) return { ok: false, segments: [], error: (r && r.error && (r.error.code || r.error.detail)) || 'OpenVibe.AI did not transcribe' };
++ return { ok: true, segments: (out.segments || []).filter(g => g && g.text).map(g => ({ start: Number(g.start) || 0, end: Number(g.end) || 0, text: String(g.text) })), error: 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' };
++ // Recordings on Media are transcribed by OpenVibe.AI; only a local file (legacy) is still decoded here.
++ const viaAi = /^https?:\/\//i.test(String(src || '')) && require('./ai-service').enabled();
++ if (!viaAi && !transcribe.available()) return { text: '', segments: [], ok: false, error: 'whisper unavailable' };
++ if (viaAi && duration > 0 && duration <= 200) {
++ const w = await _aiWindow(src, 0, 0);
++ return { text: w.segments.map(g => g.text).join(' ').replace(/\s+/g, ' ').trim(), segments: w.segments, ok: w.ok, error: w.error };
++ }
if (duration > 0 && duration <= 200) {
return await transcribe.transcribeMediaDetailed(src, { seconds: 0, timeoutMs: 300000 });
}
@@@ -244,8 +265,17 @@ async function _transcribeSpan(src, duration, { resumeFromSec = 0, priorSegments
windows.length = MAX_WINDOWS;
}
for (const start of windows) {
-- const wav = _tmp('wav');
const len = Math.min(WINDOW_SEC, total - start);
++ if (viaAi) {
++ const w = await _aiWindow(src, start, len);
++ if (w.ok) anyOk = true; else lastErr = w.error || lastErr;
++ if (w.segments.length) segments.push(...w.segments);
++ if (w.ok && typeof onWindow === 'function') {
++ try { await onWindow(Math.min(total, start + len), segments); } catch { /* progress is best-effort */ }
++ }
++ continue;
++ }
++ const wav = _tmp('wav');
const ffOk = await _runFf('ffmpeg', ['-y', '-ss', String(start), '-i', src, '-t', String(len), '-vn', '-ac', '1', '-ar', '16000', '-af', 'loudnorm=I=-16:TP=-1.5:LRA=11', '-f', 'wav', wav], 180000);
if (ffOk) {
const r = await transcribe.transcribeWavDetailed(wav, { timeoutMs: 600000, offsetSec: start });
ddiff --git a/server/ai/transcribe.js b/server/ai/transcribe.js
iindex 7081e0d9..d818cc85 100644
---- a/server/ai/transcribe.js
++++ b/server/ai/transcribe.js
@@@ -328,4 +328,6 @@ function describe() {
};
}
--module.exports = { available, describe, laneStatus, multilingualModel, transcribeWav, transcribeWavDetailed, transcribeMedia, transcribeMediaDetailed, killActive, setLowPower };
++function isLowPower() { return _lowPower; }
++
++module.exports = { available, describe, laneStatus, multilingualModel, transcribeWav, transcribeWavDetailed, transcribeMedia, transcribeMediaDetailed, killActive, setLowPower, isLowPower };
ddiff --git a/test/transcript-ai.test.js b/test/transcript-ai.test.js
nnew file mode 100644
iindex 00000000..c1d6e0b0
---- /dev/null
++++ b/test/transcript-ai.test.js
@@@ -0,0 +1,93 @@
++'use strict';
++// VOD and clip transcripts run on OpenVibe.AI (roadmap WS-O task 2): against a stub Network and a stub AI, a Media
++// recording (a URL) is transcribed by live.media.transcribe one 5-minute window at a time. Each run has media_url,
++// start_sec and seconds; low_power follows the backfill's live-stream state; timestamps are kept as AI gives them.
++// A resumed pass skips finished windows; a failed window is reported, not fatal; local whisper is not touched.
++// Needs ffmpeg/ffprobe (it makes a 10-minute test recording); skipped without them.
++// node test/transcript-ai.test.js
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++const crypto = require('crypto');
++const { spawnSync } = require('child_process');
++const { serviceAuth } = require('openvibe-contracts');
++
++if (spawnSync('ffprobe', ['-version']).status !== 0 || spawnSync('ffmpeg', ['-version']).status !== 0) {
++ console.log('transcript via AI: skipped (no ffmpeg/ffprobe)');
++ process.exit(0);
++}
++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-live-tx-ai-'));
++process.env.DB_PATH = path.join(tmp, 'live.db');
++process.env.NODE_ENV = 'test';
++process.env.OV_OAUTH_CLIENT_ID = 'live';
++process.env.OV_OAUTH_CLIENT_SECRET = 'live-secret';
++console.warn = () => {};
++const quiet = console.log;
++console.log = (...a) => { if (!/^\[/.test(String(a[0]))) quiet(...a); };
++
++const rec = path.join(tmp, 'rec.m4a');
++spawnSync('ffmpeg', ['-y', '-f', 'lavfi', '-i', 'sine=frequency=440:duration=610', '-c:a', 'aac', '-b:a', '16k', rec]);
++const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048 });
++const runs = [];
++let failAt = null;
++const network = http.createServer((req, res) => {
++ let raw = ''; req.on('data', (c) => { raw += c; });
++ req.on('end', () => {
++ const f = new URLSearchParams(raw); const now = Math.floor(Date.now() / 1000);
++ res.setHeader('Content-Type', 'application/json');
++ res.end(JSON.stringify({ access_token: serviceAuth.signServiceToken({ iss: 'https://openvibe.network', sub: 'svc:live', actor_type: 'service', aud: [f.get('audience')], cap: ['ai.run.create', 'ai.run.read'], iat: now, exp: now + 300, jti: `tok_${crypto.randomBytes(6).toString('hex')}` }, keys.privateKey), token_type: 'Bearer', expires_in: 300 }));
++ });
++});
++const ai = http.createServer((req, res) => {
++ let raw = ''; req.on('data', (c) => { raw += c; });
++ req.on('end', () => {
++ const body = JSON.parse(raw || '{}');
++ runs.push(body);
++ res.setHeader('Content-Type', 'application/json');
++ const s = body.input.start_sec;
++ if (s === failAt) { res.statusCode = 201; return res.end(JSON.stringify({ run: { id: 'run_x', status: 'failed', error: { code: 'provider.error' } } })); }
++ res.statusCode = 201;
++ res.end(JSON.stringify({ run: { id: 'run_01JAB2C3D4E5F6G7H8J9K0MNPA', status: 'succeeded', synthetic: false, workflow: { key: body.workflow }, output: { text: `words at ${s}`, language: 'en', segments: [{ start: s + 1.5, end: s + 4, text: `words at ${s}` }] }, usage: { tokens_in: 0, tokens_out: 0, cost_usd: 0 } } }));
++ });
++});
++const media = http.createServer((req, res) => {
++ const buf = fs.readFileSync(rec);
++ const m = /^bytes=(\d+)-(\d*)$/.exec(req.headers.range || '');
++ if (m) { const a = Number(m[1]); const b = m[2] ? Number(m[2]) : buf.length - 1; res.writeHead(206, { 'Content-Type': 'audio/mp4', 'Content-Range': `bytes ${a}-${b}/${buf.length}`, 'Content-Length': b - a + 1, 'Accept-Ranges': 'bytes' }); return res.end(buf.subarray(a, b + 1)); }
++ res.writeHead(200, { 'Content-Type': 'audio/mp4', 'Content-Length': buf.length, 'Accept-Ranges': 'bytes' }); res.end(buf);
++});
++
++(async () => {
++ for (const s of [network, ai, media]) await new Promise((r) => s.listen(0, '127.0.0.1', r));
++ process.env.OV_NETWORK_INTERNAL_URL = `http://127.0.0.1:${network.address().port}`;
++ process.env.OV_AI_INTERNAL_URL = `http://127.0.0.1:${ai.address().port}`;
++ const url = `http://127.0.0.1:${media.address().port}/v/42`;
++ const ma = require('../server/ai/media-analysis');
++ const tx = require('../server/ai/transcribe');
++ try {
++ const progress = [];
++ tx.setLowPower(true);
++ let r = await ma.transcribeOnly(url, { onWindow: (sec, segs) => progress.push([sec, segs.length]) });
++ assert.deepStrictEqual(runs.map((x) => [x.workflow, x.input.start_sec, x.input.seconds, x.input.low_power, x.input.media_url]), [
++ ['live.media.transcribe', 0, 300, true, url], ['live.media.transcribe', 300, 300, true, url], ['live.media.transcribe', 600, 10, true, url],
++ ]);
++ assert.deepStrictEqual([r.ok, r.text, r.segments.map((g) => g.start)], [true, 'words at 0 words at 300 words at 600', [1.5, 301.5, 601.5]]);
++ assert.deepStrictEqual(progress, [[300, 1], [600, 2], [610, 3]]);
++ // Resume after a restart: finished windows are not run again.
++ runs.length = 0; tx.setLowPower(false);
++ r = await ma.transcribeOnly(url, { resumeFromSec: 600, priorSegments: [{ start: 1, end: 2, text: 'kept' }] });
++ assert.deepStrictEqual(runs.map((x) => [x.input.start_sec, x.input.low_power]), [[600, false]]);
++ assert.strictEqual(r.text, 'kept words at 600');
++ // A failed window is reported; the others still count.
++ runs.length = 0; failAt = 300;
++ r = await ma.transcribeOnly(url);
++ assert.deepStrictEqual([r.ok, r.segments.length], [true, 2]);
++ } finally {
++ for (const s of [network, ai, media]) s.close();
++ fs.rmSync(tmp, { recursive: true, force: true });
++ }
++ console.log('transcript via AI: all checks passed');
++ process.exit(0);
++})().catch((e) => { console.error(e); process.exit(1); });