openvibers/openvibe.ai Commits
Citations or explicit gaps on every run (WS-O task 3): each run records grounding { cited, gaps }: the source ordinals its output cites, and the gaps it names; an output that cites nothing and names no gap gets one written for it (no sources, or sources not cited), so nothing reads as verified when it is not. Covers live.* and network.site_copy like every other workflow; a cached run carries the grounding of the run it reuses; runs.grounding is an added column
Changes
@@@ -306,6 +306,9 @@ function openDb(dbPath) {db.pragma('foreign_keys = ON');db.pragma('busy_timeout = 5000');db.exec(SCHEMA);++ // Added after the first release: ADD COLUMN only, so an older build ignores it.++ const cols = new Set(db.prepare("SELECT name FROM pragma_table_info('runs')").all().map((c) => c.name));++ if (!cols.has('grounding')) db.exec('ALTER TABLE runs ADD COLUMN grounding TEXT'); // { cited, gaps } (WS-O task 3)return db;}@@@ -86,6 +86,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now},usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts },citations_count: r.citations_count,++ grounding: parseJson(r.grounding, null),retry_of: r.retry_of,idempotency_key: r.idempotency_key,trace_id: r.trace_id,@@@ -202,6 +203,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now.run({ ...base, output: JSON.stringify(hit.output), model_key: hit.model_key, cache_key: cacheKey, cached_from: hit.run_id });const src = citations(hit.run_id);if (src.length) addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');++ // The reused output carries the grounding it was produced with.++ db.prepare('UPDATE runs SET grounding = (SELECT grounding FROM runs WHERE id = ?) WHERE id = ?').run(hit.run_id, id);events.runChanged(getRow(id));})();registry.audit(principal.sub, 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, status: 'cached', cached_from: hit.run_id } });@@@ -285,9 +288,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowif (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');// The status, its citations and the ai.run.succeeded event commit together.const done = db.transaction(() => {-- const d = db.prepare(`UPDATE runs SET status = 'succeeded', output = ?, synthetic = ?, provider_key = ?, model_key = ?, fallback_used = ?, attempts = ?, tokens_in = ?, tokens_out = ?, cost_usd = ?,++ const d = db.prepare(`UPDATE runs SET status = 'succeeded', output = ?, grounding = ?, synthetic = ?, provider_key = ?, model_key = ?, fallback_used = ?, attempts = ?, tokens_in = ?, tokens_out = ?, cost_usd = ?,route_key = COALESCE(?, route_key), route_version = COALESCE(?, route_version), finished_at = ? WHERE id = ? AND status = 'running'`)-- .run(JSON.stringify(r.output), r.synthetic ? 1 : 0, r.provider, r.model, r.fallbackUsed ? 1 : 0, attempts, r.usage.input, r.usage.output, r.cost,++ .run(JSON.stringify(r.output), JSON.stringify(r.grounding || null), r.synthetic ? 1 : 0, r.provider, r.model, r.fallbackUsed ? 1 : 0, attempts, r.usage.input, r.usage.output, r.cost,r.route ? r.route.key : null, r.route ? r.route.version : null, iso(clock.now()), id);if (!d.changes) return d;if (r.citations.length) addCitations(id, r.citations, 'workflow');@@@ -14,6 +14,11 @@* The final output is validated against the workflow's output schema. An empty or invalid answer* fails the run (output.empty / output.invalid); nothing is ever filled in. input.sources become* citation rows, marked cited when the output's `citations` arrays point at them.++ *++ * Every run also gets `grounding` { cited, gaps } (roadmap WS-O task 3): the source ordinals the++ * output cites, and what it cannot back up. Gaps come from the output's own `gaps` arrays; an output++ * that cites nothing and names no gap gets one written for it (no sources, or sources not cited), so++ * no output ever reads as verified when it is not.*/const fs = require('fs');const { AiError, sha256, parseJsonLoose } = require('../util');@@@ -24,6 +29,29 @@ const { PREPARE, POSTPROCESS } = require('./hooks');const ROLE_TIMEOUT = { chat: 20000, vision: 30000, director: 25000, summary: 30000, legacy: 30000 };++/** Every string in every `gaps` array of an output, in order, without repeats. */++function collectGaps(v, out = []) {++ if (Array.isArray(v)) { v.forEach((x) => collectGaps(x, out)); return out; }++ if (!v || typeof v !== 'object') return out;++ for (const [k, x] of Object.entries(v)) {++ if (k === 'gaps' && Array.isArray(x)) x.forEach((g) => { const s = typeof g === 'string' ? g.trim() : ''; if (s && !out.includes(s)) out.push(s.slice(0, 300)); });++ else collectGaps(x, out);++ }++ return out;++}++++/** { cited, gaps } for one output: never empty-handed, so nothing reads as verified by default. */++function groundingOf(output, sourceCount, citeStep) {++ const cited = citeStep ? [...collectCited(output)].filter((i) => i < sourceCount).sort((a, b) => a - b) : [];++ const gaps = collectGaps(output).slice(0, 20);++ if (!cited.length && !gaps.length) {++ gaps.push(sourceCount === 0++ ? 'No sources were given: this is based on the input alone, and nothing in it is independently verified.'++ : citeStep ? 'None of the given sources is cited: nothing in this is backed by them.' : 'Sources were given but this workflow does not cite them: nothing in this is backed by them.');++ }++ return { cited, gaps };++}++/** Deep copy of a schema with every `citations` array narrowed to valid source indices. */function narrowCitations(schema, count) {if (!schema || typeof schema !== 'object') return schema;@@@ -212,6 +240,7 @@ function createEngine({ registry, pool, fetcher }) {}));return {output: last.output, citations, jsonSchema,++ grounding: groundingOf(last.output, sources.length, citeStep),synthetic: Boolean(exec && exec.synthetic),provider: exec ? exec.provider : null, model: exec ? exec.model : null, fallbackUsed: Boolean(exec && exec.fallbackUsed),usage, cost,@@@ -222,4 +251,4 @@ function createEngine({ registry, pool, fetcher }) {return { execute, narrowCitations };}--module.exports = { createEngine, narrowCitations, collectCited };++module.exports = { createEngine, narrowCitations, collectCited, collectGaps, groundingOf };@@@ -34,6 +34,8 @@ t.test("actor A's repeat is served from A's cache without a provider call", asynassert.strictEqual(again.body.run.provenance.cached_from, first.body.run.id);assert.deepStrictEqual(again.body.run.output, first.body.run.output);assert.strictEqual(seam.calls, calls);++ assert.ok(first.body.run.grounding && first.body.run.grounding.gaps.length, 'the first run says what backs it (WS-O task 3)');++ assert.deepStrictEqual(again.body.run.grounding, first.body.run.grounding, 'a cached run carries the grounding of the run it reuses');});t.test("actor B never receives actor A's private cache entry", async () => {@@@ -90,6 +90,9 @@ t.test('every workflow runs end to end on the stub and returns schema-valid outpassert.strictEqual(run.provenance.workflow, w.key);assert.ok(run.provenance.model, `${w.key} has a model`);assert.ok(schemas.validate(w.output_schema, run.output).valid, `${w.key} output invalid`);++ // WS-O task 3: every output says what backs it: cited sources, or explicit gaps.++ assert.ok(run.grounding && Array.isArray(run.grounding.cited) && Array.isArray(run.grounding.gaps), `${w.key} has grounding`);++ assert.ok(run.grounding.cited.length || run.grounding.gaps.length, `${w.key}: citations or explicit gaps`);if (EXAMPLES[w.key].sources) {assert.strictEqual(run.citations_count, EXAMPLES[w.key].sources.length, `${w.key} citations`);const d = await request(h.base, 'GET', `/api/v1/runs/${run.id}`, { tok });@@@ -99,6 +102,16 @@ t.test('every workflow runs end to end on the stub and returns schema-valid outp}});++t.test('grounding: cited sources, the output\'s own gaps, or a gap written for it (WS-O task 3)', async () => {++ const { groundingOf } = require('../server/workflows/engine');++ assert.deepStrictEqual(groundingOf({ summary: 'x', citations: [1, 0, 1, 7] }, 2, true), { cited: [0, 1], gaps: [] }, 'cited ordinals in range, sorted, once');++ assert.deepStrictEqual(groundingOf({ items: [{ gaps: ['no price given'] }], gaps: ['no price given', 'date unclear'] }, 1, true).gaps, ['no price given', 'date unclear'], 'the output\'s gaps, once each');++ assert.match(groundingOf({ text: 'hola' }, 0, false).gaps[0], /No sources were given/);++ assert.match(groundingOf({ summary: 'x', citations: [] }, 3, true).gaps[0], /None of the given sources is cited/);++ assert.match(groundingOf({ summary: 'x' }, 2, false).gaps[0], /does not cite them/);++ // (A cached run carrying its source run's grounding: test/cache.test.js; stub output is never cached.)++});++t.test('citations index only the provided sources', async () => {const r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok, body: { workflow: 'news.summarize_story', input: { sources: SRC(2) } } });const seen = JSON.stringify(r.body.run.output).match(/"citations":\[([0-9,]*)\]/g) || [];
@@ -306,6 +306,9 @@ function openDb(dbPath) {
@@ -306,6 +306,9 @@ function openDb(dbPath) {
db.pragma('foreign_keys = ON');
db.pragma('foreign_keys = ON');
db.pragma('busy_timeout = 5000');
db.pragma('busy_timeout = 5000');
db.exec(SCHEMA);
db.exec(SCHEMA);
+ // Added after the first release: ADD COLUMN only, so an older build ignores it.
+ const cols = new Set(db.prepare("SELECT name FROM pragma_table_info('runs')").all().map((c) => c.name));
+ if (!cols.has('grounding')) db.exec('ALTER TABLE runs ADD COLUMN grounding TEXT'); // { cited, gaps } (WS-O task 3)
return db;
return db;
}
}
@@ -86,6 +86,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -86,6 +86,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
},
},
usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts },
usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts },
citations_count: r.citations_count,
citations_count: r.citations_count,
+ grounding: parseJson(r.grounding, null),
retry_of: r.retry_of,
retry_of: r.retry_of,
idempotency_key: r.idempotency_key,
idempotency_key: r.idempotency_key,
trace_id: r.trace_id,
trace_id: r.trace_id,
@@ -202,6 +203,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -202,6 +203,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
.run({ ...base, output: JSON.stringify(hit.output), model_key: hit.model_key, cache_key: cacheKey, cached_from: hit.run_id });
.run({ ...base, output: JSON.stringify(hit.output), model_key: hit.model_key, cache_key: cacheKey, cached_from: hit.run_id });
const src = citations(hit.run_id);
const src = citations(hit.run_id);
if (src.length) addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');
if (src.length) addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');
+ // The reused output carries the grounding it was produced with.
+ db.prepare('UPDATE runs SET grounding = (SELECT grounding FROM runs WHERE id = ?) WHERE id = ?').run(hit.run_id, id);
events.runChanged(getRow(id));
events.runChanged(getRow(id));
})();
})();
registry.audit(principal.sub, 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, status: 'cached', cached_from: hit.run_id } });
registry.audit(principal.sub, 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, status: 'cached', cached_from: hit.run_id } });
@@ -285,9 +288,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -285,9 +288,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');
if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');
// The status, its citations and the ai.run.succeeded event commit together.
// The status, its citations and the ai.run.succeeded event commit together.
const done = db.transaction(() => {
const done = db.transaction(() => {
- const d = db.prepare(`UPDATE runs SET status = 'succeeded', output = ?, synthetic = ?, provider_key = ?, model_key = ?, fallback_used = ?, attempts = ?, tokens_in = ?, tokens_out = ?, cost_usd = ?,
+ const d = db.prepare(`UPDATE runs SET status = 'succeeded', output = ?, grounding = ?, synthetic = ?, provider_key = ?, model_key = ?, fallback_used = ?, attempts = ?, tokens_in = ?, tokens_out = ?, cost_usd = ?,
route_key = COALESCE(?, route_key), route_version = COALESCE(?, route_version), finished_at = ? WHERE id = ? AND status = 'running'`)
route_key = COALESCE(?, route_key), route_version = COALESCE(?, route_version), finished_at = ? WHERE id = ? AND status = 'running'`)
- .run(JSON.stringify(r.output), r.synthetic ? 1 : 0, r.provider, r.model, r.fallbackUsed ? 1 : 0, attempts, r.usage.input, r.usage.output, r.cost,
+ .run(JSON.stringify(r.output), JSON.stringify(r.grounding || null), r.synthetic ? 1 : 0, r.provider, r.model, r.fallbackUsed ? 1 : 0, attempts, r.usage.input, r.usage.output, r.cost,
r.route ? r.route.key : null, r.route ? r.route.version : null, iso(clock.now()), id);
r.route ? r.route.key : null, r.route ? r.route.version : null, iso(clock.now()), id);
if (!d.changes) return d;
if (!d.changes) return d;
if (r.citations.length) addCitations(id, r.citations, 'workflow');
if (r.citations.length) addCitations(id, r.citations, 'workflow');
@@ -14,6 +14,11 @@
@@ -14,6 +14,11 @@
* The final output is validated against the workflow's output schema. An empty or invalid answer
* The final output is validated against the workflow's output schema. An empty or invalid answer
* fails the run (output.empty / output.invalid); nothing is ever filled in. input.sources become
* fails the run (output.empty / output.invalid); nothing is ever filled in. input.sources become
* citation rows, marked cited when the output's `citations` arrays point at them.
* citation rows, marked cited when the output's `citations` arrays point at them.
+ *
+ * Every run also gets `grounding` { cited, gaps } (roadmap WS-O task 3): the source ordinals the
+ * output cites, and what it cannot back up. Gaps come from the output's own `gaps` arrays; an output
+ * that cites nothing and names no gap gets one written for it (no sources, or sources not cited), so
+ * no output ever reads as verified when it is not.
*/
*/
const fs = require('fs');
const fs = require('fs');
const { AiError, sha256, parseJsonLoose } = require('../util');
const { AiError, sha256, parseJsonLoose } = require('../util');
@@ -24,6 +29,29 @@ const { PREPARE, POSTPROCESS } = require('./hooks');
@@ -24,6 +29,29 @@ const { PREPARE, POSTPROCESS } = require('./hooks');
const ROLE_TIMEOUT = { chat: 20000, vision: 30000, director: 25000, summary: 30000, legacy: 30000 };
const ROLE_TIMEOUT = { chat: 20000, vision: 30000, director: 25000, summary: 30000, legacy: 30000 };
+/** Every string in every `gaps` array of an output, in order, without repeats. */
+function collectGaps(v, out = []) {
+ if (Array.isArray(v)) { v.forEach((x) => collectGaps(x, out)); return out; }
+ if (!v || typeof v !== 'object') return out;
+ for (const [k, x] of Object.entries(v)) {
+ if (k === 'gaps' && Array.isArray(x)) x.forEach((g) => { const s = typeof g === 'string' ? g.trim() : ''; if (s && !out.includes(s)) out.push(s.slice(0, 300)); });
+ else collectGaps(x, out);
+ }
+ return out;
+}
+
+/** { cited, gaps } for one output: never empty-handed, so nothing reads as verified by default. */
+function groundingOf(output, sourceCount, citeStep) {
+ const cited = citeStep ? [...collectCited(output)].filter((i) => i < sourceCount).sort((a, b) => a - b) : [];
+ const gaps = collectGaps(output).slice(0, 20);
+ if (!cited.length && !gaps.length) {
+ gaps.push(sourceCount === 0
+ ? 'No sources were given: this is based on the input alone, and nothing in it is independently verified.'
+ : citeStep ? 'None of the given sources is cited: nothing in this is backed by them.' : 'Sources were given but this workflow does not cite them: nothing in this is backed by them.');
+ }
+ return { cited, gaps };
+}
+
/** Deep copy of a schema with every `citations` array narrowed to valid source indices. */
/** Deep copy of a schema with every `citations` array narrowed to valid source indices. */
function narrowCitations(schema, count) {
function narrowCitations(schema, count) {
if (!schema || typeof schema !== 'object') return schema;
if (!schema || typeof schema !== 'object') return schema;
@@ -212,6 +240,7 @@ function createEngine({ registry, pool, fetcher }) {
@@ -212,6 +240,7 @@ function createEngine({ registry, pool, fetcher }) {
}));
}));
return {
return {
output: last.output, citations, jsonSchema,
output: last.output, citations, jsonSchema,
+ grounding: groundingOf(last.output, sources.length, citeStep),
synthetic: Boolean(exec && exec.synthetic),
synthetic: Boolean(exec && exec.synthetic),
provider: exec ? exec.provider : null, model: exec ? exec.model : null, fallbackUsed: Boolean(exec && exec.fallbackUsed),
provider: exec ? exec.provider : null, model: exec ? exec.model : null, fallbackUsed: Boolean(exec && exec.fallbackUsed),
usage, cost,
usage, cost,
@@ -222,4 +251,4 @@ function createEngine({ registry, pool, fetcher }) {
@@ -222,4 +251,4 @@ function createEngine({ registry, pool, fetcher }) {
return { execute, narrowCitations };
return { execute, narrowCitations };
}
}
-module.exports = { createEngine, narrowCitations, collectCited };
+module.exports = { createEngine, narrowCitations, collectCited, collectGaps, groundingOf };
@@ -34,6 +34,8 @@ t.test("actor A's repeat is served from A's cache without a provider call", asyn
@@ -34,6 +34,8 @@ t.test("actor A's repeat is served from A's cache without a provider call", asyn
assert.strictEqual(again.body.run.provenance.cached_from, first.body.run.id);
assert.strictEqual(again.body.run.provenance.cached_from, first.body.run.id);
assert.deepStrictEqual(again.body.run.output, first.body.run.output);
assert.deepStrictEqual(again.body.run.output, first.body.run.output);
assert.strictEqual(seam.calls, calls);
assert.strictEqual(seam.calls, calls);
+ assert.ok(first.body.run.grounding && first.body.run.grounding.gaps.length, 'the first run says what backs it (WS-O task 3)');
+ assert.deepStrictEqual(again.body.run.grounding, first.body.run.grounding, 'a cached run carries the grounding of the run it reuses');
});
});
t.test("actor B never receives actor A's private cache entry", async () => {
t.test("actor B never receives actor A's private cache entry", async () => {
@@ -90,6 +90,9 @@ t.test('every workflow runs end to end on the stub and returns schema-valid outp
@@ -90,6 +90,9 @@ t.test('every workflow runs end to end on the stub and returns schema-valid outp
assert.strictEqual(run.provenance.workflow, w.key);
assert.strictEqual(run.provenance.workflow, w.key);
assert.ok(run.provenance.model, `${w.key} has a model`);
assert.ok(run.provenance.model, `${w.key} has a model`);
assert.ok(schemas.validate(w.output_schema, run.output).valid, `${w.key} output invalid`);
assert.ok(schemas.validate(w.output_schema, run.output).valid, `${w.key} output invalid`);
+ // WS-O task 3: every output says what backs it: cited sources, or explicit gaps.
+ assert.ok(run.grounding && Array.isArray(run.grounding.cited) && Array.isArray(run.grounding.gaps), `${w.key} has grounding`);
+ assert.ok(run.grounding.cited.length || run.grounding.gaps.length, `${w.key}: citations or explicit gaps`);
if (EXAMPLES[w.key].sources) {
if (EXAMPLES[w.key].sources) {
assert.strictEqual(run.citations_count, EXAMPLES[w.key].sources.length, `${w.key} citations`);
assert.strictEqual(run.citations_count, EXAMPLES[w.key].sources.length, `${w.key} citations`);
const d = await request(h.base, 'GET', `/api/v1/runs/${run.id}`, { tok });
const d = await request(h.base, 'GET', `/api/v1/runs/${run.id}`, { tok });
@@ -99,6 +102,16 @@ t.test('every workflow runs end to end on the stub and returns schema-valid outp
@@ -99,6 +102,16 @@ t.test('every workflow runs end to end on the stub and returns schema-valid outp
}
}
});
});
+t.test('grounding: cited sources, the output\'s own gaps, or a gap written for it (WS-O task 3)', async () => {
+ const { groundingOf } = require('../server/workflows/engine');
+ assert.deepStrictEqual(groundingOf({ summary: 'x', citations: [1, 0, 1, 7] }, 2, true), { cited: [0, 1], gaps: [] }, 'cited ordinals in range, sorted, once');
+ assert.deepStrictEqual(groundingOf({ items: [{ gaps: ['no price given'] }], gaps: ['no price given', 'date unclear'] }, 1, true).gaps, ['no price given', 'date unclear'], 'the output\'s gaps, once each');
+ assert.match(groundingOf({ text: 'hola' }, 0, false).gaps[0], /No sources were given/);
+ assert.match(groundingOf({ summary: 'x', citations: [] }, 3, true).gaps[0], /None of the given sources is cited/);
+ assert.match(groundingOf({ summary: 'x' }, 2, false).gaps[0], /does not cite them/);
+ // (A cached run carrying its source run's grounding: test/cache.test.js; stub output is never cached.)
+});
+
t.test('citations index only the provided sources', async () => {
t.test('citations index only the provided sources', async () => {
const r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok, body: { workflow: 'news.summarize_story', input: { sources: SRC(2) } } });
const r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok, body: { workflow: 'news.summarize_story', input: { sources: SRC(2) } } });
const seen = JSON.stringify(r.body.run.output).match(/"citations":\[([0-9,]*)\]/g) || [];
const seen = JSON.stringify(r.body.run.output).match(/"citations":\[([0-9,]*)\]/g) || [];