openvibers/openvibe.ai Commits

Per-streamer AI budgets are AI quotas (roadmap WS-O task 2; openvibe-contracts 0.74.0 -> 0.75.0 ai.quota.attribution.manage). server/api/attribution-quotas.js: a service sets (PUT /api/v1/attribution-quotas/:attribution, hour or day, cost and/or requests, optional workflow prefix), reads with the window's use, or removes caps on its own attributions only (Live: live:user:<id>, prefix live.viewers.). The rows are quotas with scope attribution and origin service:<svc>, so reserve() refuses a run over the cap with 429 quota.exceeded before any provider call, the operator console lists them, and upsert audits every change. test/attribution-quotas.test.js: capability and ownership, validation, a run over the cap refused while other workflows and streamers run, removal. npm test passes.

OP OpenVibers committed 28 Sept 2026 · 960f14a8a3 · 5 files, +130 −5


Changes

ddiff --git a/package-lock.json b/package-lock.json
iindex aef230d..0d04f39 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -13,7 +13,7 @@
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"express": "^4.21.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.74.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.75.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
},
@@@ -1395,9 +1395,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.74.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.74.0",
-- "integrity": "sha512-ZwlzNihXHXd/qEzBqzsmu5sPZLMS2LUcaX9f3eJcqFYMfhrXzqcmQYbRzLVxN8UvAgDTUiH9WIcTdblPmflNmA==",
++ "version": "0.75.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.75.0",
++ "integrity": "sha512-l+VZGYvVpX2IGsEFPAFwJa8upFRBpZ0ufZxTSua72UeGSj7z2ceHxNjuua0oQ1fnsshl4hCvXHyen1eHjic/eA==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex bd4f6ea..a261b80 100644
---- a/package.json
++++ b/package.json
@@@ -19,7 +19,7 @@
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"express": "^4.21.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.74.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.75.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
},
ddiff --git a/server/api/attribution-quotas.js b/server/api/attribution-quotas.js
nnew file mode 100644
iindex 0000000..b31567a
---- /dev/null
++++ b/server/api/attribution-quotas.js
@@@ -0,0 +1,73 @@
++'use strict';
++/**
++ * Caps a service sets on its own attributions (roadmap WS-O task 2; Contracts 0.75.0 ai.quota.attribution.manage,
++ * ai.attribution-quota-put@1, ai.attribution-quota@1): Live caps what a streamer's AI viewers may cost a day
++ * (attribution live:user:<id>, workflow prefix live.viewers.), so the per-streamer budget is enforced here, before
++ * any provider call (server/quota.js, scope attribution), instead of only in Live.
++ *
++ * PUT /api/v1/attribution-quotas/:attribution set the cap for one window (hour or day) and optional prefix
++ * GET /api/v1/attribution-quotas/:attribution the caps with what each window has used ([] when none)
++ * DELETE /api/v1/attribution-quotas/:attribution remove them all
++ *
++ * The attribution's service must be the calling service. Rows are quotas (origin service:<svc>), so the operator
++ * console lists them with the rest; every change is audited by quotas.upsert.
++ */
++const express = require('express');
++const { validate } = require('openvibe-contracts');
++const { AiError, iso } = require('../util');
++
++const ATTR_RE = /^([a-z][a-z0-9-]{1,31}):([a-z][a-z0-9_]{1,31}):([A-Za-z0-9_.-]{1,80})$/;
++const WINDOW_S = { hour: 3600, day: 86400 };
++
++function attributionQuotasRouter({ db, quotas, auth, clock = { now: () => Date.now() }, sendError }) {
++ const r = express.Router();
++ const guard = auth.requireCap('ai.quota.attribution.manage');
++
++ function own(req) {
++ const m = ATTR_RE.exec(String(req.params.attribution || ''));
++ if (!m) throw new AiError(422, 'input.invalid', 'attribution must be <service>:<type>:<id>');
++ const s = req.principal && req.principal.subject;
++ if (!s || s.type !== 'service' || s.id !== m[1]) throw new AiError(403, 'capability.denied', 'a service caps only its own attributions');
++ return m[0];
++ }
++
++ function view(row) {
++ const now = Math.floor(clock.now() / 1000);
++ const len = WINDOW_S[row.window];
++ const start = Math.floor(now / len) * len;
++ const c = db.prepare('SELECT requests, cost_usd FROM usage_counters WHERE scope_type = ? AND scope_id = ? AND window = ? AND window_start = ? AND workflow_prefix = ?')
++ .get('attribution', row.scope_id, row.window, start, row.workflow_prefix || '') || { requests: 0, cost_usd: 0 };
++ return {
++ attribution: row.scope_id, window: row.window, workflow_prefix: row.workflow_prefix || null,
++ max_cost_usd: row.max_cost_usd == null ? null : row.max_cost_usd, max_requests: row.max_requests == null ? null : row.max_requests,
++ used: { requests: c.requests || 0, cost_usd: Math.round((c.cost_usd || 0) * 1e6) / 1e6 },
++ window_resets_at: iso((start + len) * 1000), updated_at: row.updated_at,
++ };
++ }
++ const rows = (attr) => db.prepare("SELECT * FROM quotas WHERE scope_type = 'attribution' AND scope_id = ? AND status = 'active' ORDER BY window, workflow_prefix").all(attr);
++
++ r.put('/api/v1/attribution-quotas/:attribution', guard, (req, res) => {
++ try {
++ const attr = own(req);
++ const body = req.body || {};
++ const v = validate('ai.attribution-quota-put@1', body);
++ if (!v.valid) throw new AiError(422, 'input.invalid', 'body does not match ai.attribution-quota-put@1', { errors: v.errors });
++ const row = quotas.upsert({ scope_type: 'attribution', scope_id: attr, window: body.window, max_cost_usd: body.max_cost_usd, max_requests: body.max_requests, workflow_prefix: body.workflow_prefix || null, status: 'active' },
++ { actor: req.principal.sub, origin: `service:${req.principal.subject.id}`, trace: req.ov && req.ov.traceId });
++ res.set('Cache-Control', 'no-store').json(view(row));
++ } catch (err) { sendError(res, err, req.ov); }
++ });
++ r.get('/api/v1/attribution-quotas/:attribution', guard, (req, res) => {
++ try { res.set('Cache-Control', 'no-store').json({ quotas: rows(own(req)).map(view) }); } catch (err) { sendError(res, err, req.ov); }
++ });
++ r.delete('/api/v1/attribution-quotas/:attribution', guard, (req, res) => {
++ try {
++ const attr = own(req);
++ const n = db.prepare("UPDATE quotas SET status = 'disabled', updated_at = ? WHERE scope_type = 'attribution' AND scope_id = ? AND status = 'active'").run(iso(clock.now()), attr).changes;
++ res.status(n ? 204 : 404).end();
++ } catch (err) { sendError(res, err, req.ov); }
++ });
++ return r;
++}
++
++module.exports = { attributionQuotasRouter };
ddiff --git a/server/app.js b/server/app.js
iindex abb8441..dfbc2c3 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -46,6 +46,7 @@ function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys
// Operator actions shared by the admin API and the operator console (one implementation, one audit row).
const ops = createOps({ db, registry, pool, quotas, cache, runs });
app.use(runsRouter({ runs, registry, auth, config, log }));
++ app.use(require('./api/attribution-quotas').attributionQuotasRouter({ db, quotas, auth, clock, sendError: (res, err, ctx) => require('./api/runs').sendError(res, err, ctx, log) }));
if (credentials) app.use(require('./credentials').credentialsRouter({ credentials, auth, registry, sendError: (res, err, ctx) => require('./api/runs').sendError(res, err, ctx, log) }));
app.use(adminRouter({ db, registry, pool, quotas, cache, runs, auth, ops, log }));
// The operator console (/console, /auth/*): Network staff, server-rendered, no scripts (server/console).
ddiff --git a/test/attribution-quotas.test.js b/test/attribution-quotas.test.js
nnew file mode 100644
iindex 0000000..8cd4aca
---- /dev/null
++++ b/test/attribution-quotas.test.js
@@@ -0,0 +1,51 @@
++'use strict';
++// Caps a service sets on its own attributions (roadmap WS-O task 2; Contracts 0.75.0 ai.quota.attribution.manage):
++// Live caps a streamer's AI viewers (live:user:<id>, prefix live.viewers.); a run over the cap is refused 429 before
++// any provider call; runs of other workflows or other streamers are not counted against it; another service cannot
++// set, read or remove Live's caps; removing the cap lets runs through again.
++const assert = require('assert');
++const { validate } = require('openvibe-contracts');
++const { boot, request, token, suite, ALL } = require('./helpers');
++
++const t = suite('attribution-quotas');
++const live = token('live', [...ALL, 'ai.quota.attribution.manage']);
++const tools = token('tools', [...ALL, 'ai.quota.attribution.manage']);
++const plain = token('live', ALL);
++let h;
++const run = (workflow, userId) => request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: live, body: { workflow, input: { role: 'chat', user: 'hi' }, attribution: { service: 'live', type: 'user', id: String(userId) } } });
++
++t.test('boot', async () => { h = await boot(); });
++
++t.test('a service sets, reads and removes caps on its own attributions only', async () => {
++ let r = await request(h.base, 'PUT', '/api/v1/attribution-quotas/live:user:42', { tok: plain, body: { window: 'day', max_requests: 1 } });
++ assert.strictEqual(r.status, 403, 'needs the capability');
++ r = await request(h.base, 'PUT', '/api/v1/attribution-quotas/live:user:42', { tok: tools, body: { window: 'day', max_requests: 1 } });
++ assert.deepStrictEqual([r.status, r.body.code], [403, 'capability.denied'], "tools cannot cap Live's streamers");
++ r = await request(h.base, 'PUT', '/api/v1/attribution-quotas/live:user:42', { tok: live, body: { window: 'day' } });
++ assert.strictEqual(r.status, 422, 'a cap needs a limit');
++ r = await request(h.base, 'PUT', '/api/v1/attribution-quotas/live:user:42', { tok: live, body: { window: 'day', max_requests: 1, workflow_prefix: 'live.viewers.' } });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.ok(validate('ai.attribution-quota@1', r.body).valid, JSON.stringify(validate('ai.attribution-quota@1', r.body).errors));
++ assert.deepStrictEqual([r.body.attribution, r.body.max_requests, r.body.used.requests], ['live:user:42', 1, 0]);
++ r = await request(h.base, 'GET', '/api/v1/attribution-quotas/live:user:42', { tok: tools });
++ assert.strictEqual(r.status, 403);
++});
++
++t.test('a run over the cap is refused; other workflows and streamers are not counted', async () => {
++ let r = await run('live.viewers.line', 42);
++ assert.ok([200, 201].includes(r.status), r.text);
++ r = await run('live.viewers.line', 42);
++ assert.deepStrictEqual([r.status, r.body.code], [429, 'quota.exceeded'], 'the second viewer run today is over the cap');
++ r = await run('live.chat.insight', 42);
++ assert.ok([200, 201].includes(r.status), 'another workflow of the same streamer runs');
++ r = await run('live.viewers.line', 43);
++ assert.ok([200, 201].includes(r.status), 'another streamer runs');
++ r = await request(h.base, 'GET', '/api/v1/attribution-quotas/live:user:42', { tok: live });
++ assert.deepStrictEqual(r.body.quotas.map((q) => [q.window, q.workflow_prefix, q.used.requests]), [['day', 'live.viewers.', 1]]);
++ assert.strictEqual((await request(h.base, 'DELETE', '/api/v1/attribution-quotas/live:user:42', { tok: live })).status, 204);
++ r = await run('live.viewers.line', 42);
++ assert.ok([200, 201].includes(r.status), 'removed: runs again');
++ await h.stop();
++});
++
++t.run();