openvibers/openvibe.live Commits

A clip whose speech-to-text failed is retried instead of keeping an empty transcript; the moments job reads chat spikes over the stream's own span (#337)

OP OpenVibers committed 21 days ago · 61d0bb5074 · 6 files, +107 −16

Changes

ddiff --git a/server/ai/ai-analysis.js b/server/ai/ai-analysis.js
iindex 7e792198..a9f8661d 100644
---- a/server/ai/ai-analysis.js
++++ b/server/ai/ai-analysis.js
@@@ -357,14 +357,24 @@ async function generateClipOverview(clip) {
});
const overview = (r && r.overview) ? r.overview : ' ';
const transcript = r ? r.transcript : '';
-- try { await db.setClipAiOverview(clip.id, { overview, transcript: transcript || null, segments: r ? r.segments : null }); } catch { /* */ }
-- // The overview pass already ran whisper — record it as done so the transcript poller
-- // doesn't re-transcribe the same clip. (Only when we actually got speech; a blank
-- // result is left for the dedicated, retrying transcript path to handle properly.)
-- try { if (transcript && transcript.trim()) await db.setClipTranscriptStatus(clip.id, 'done'); } catch { /* */ }
++ const speech = clipOverviewSpeech(r);
++ try { await db.setClipAiOverview(clip.id, { overview, transcript: transcript || null, segments: speech.segments }); } catch { /* */ }
++ try { if (speech.status) await db.setClipTranscriptStatus(clip.id, speech.status); } catch { /* */ }
return { overview: r ? r.overview : null, transcript };
}
++/**
++ * What a clip's overview pass records about speech. It already ran speech-to-text, so when that heard the clip its
++ * segments are stored and the status settled (done, or empty for no speech), and the transcript poller leaves the clip
++ * alone. When speech-to-text failed nothing is stored: segments null keeps ai_transcript_json NULL (never attempted),
++ * the poller's retrying path takes the clip. Storing [] there took 172 clips out of the queue for good (2026-10-23:
++ * AI could not read Media's virtual clips, and each was left `pending` or `retry` with an empty transcript).
++ */
++function clipOverviewSpeech(r) {
++ if (!r || !r.transcriptOk) return { segments: null, status: null };
++ return { segments: r.segments || [], status: String(r.transcript || '').trim() ? 'done' : 'empty' };
++}
++
/** Report AI config + optionally probe OpenVibe.AI with a one-word run. */
async function testStatus({ probe = true } = {}) {
const cfg = {
@@@ -388,7 +398,7 @@ module.exports = {
isEnabled, withinBudget, pasteAnalysisEnabled, streamMemoryEnabled, transcriptionEnabled, captureIntervalSec,
analyzeImagePaste, analyzeTextPaste, analyzeStreamFrame, summarizeStreamMemories,
generateStreamerOverview, generateVodOverview, generateClipOverview, ensureVodTimeline,
-- generateVodTranscript, generateClipTranscript, testStatus,
++ generateVodTranscript, generateClipTranscript, clipOverviewSpeech, testStatus,
sharedKeyReady,
complete: llm.complete, llm,
};
ddiff --git a/server/ai/ai-moments-job.js b/server/ai/ai-moments-job.js
iindex 225e9bf7..1debd8b5 100644
---- a/server/ai/ai-moments-job.js
++++ b/server/ai/ai-moments-job.js
@@@ -151,9 +151,11 @@ async function _momentContext(streamId, vodId) {
const transcript = _sample(await db.getStreamTranscriptSegments(streamId) || [], 60);
// Where viewers clipped: the clips live in OpenVibe.Media.
const clipTimes = await require('../media-proxy/lookups').clipStartTimes(streamId, vodId);
-- // Chat's own buckets since the stream began (offsets are relative to started_at, as before).
-- const started = await (async () => { try { const r = await db.get('SELECT started_at FROM streams WHERE id = ?', [streamId]); return r && r.started_at ? Date.parse(String(r.started_at).replace(' ', 'T') + 'Z') : 0; } catch { return 0; } })();
-- const spikes = started > 0 ? (await chatReads.spikeOffsets(streamId, 30, 8, started) || []) : [];
++ // Chat's own buckets over the stream (offsets are relative to started_at, as before; it ends at ended_at, or now).
++ const ms = (t) => (t ? Date.parse(String(t).replace(' ', 'T') + 'Z') || 0 : 0);
++ const span = await (async () => { try { return await db.get('SELECT started_at, ended_at FROM streams WHERE id = ?', [streamId]) || {}; } catch { return {}; } })();
++ const started = ms(span.started_at);
++ const spikes = started > 0 ? (await chatReads.spikeOffsets(streamId, 30, 8, started, ms(span.ended_at) || undefined) || []) : [];
// Non-speech sounds are strong moment candidates — an explosion or a burst of
// laughter marks a highlight as reliably as anything said out loud.
let sounds = [];
ddiff --git a/server/ai/media-analysis.js b/server/ai/media-analysis.js
iindex 60606db4..3ca6907c 100644
---- a/server/ai/media-analysis.js
++++ b/server/ai/media-analysis.js
@@@ -313,7 +313,7 @@ async function _transcribeSpan(src, duration, { resumeFromSec = 0, priorSegments
/**
* Analyze a media source (ffmpeg-consumable path or URL): spread frames → vision
* descriptions (optionally stored as memories) + sampled transcript → overview.
-- * @returns {{ overview, transcript, frames, duration }}
++ * @returns {{ overview, transcript, segments, transcriptOk, frames, duration }}
*/
async function analyzeMedia(src, { streamId = null, userId = null, numFrames = null, storeMemories = false, offsetBase = 0 } = {}) {
if (!src) return { overview: null, transcript: '', frames: [], duration: 0 };
@@@ -331,7 +331,7 @@ async function analyzeMedia(src, { streamId = null, userId = null, numFrames = n
frames = await captureFrameMemories(src, times, { streamId, userId, offsetBase, store: storeMemories });
}
-- const { text: transcript, segments } = await _transcribeSpan(src, duration);
++ const { text: transcript, segments, ok: transcriptOk } = await _transcribeSpan(src, duration);
let overview = null;
if ((frames.length || transcript) && await ai.isEnabled() && await ai.withinBudget()) {
@@@ -339,7 +339,8 @@ async function analyzeMedia(src, { streamId = null, userId = null, numFrames = n
const o = await require('./ai-service').structured('live.media.overview', { frames: frames.map(f => String(f.description || '').slice(0, 1000)).filter(Boolean).slice(0, 60), transcript: String(transcript || '').slice(0, 200000) }, { meter: { kind: 'media_overview', role: 'legacy', ownerUserId: userId } });
overview = o && o.overview ? String(o.overview).slice(0, 2000) : null;
}
-- return { overview: overview || null, transcript: transcript || '', segments: segments || [], frames, duration };
++ // transcriptOk: speech-to-text heard the media (an empty transcript then means no speech, not a failure).
++ return { overview: overview || null, transcript: transcript || '', segments: segments || [], transcriptOk: !!transcriptOk, frames, duration };
}
/**
ddiff --git a/server/chat/chat-reads.js b/server/chat/chat-reads.js
iindex 8b381e4f..2b49ea5d 100644
---- a/server/chat/chat-reads.js
++++ b/server/chat/chat-reads.js
@@@ -31,6 +31,7 @@ const client = require('./chat-client');
const CACHE_TTL_MS = 15_000; // how long an answer is used without asking Chat again
const CACHE_KEEP_MS = 15 * 60_000; // how long a last-good value answers peeks and failures
const FAIL_TTL_MS = 2_000; // how long a failure is not retried (it never replaces the value)
++const SPIKE_MAX_BUCKETS = 10_000; // Chat's timeline answers at most this many buckets
const cache = new Map(); // key -> { at, value, failed?, failedAt? }
const inflight = new Map(); // key -> Promise
@@@ -478,13 +479,16 @@ function recentChatText(streamId, sinceSec = 120, limit = 40) {
}, () => []);
}
--/** The busiest `bucketSec` time-buckets of a stream, as offsets from `sinceMs`: [{ offset, count }]. */
--function spikeOffsets(streamId, bucketSec = 30, topN = 8, sinceMs) {
++/** The busiest `bucketSec` time-buckets of a stream between `sinceMs` and `untilMs` (default now), as offsets from
++ * `sinceMs`: [{ offset, count }]. Chat answers at most 10,000 buckets, so a stream that ended gives its end: up to
++ * now, a stream from two months ago asked for 184,000 and got a 400 (2026-10-23). */
++function spikeOffsets(streamId, bucketSec = 30, topN = 8, sinceMs, untilMs) {
const id = Number(streamId) || 0;
if (!id) return Promise.resolve([]);
-- const key = `sp:${id}:${bucketSec}:${topN}:${sinceMs || 0}`;
++ const until = untilMs > (sinceMs || 0) ? Math.min(untilMs, (sinceMs || 0) + SPIKE_MAX_BUCKETS * bucketSec * 1000) : null;
++ const key = `sp:${id}:${bucketSec}:${topN}:${sinceMs || 0}:${until || 0}`;
return load(key, async () => {
-- const out = await client.readTimeline({ stream_id: id, since: sinceMs || 0, bucket_ms: bucketSec * 1000 });
++ const out = await client.readTimeline({ stream_id: id, since: sinceMs || 0, until, bucket_ms: bucketSec * 1000 });
if (!out || !Array.isArray(out.buckets)) throw unavailable('timeline');
const base = sinceMs || 0;
return out.buckets
ddiff --git a/test/chat-spike-window.test.js b/test/chat-spike-window.test.js
nnew file mode 100644
iindex 00000000..356f59fa
---- /dev/null
++++ b/test/chat-spike-window.test.js
@@@ -0,0 +1,35 @@
++'use strict';
++// The AI moments job's chat spikes (chat-reads.spikeOffsets) ask Chat's timeline for the stream's own span: from
++// started_at to ended_at, never more than Chat's 10,000 buckets. Asked up to now, a stream from two months ago needed
++// 184,000 thirty-second buckets and Chat answered 400 "Invalid bucket_ms" (2026-10-23: 12 recordings had no spikes).
++const assert = require('assert');
++console.warn = () => {};
++
++const client = require('../server/chat/chat-client');
++const calls = [];
++client.available = () => true;
++client.readTimeline = async (q) => { calls.push(q); return { buckets: [{ t: q.since + 60000, count: 3 }, { t: q.since + 30000, count: 9 }] }; };
++const chatReads = require('../server/chat/chat-reads');
++
++(async () => {
++ const started = Date.parse('2026-08-20T12:00:00Z');
++ const ended = started + 2 * 3600 * 1000;
++ const spikes = await chatReads.spikeOffsets(2125, 30, 8, started, ended);
++ assert.deepStrictEqual(spikes, [{ offset: 30, count: 9 }, { offset: 60, count: 3 }]);
++ assert.deepStrictEqual(calls.pop(), { stream_id: 2125, since: started, until: ended, bucket_ms: 30000 }, 'an ended stream is read up to its end');
++
++ // A stream still live (no end): up to now, as before; the client leaves a null `until` out of the query.
++ await chatReads.spikeOffsets(2126, 30, 8, started);
++ assert.strictEqual(calls.pop().until, null);
++
++ // An end past Chat's limit is held to 10,000 buckets from the start.
++ await chatReads.spikeOffsets(2127, 30, 8, started, started + 400 * 86400 * 1000);
++ const q = calls.pop();
++ assert.strictEqual((q.until - q.since) / q.bucket_ms, 10000);
++
++ // An end before the start is ignored (asked up to now).
++ await chatReads.spikeOffsets(2128, 30, 8, started, started - 1000);
++ assert.strictEqual(calls.pop().until, null);
++ console.log('chat-spike-window: ok');
++ process.exit(0);
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/clip-overview-speech.test.js b/test/clip-overview-speech.test.js
nnew file mode 100644
iindex 00000000..b60b0c13
---- /dev/null
++++ b/test/clip-overview-speech.test.js
@@@ -0,0 +1,39 @@
++'use strict';
++// A clip's overview pass records speech only when speech-to-text heard the clip (ai-analysis.clipOverviewSpeech):
++// heard with words → the segments and 'done'; heard, no words → [] and 'empty'; not heard (AI could not read the
++// media, whisper down) → nothing, so ai_transcript_json stays NULL and the transcript poller retries the clip. Storing
++// [] for a failure took 172 clips out of the queue for good (2026-10-23). Against the database: the stored row, and
++// getClipsNeedingTranscript picking the clip up only in the failure case.
++const assert = require('assert');
++console.warn = () => {};
++
++const db = require('../server/db/database');
++const { clipOverviewSpeech } = require('../server/ai/ai-analysis');
++
++(async () => {
++ await db.initDb();
++ const seg = [{ start: 0, end: 2, text: 'hello chat' }];
++ assert.deepStrictEqual(clipOverviewSpeech({ transcriptOk: true, transcript: 'hello chat', segments: seg }), { segments: seg, status: 'done' });
++ assert.deepStrictEqual(clipOverviewSpeech({ transcriptOk: true, transcript: '', segments: [] }), { segments: [], status: 'empty' });
++ assert.deepStrictEqual(clipOverviewSpeech({ transcriptOk: false, transcript: '', segments: [] }), { segments: null, status: null });
++ assert.deepStrictEqual(clipOverviewSpeech(null), { segments: null, status: null });
++
++ // What the overview pass then writes (generateClipOverview's two calls), for three clips.
++ const store = async (id, r) => {
++ const speech = clipOverviewSpeech(r);
++ await db.setClipAiOverview(id, { overview: 'A cooking stream.', segments: speech.segments });
++ if (speech.status) await db.setClipTranscriptStatus(id, speech.status);
++ };
++ await store(9101, { transcriptOk: true, transcript: 'hello chat', segments: seg });
++ await store(9102, { transcriptOk: true, transcript: '', segments: [] });
++ await store(9103, { transcriptOk: false, transcript: '', segments: [] });
++ const row = async (id) => db.get('SELECT ai_transcript_json, transcript_status FROM clip_ai_state WHERE clip_id = ?', [id]);
++ assert.deepStrictEqual({ ...await row(9101) }, { ai_transcript_json: JSON.stringify(seg), transcript_status: 'done' });
++ assert.deepStrictEqual({ ...await row(9102) }, { ai_transcript_json: '[]', transcript_status: 'empty' });
++ assert.strictEqual((await row(9103)).ai_transcript_json, null, 'a failed pass stores no transcript');
++ const queued = (await db.getClipsNeedingTranscript(50)).map((r) => Number(r.id));
++ assert.ok(queued.includes(9103), 'the poller retries the clip speech-to-text could not hear');
++ assert.ok(!queued.includes(9101) && !queued.includes(9102), 'a heard clip is not transcribed again');
++ console.log('clip-overview-speech: ok');
++ process.exit(0);
++})().catch((e) => { console.error(e); process.exit(1); });