openvibers/openvibe.ai Commits

Deep dive fixes: capture worker on 4711, and a report when the synthesis model fails (#495)

OP OpenVibers committed 1 days ago · 064d294e95 · 6 files, +90 −13

* Deep dive: the capture worker listens on 4711 (Search owns 4710)

* Deep dive: a synthesis that does not answer still reports from the analyses

* Deep dive: Media reads need media.object.read on the service token

Changes

ddiff --git a/deploy/systemd/openvibe-ai-capture.service b/deploy/systemd/openvibe-ai-capture.service
iindex 092e306..575e695 100644
---- a/deploy/systemd/openvibe-ai-capture.service
++++ b/deploy/systemd/openvibe-ai-capture.service
@@@ -1,6 +1,6 @@
[Unit]
Description=OpenVibe.AI capture worker — headless Chrome for deep dives (one browser at a time, server/capture/)
--# The API calls this unit on 127.0.0.1:4710 (AI_CAPTURE_URL in openvibe-ai.service). It has its own unit so a page
++# The API calls this unit on 127.0.0.1:4711 (AI_CAPTURE_URL in openvibe-ai.service). It has its own unit so a page
# that eats memory is killed by MemoryMax without taking the API down: the host has about 5 GB of free memory and
# never starts a second browser. AI_DRILL/LIVE_DRILL turns it off in code (POST /capture answers 503 capture.drill).
After=network.target
@@@ -14,7 +14,7 @@ WorkingDirectory=/opt/openvibe.ai
EnvironmentFile=/etc/openvibe/ai.env
Environment=NODE_ENV=production
Environment=AI_CAPTURE_HOST=127.0.0.1
--Environment=AI_CAPTURE_PORT=4710
++Environment=AI_CAPTURE_PORT=4711
Environment=AI_CAPTURE_CHROME=/usr/bin/google-chrome
Environment=OV_MEDIA_INTERNAL_URL=http://127.0.0.1:4100
# Chrome writes its profile and its caches under HOME; the code tree stays read-only.
ddiff --git a/deploy/systemd/openvibe-ai.service b/deploy/systemd/openvibe-ai.service
iindex 19be734..695f006 100644
---- a/deploy/systemd/openvibe-ai.service
++++ b/deploy/systemd/openvibe-ai.service
@@@ -31,8 +31,8 @@ ProtectHome=read-only
StateDirectory=openvibe-ai
# media.analyze's work directory under StateDirectory (writable; the code tree is read-only).
Environment=AI_MEDIA_ANALYSIS_DIR=/var/lib/openvibe-ai/media-tmp
--# Deep dives' screenshots are taken by the capture worker (openvibe-ai-capture.service, 127.0.0.1:4710).
--Environment=AI_CAPTURE_URL=http://127.0.0.1:4710
++# Deep dives' screenshots are taken by the capture worker (openvibe-ai-capture.service, 127.0.0.1:4711).
++Environment=AI_CAPTURE_URL=http://127.0.0.1:4711
# Media's internal origin (its unit's port 4100): screenshots and the deep dive's private shots are Media objects.
Environment=OV_MEDIA_INTERNAL_URL=http://127.0.0.1:4100
ddiff --git a/server/config.js b/server/config.js
iindex 9ae03f7..703eb52 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -389,6 +389,8 @@ function load(env = process.env) {
maxSources: { quick: 40, thorough: 80, expert: 140 },
analyseViewports: { quick: 1, thorough: 1, expert: 2 },
maxOutputTokens: Math.max(400, Math.min(8000, int(env.AI_DEEP_DIVE_MAX_OUTPUT_TOKENS, 3000))),
++ // The longest answer: the background JSON route may wait for the cheapest correct model.
++ synthesisRoute: String(env.AI_DEEP_DIVE_SYNTHESIS_ROUTE || 'background.json').trim(),
},
// The capture worker (server/capture/): headless Chrome through playwright-core, one browser at a time.
// AI_CAPTURE_URL points at the worker's loopback unit; unset runs the same capture in this process (tests).
ddiff --git a/server/deep-dive/index.js b/server/deep-dive/index.js
iindex bb8a904..37f156c 100644
---- a/server/deep-dive/index.js
++++ b/server/deep-dive/index.js
@@@ -108,6 +108,50 @@ function createDeepDive({ config = {}, db, clock = { now: () => Date.now() }, lo
return [{ claim: oneLine(item.claim || 'observed on the site', 600), source_id: site.source_id, capture_id: cap ? cap.id : null, crop_id: cap && cap.crops && cap.crops[0] ? cap.crops[0].id : null, url: site.url, read_at: cap ? cap.captured_at : iso(clock.now()) }];
}
++ /**
++ * A synthesis that does not depend on the model: used when the synthesis call fails (a truncated JSON answer,
++ * a provider refusal) so a run that already read real sites still reports what it found, with a gap naming the
++ * missing step. It is assembled from the per-site analyses and the rules register only; nothing is invented.
++ */
++ function fallbackSynthesis({ analysis, capturesBySite, rules, question, tokens }) {
++ const sites = analysis.map((a) => ({ source_id: a.source_id, url: a.url }));
++ const ev = (a, claim) => {
++ const cap = (capturesBySite.get(a.source_id) || [])[0];
++ return [{ claim: oneLine(claim, 600), source_id: a.source_id, capture_id: cap ? cap.id : null, crop_id: cap && cap.crops && cap.crops[0] ? cap.crops[0].id : null, url: a.url, read_at: cap ? cap.captured_at : iso(clock.now()) }];
++ };
++ const strengths = new Map();
++ const weaknesses = [];
++ for (const a of analysis) {
++ for (const x of a.strengths || []) strengths.set(x, [...(strengths.get(x) || []), a]);
++ for (const x of a.weaknesses || []) weaknesses.push({ a, text: x });
++ }
++ const patterns = [...strengths.entries()].slice(0, 10).map(([name, list]) => ({ name: oneLine(name, 160), what: `Seen on ${list.length} of ${analysis.length} sites in this run.`, sites: list.map((a) => a.source_id), usage_count: list.length, evidence: list.slice(0, 2).flatMap((a) => ev(a, name)) }));
++ if (!patterns.length) patterns.push({ name: 'What the sites are', what: analysis.map((a) => a.summary).filter(Boolean).slice(0, 3).join(' ').slice(0, 1200), sites: sites.map((x) => x.source_id).slice(0, 30), usage_count: sites.length, evidence: analysis.slice(0, 2).flatMap((a) => ev(a, a.summary)) });
++ const best = analysis.slice().sort((a, b) => (b.strengths || []).length - (a.strengths || []).length)[0] || null;
++ const recs = weaknesses.slice(0, 6).map((w, i) => ({
++ id: `rec_${i + 1}`, what: `Avoid: ${oneLine(w.text, 500)}`, why: `Seen on ${w.a.url}; the site analysis says it fails there.`,
++ rule: { id: null, status: 'none', text: '' }, evidence: ev(w.a, w.text),
++ }));
++ if (!recs.length && best) recs.push({ id: 'rec_1', what: `The best version is ${best.url}: keep what it does well.`, why: 'It has the most strengths among the sites read.', rule: { id: null, status: 'none', text: '' }, evidence: ev(best, best.summary) });
++ return {
++ patterns,
++ comparison: { columns: ['Site', 'Summary', 'Good', 'Fails'], rows: analysis.slice(0, 30).map((a) => [a.url, oneLine(a.summary, 600), (a.strengths || []).join('; ').slice(0, 800), (a.weaknesses || []).join('; ').slice(0, 800)]) },
++ best: best ? { source_id: best.source_id, what: best.summary, evidence: ev(best, best.summary) } : { source_id: null, what: '', evidence: [] },
++ mistakes: weaknesses.slice(0, 12).map((w) => ({ what: w.text, evidence: ev(w.a, w.text) })),
++ gap: `A gap in this run: the synthesis model did not answer, so this is assembled from the site analyses. ${oneLine(question, 200)} was not compared by a model.`,
++ reference_tokens: {
++ palettes: tokens.map((t) => ({ source_id: t.source_id, colours: t.palettes })).filter((x) => x.colours.length).slice(0, 30),
++ type_scales: tokens.map((t) => ({ source_id: t.source_id, families: t.families, sizes: t.sizes })).filter((x) => x.families.length || x.sizes.length).slice(0, 30),
++ spacing: tokens.map((t) => ({ source_id: t.source_id, steps: t.spacing })).filter((x) => x.steps.length).slice(0, 30),
++ },
++ do: [...strengths.keys()].slice(0, 12).map((x) => oneLine(x, 600)),
++ dont: weaknesses.slice(0, 12).map((w) => oneLine(w.text, 600)),
++ accessibility: analysis.map((a) => a.accessibility).filter(Boolean).slice(0, 8),
++ speed: analysis.map((a) => a.speed).filter(Boolean).slice(0, 8),
++ recommendations: recs,
++ };
++ }
++
/** Run one deep dive for one run row. Returns { output, usage, cost, models } for the workflow step. */
async function run({ run, input, ctx = {} }) {
const question = oneLine(input.question, 500);
@@@ -336,13 +380,24 @@ function createDeepDive({ config = {}, db, clock = { now: () => Date.now() }, lo
spacing: (dom.spacing || []).slice(0, 8).map((x) => x.value),
};
});
-- const synthCall = await llm.call({ routeKey: 'default.json', ...prompts.synthesisPrompt({ question, depth, analyses: analysis, sources: allSources, rules: gathered.rules || [], components: gathered.components || [], tokens }), schema: SYNTHESIS_OUTPUT, schemaName: 'deep_dive_synthesis', maxTokens: Math.min(3500, maxOut), signal: ctx.signal, ctx });
-- addCall('synthesise', synthCall);
-- const s = synthCall.json || {};
++ // The synthesis is the longest answer: a bigger budget than the other steps, on the background JSON
++ // route (it may wait for the cheapest correct model). If the model still truncates or refuses, the
++ // report is assembled from the site analyses and names the missing synthesis as a gap.
++ let s = null;
++ try {
++ const synthCall = await llm.call({ routeKey: cfg.synthesisRoute || 'background.json', ...prompts.synthesisPrompt({ question, depth, analyses: analysis, sources: allSources, rules: gathered.rules || [], components: gathered.components || [], tokens }), schema: SYNTHESIS_OUTPUT, schemaName: 'deep_dive_synthesis', maxTokens: Math.min(5000, Math.max(4000, maxOut)), signal: ctx.signal, ctx });
++ addCall('synthesise', synthCall);
++ s = synthCall.json || null;
++ } catch (err) {
++ if (err && err.code === 'run.cancelled') throw err;
++ warnings.push(`synthesis: ${String((err && err.message) || err).slice(0, 240)}`);
++ gaps.push('The synthesis model did not answer; the synthesis was assembled from the site analyses.');
++ s = fallbackSynthesis({ analysis, capturesBySite, rules: gathered.rules || [], question, tokens });
++ }
const sitesForEvidence = analysis.map((a) => ({ source_id: a.source_id, url: a.url }));
const ruleIds = new Set((gathered.rules || []).map((r) => r.id));
-- const mapEvidence = (list) => (Array.isArray(list) ? list : []).flatMap((e) => evidenceOf(e, sitesForEvidence, capturesBySite)).slice(0, 20);
-- const bestSite = sitesForEvidence[(Number(s.best && s.best.site) || 1) - 1] || sitesForEvidence[0];
++ const mapEvidence = (list) => (Array.isArray(list) ? list : []).flatMap((e) => (e && e.source_id && e.url ? [e] : evidenceOf(e, sitesForEvidence, capturesBySite))).slice(0, 20);
++ const bestSite = (s.best && s.best.source_id ? sitesForEvidence.find((x) => x.source_id === s.best.source_id) : sitesForEvidence[(Number(s.best && s.best.site) || 1) - 1]) || sitesForEvidence[0];
const synthesis = {
patterns: (s.patterns || []).slice(0, 40).map((p) => ({
name: oneLine(p.name, 160), what: oneLine(p.what, 1200),
@@@ -354,9 +409,9 @@ function createDeepDive({ config = {}, db, clock = { now: () => Date.now() }, lo
mistakes: (s.mistakes || []).slice(0, 20).map((m) => ({ what: oneLine(m.what, 1200), evidence: mapEvidence(m.evidence) })),
gap: oneLine(s.gap, 2000),
reference_tokens: {
-- palettes: (s.reference_tokens && s.reference_tokens.palettes || []).map((p) => ({ source_id: (sitesForEvidence[(Number(p.site) || 1) - 1] || {}).source_id, colours: (p.colours || []).slice(0, 16) })).filter((p) => p.source_id).slice(0, 30),
-- type_scales: (s.reference_tokens && s.reference_tokens.type_scales || []).map((p) => ({ source_id: (sitesForEvidence[(Number(p.site) || 1) - 1] || {}).source_id, families: (p.families || []).slice(0, 8), sizes: (p.sizes || []).slice(0, 16) })).filter((p) => p.source_id).slice(0, 30),
-- spacing: (s.reference_tokens && s.reference_tokens.spacing || []).map((p) => ({ source_id: (sitesForEvidence[(Number(p.site) || 1) - 1] || {}).source_id, steps: (p.steps || []).slice(0, 16) })).filter((p) => p.source_id).slice(0, 30),
++ palettes: (s.reference_tokens && s.reference_tokens.palettes || []).map((p) => ({ source_id: p.source_id || (sitesForEvidence[(Number(p.site) || 1) - 1] || {}).source_id, colours: (p.colours || []).slice(0, 16) })).filter((p) => p.source_id).slice(0, 30),
++ type_scales: (s.reference_tokens && s.reference_tokens.type_scales || []).map((p) => ({ source_id: p.source_id || (sitesForEvidence[(Number(p.site) || 1) - 1] || {}).source_id, families: (p.families || []).slice(0, 8), sizes: (p.sizes || []).slice(0, 16) })).filter((p) => p.source_id).slice(0, 30),
++ spacing: (s.reference_tokens && s.reference_tokens.spacing || []).map((p) => ({ source_id: p.source_id || (sitesForEvidence[(Number(p.site) || 1) - 1] || {}).source_id, steps: (p.steps || []).slice(0, 16) })).filter((p) => p.source_id).slice(0, 30),
},
do: (s.do || []).slice(0, 20).map((x) => oneLine(x, 600)),
dont: (s.dont || []).slice(0, 20).map((x) => oneLine(x, 600)),
ddiff --git a/server/media-store.js b/server/media-store.js
iindex e182127..f39438b 100644
---- a/server/media-store.js
++++ b/server/media-store.js
@@@ -26,7 +26,8 @@ function createMediaStore({ config, env = process.env, fetchImpl = globalThis.fe
client = createServiceTokenClient({
tokenUrl: `${String(config.networkInternalUrl).replace(/\/+$/, '')}/oauth/token`,
clientId: env.OV_OAUTH_CLIENT_ID || 'ai', clientSecret: secret,
-- audience: 'openvibe.media', scope: 'media.object.upload', fetch: fetchImpl,
++ // Upload for a picture or a deep dive's screenshot; read for the signed URL a private object needs.
++ audience: 'openvibe.media', scope: 'media.object.upload media.object.read', fetch: fetchImpl,
});
return client;
};
ddiff --git a/test/deep-dive.test.js b/test/deep-dive.test.js
iindex 4a9465a..c3026e5 100644
---- a/test/deep-dive.test.js
++++ b/test/deep-dive.test.js
@@@ -197,4 +197,23 @@ async function waitFor(h, id, tok, wanted, tries = 120) {
assert.strictEqual(slow.calls.length, calls, 'cancelling stopped every later capture');
process.stdout.write('deep dive: cancel stops the next capture and model call\n');
} finally { await h2.close(); }
++
++ // ── A synthesis model that does not answer still leaves a report, assembled from the analyses ──
++ const f3 = fixtures();
++ const h3 = await boot({ env: env({ AI_DEEP_DIVE_SYNTHESIS_ROUTE: 'no.such.route' }), research: f3.research, search: f3.search, capture: fakeCapture(), deepDiveFetch: f3.sourceFetch });
++ try {
++ const exchange = token('exchange', ['ai.run.create'], { sub: 'svc:exchange', ns: ['ai.complete'] });
++ const r = await request(h3.base, 'POST', '/api/v1/deep-dives', { tok: exchange, body: { question: 'a stream overlay deck', depth: 'quick', on_behalf_of: { type: 'user', id: USER } } });
++ assert.strictEqual(r.status, 202);
++ const id = r.body.deep_dive.id;
++ const done = await waitFor(h3, id, exchange, ['succeeded', 'failed', 'cancelled']);
++ assert.strictEqual(done.status, 'succeeded', JSON.stringify(done.error));
++ const full = await request(h3.base, 'GET', `/api/v1/deep-dives/${id}`, { tok: exchange });
++ const report = full.body.deep_dive.report;
++ assert.ok(report, 'a report exists even when the synthesis model does not answer');
++ assert.ok(report.gaps.some((g) => /The synthesis model did not answer/.test(g)), 'the gap names the missing synthesis');
++ assert.ok(report.synthesis.patterns.length >= 1 && report.recommendations.length >= 1, 'the synthesis was assembled from the site analyses');
++ assert.ok(report.synthesis.patterns[0].evidence.length >= 1, 'the assembled synthesis still cites its evidence');
++ process.stdout.write('deep dive: a failed synthesis still reports from the analyses, with a gap\n');
++ } finally { await h3.close(); }
})().catch((e) => { process.stderr.write(String(e && e.stack || e) + '\n'); process.exit(1); });