openvibers/openvibe.ai Commits

Route remaining AI capabilities through placement fallback pools (#10)

OP OpenVibers committed 3 Oct 2026 · b74425b2d7 · 7 files, +142 −27

checks run by GitHub CI (test, test-pg) · CI ✓ 3/3 · review SHIP by DeepSeek harness · 2 fix rounds

https://github.com/OpenVibers/OpenVibe.AI/pull/10

Changes

ddiff --git a/README.md b/README.md
iindex 986b5ee..8eed68e 100644
---- a/README.md
++++ b/README.md
@@@ -211,9 +211,10 @@ provider that is disabled, missing credentials, circuit-open or unable to do the
failure; a fallback that answers is recorded on the run (`fallback_used`), in the request log
(`fallback = 1`) and in the audit log. Every run response and `GET /api/v1/runs/:id` carry `explain`
(objective, reasons, selected, and every candidate with its estimated cost, latency and excluded
--reason). Speech and embeddings stay single-provider until a second upstream exists: with one provider
--they answer the explicit `503 provider.unavailable`, never a synthetic answer. The stub joins every
--route as a last resort outside production only (`AI_STUB_FALLBACK`).
++reason). Speech and embeddings use placement pools of registered `stt` and `embedding` models; until
++a second upstream exists, an unavailable sole provider answers `503 provider.unavailable`. Media
++uses separate local and paid pools so paid work still requires its explicit budget. The synthetic
++stub can join general text routes as a last resort outside production only (`AI_STUB_FALLBACK`).
## Workflows
@@@ -240,10 +241,10 @@ route as a last resort outside production only (`AI_STUB_FALLBACK`).
`AI_MEDIA_SCENE_THRESHOLD`), black and frozen picture, silence, and loudness. Loudness has the integrated level
and range, the typical level (the median audible second) and loud moments at least 6 LU above it.
2. **Scenes** from the scene changes (at most 300).
-- 3. **Speech**: whisper.cpp through `live.stt`.
++ 3. **Speech**: whisper.cpp through `live.stt`, with another registered STT model eligible for fallback.
4. **Highlights**: windows scored by loud, scene changes and speech, each with its reasons and transcript excerpt.
-- 5. **An overview** from the local model (`media.local`), else extractive from the measured facts. A paid provider
-- (`media.paid`, the shared provider) writes it only with `allow_paid: true` **and** an active quota with
++ 5. **An overview** from a local model (`media.local`), else extractive from the measured facts. A paid provider
++ (`media.paid`, including the shared provider) writes it only with `allow_paid: true` **and** an active quota with
`workflow_prefix` `media.paid` and `max_cost_usd` > 0, while today's `media.*` spend is under it. That quota
gates only this call, never whole runs.
ddiff --git a/server/providers/index.js b/server/providers/index.js
iindex d8517f1..74bbca0 100644
---- a/server/providers/index.js
++++ b/server/providers/index.js
@@@ -97,10 +97,10 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
async function resetHealth(key) { await db.prepare('DELETE FROM provider_health WHERE provider_key = ?').run(key); }
// ── Pricing: platform.rate-card@1 per metric (server/providers/rate-cards.js; models row -> AI_PRICING_JSON -> flat) ──
-- /** The { in, cached, out } rate cards for a provider + model, or null for a provider that is never billed (stub, whisper). */
++ /** The { in, cached, out } rate cards for a provider + model, or null for a provider that is never billed. */
async function rateCardsFor(providerKey, model) {
const p = await registry.getProvider(providerKey);
-- if (!p || p.kind === 'stub' || p.kind === 'whisper') return null;
++ if (!p || isFreeProvider(p)) return null;
const row = model ? await registry.getModel(providerKey, model) : null;
return rates.buildCards({ providerKey, model: model || null, row, pricing: config.pricing });
}
@@@ -130,9 +130,13 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
}
// ── Routing ────────────────────────────────────────────
-- /** A provider that is never billed: the deterministic stub and the local servers (llama.cpp, whisper.cpp). */
++ /** Local execution never enters the media paid pool, even if a record has no explicit price flag. */
++ function isLocalProvider(p) {
++ return p.kind === 'whisper' || Boolean(p.metadata && p.metadata.local);
++ }
++ /** A provider that is never billed: the deterministic stub and free or local servers. */
function isFreeProvider(p) {
-- return p.kind === 'stub' || p.kind === 'whisper' || Boolean(p.metadata && p.metadata.paid === false);
++ return p.kind === 'stub' || isLocalProvider(p) || Boolean(p.metadata && p.metadata.paid === false);
}
const offerId = (provider, model) => (model ? `${provider}:${model}` : provider);
const splitOfferId = (id) => {
@@@ -189,9 +193,20 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
(cardsByProvider[p.key] || (cardsByProvider[p.key] = [])).push(cards.in, cards.cached, cards.out);
}
};
++ const scope = route.constraints && route.constraints.provider_scope;
for (const p of await registry.listProviders()) {
if (p.key === 'stub' || !p.capabilities.some(f => features.includes(f))) continue;
-- await addOffer(p, (route.constraints && route.constraints.model) || p.default_model || null, p.capabilities);
++ if (scope === 'local' && !isLocalProvider(p)) continue;
++ if (scope === 'paid' && isFreeProvider(p)) continue;
++ if (operation === 'embed' || operation === 'transcribe') {
++ // A chat model advertised by an OpenAI-shaped provider is not an embedding or STT model.
++ const type = operation === 'embed' ? 'embedding' : 'stt';
++ for (const m of await registry.listModels({ provider: p.key })) {
++ if (m.type === type && m.status === 'active') await addOffer(p, m.model_key, p.capabilities);
++ }
++ } else {
++ await addOffer(p, (route.constraints && route.constraints.model) || p.default_model || null, p.capabilities);
++ }
}
// A pinned candidate stays in the pool even when the provider's capability list is incomplete (migration):
// the pin is an explicit admin assertion that it can serve the route's capability.
@@@ -199,6 +214,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
if (!pin || !pin.provider || pin.provider === 'stub') continue;
const p = await registry.getProvider(pin.provider);
if (!p) continue;
++ if (scope === 'local' && !isLocalProvider(p)) continue;
++ if (scope === 'paid' && isFreeProvider(p)) continue;
const caps = route.capability && !p.capabilities.includes(route.capability) ? [...p.capabilities, route.capability] : p.capabilities;
await addOffer(p, pin.model || p.default_model || null, caps);
}
@@@ -263,7 +280,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
const { order, explain, placement: result = null, ...rest } = route.capability && operation
? await poolCandidates(route, features, operation)
: await pinnedCandidates(route, features);
-- if (config.stubFallback && !order.some(c => c.provider === 'stub') && await registry.getProvider('stub')) {
++ if (config.stubFallback && !(route.constraints && route.constraints.provider_scope)
++ && !order.some(c => c.provider === 'stub') && await registry.getProvider('stub')) {
order.push({ provider: 'stub', model: null });
explain.candidates.push({ provider: 'stub', model: null, eligible: true, excluded_reason: null, cost: 0, latency: null });
}
@@@ -360,6 +378,9 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
if (typeof fn !== 'function') throw new ProviderError(`${operation} not implemented`, { code: 'provider.unsupported' });
statFor(c.provider).calls += p.kind === 'stub' ? 0 : 1; // stub counts its own calls
const result = await fn({ ...req, model, signal: ctx.signal, timeoutMs });
++ if (operation === 'embed' && (!Array.isArray(result.vectors) || result.vectors.length !== req.input.length
++ || result.vectors.some(v => !Array.isArray(v) || !v.length || v.some(n => typeof n !== 'number' || !Number.isFinite(n)))))
++ throw new ProviderError('embedding response has invalid vectors', { code: 'provider.output' });
const latency = Date.now() - t0;
const usage = result.usage || { input: 0, output: 0, cached: 0 };
const cost = await costOf(c.provider, result.model || model, usage);
ddiff --git a/server/providers/openai.js b/server/providers/openai.js
iindex b2474e6..bc8e2f2 100644
---- a/server/providers/openai.js
++++ b/server/providers/openai.js
@@@ -81,9 +81,10 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc
async function transcribe(req) {
if (!req.filePath || !fs.existsSync(req.filePath)) throw new ProviderError('transcription audio file missing', { code: 'provider.input' });
++ if (!req.model) throw new ProviderError('transcription model is not configured', { code: 'provider.model' });
const form = new FormData();
form.append('file', new Blob([fs.readFileSync(req.filePath)], { type: 'audio/wav' }), path.basename(req.filePath));
-- form.append('model', req.model || 'whisper-1');
++ form.append('model', req.model);
form.append('response_format', 'verbose_json');
if (req.language && req.language !== 'auto') form.append('language', req.language);
let res;
@@@ -97,10 +98,13 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc
// verbose_json has segments; some providers still answer plain text.
try {
const j = JSON.parse(raw);
++ if (!j || typeof j !== 'object' || (typeof j.text !== 'string' && typeof j.transcript !== 'string'))
++ throw new ProviderError('transcription response has no text field', { code: 'provider.output' });
const segments = Array.isArray(j.segments) ? j.segments.map(s => ({ start: Number(s.start) || 0, end: Number(s.end) || 0, text: String(s.text || '').trim() })) : [];
-- return { text: String(j.text || j.transcript || '').trim(), segments, model: req.model || 'whisper-1', usage: { input: 0, output: 0, cached: 0 } };
-- } catch {
-- return { text: String(raw || '').trim(), segments: [], model: req.model || 'whisper-1', usage: { input: 0, output: 0, cached: 0 } };
++ return { text: String(j.text || j.transcript || '').trim(), segments, model: req.model, usage: { input: 0, output: 0, cached: 0 } };
++ } catch (err) {
++ if (err instanceof ProviderError) throw err;
++ return { text: String(raw || '').trim(), segments: [], model: req.model, usage: { input: 0, output: 0, cached: 0 } };
}
}
ddiff --git a/server/registry.js b/server/registry.js
iindex 6ffd83b..fc4185e 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -205,6 +205,8 @@ function createRegistry(db, { clock = { now: () => Date.now() }, env = process.e
}
if (!Array.isArray(r.fallbacks)) throw new AiError(422, 'ai.invalid', 'fallbacks must be an array');
if (r.constraints && typeof r.constraints !== 'object') throw new AiError(422, 'ai.invalid', 'constraints must be an object');
++ if (r.constraints && r.constraints.provider_scope !== undefined && !['local', 'paid'].includes(r.constraints.provider_scope))
++ throw new AiError(422, 'ai.invalid', 'constraints.provider_scope must be local or paid');
if (!['text', 'json'].includes(r.response_format)) throw new AiError(422, 'ai.invalid', 'response_format must be text or json');
if (!['active', 'disabled'].includes(r.status)) throw new AiError(422, 'ai.invalid', 'route status must be active or disabled');
const version = await nextVersion('routes', key);
ddiff --git a/server/workflows/seed.js b/server/workflows/seed.js
iindex 8de63d0..e619ea2 100644
---- a/server/workflows/seed.js
++++ b/server/workflows/seed.js
@@@ -6,8 +6,8 @@
* AI_MODEL), http-seam (AI_HTTP_SEAM_URL), whisper (WHISPER_*), local (AI_LOCAL_LLM_URL)
* — records whose origin is 'seed' follow the environment; admin-edited ones are left alone
* models the configured default model
-- * routes live.<role> for Live's roles, live.stt, default.chat|json|embedding as capability pools
-- * ordered by openvibe-sdk/placement; media.local|paid stay pinned; and the historical
++ * routes live.<role> for Live's roles, live.stt, default.chat|json|embedding and media.local|paid
++ * as capability pools ordered by openvibe-sdk/placement; and the historical
* product route keys as explicit aliases of default.json
* templates, workflows from server/workflows/{live,core,products}.js — a code change becomes a
* new version unless an admin has versioned that key since (admin edits win)
@@@ -77,28 +77,30 @@ async function seed({ registry, quotas, config, env = process.env, db }) {
for (const m of models) if (m && !await registry.getModel('shared', m)) await registry.upsertModel({ provider_key: 'shared', model_key: m, type: 'chat', supports: { json: true, vision: true } }, { actor: 'seed' });
if (!await registry.getModel('stub', 'stub-1')) await registry.upsertModel({ provider_key: 'stub', model_key: 'stub-1', type: 'chat', supports: { json: true, vision: true }, metadata: { synthetic: true } }, { actor: 'seed' });
if (!await registry.getModel('whisper', 'whisper.cpp')) await registry.upsertModel({ provider_key: 'whisper', model_key: 'whisper.cpp', type: 'stt', supports: { json: false } }, { actor: 'seed' });
++ const embeddingModel = env.AI_EMBEDDING_MODEL || 'text-embedding-3-small';
++ if (!await registry.getModel('shared', embeddingModel)) await registry.upsertModel({ provider_key: 'shared', model_key: embeddingModel, type: 'embedding' }, { actor: 'seed' });
// Routes. A capability pool covers every provider that can serve the capability, ordered by
// openvibe-sdk/placement (brief §4/§5): no static primary/fallbacks and no AI_MODEL_<ROLE> pins.
// The shared provider is the pool's authority (objective correctness): it serves while it is healthy and
// the planner fails over to the rest of the pool (the local model) only when it is not, so a small local
-- // model never wins Live's text work on price. media.local and media.paid stay pinned: media.analyze tries
-- // the local model first and only reaches the paid route under its own budget, never as a pool fallback.
++ // model never wins Live's text work on price. The media pools remain separate: media.analyze tries
++ // local providers first and only reaches paid providers under its own budget.
const authority = shared ? (shared.default_model ? `shared:${shared.default_model}` : 'shared') : null;
// This route field names a model operation; the contracts scanner checks unquoted capability keys as auth grants.
const pool = (timeoutMs, extra = {}) => ({ 'capability': 'chat', constraints: { objective: 'correctness', latency_class: 'interactive', ...(authority ? { authority } : {}) }, pinned: [], fallbacks: [], options: {}, max_output_tokens: null, response_format: 'text', timeout_ms: timeoutMs, alias_of: null, ...extra });
for (const role of live.ROLES) {
await registry.seedVersioned('route', `live.${role}`, pool(ROLE_TIMEOUT[role]));
}
-- // Speech and embeddings stay single-provider until O9 brings a second upstream (decision 8): a pool there
-- // would pull in any OpenAI-shaped provider and answer 200 with nothing, so they answer the explicit 503.
-- await registry.seedVersioned('route', 'live.stt', { primary: { provider: 'whisper', model: null }, fallbacks: [], options: {}, max_output_tokens: null, response_format: 'text', timeout_ms: 600000, alias_of: null });
++ // Speech and embeddings only admit registered models of the matching type. Until O9 adds another
++ // upstream, an unavailable sole provider still answers the explicit 503.
++ await registry.seedVersioned('route', 'live.stt', pool(600000, { 'capability': 'transcribe', constraints: { objective: 'balanced' } }));
await registry.seedVersioned('route', 'default.chat', pool(30000, { max_output_tokens: 800 }));
await registry.seedVersioned('route', 'default.json', pool(60000, { options: { temperature: 0.3 }, max_output_tokens: 2400, response_format: 'json' }));
-- await registry.seedVersioned('route', 'default.embedding', { primary: { provider: 'shared', model: env.AI_EMBEDDING_MODEL || 'text-embedding-3-small' }, fallbacks: [], options: {}, max_output_tokens: null, response_format: 'text', timeout_ms: 30000, alias_of: null });
++ await registry.seedVersioned('route', 'default.embedding', pool(30000, { 'capability': 'embed', constraints: { objective: 'balanced' } }));
// media.analyze (WS-O task 5): the local model when there is one; the paid route is used only under a media.paid budget.
-- if (config.localLlm && config.localLlm.url) await registry.seedVersioned('route', 'media.local', { primary: { provider: 'local', model: config.localLlm.model || null }, fallbacks: [], options: { temperature: 0.3 }, max_output_tokens: 350, response_format: 'text', timeout_ms: config.localLlm.timeoutMs, alias_of: null });
-- await registry.seedVersioned('route', 'media.paid', { primary: { provider: 'shared', model: null }, fallbacks: [], options: { temperature: 0.3 }, max_output_tokens: 350, response_format: 'text', timeout_ms: 60000, alias_of: null });
++ if (config.localLlm && config.localLlm.url) await registry.seedVersioned('route', 'media.local', pool(config.localLlm.timeoutMs, { 'capability': 'summarize', constraints: { objective: 'balanced', provider_scope: 'local' }, options: { temperature: 0.3 }, max_output_tokens: 350 }));
++ await registry.seedVersioned('route', 'media.paid', pool(60000, { 'capability': 'summarize', constraints: { objective: 'correctness', provider_scope: 'paid', ...(authority ? { authority } : {}) }, options: { temperature: 0.3 }, max_output_tokens: 350 }));
for (const key of HISTORICAL_ROUTES) await registry.seedVersioned('route', key, { primary: { provider: 'shared', model: null }, fallbacks: [], options: {}, max_output_tokens: null, response_format: 'json', timeout_ms: null, alias_of: 'default.json' });
// Templates, then workflows (workflows reference templates)
ddiff --git a/test/provider-fabric.test.js b/test/provider-fabric.test.js
nnew file mode 100644
iindex 0000000..8cba793
---- /dev/null
++++ b/test/provider-fabric.test.js
@@@ -0,0 +1,85 @@
++'use strict';
++// Every upstream capability can advance from a failing placed offer to the next real offer.
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const { boot, seamServer, suite } = require('./helpers');
++
++const t = suite('provider-fabric');
++
++t.test('seeded upstream routes use placement and keep media budgets separate', async () => {
++ const h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_LOCAL_LLM_URL: 'http://127.0.0.1:1/v1' } });
++ try {
++ for (const [key, capability, scope] of [
++ ['live.stt', 'transcribe', undefined],
++ ['default.embedding', 'embed', undefined],
++ ['media.local', 'summarize', 'local'],
++ ['media.paid', 'summarize', 'paid'],
++ ]) {
++ const r = await h.registry.resolveRoute(key);
++ assert.strictEqual(r.capability, capability, key);
++ assert.strictEqual(r.constraints.provider_scope, scope, key);
++ assert.deepStrictEqual(r.fallbacks, [], key);
++ }
++ const embed = await h.pool.candidates(await h.registry.resolveRoute('default.embedding'), ['embed'], 'embed');
++ assert.ok(embed.explain.candidates.some(c => c.provider === 'shared' && c.model === 'text-embedding-3-small'));
++ assert.ok(!embed.explain.candidates.some(c => c.model === 'gpt-4o-mini'), 'chat model cannot serve embeddings');
++ } finally { await h.stop(); }
++});
++
++t.test('primary upstream fails; next offer serves chat, summarize, embed and transcribe with explain', async () => {
++ const failed = await seamServer((body, req) => {
++ if (req.url.endsWith('/embeddings')) return { data: [] };
++ if (req.url.endsWith('/audio/transcriptions')) return {};
++ return { status: 503, body: { error: { message: 'forced failure' } } };
++ });
++ const answered = await seamServer((body, req) => {
++ if (req.url.endsWith('/embeddings')) return { data: [{ index: 0, embedding: [0.1, 0.2] }], model: 'embed-good' };
++ if (req.url.endsWith('/audio/transcriptions')) return { text: 'spoken answer', segments: [{ start: 0, end: 1, text: 'spoken answer' }] };
++ return { choices: [{ message: { content: 'real answer' } }], model: body.model || 'chat-good' };
++ });
++ const h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '50' } });
++ const audio = path.join(h.dir, 'audio.wav');
++ fs.writeFileSync(audio, 'test audio');
++ try {
++ for (const [key, url] of [['dead', failed.url], ['good', answered.url]]) {
++ await h.registry.upsertProvider({ key, kind: 'openai', status: 'active', auth_mode: 'none', base_url: url,
++ default_model: `chat-${key}`, capabilities: ['chat', 'summarize', 'embed', 'transcribe'], timeout_ms: 3000,
++ metadata: { paid: true } });
++ for (const [name, type] of [['chat', 'chat'], ['embed', 'embedding'], ['speech', 'stt']]) {
++ await h.registry.upsertModel({ provider_key: key, model_key: `${name}-${key}`, type,
++ cost: { in_per_mtok: key === 'dead' ? 1 : 20, out_per_mtok: key === 'dead' ? 1 : 20 } });
++ }
++ }
++ for (const [key, url] of [['localdead', failed.url], ['localgood', answered.url]]) {
++ await h.registry.upsertProvider({ key, kind: 'openai', status: 'active', auth_mode: 'none', base_url: url,
++ default_model: `chat-${key}`, capabilities: ['chat', 'summarize'], timeout_ms: 3000,
++ metadata: { local: true, paid: false } });
++ await h.registry.upsertModel({ provider_key: key, model_key: `chat-${key}`, type: 'chat' });
++ }
++ await h.registry.upsertProvider({ key: 'whisper', status: 'disabled' });
++ for (const [name, capability, operation, model, req, scope, primary, next] of [
++ ['chat', 'chat', 'chat', 'chat', { messages: [{ role: 'user', content: 'hi' }] }, 'paid', 'dead', 'good'],
++ ['summary', 'summarize', 'summarize', 'chat', { messages: [{ role: 'user', content: 'hi' }] }, 'paid', 'dead', 'good'],
++ ['local', 'summarize', 'summarize', 'chat', { messages: [{ role: 'user', content: 'hi' }] }, 'local', 'localdead', 'localgood'],
++ ['embedding', 'embed', 'embed', 'embed', { input: ['hi'] }, undefined, 'dead', 'good'],
++ ['speech', 'transcribe', 'transcribe', 'speech', { filePath: audio }, undefined, 'dead', 'good'],
++ ]) {
++ const route = await h.registry.createRouteVersion(`test.fabric.${name}`, {
++ capability, constraints: { objective: 'correctness', authority: `${primary}:${model}-${primary}`, ...(scope ? { provider_scope: scope } : {}) }, timeout_ms: 3000,
++ });
++ const logs = [];
++ const exec = await h.pool.execute(route, operation, { ...req, timeoutMs: 3000 }, { routeKey: route.key, routeVersion: route.version, logRequest: e => logs.push(e) });
++ assert.strictEqual(exec.provider, next, name);
++ assert.strictEqual(exec.fallbackUsed, true, name);
++ assert.strictEqual(exec.explain.selected, next, name);
++ assert.ok(exec.explain.candidates.some(c => c.provider === primary), `${name}: primary is explained`);
++ if (scope === 'paid') assert.ok(!exec.explain.candidates.some(c => c.provider.startsWith('local')), 'paid media excludes local offers');
++ if (scope === 'local') assert.ok(!exec.explain.candidates.some(c => c.provider === 'dead' || c.provider === 'good'), 'local media excludes paid offers');
++ assert.ok(logs.some(e => e.provider_key === primary && e.status === 'error'), `${name}: primary failed`);
++ assert.ok(logs.some(e => e.provider_key === next && e.status === 'ok' && e.fallback === 1), `${name}: fallback served`);
++ }
++ } finally { await h.stop(); await failed.close(); await answered.close(); }
++});
++
++t.run();
ddiff --git a/test/provider-router.test.js b/test/provider-router.test.js
iindex 8cdda73..60f058a 100644
---- a/test/provider-router.test.js
++++ b/test/provider-router.test.js
@@@ -74,7 +74,7 @@ t.test('forced failure on every route: a logged fallback, or an explicit 503 (ne
}
}
}
-- assert.deepStrictEqual(failed.sort(), ['default.embedding', 'live.stt', 'media.paid'], `failed routes: ${failed}`);
++ assert.deepStrictEqual(failed.sort(), ['default.embedding', 'live.stt'], `failed routes: ${failed}`);
await h.stop(); await ok.close();
});