openvibers/openvibe.ai Commits
Capability pools placed by the SDK planner, with explain on every run (plan T6 phase 1; openvibe-sdk v0.25.2): a route may name a capability instead of a fixed primary and fallbacks, and the router asks openvibe-sdk/placement for the order over every provider that can serve it (price from the model rate cards, latency and health from provider_stats_daily and the breaker, hysteresis through placement_state), then runs its unchanged attempt/retry/breaker/log loop over that order. The seeded text routes (Live's roles, default.chat, default.json) are pools whose authority is the shared provider (objective correctness): it serves while healthy and the local model takes over only when it is not, so a small local model never wins Live's text work on price. media.local/media.paid stay pinned; speech and embeddings answer an explicit 503 provider.unavailable until a second provider exists (never synthetic output). Every run response and GET /api/v1/runs/:id carry explain {objective, reasons, selected, candidates}. Migration 0003: routes.capability/constraints/pinned, runs.explain. AI_FALLBACK_* and AI_MODEL_<ROLE> are no longer read. npm test and test:pg 27/27.
Changes
@@@ -196,17 +196,23 @@ definition, but never over an admin's version.| Kind | What | Configured by ||---|---|---|| `stub` | Deterministic, no key, realistically shaped output (schema-shaped JSON, sentences, unit-vector embeddings, timed transcript segments), clearly marked synthetic (`(synthetic)` text, `synthetic: true` on the run). Never cached. | always present |--| `openai` | Any OpenAI-compatible API (OpenAI, OpenRouter, Groq, Together, Ollama, LM Studio, llama.cpp): chat with structured-output step-down, embeddings, Whisper-style transcription. Ported from Live's `llm.js` / `ai-provider.js`. | `AI_PROVIDER`, `AI_BASE_URL`, `AI_API_KEY`, `AI_MODEL`, `AI_MODEL_<ROLE>` — Live's setting names |++| `openai` | Any OpenAI-compatible API (OpenAI, OpenRouter, Groq, Together, Ollama, LM Studio, llama.cpp): chat with structured-output step-down, embeddings, Whisper-style transcription. Ported from Live's `llm.js` / `ai-provider.js`. | `AI_PROVIDER`, `AI_BASE_URL`, `AI_API_KEY`, `AI_MODEL` — Live's setting names || `anthropic` | Messages API with cached system blocks and forced-tool JSON. Ported from Live's `llm.js`. | same, with `AI_PROVIDER=anthropic` || `http` | The local HTTP seam: `POST {operation, …}` → `{text, json, usage}` | `AI_HTTP_SEAM_URL` or the admin API || `whisper` | whisper.cpp on this host, with Live's VAD, hallucination filter, multilingual model and live/batch lanes | `WHISPER_*` — Live's names || `openai`, key `local` | A model server on this host speaking the OpenAI API (llama.cpp's `llama-server`, Ollama's `/v1`): no key, never paid; route `media.local` | `AI_LOCAL_LLM_URL`, `AI_LOCAL_LLM_MODEL`, `AI_LOCAL_LLM_TIMEOUT_MS` |--Routing tries the route's primary, then each fallback; a provider that is disabled, missing--credentials, circuit-open or unable to do the operation is **skipped and logged**; every call has a--strict timeout, cancellation and one retry on a transient failure; a fallback that answers is--recorded on the run (`fallback_used`), in the request log (`fallback = 1`) and in the audit log. The--stub joins every route as a last resort outside production only (`AI_STUB_FALLBACK`).++Routing: a route either names a `capability` and the router builds its pool from every provider that++can serve it, ordered by `openvibe-sdk/placement` (objective, price, measured latency, health, with++hysteresis through the route's current placement), or pins a `[primary, ...fallbacks]` list. A++provider that is disabled, missing credentials, circuit-open or unable to do the operation is++**skipped and logged**; every call has a strict timeout, cancellation and one retry on a transient++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`).## Workflows@@@ -340,8 +346,10 @@ Deployed: `/opt/openvibe.ai`, env `/etc/openvibe/ai.env` (0600), unit `deploy/syof Network OAuth client `ai`, and `AI_CONSOLE_SESSION_SECRET` in `/etc/openvibe/ai.env`.- Import holds from the 2026-09-23 run: a streamer's own provider key stays in Live, and 3,021 Livetranslations cannot become cache entries (they have no source text) and stay in Live.--- No fallback is declared on any production route, so an outage of the one real provider fails every-- Live AI feature. On 2026-09-23 `live.*` and `network.site_copy` outputs carried no citations or gaps++- Text routes now run a capability pool (`shared` plus a local model when one is configured) instead++ of a declared fallback, but speech and embeddings still have a single upstream, so an outage there++ fails that feature until O9 adds a second provider. On 2026-09-23 `live.*` and `network.site_copy`++ outputs carried no citations or gaps(0 citation rows); runs record grounding (cited sources and named gaps) since 2026-09-25.- The host has a backup of `ov_ai` but no restore drill has run for it; `/metrics` is built but notdeployed yet.nnew file mode 100644@@@ -0,0 +1,12 @@++-- phase: expand++-- T6 provider router, phase 1 (brief-t6-ai.out.md §4/§7, decisions-t4-t6.md "T6"):++-- a route either pins [primary, ...fallbacks] (unchanged) or names a capability and lets the router++-- assemble the pool from every provider that can serve it, ordered by openvibe-sdk/placement. The++-- run keeps the placement explanation the router produced (objective, reasons, selected, candidates),++-- so POST /runs, the direct operations and GET /runs/:id all carry it.++++ALTER TABLE routes ADD COLUMN capability text COLLATE "C"; -- a feature the pool must cover (chat|transcribe|embed|...) or NULL for pinned routes++ALTER TABLE routes ADD COLUMN constraints text COLLATE "C" NOT NULL DEFAULT '{}'; -- placement requirements: objective, latency_class, max_latency_ms, max_cost_usd, trust, model, authority++ALTER TABLE routes ADD COLUMN pinned text COLLATE "C" NOT NULL DEFAULT '[]'; -- [{provider, model}] always kept in the pool (migration)++++ALTER TABLE runs ADD COLUMN explain text COLLATE "C"; -- platform.placement-result@1 plus the candidate view, as the router returned it@@@ -13,7 +13,7 @@"dotenv": "^16.4.7","express": "^4.21.2","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.23.1",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0","pg": "^8.23.0"},@@@ -1464,9 +1464,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.23.1",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.23.1",-- "integrity": "sha512-6u1Ym8u+nSKyq+uJjel4/3QDQ9HOjpQzFa8oMasorXQhgHUlDEQ7X6F6Xf1zZTuuVdMVkxOeqJqOnQbraktosg==",++ "version": "0.25.2",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2",++ "integrity": "sha512-rbBj0NXUFNEOTDEtXYS5mD2uO5VJw6ZGC9zME1nbcEqrATMWbKeOt2cO02T/7QDP2iR+lJP97QhYTEF8rlyMog==","license": "MIT","engines": {"node": ">=20"@@@ -25,7 +25,7 @@"dotenv": "^16.4.7","express": "^4.21.2","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.23.1",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0","pg": "^8.23.0"},@@@ -33,11 +33,6 @@ function load(env = process.env) {const port = int(env.PORT, 4700);const aiProvider = String(env.AI_PROVIDER || '').trim().toLowerCase();const aiBaseUrl = trimUrl(env.AI_BASE_URL);-- const roleModels = {};-- for (const r of LIVE_ROLES) {-- const v = env[`AI_MODEL_${r.toUpperCase()}`];-- if (v && String(v).trim()) roleModels[r] = String(v).trim();-- }let pricing = {};if (env.AI_PRICING_JSON) {try { pricing = JSON.parse(env.AI_PRICING_JSON) || {}; } catch { throw new Error('AI_PRICING_JSON is not valid JSON'); }@@@ -89,15 +84,15 @@ function load(env = process.env) {baseUrl: aiBaseUrl,apiKeyRef: 'env:AI_API_KEY', // secret reference, never the valuemodel: String(env.AI_MODEL || '').trim(),-- roleModels,timeoutMs: int(env.AI_PROVIDER_TIMEOUT_MS, 30000),},-- // Optional second OpenAI-compatible provider used as the fallback of every route.-- fallback: {-- baseUrl: trimUrl(env.AI_FALLBACK_BASE_URL),-- apiKeyRef: 'env:AI_FALLBACK_API_KEY',-- model: String(env.AI_FALLBACK_MODEL || '').trim(),-- kind: String(env.AI_FALLBACK_PROVIDER || '').trim().toLowerCase(),++ // ── Placement (T6 provider router): openvibe-sdk/placement orders a route's capability pool ──++ placement: {++ // A candidate replaces the route's current placement only when it scores this much better,++ // so a small price/latency wobble does not make the router oscillate.++ minGain: Math.max(0, Math.min(1, float(env.AI_PLACEMENT_MIN_GAIN, 0.15))),++ costWeight: float(env.AI_PLACEMENT_COST_WEIGHT, 0.5),++ latencyWeight: float(env.AI_PLACEMENT_LATENCY_WEIGHT, 0.5),},// Configurable local HTTP seam: POST {operation, request} JSON to this URL.httpSeamUrl: trimUrl(env.AI_HTTP_SEAM_URL),@@@ -5,16 +5,22 @@* pool.adapter(key) the adapter for a provider record (rebuilt when the record changes)* pool.health(key) circuit-breaker state* pool.execute(route, op, req, ctx)-- * tries the route's primary, then each fallback, skipping candidates that are disabled,-- * circuit-open, missing credentials or that do not support the operation; applies a strict-- * timeout and cancellation to every call and one retry on a transient failure; records one-- * request-log row per attempt or skip (fallback = 1 for every non-primary candidate) and-- * returns { result, provider, model, fallbackUsed }. When nothing answers it throws++ * tries the route's candidates in order, skipping ones that are disabled, circuit-open,++ * missing credentials or that do not support the operation; applies a strict timeout and++ * cancellation to every call and one retry on a transient failure; records one request-log++ * row per attempt or skip (fallback = 1 for every non-first candidate) and returns++ * { result, provider, model, fallbackUsed, explain }. When nothing answers it throws* AiError 503 provider.unavailable — an explicit state, never a made-up answer.*++ * A route either names a `capability` and the pool is built from every provider that can serve it,++ * ordered by openvibe-sdk/placement (objective, price, measured latency, health, with hysteresis++ * through placement_state.current); or it pins `[primary, ...fallbacks]`. `explain` (objective,++ * reasons, selected, candidates with their excluded reasons) rides every run response.++ ** Circuit breaker per provider: `failureThreshold` consecutive failures open it for `cooldownMs`;* after that one half-open probe decides whether it closes again.*/++const placement = require('openvibe-sdk/placement');const { AiError, resolveSecret, sha256 } = require('../util');const { retryable, sleep, ProviderError } = require('./common');const { createStubProvider } = require('./stub');@@@ -29,6 +35,11 @@ function callerFault(err) {return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 403, 404, 408, 429].includes(st);}++/** An upstream 429 is rate-limit capacity, not provider ill health: it shifts to the next candidate but must not open the circuit. */++function rateLimited(err) {++ return (err && err.status === 429) || (err && err.code === 'provider.rate_limited');++}++function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, credentialFetch = null }) {const cache = new Map(); // key -> { stamp, adapter }const stats = {}; // key -> { calls, failures, ... } (in memory; tests + /api/ready)@@@ -113,10 +124,129 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no}// ── Routing ────────────────────────────────────────────-- async function candidates(route) {++ /** A provider that is never billed: the deterministic stub and the local servers (llama.cpp, whisper.cpp). */++ function isFreeProvider(p) {++ return p.kind === 'stub' || p.kind === 'whisper' || Boolean(p.metadata && p.metadata.paid === false);++ }++ const offerId = (provider, model) => (model ? `${provider}:${model}` : provider);++ const splitOfferId = (id) => {++ const i = String(id).indexOf(':');++ return i < 0 ? { provider: id, model: null } : { provider: id.slice(0, i), model: id.slice(i + 1) };++ };++++ // ── Capability pool (T6 phase 1): placement over every provider that can serve the capability ──++ const getRouteState = db.prepare('SELECT * FROM placement_state WHERE route_key = ?');++ const getProviderStats = db.prepare(`SELECT provider, MAX(latency_p95_ms) AS p95, SUM(requests)::bigint AS requests, SUM(errors)::bigint AS errors++ FROM provider_stats_daily WHERE day >= ? GROUP BY provider`);++ async function statsByProvider(days = 7) {++ const from = new Date(clock.now() - days * 86400000).toISOString().slice(0, 10);++ const m = new Map();++ for (const r of await getProviderStats.all(from)) m.set(r.provider, { p95: r.p95 == null ? null : Number(r.p95), requests: Number(r.requests), errors: Number(r.errors) });++ return m;++ }++++ /**++ * The order the router tries for a capability route: openvibe-sdk/placement ranks every provider whose++ * capabilities cover the operation, by objective, price (the model rate card), measured latency and++ * health (placement_state error rate + the breaker); hysteresis through the route's current placement.++ * The stub is never a placement candidate (synthetic output must not win on price) — it is appended last.++ */++ async function poolCandidates(route, features, operation) {++ const stats = await statsByProvider();++ const offers = [];++ const rateCards = [];++ const seen = new Set();++ const addOffer = async (p, model, caps) => {++ const id = offerId(p.key, model);++ if (seen.has(id)) return;++ seen.add(id);++ const creds = p.auth_mode === 'none' || Boolean(resolveSecret(p.secret_ref, env));++ const h = await health(p.key);++ const st = stats.get(p.key) || { p95: null, requests: 0, errors: 0 };++ const errorRate = st.requests ? st.errors / st.requests : 0;++ const status = p.status !== 'active' || !creds ? 'down' : h.state === 'open' ? 'down' : (h.state === 'half_open' || errorRate > 0.5) ? 'degraded' : 'up';++ const free = isFreeProvider(p);++ offers.push({++ offer_id: id, kind: 'provider', provider: p.key, region: 'global',++ trust: (p.metadata && p.metadata.local) || p.kind === 'stub' ? 'community' : 'first-party',++ capabilities: caps, latency_ms: st.p95 ? { p95: st.p95 } : {}, health: { status },++ pricing: free ? { model: 'prepaid' } : { model: 'metered', rate_card: id },++ });++ if (!free) {++ const price = await priceFor(p.key, model);++ rateCards.push({ id, provider: p.key, metric: 'tokens', unit_size: 1e6, unit_price_usd: (price.in + price.out) / 2, free_allowance: 0, reset_period: 'month' });++ }++ };++ 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);++ }++ // 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.++ for (const pin of route.pinned || []) {++ if (!pin || !pin.provider || pin.provider === 'stub') continue;++ const p = await registry.getProvider(pin.provider);++ if (!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);++ }++ const c = route.constraints || {};++ const requirements = {++ kind: 'ai', mobility: 'request', latency_class: c.latency_class || 'interactive', objective: c.objective || 'balanced',++ capabilities: features, units: Number(c.units) || 1, latency_op: operation,++ ...(c.max_latency_ms != null ? { max_latency_ms: Number(c.max_latency_ms) } : {}),++ ...(c.max_cost_usd != null ? { max_cost_usd: Number(c.max_cost_usd) } : {}),++ ...(c.trust ? { trust: c.trust } : {}),++ ...(c.authority ? { authority: c.authority } : {}),++ };++ const state = await getRouteState.get(route.key);++ const current = state && state.current_provider ? offerId(state.current_provider, state.current_model) : null;++ const result = placement.plan(requirements, offers, {++ rateCards, now: clock.now(), current, minGain: config.placement.minGain,++ weights: { cost: config.placement.costWeight, latency: config.placement.latencyWeight },++ });++ const byId = new Map(offers.map(o => [o.offer_id, o]));++ const explain = {++ objective: result.objective, reasons: result.reasons, selected: result.selected,++ candidates: result.candidates.map((x) => {++ const o = byId.get(x.id) || {};++ return { provider: o.provider || splitOfferId(x.id).provider, model: splitOfferId(x.id).model, eligible: Boolean(x.eligible),++ excluded_reason: x.excluded_because || null, cost: x.estimated_cost_usd == null ? null : x.estimated_cost_usd, latency: x.estimated_latency_ms == null ? null : x.estimated_latency_ms };++ }),++ };++ const order = [];++ if (result.selected) order.push(result.selected);++ const rest = result.candidates.filter(x => x.id !== result.selected).sort((a, b) => {++ const ea = a.eligible && Number.isFinite(a.estimated_cost_usd) ? 0 : 1;++ const eb = b.eligible && Number.isFinite(b.estimated_cost_usd) ? 0 : 1;++ return ea - eb || (a.score ?? Infinity) - (b.score ?? Infinity);++ });++ for (const x of rest) order.push(x.id);++ return { order: order.map(splitOfferId), explain };++ }++++ /** A pinned route: [primary, ...fallbacks], exactly as before. */++ async function pinnedCandidates(route, features) {const list = [route.primary, ...(route.fallbacks || [])].filter(c => c && c.provider);-- if (config.stubFallback && !list.some(c => c.provider === 'stub') && await registry.getProvider('stub')) list.push({ provider: 'stub', model: null, auto: true });-- return list;++ const explain = { objective: null, reasons: ['pinned route: the primary, then its fallbacks'], selected: null, candidates: [] };++ for (const c of list) {++ const entry = await adapter(c.provider);++ const p = entry && entry.record;++ const why = await skipReason(p, entry && entry.adapter, features);++ explain.candidates.push({ provider: c.provider, model: c.model || (p && p.default_model) || null, eligible: !why, excluded_reason: why, cost: null, latency: null });++ }++ return { order: list.map(c => ({ provider: c.provider, model: c.model || null })), explain };++ }++++ async function candidates(route, features, operation) {++ const { order, explain } = 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')) {++ order.push({ provider: 'stub', model: null });++ explain.candidates.push({ provider: 'stub', model: null, eligible: true, excluded_reason: null, cost: 0, latency: null });++ }++ return { order, explain };}async function skipReason(p, a, features) {@@@ -191,7 +321,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.noif (req.image) features.push('vision');if (req.json) features.push('json');const tried = [];-- const list = await candidates(route);++ const { order: list, explain } = await candidates(route, features, operation);for (let i = 0; i < list.length; i++) {const c = list[i];const entry = await adapter(c.provider);@@@ -233,7 +363,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.nodebug_response: ctx.debugRaw ? String(result.text || '').slice(0, 200000) : null,});}-- return { result, provider: c.provider, providerKind: p.kind, model: result.model || model, fallbackUsed: fallback, usage, cost, latencyMs: latency, synthetic: Boolean(result.synthetic || a.synthetic), tried, startedAt: started };++ explain.selected = c.provider;++ return { result, provider: c.provider, providerKind: p.kind, model: result.model || model, fallbackUsed: fallback, usage, cost, latencyMs: latency, synthetic: Boolean(result.synthetic || a.synthetic), tried, startedAt: started, explain };} catch (err) {lastErr = err;if (ctx.signal && ctx.signal.aborted) {@@@ -247,12 +378,15 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no}}// A request the provider rejected as malformed is the caller's fault, not a sign the-- // provider is unhealthy: it must not open the circuit other callers depend on.-- if (!callerFault(lastErr)) await recordFailure(c.provider, lastErr);++ // provider is unhealthy; a 429 is capacity, not health either, and moves on without++ // opening the circuit. Neither must take a shared provider away from other callers.++ if (!callerFault(lastErr) && !rateLimited(lastErr)) await recordFailure(c.provider, lastErr);tried.push({ provider: c.provider, error: String(lastErr && lastErr.message || lastErr).slice(0, 200) });log.warn(`[ai] ${operation} via ${c.provider}/${model || '-'} failed: ${lastErr && lastErr.message}`);}-- throw new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried });++ explain.selected = null;++ explain.reasons = [...(explain.reasons || []), 'no provider on this route could answer'];++ throw new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried, explain });}return { adapter, health, resetHealth, execute, executeWithCredential, priceFor, costOf, stats, statFor, candidates };@@@ -156,16 +156,23 @@ function createRegistry(db, { clock = { now: () => Date.now() }, env = process.ereturn {key: r.key, version: r.version, status: r.status, primary: { provider: r.primary_provider, model: r.primary_model },fallbacks: parseJson(r.fallbacks, []), options: parseJson(r.options, {}), max_output_tokens: r.max_output_tokens,-- response_format: r.response_format, timeout_ms: r.timeout_ms, alias_of: r.alias_of, created_by: r.created_by, created_at: r.created_at,++ response_format: r.response_format, timeout_ms: r.timeout_ms, alias_of: r.alias_of,++ capability: r.capability || null, constraints: parseJson(r.constraints, {}), pinned: parseJson(r.pinned, []),++ created_by: r.created_by, created_at: r.created_at,};}-- const routeContent = (r) => sha256({ p: r.primary, f: r.fallbacks, o: r.options, m: r.max_output_tokens, rf: r.response_format, t: r.timeout_ms, a: r.alias_of, s: r.status });++ const routeContent = (r) => sha256({ p: r.primary, f: r.fallbacks, o: r.options, m: r.max_output_tokens, rf: r.response_format, t: r.timeout_ms, a: r.alias_of, s: r.status, c: r.capability || null, x: r.constraints || {}, pi: r.pinned || [] });async function createRouteVersion(key, input, { actor = 'system', trace = null } = {}) {if (!KEY_RE.test(key || '')) throw new AiError(422, 'ai.invalid', 'route key must be lowercase dotted');const prev = decodeRoute(await latestAny('routes', key));const r = {primary: input.primary || (prev && prev.primary),fallbacks: input.fallbacks !== undefined ? input.fallbacks : (prev ? prev.fallbacks : []),++ // A pool route names a capability; sending a version without one is a deliberate pin (an admin edit++ // that states primary/fallbacks), so capability is taken from the input only, never inherited.++ capability: input.capability !== undefined ? (input.capability || null) : (prev && prev.capability && input.primary === undefined ? prev.capability : null),++ constraints: input.constraints !== undefined ? input.constraints : ((input.primary === undefined && prev) ? prev.constraints : {}),++ pinned: input.pinned !== undefined ? input.pinned : ((input.primary === undefined && prev) ? prev.pinned : []),options: input.options !== undefined ? input.options : (prev ? prev.options : {}),max_output_tokens: input.max_output_tokens !== undefined ? input.max_output_tokens : (prev ? prev.max_output_tokens : null),response_format: input.response_format || (prev && prev.response_format) || 'text',@@@ -173,21 +180,29 @@ function createRegistry(db, { clock = { now: () => Date.now() }, env = process.ealias_of: input.alias_of !== undefined ? input.alias_of : (prev ? prev.alias_of : null),status: input.status || 'active',};-- if (!r.alias_of) {-- if (!r.primary || !r.primary.provider) throw new AiError(422, 'ai.invalid', 'route needs primary.provider');++ if (r.capability && !FEATURES.includes(r.capability)) throw new AiError(422, 'ai.invalid', `route capability must be one of ${FEATURES.join(', ')}`);++ if (!r.alias_of && !r.capability) {++ if (!r.primary || !r.primary.provider) throw new AiError(422, 'ai.invalid', 'route needs primary.provider or a capability');for (const c of [r.primary, ...(r.fallbacks || [])]) {if (!c || !await getProvider(c.provider)) throw new AiError(422, 'ai.invalid', `route references unknown provider ${c && c.provider}`);}-- } else if (!await latestAny('routes', r.alias_of)) throw new AiError(422, 'ai.invalid', `alias target ${r.alias_of} does not exist`);++ } else if (r.alias_of && !await latestAny('routes', r.alias_of)) throw new AiError(422, 'ai.invalid', `alias target ${r.alias_of} does not exist`);++ for (const c of r.pinned || []) {++ if (!c || !await getProvider(c.provider)) throw new AiError(422, 'ai.invalid', `pinned candidate references unknown provider ${c && c.provider}`);++ }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 (!['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);-- await db.prepare(`INSERT INTO routes (key, version, status, primary_provider, primary_model, fallbacks, options, max_output_tokens, response_format, timeout_ms, alias_of, created_by, created_at)-- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)-- .run(key, version, r.status, r.alias_of ? (r.primary && r.primary.provider) || 'stub' : r.primary.provider, r.primary ? r.primary.model || null : null,-- JSON.stringify(r.fallbacks || []), JSON.stringify(r.options || {}), r.max_output_tokens, r.response_format, r.timeout_ms, r.alias_of, actor, now());-- await audit(actor, 'route.version', 'route', key, { trace, metadata: { version, previous: prev ? prev.version : null, status: r.status, alias_of: r.alias_of } });++ const primaryProvider = r.alias_of ? ((r.primary && r.primary.provider) || 'stub')++ : (r.primary && r.primary.provider) || ((r.pinned && r.pinned[0] && r.pinned[0].provider) || 'stub');++ await db.prepare(`INSERT INTO routes (key, version, status, primary_provider, primary_model, fallbacks, options, max_output_tokens, response_format, timeout_ms, alias_of, capability, constraints, pinned, created_by, created_at)++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)++ .run(key, version, r.status, primaryProvider, (r.primary && r.primary.model) || null,++ JSON.stringify(r.fallbacks || []), JSON.stringify(r.options || {}), r.max_output_tokens, r.response_format, r.timeout_ms, r.alias_of,++ r.capability || null, JSON.stringify(r.constraints || {}), JSON.stringify(r.pinned || []), actor, now());++ await audit(actor, 'route.version', 'route', key, { trace, metadata: { version, previous: prev ? prev.version : null, status: r.status, alias_of: r.alias_of, capability: r.capability || null } });return await getRoute(key, version);}async function getRoute(key, version) {@@@ -85,6 +85,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowprovider: r.provider_key, model: r.model_key, fallback_used: Boolean(r.fallback_used), run_id: r.id, cached_from: r.cached_from, synthetic: Boolean(r.synthetic),},usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts },++ explain: parseJson(r.explain, null),citations_count: r.citations_count,grounding: parseJson(r.grounding, null),retry_of: r.retry_of,@@@ -203,11 +204,12 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowconst hit = await cache.get(cacheKey, scope);if (hit) {await db.tx(async () => {++ const cacheExplain = { objective: 'cache', reasons: ["the answer was already in this scope's cache; no provider was called"], selected: null, candidates: [] };await db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, requester_type, requester_id, on_behalf_of, attribution,-- source_service, target, target_key, input, input_hash, output, model_key, cache_key, cached_from, retry_of, idempotency_key, trace_id, request_id, options, created_at, started_at, finished_at, citations_count)++ source_service, target, target_key, input, input_hash, output, model_key, cache_key, cached_from, retry_of, idempotency_key, trace_id, request_id, options, explain, created_at, started_at, finished_at, citations_count)VALUES (@id, @workflow_key, @workflow_version, @template_key, @template_version, @route_key, @route_version, 'cached', @requester_type, @requester_id, @on_behalf_of, @attribution,-- @source_service, @target, @target_key, @input, @input_hash, @output, @model_key, @cache_key, @cached_from, @retry_of, @idempotency_key, @trace_id, @request_id, @options, @created_at, @created_at, @created_at, 0)`)-- .run({ ...base, output: JSON.stringify(hit.output), model_key: hit.model_key, cache_key: cacheKey, cached_from: hit.run_id });++ @source_service, @target, @target_key, @input, @input_hash, @output, @model_key, @cache_key, @cached_from, @retry_of, @idempotency_key, @trace_id, @request_id, @options, @explain, @created_at, @created_at, @created_at, 0)`)++ .run({ ...base, output: JSON.stringify(hit.output), model_key: hit.model_key, cache_key: cacheKey, cached_from: hit.run_id, explain: JSON.stringify(cacheExplain) });const src = await citations(hit.run_id);if (src.length) await 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.@@@ -301,9 +303,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now// The status, its citations and the ai.run.succeeded event commit together.const done = await db.tx(async () => {const d = await 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), explain = ?, finished_at = ? WHERE id = ? AND status = 'running'`).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, r.explain ? JSON.stringify(r.explain) : null, iso(clock.now()), id);if (!d.changes) return d;if (r.citations.length) await addCitations(id, r.citations, 'workflow');await events.runChanged(await getRow(id));@@@ -323,9 +325,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowconst detail = interrupted ? 'the service shut down while this run was in progress' : cancelled ? 'cancelled' : (err instanceof AiError ? err.detail : 'internal error');if (!(err instanceof AiError) && !cancelled) log.error(`[runs] ${id} crashed: ${err && err.stack || err}`);const extra = err instanceof AiError && err.extra ? JSON.stringify(err.extra).slice(0, 2000) : null;++ const explain = err instanceof AiError && err.extra && err.extra.explain ? JSON.stringify(err.extra.explain) : null;await db.tx(async () => {-- const f = await db.prepare(`UPDATE runs SET status = ?, error_code = ?, error_detail = ?, attempts = ?, finished_at = ? WHERE id = ? AND status IN ('queued', 'running')`)-- .run(cancelled ? 'cancelled' : 'failed', code, extra ? `${detail} ${extra}` : detail, attempts, iso(clock.now()), id);++ const f = await db.prepare(`UPDATE runs SET status = ?, error_code = ?, error_detail = ?, attempts = ?, explain = COALESCE(?, explain), finished_at = ? WHERE id = ? AND status IN ('queued', 'running')`)++ .run(cancelled ? 'cancelled' : 'failed', code, extra ? `${detail} ${extra}` : detail, attempts, explain, iso(clock.now()), id);if (f.changes) await events.runChanged(await getRow(id));});const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);@@@ -258,6 +258,7 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spaprovider: exec ? exec.provider : null, model: exec ? exec.model : null, fallbackUsed: Boolean(exec && exec.fallbackUsed),usage, cost,route: last.route || null,++ explain: exec ? exec.explain || null : null,};}@@@ -3,11 +3,12 @@* Boot-time seeding from configuration and code. Idempotent: every boot re-applies it.** providers stub (always), shared (Live's shared key: AI_PROVIDER / AI_BASE_URL / AI_API_KEY /-- * AI_MODEL), fallback (AI_FALLBACK_*), http-seam (AI_HTTP_SEAM_URL), whisper (WHISPER_*)++ * 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 and per-role models-- * routes live.<role> for Live's roles (AI_MODEL_<ROLE> overrides), live.stt, default.chat|json|-- * embedding, and the historical product route keys as explicit aliases of default.json++ * 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++ * 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)* quotas AI_MAX_COST_USD_PER_DAY (global/day, Live's ai_max_cost_usd_per_day) and a per-service@@@ -50,10 +51,6 @@ function providerRecords(config, env) {sharedProviderRecord(config, env),{ key: 'whisper', display_name: 'whisper.cpp (local speech-to-text)', kind: 'whisper', status: 'active', auth_mode: 'none', default_model: 'whisper.cpp', capabilities: ['transcribe'], timeout_ms: 600000, priority: 20, metadata: { configured_by: 'WHISPER_*' } },];-- if (config.fallback.baseUrl) {-- const kind = config.fallback.kind === 'anthropic' ? 'anthropic' : 'openai';-- out.push({ key: 'fallback', display_name: 'Fallback provider', kind, status: 'active', base_url: config.fallback.baseUrl, auth_mode: resolveSecret(config.fallback.apiKeyRef, env) ? (kind === 'anthropic' ? 'x-api-key' : 'bearer') : 'none', secret_ref: config.fallback.apiKeyRef, default_model: config.fallback.model || null, capabilities: [], timeout_ms: 30000, priority: 50, metadata: { configured_by: 'AI_FALLBACK_*' } });-- }if (config.localLlm && config.localLlm.url) {// A local model server (llama.cpp, Ollama) speaking the OpenAI API: no key, never paid (WS-O task 5).out.push({ key: 'local', display_name: 'Local model (OpenAI-compatible server on this host)', kind: 'openai', status: 'active', base_url: config.localLlm.url, auth_mode: 'none', secret_ref: null, default_model: config.localLlm.model || 'local', capabilities: ['chat', 'generate', 'summarize', 'classify', 'extract'], timeout_ms: config.localLlm.timeoutMs, priority: 5, metadata: { configured_by: 'AI_LOCAL_LLM_URL', local: true, paid: false } });@@@ -74,29 +71,33 @@ async function seed({ registry, quotas, config, env = process.env, db }) {}}const shared = await registry.getProvider('shared');-- const hasFallback = Boolean(await registry.getProvider('fallback'));-- const fallbacks = hasFallback ? [{ provider: 'fallback', model: null }] : [];// Models-- const models = new Set([shared.default_model, ...Object.values(config.shared.roleModels)]);++ const models = new Set([shared.default_model]);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' });-- // Routes++ // 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.++ const authority = shared ? (shared.default_model ? `shared:${shared.default_model}` : 'shared') : null;++ 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}`, {-- primary: { provider: 'shared', model: config.shared.roleModels[role] || null }, fallbacks,-- options: {}, max_output_tokens: null, response_format: 'text', timeout_ms: ROLE_TIMEOUT[role], alias_of: null,-- });++ 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 });-- await registry.seedVersioned('route', 'default.chat', { primary: { provider: 'shared', model: null }, fallbacks, options: {}, max_output_tokens: 800, response_format: 'text', timeout_ms: 30000, alias_of: null });-- await registry.seedVersioned('route', 'default.json', { primary: { provider: 'shared', model: null }, fallbacks, options: { temperature: 0.3 }, max_output_tokens: 2400, response_format: 'json', timeout_ms: 60000, alias_of: null });++ 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 });// 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: config.shared.roleModels.summary || null }, fallbacks, options: { temperature: 0.3 }, max_output_tokens: 350, response_format: 'text', timeout_ms: 60000, 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 });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)nnew file mode 100644@@@ -0,0 +1,82 @@++'use strict';++// T6 placement (openvibe-sdk/placement, used by the provider router): ordering by price, latency and++// objective; hard constraints exclude a candidate with a reason (capability, health, trust, ceiling);++// an unpriced paid provider is never assumed free; hysteresis keeps the current placement unless a++// candidate is clearly better; correctness keeps the named authority. Pure offers — no network.++const assert = require('assert');++const { plan } = require('openvibe-sdk/placement');++const { suite } = require('./helpers');++++const t = suite('placement');++const card = (id, provider, price) => ({ id, provider, metric: 'tokens', unit_size: 1e6, unit_price_usd: price, free_allowance: 0, reset_period: 'month' });++const offer = (id, opts = {}) => ({++ offer_id: id, kind: 'provider', region: 'global', trust: 'first-party', capabilities: ['chat'],++ health: { status: 'up' }, pricing: { model: 'metered', rate_card: id }, ...opts,++});++const req = (over = {}) => ({ kind: 'ai', mobility: 'request', latency_class: 'interactive', objective: 'balanced', capabilities: ['chat'], units: 1, ...over });++++t.test('cheapest picks the lower-priced offer', () => {++ const offers = [offer('pricey'), offer('cheap')];++ const r = plan(req({ objective: 'cheapest' }), offers, { rateCards: [card('pricey', 'pricey', 30), card('cheap', 'cheap', 5)] });++ assert.strictEqual(r.selected, 'cheap');++ assert.deepStrictEqual(r.candidates.map((c) => [c.id, c.eligible]), [['pricey', true], ['cheap', true]]);++});++++t.test('lowest-latency picks the faster offer', () => {++ const offers = [offer('slow', { latency_ms: { p95: 900 } }), offer('fast', { latency_ms: { p95: 90 } })];++ const r = plan(req({ objective: 'lowest-latency' }), offers, { rateCards: [card('slow', 'slow', 1), card('fast', 'fast', 1)] });++ assert.strictEqual(r.selected, 'fast');++});++++t.test('hard constraints exclude with a reason; the eligible one is chosen', () => {++ const offers = [++ offer('nocap', { capabilities: ['vision'] }),++ offer('sick', { health: { status: 'down' } }),++ offer('outsider', { trust: 'community' }),++ offer('slow', { latency_ms: { p95: 5000 } }),++ offer('good'),++ ];++ const r = plan(req({ trust: ['first-party'], max_latency_ms: 1000 }), offers, { rateCards: offers.map((o) => card(o.offer_id, o.offer_id, 1)) });++ const why = Object.fromEntries(r.candidates.map((c) => [c.id, c.excluded_because]));++ assert.match(why.nocap, /lacks chat/);++ assert.match(why.sick, /health down/);++ assert.match(why.outsider, /trust/);++ assert.match(why.slow, /over 1000 ms/);++ assert.strictEqual(r.selected, 'good');++});++++t.test('an unpriced paid provider is excluded, never assumed free', () => {++ const unpriced = { offer_id: 'free?', kind: 'provider', region: 'global', trust: 'first-party', capabilities: ['chat'], health: { status: 'up' }, pricing: { model: 'metered' } };++ const offers = [offer('priced'), unpriced];++ const r = plan(req({ objective: 'cheapest' }), offers, { rateCards: [card('priced', 'priced', 5)] });++ const u = r.candidates.find((c) => c.id === 'free?');++ assert.strictEqual(u.eligible, true, 'not a hard-constraint failure');++ assert.ok(!Number.isFinite(u.estimated_cost_usd), 'no price known, so no finite cost estimate');++ assert.strictEqual(r.selected, 'priced');++});++++t.test('hysteresis: a slightly better candidate does not move the current placement', () => {++ const offers = [offer('current'), offer('rival')];++ const rateCards = [card('current', 'current', 10), card('rival', 'rival', 9)];++ const stays = plan(req({ objective: 'cheapest' }), offers, { rateCards, current: 'current', minGain: 0.5 });++ assert.strictEqual(stays.selected, 'current', stays.reasons.join('; '));++ assert.ok(stays.reasons.some((x) => /stays on current/.test(x)));++ const moves = plan(req({ objective: 'cheapest' }), offers, { rateCards, current: 'current', minGain: 0.05 });++ assert.strictEqual(moves.selected, 'rival');++});++++t.test('correctness keeps the named authority, whatever the price', () => {++ const offers = [offer('authority'), offer('cheaper')];++ const r = plan(req({ objective: 'correctness', authority: 'authority' }), offers, { rateCards: [card('authority', 'authority', 50), card('cheaper', 'cheaper', 1)] });++ assert.strictEqual(r.selected, 'authority');++ assert.ok(r.reasons.some((x) => /authority keeps critical work/.test(x)));++});++++t.test('a candidate that is no longer eligible fails over immediately (no minGain hold)', () => {++ const offers = [offer('current', { health: { status: 'down' } }), offer('other')];++ const r = plan(req({ objective: 'cheapest' }), offers, { rateCards: [card('current', 'current', 1), card('other', 'other', 9)], current: 'current', minGain: 0.9 });++ assert.strictEqual(r.selected, 'other');++ assert.ok(r.reasons.some((x) => /no longer eligible/.test(x)));++});++++t.run();nnew file mode 100644@@@ -0,0 +1,199 @@++'use strict';++// T6 provider router (phase 1): a route names a capability and the router tries its pool in++// openvibe-sdk/placement order, over the existing retry/breaker/log loop.++// - forced failure on EVERY route from registry.listRoutes(): the primary fails and the pool falls++// back to another member (logged), or the route answers the explicit 503 provider.unavailable for++// a single-provider capability — never synthetic output;++// - two seam providers on one capability: the cheaper is chosen; a circuit-open provider is excluded;++// an upstream 429 shifts to the other without opening the circuit;++// - explain{objective,reasons,selected,candidates} rides every run response and GET /runs/:id, with++// the per-attempt request log still there.++const assert = require('assert');++const { boot, request, token, suite, seamServer, ALL } = require('./helpers');++++const t = suite('provider-router');++const tok = token('live', ALL);++let h;++const seam = () => seamServer((body) => (body.operation === 'embed'++ ? { vectors: [[0.1, 0.2]], model: 'seam-model', usage: { input: 1, output: 0 } }++ : { text: 'a real answer', model: 'seam-model', usage: { input: 5, output: 2 } }));++++async function provider(body) {++ const r = await request(h.base, 'POST', '/api/v1/providers', { tok, body: { kind: 'http', auth_mode: 'none', timeout_ms: 3000, ...body } });++ assert.strictEqual(r.status, 201, r.text);++}++async function model(providerKey, modelKey, price) {++ const r = await request(h.base, 'POST', '/api/v1/models', { tok, body: { provider_key: providerKey, model_key: modelKey, type: 'chat', cost: { in_per_mtok: price, out_per_mtok: price } } });++ assert.strictEqual(r.status, 201, r.text);++}++async function route(key, body) {++ const r = await request(h.base, 'POST', `/api/v1/routes/${key}/versions`, { tok, body });++ assert.strictEqual(r.status, 201, r.text);++ return r.body.route;++}++const CANDIDATE_KEYS = ['cost', 'eligible', 'excluded_reason', 'latency', 'model', 'provider'];++++t.test('forced failure on every route: a logged fallback, or an explicit 503 (never synthetic)', async () => {++ const ok = await seamServer(() => ({ text: 'a real answer', model: 'seam-model', usage: { input: 5, output: 2 } }));++ h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '50' } });++ await provider({ key: 'dead', base_url: 'http://127.0.0.1:1/x', capabilities: ['chat', 'generate', 'summarize', 'extract', 'json'] });++ await provider({ key: 'ok', base_url: ok.url, capabilities: ['chat', 'generate', 'summarize', 'extract', 'json'] });++++ const routes = new Map();++ for (const r of await h.registry.listRoutes()) {++ const rr = await h.registry.resolveRoute(r.key);++ if (rr && !rr.disabled) routes.set(rr.key, rr);++ }++ assert.ok(routes.size >= 8, `routes discovered: ${[...routes.keys()].join(', ')}`);++++ const failed = [];++ for (const rr of routes.values()) {++ const op = rr.capability === 'transcribe' || rr.key === 'live.stt' ? 'transcribe' : rr.capability === 'embed' ? 'embed' : 'chat';++ const req = op === 'embed' ? { input: ['x'], timeoutMs: 2000 }++ : op === 'transcribe' ? { filePath: '/nope.mp4', language: 'en', timeoutMs: 2000 }++ : { messages: [{ role: 'user', content: 'hi' }], timeoutMs: 2000 };++ const logs = [];++ let exec = null; let err = null;++ try { exec = await h.pool.execute(rr, op, req, { routeKey: rr.key, routeVersion: rr.version, logRequest: (e) => logs.push(e) }); } catch (e) { err = e; }++++ const explain = exec ? exec.explain : (err && err.extra && err.extra.explain);++ assert.ok(explain && Array.isArray(explain.candidates), `${rr.key}: explain present`);++ for (const c of explain.candidates) assert.deepStrictEqual(Object.keys(c).sort(), CANDIDATE_KEYS, `${rr.key}: candidate shape`);++ if (err) {++ assert.strictEqual(err.status, 503, `${rr.key}: explicit 503 (${err.message})`);++ assert.strictEqual(err.code, 'provider.unavailable', `${rr.key}: stable code`);++ assert.ok(logs.some((e) => e.status !== 'ok'), `${rr.key}: the failure was logged`);++ failed.push(rr.key);++ } else {++ assert.ok(!exec.synthetic, `${rr.key}: never a synthetic answer`);++ assert.ok(exec.provider, `${rr.key}: a provider answered`);++ if (rr.capability === 'chat') {++ assert.ok(logs.some((e) => e.status === 'ok' && e.fallback === 1), `${rr.key}: the dead primary failed and the pool fell back`);++ assert.strictEqual(exec.explain.selected, 'ok', `${rr.key}: explain names the winner`);++ }++ }++ }++ assert.deepStrictEqual(failed.sort(), ['default.embedding', 'live.stt', 'media.paid'], `failed routes: ${failed}`);++ await h.stop(); await ok.close();++});++++t.test('two providers on one capability: the cheaper is chosen; a circuit-open one is excluded', async () => {++ const ok = await seam();++ const flaky = await seamServer(() => ({ status: 500, body: { error: { message: 'boom' } } }));++ h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '2', AI_BREAKER_COOLDOWN_MS: '600000', AI_PLACEMENT_MIN_GAIN: '0' } });++ await provider({ key: 'cheap', base_url: ok.url, capabilities: ['classify'], default_model: 'cheap-m' });++ await model('cheap', 'cheap-m', 1);++ await provider({ key: 'pricey', base_url: ok.url, capabilities: ['classify'], default_model: 'pricey-m' });++ await model('pricey', 'pricey-m', 20);++ const r = await route('test.pool-cheap', { capability: 'classify', constraints: { objective: 'cheapest' }, timeout_ms: 3000 });++ assert.strictEqual(r.capability, 'classify');++++ const exec = await h.pool.execute(r, 'classify', { messages: [{ role: 'user', content: 'x' }], timeoutMs: 3000 }, { routeKey: r.key, routeVersion: r.version });++ assert.strictEqual(exec.provider, 'cheap', 'the cheaper provider is chosen');++ assert.strictEqual(exec.explain.selected, 'cheap');++ const byProvider = Object.fromEntries(exec.explain.candidates.map((c) => [c.provider, c]));++ assert.ok(byProvider.cheap.cost < byProvider.pricey.cost, 'rate cards carry the real prices');++++ // A provider that keeps failing opens its circuit; placement then excludes it and the other answers.++ await provider({ key: 'flaky', base_url: flaky.url, capabilities: ['extract'], default_model: 'flaky-m' });++ await model('flaky', 'flaky-m', 1);++ await provider({ key: 'good', base_url: ok.url, capabilities: ['extract'], default_model: 'good-m' });++ await model('good', 'good-m', 20);++ const rb = await route('test.pool-breaker', { capability: 'extract', constraints: { objective: 'cheapest' }, timeout_ms: 3000 });++ const extract = { messages: [{ role: 'user', content: 'x' }], timeoutMs: 3000 };++ for (let i = 0; i < 2; i++) {++ const e = await h.pool.execute(rb, 'extract', extract, { routeKey: rb.key, routeVersion: rb.version });++ assert.strictEqual(e.provider, 'good', 'the failing provider falls back to good');++ assert.strictEqual(e.explain.selected, 'good');++ }++ assert.strictEqual((await h.pool.health('flaky')).state, 'open');++ const e3 = await h.pool.execute(rb, 'extract', extract, { routeKey: rb.key, routeVersion: rb.version });++ const flakyCand = e3.explain.candidates.find((c) => c.provider === 'flaky');++ assert.strictEqual(flakyCand.eligible, false, 'the open circuit excludes it');++ assert.match(flakyCand.excluded_reason, /health down/);++ assert.strictEqual(e3.provider, 'good');++ await ok.close(); await flaky.close();++});++++t.test('an upstream 429 shifts to another provider without opening the circuit', async () => {++ const ok = await seamServer(() => ({ text: 'steady answer', model: 'steady-m', usage: { input: 3, output: 1 } }));++ const busy = await seamServer(() => ({ status: 429, body: { error: { message: 'rate limited' } } }));++ await provider({ key: 'busy', base_url: busy.url, capabilities: ['enrich'], default_model: 'busy-m' });++ await model('busy', 'busy-m', 0.5);++ await provider({ key: 'steady', base_url: ok.url, capabilities: ['enrich'], default_model: 'steady-m' });++ await model('steady', 'steady-m', 20);++ const r = await route('test.pool-429', { capability: 'enrich', constraints: { objective: 'cheapest' }, timeout_ms: 3000 });++++ const exec = await h.pool.execute(r, 'enrich', { messages: [{ role: 'user', content: 'x' }], timeoutMs: 3000 }, { routeKey: r.key, routeVersion: r.version });++ assert.strictEqual(exec.provider, 'steady', 'the 429 provider is skipped for this call');++ assert.strictEqual(exec.explain.selected, 'steady');++ assert.strictEqual((await h.pool.health('busy')).state, 'closed', 'a 429 is capacity, not ill health');++ await ok.close(); await busy.close();++});++++t.test('explain rides the run response and GET /runs/:id; the request log stays', async () => {++ const fails = await seamServer(() => ({ status: 500, body: { error: { message: 'down' } } }));++ const ok = await seamServer(() => ({ text: 'the fallback answer', model: 'p2-m', usage: { input: 4, output: 2 } }));++ await provider({ key: 'p1', base_url: fails.url, capabilities: ['chat', 'generate'], default_model: 'p1-m' });++ await model('p1', 'p1-m', 1);++ await provider({ key: 'p2', base_url: ok.url, capabilities: ['chat', 'generate'], default_model: 'p2-m' });++ await model('p2', 'p2-m', 20);++ const r = await route('default.chat', { capability: 'chat', constraints: { objective: 'cheapest' }, timeout_ms: 3000, response_format: 'text' });++ assert.strictEqual(r.capability, 'chat');++++ const res = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok, body: { prompt: 'hello', options: { cache: false } } });++ assert.strictEqual(res.status, 201, res.text);++ const run = res.body.run;++ assert.strictEqual(run.status, 'succeeded');++ assert.strictEqual(run.provenance.provider, 'p2');++ assert.ok(run.explain, 'the run carries explain');++ assert.strictEqual(run.explain.objective, 'cheapest');++ assert.ok(Array.isArray(run.explain.reasons) && run.explain.reasons.length);++ assert.strictEqual(run.explain.selected, 'p2');++ for (const c of run.explain.candidates) assert.deepStrictEqual(Object.keys(c).sort(), CANDIDATE_KEYS);++ const shared = run.explain.candidates.find((c) => c.provider === 'shared');++ assert.ok(shared && shared.eligible === false && shared.excluded_reason, 'the credential-less shared provider is excluded with a reason');++++ const got = await request(h.base, 'GET', `/api/v1/runs/${run.id}`, { tok });++ assert.deepStrictEqual(got.body.run.explain, run.explain, 'GET /runs/:id returns the same explain');++ const reqs = got.body.requests;++ assert.ok(reqs.some((x) => x.provider_key === 'p1' && x.status === 'error'), 'the failed primary is in the request log');++ const fb = reqs.find((x) => x.provider_key === 'p2');++ assert.strictEqual(fb.status, 'ok');++ assert.strictEqual(fb.fallback, 1, 'the fallback attempt is marked');++++ // A run that nothing can answer still carries explain (selected null) and the explicit 503 code.++ await request(h.base, 'POST', '/api/v1/providers/p2/disable', { tok });++ const bad = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok, body: { prompt: 'nobody', options: { cache: false } } });++ assert.strictEqual(bad.body.run.status, 'failed', bad.text);++ assert.strictEqual(bad.body.run.error.code, 'provider.unavailable');++ assert.ok(bad.body.run.explain && bad.body.run.explain.selected === null, 'a failed run still explains itself');++ await fails.close(); await ok.close();++});++++t.test('seeded text pools keep the shared provider as authority: a free local model is only the failover', async () => {++ const { plan } = require('openvibe-sdk/placement');++ const shared = await h.registry.getProvider('shared');++ const authority = shared.default_model ? `shared:${shared.default_model}` : 'shared';++ let checked = 0;++ for (const r of await h.registry.listRoutes()) {++ const rr = await h.registry.resolveRoute(r.key);++ if (!rr || rr.disabled || rr.capability !== 'chat' || rr.created_by !== 'seed' || !/^(live\.|default\.(chat|json)$)/.test(rr.key)) continue;++ checked++;++ assert.strictEqual(rr.constraints.objective, 'correctness', `${rr.key}: correctness keeps the authority`);++ assert.strictEqual(rr.constraints.authority, authority, `${rr.key}: the shared provider is the authority`);++ }++ assert.ok(checked >= 5, `seeded text pools checked: ${checked}`);++ // The planner under those constraints: the paid shared offer beats the free local one while it is up, and++ // the local one takes over the moment shared is down.++ const offer = (id, status, pricing) => ({ offer_id: id, kind: 'provider', provider: id.split(':')[0], region: 'global', trust: 'first-party', capabilities: ['chat'], latency_ms: {}, health: { status }, pricing });++ const req = { kind: 'ai', mobility: 'request', latency_class: 'interactive', objective: 'correctness', capabilities: ['chat'], units: 1, authority: 'shared:m' };++ const cards = [{ id: 'shared:m', provider: 'shared', metric: 'tokens', unit_size: 1e6, unit_price_usd: 2, free_allowance: 0, reset_period: 'month' }];++ const up = plan(req, [offer('shared:m', 'up', { model: 'metered', rate_card: 'shared:m' }), offer('local:l', 'up', { model: 'prepaid' })], { rateCards: cards });++ assert.strictEqual(up.selected, 'shared:m', JSON.stringify(up.reasons));++ const down = plan(req, [offer('shared:m', 'down', { model: 'metered', rate_card: 'shared:m' }), offer('local:l', 'up', { model: 'prepaid' })], { rateCards: cards });++ assert.strictEqual(down.selected, 'local:l', JSON.stringify(down.reasons));++});++++t.test('shutdown', async () => { await h.stop(); });++t.run();
@@ -196,17 +196,23 @@ definition, but never over an admin's version.
@@ -196,17 +196,23 @@ definition, but never over an admin's version.
| Kind | What | Configured by |
| Kind | What | Configured by |
|---|---|---|
|---|---|---|
| `stub` | Deterministic, no key, realistically shaped output (schema-shaped JSON, sentences, unit-vector embeddings, timed transcript segments), clearly marked synthetic (`(synthetic)` text, `synthetic: true` on the run). Never cached. | always present |
| `stub` | Deterministic, no key, realistically shaped output (schema-shaped JSON, sentences, unit-vector embeddings, timed transcript segments), clearly marked synthetic (`(synthetic)` text, `synthetic: true` on the run). Never cached. | always present |
-| `openai` | Any OpenAI-compatible API (OpenAI, OpenRouter, Groq, Together, Ollama, LM Studio, llama.cpp): chat with structured-output step-down, embeddings, Whisper-style transcription. Ported from Live's `llm.js` / `ai-provider.js`. | `AI_PROVIDER`, `AI_BASE_URL`, `AI_API_KEY`, `AI_MODEL`, `AI_MODEL_<ROLE>` — Live's setting names |
+| `openai` | Any OpenAI-compatible API (OpenAI, OpenRouter, Groq, Together, Ollama, LM Studio, llama.cpp): chat with structured-output step-down, embeddings, Whisper-style transcription. Ported from Live's `llm.js` / `ai-provider.js`. | `AI_PROVIDER`, `AI_BASE_URL`, `AI_API_KEY`, `AI_MODEL` — Live's setting names |
| `anthropic` | Messages API with cached system blocks and forced-tool JSON. Ported from Live's `llm.js`. | same, with `AI_PROVIDER=anthropic` |
| `anthropic` | Messages API with cached system blocks and forced-tool JSON. Ported from Live's `llm.js`. | same, with `AI_PROVIDER=anthropic` |
| `http` | The local HTTP seam: `POST {operation, …}` → `{text, json, usage}` | `AI_HTTP_SEAM_URL` or the admin API |
| `http` | The local HTTP seam: `POST {operation, …}` → `{text, json, usage}` | `AI_HTTP_SEAM_URL` or the admin API |
| `whisper` | whisper.cpp on this host, with Live's VAD, hallucination filter, multilingual model and live/batch lanes | `WHISPER_*` — Live's names |
| `whisper` | whisper.cpp on this host, with Live's VAD, hallucination filter, multilingual model and live/batch lanes | `WHISPER_*` — Live's names |
| `openai`, key `local` | A model server on this host speaking the OpenAI API (llama.cpp's `llama-server`, Ollama's `/v1`): no key, never paid; route `media.local` | `AI_LOCAL_LLM_URL`, `AI_LOCAL_LLM_MODEL`, `AI_LOCAL_LLM_TIMEOUT_MS` |
| `openai`, key `local` | A model server on this host speaking the OpenAI API (llama.cpp's `llama-server`, Ollama's `/v1`): no key, never paid; route `media.local` | `AI_LOCAL_LLM_URL`, `AI_LOCAL_LLM_MODEL`, `AI_LOCAL_LLM_TIMEOUT_MS` |
-Routing tries the route's primary, then each fallback; a provider that is disabled, missing
-credentials, circuit-open or unable to do the operation is **skipped and logged**; every call has a
-strict timeout, cancellation and one retry on a transient failure; a fallback that answers is
-recorded on the run (`fallback_used`), in the request log (`fallback = 1`) and in the audit log. The
-stub joins every route as a last resort outside production only (`AI_STUB_FALLBACK`).
+Routing: a route either names a `capability` and the router builds its pool from every provider that
+can serve it, ordered by `openvibe-sdk/placement` (objective, price, measured latency, health, with
+hysteresis through the route's current placement), or pins a `[primary, ...fallbacks]` list. A
+provider that is disabled, missing credentials, circuit-open or unable to do the operation is
+**skipped and logged**; every call has a strict timeout, cancellation and one retry on a transient
+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`).
## Workflows
## Workflows
@@ -340,8 +346,10 @@ Deployed: `/opt/openvibe.ai`, env `/etc/openvibe/ai.env` (0600), unit `deploy/sy
@@ -340,8 +346,10 @@ Deployed: `/opt/openvibe.ai`, env `/etc/openvibe/ai.env` (0600), unit `deploy/sy
of Network OAuth client `ai`, and `AI_CONSOLE_SESSION_SECRET` in `/etc/openvibe/ai.env`.
of Network OAuth client `ai`, and `AI_CONSOLE_SESSION_SECRET` in `/etc/openvibe/ai.env`.
- Import holds from the 2026-09-23 run: a streamer's own provider key stays in Live, and 3,021 Live
- Import holds from the 2026-09-23 run: a streamer's own provider key stays in Live, and 3,021 Live
translations cannot become cache entries (they have no source text) and stay in Live.
translations cannot become cache entries (they have no source text) and stay in Live.
-- No fallback is declared on any production route, so an outage of the one real provider fails every
- Live AI feature. On 2026-09-23 `live.*` and `network.site_copy` outputs carried no citations or gaps
+- Text routes now run a capability pool (`shared` plus a local model when one is configured) instead
+ of a declared fallback, but speech and embeddings still have a single upstream, so an outage there
+ fails that feature until O9 adds a second provider. On 2026-09-23 `live.*` and `network.site_copy`
+ outputs carried no citations or gaps
(0 citation rows); runs record grounding (cited sources and named gaps) since 2026-09-25.
(0 citation rows); runs record grounding (cited sources and named gaps) since 2026-09-25.
- The host has a backup of `ov_ai` but no restore drill has run for it; `/metrics` is built but not
- The host has a backup of `ov_ai` but no restore drill has run for it; `/metrics` is built but not
deployed yet.
deployed yet.
new file mode 100644
new file mode 100644
@@ -0,0 +1,12 @@
@@ -0,0 +1,12 @@
+-- phase: expand
+-- T6 provider router, phase 1 (brief-t6-ai.out.md §4/§7, decisions-t4-t6.md "T6"):
+-- a route either pins [primary, ...fallbacks] (unchanged) or names a capability and lets the router
+-- assemble the pool from every provider that can serve it, ordered by openvibe-sdk/placement. The
+-- run keeps the placement explanation the router produced (objective, reasons, selected, candidates),
+-- so POST /runs, the direct operations and GET /runs/:id all carry it.
+
+ALTER TABLE routes ADD COLUMN capability text COLLATE "C"; -- a feature the pool must cover (chat|transcribe|embed|...) or NULL for pinned routes
+ALTER TABLE routes ADD COLUMN constraints text COLLATE "C" NOT NULL DEFAULT '{}'; -- placement requirements: objective, latency_class, max_latency_ms, max_cost_usd, trust, model, authority
+ALTER TABLE routes ADD COLUMN pinned text COLLATE "C" NOT NULL DEFAULT '[]'; -- [{provider, model}] always kept in the pool (migration)
+
+ALTER TABLE runs ADD COLUMN explain text COLLATE "C"; -- platform.placement-result@1 plus the candidate view, as the router returned it
@@ -13,7 +13,7 @@
@@ -13,7 +13,7 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.23.1",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
@@ -1464,9 +1464,9 @@
@@ -1464,9 +1464,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.23.1",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.23.1",
- "integrity": "sha512-6u1Ym8u+nSKyq+uJjel4/3QDQ9HOjpQzFa8oMasorXQhgHUlDEQ7X6F6Xf1zZTuuVdMVkxOeqJqOnQbraktosg==",
+ "version": "0.25.2",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2",
+ "integrity": "sha512-rbBj0NXUFNEOTDEtXYS5mD2uO5VJw6ZGC9zME1nbcEqrATMWbKeOt2cO02T/7QDP2iR+lJP97QhYTEF8rlyMog==",
"license": "MIT",
"license": "MIT",
"engines": {
"engines": {
"node": ">=20"
"node": ">=20"
@@ -25,7 +25,7 @@
@@ -25,7 +25,7 @@
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.76.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.23.1",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.25.2",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.27.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
@@ -33,11 +33,6 @@ function load(env = process.env) {
@@ -33,11 +33,6 @@ function load(env = process.env) {
const port = int(env.PORT, 4700);
const port = int(env.PORT, 4700);
const aiProvider = String(env.AI_PROVIDER || '').trim().toLowerCase();
const aiProvider = String(env.AI_PROVIDER || '').trim().toLowerCase();
const aiBaseUrl = trimUrl(env.AI_BASE_URL);
const aiBaseUrl = trimUrl(env.AI_BASE_URL);
- const roleModels = {};
- for (const r of LIVE_ROLES) {
- const v = env[`AI_MODEL_${r.toUpperCase()}`];
- if (v && String(v).trim()) roleModels[r] = String(v).trim();
- }
let pricing = {};
let pricing = {};
if (env.AI_PRICING_JSON) {
if (env.AI_PRICING_JSON) {
try { pricing = JSON.parse(env.AI_PRICING_JSON) || {}; } catch { throw new Error('AI_PRICING_JSON is not valid JSON'); }
try { pricing = JSON.parse(env.AI_PRICING_JSON) || {}; } catch { throw new Error('AI_PRICING_JSON is not valid JSON'); }
@@ -89,15 +84,15 @@ function load(env = process.env) {
@@ -89,15 +84,15 @@ function load(env = process.env) {
baseUrl: aiBaseUrl,
baseUrl: aiBaseUrl,
apiKeyRef: 'env:AI_API_KEY', // secret reference, never the value
apiKeyRef: 'env:AI_API_KEY', // secret reference, never the value
model: String(env.AI_MODEL || '').trim(),
model: String(env.AI_MODEL || '').trim(),
- roleModels,
timeoutMs: int(env.AI_PROVIDER_TIMEOUT_MS, 30000),
timeoutMs: int(env.AI_PROVIDER_TIMEOUT_MS, 30000),
},
},
- // Optional second OpenAI-compatible provider used as the fallback of every route.
- fallback: {
- baseUrl: trimUrl(env.AI_FALLBACK_BASE_URL),
- apiKeyRef: 'env:AI_FALLBACK_API_KEY',
- model: String(env.AI_FALLBACK_MODEL || '').trim(),
- kind: String(env.AI_FALLBACK_PROVIDER || '').trim().toLowerCase(),
+ // ── Placement (T6 provider router): openvibe-sdk/placement orders a route's capability pool ──
+ placement: {
+ // A candidate replaces the route's current placement only when it scores this much better,
+ // so a small price/latency wobble does not make the router oscillate.
+ minGain: Math.max(0, Math.min(1, float(env.AI_PLACEMENT_MIN_GAIN, 0.15))),
+ costWeight: float(env.AI_PLACEMENT_COST_WEIGHT, 0.5),
+ latencyWeight: float(env.AI_PLACEMENT_LATENCY_WEIGHT, 0.5),
},
},
// Configurable local HTTP seam: POST {operation, request} JSON to this URL.
// Configurable local HTTP seam: POST {operation, request} JSON to this URL.
httpSeamUrl: trimUrl(env.AI_HTTP_SEAM_URL),
httpSeamUrl: trimUrl(env.AI_HTTP_SEAM_URL),
@@ -5,16 +5,22 @@
@@ -5,16 +5,22 @@
* pool.adapter(key) the adapter for a provider record (rebuilt when the record changes)
* pool.adapter(key) the adapter for a provider record (rebuilt when the record changes)
* pool.health(key) circuit-breaker state
* pool.health(key) circuit-breaker state
* pool.execute(route, op, req, ctx)
* pool.execute(route, op, req, ctx)
- * tries the route's primary, then each fallback, skipping candidates that are disabled,
- * circuit-open, missing credentials or that do not support the operation; applies a strict
- * timeout and cancellation to every call and one retry on a transient failure; records one
- * request-log row per attempt or skip (fallback = 1 for every non-primary candidate) and
- * returns { result, provider, model, fallbackUsed }. When nothing answers it throws
+ * tries the route's candidates in order, skipping ones that are disabled, circuit-open,
+ * missing credentials or that do not support the operation; applies a strict timeout and
+ * cancellation to every call and one retry on a transient failure; records one request-log
+ * row per attempt or skip (fallback = 1 for every non-first candidate) and returns
+ * { result, provider, model, fallbackUsed, explain }. When nothing answers it throws
* AiError 503 provider.unavailable — an explicit state, never a made-up answer.
* AiError 503 provider.unavailable — an explicit state, never a made-up answer.
*
*
+ * A route either names a `capability` and the pool is built from every provider that can serve it,
+ * ordered by openvibe-sdk/placement (objective, price, measured latency, health, with hysteresis
+ * through placement_state.current); or it pins `[primary, ...fallbacks]`. `explain` (objective,
+ * reasons, selected, candidates with their excluded reasons) rides every run response.
+ *
* Circuit breaker per provider: `failureThreshold` consecutive failures open it for `cooldownMs`;
* Circuit breaker per provider: `failureThreshold` consecutive failures open it for `cooldownMs`;
* after that one half-open probe decides whether it closes again.
* after that one half-open probe decides whether it closes again.
*/
*/
+const placement = require('openvibe-sdk/placement');
const { AiError, resolveSecret, sha256 } = require('../util');
const { AiError, resolveSecret, sha256 } = require('../util');
const { retryable, sleep, ProviderError } = require('./common');
const { retryable, sleep, ProviderError } = require('./common');
const { createStubProvider } = require('./stub');
const { createStubProvider } = require('./stub');
@@ -29,6 +35,11 @@ function callerFault(err) {
@@ -29,6 +35,11 @@ function callerFault(err) {
return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 403, 404, 408, 429].includes(st);
return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 403, 404, 408, 429].includes(st);
}
}
+/** An upstream 429 is rate-limit capacity, not provider ill health: it shifts to the next candidate but must not open the circuit. */
+function rateLimited(err) {
+ return (err && err.status === 429) || (err && err.code === 'provider.rate_limited');
+}
+
function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, credentialFetch = null }) {
function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, credentialFetch = null }) {
const cache = new Map(); // key -> { stamp, adapter }
const cache = new Map(); // key -> { stamp, adapter }
const stats = {}; // key -> { calls, failures, ... } (in memory; tests + /api/ready)
const stats = {}; // key -> { calls, failures, ... } (in memory; tests + /api/ready)
@@ -113,10 +124,129 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -113,10 +124,129 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
}
}
// ── Routing ────────────────────────────────────────────
// ── Routing ────────────────────────────────────────────
- async function candidates(route) {
+ /** A provider that is never billed: the deterministic stub and the local servers (llama.cpp, whisper.cpp). */
+ function isFreeProvider(p) {
+ return p.kind === 'stub' || p.kind === 'whisper' || Boolean(p.metadata && p.metadata.paid === false);
+ }
+ const offerId = (provider, model) => (model ? `${provider}:${model}` : provider);
+ const splitOfferId = (id) => {
+ const i = String(id).indexOf(':');
+ return i < 0 ? { provider: id, model: null } : { provider: id.slice(0, i), model: id.slice(i + 1) };
+ };
+
+ // ── Capability pool (T6 phase 1): placement over every provider that can serve the capability ──
+ const getRouteState = db.prepare('SELECT * FROM placement_state WHERE route_key = ?');
+ const getProviderStats = db.prepare(`SELECT provider, MAX(latency_p95_ms) AS p95, SUM(requests)::bigint AS requests, SUM(errors)::bigint AS errors
+ FROM provider_stats_daily WHERE day >= ? GROUP BY provider`);
+ async function statsByProvider(days = 7) {
+ const from = new Date(clock.now() - days * 86400000).toISOString().slice(0, 10);
+ const m = new Map();
+ for (const r of await getProviderStats.all(from)) m.set(r.provider, { p95: r.p95 == null ? null : Number(r.p95), requests: Number(r.requests), errors: Number(r.errors) });
+ return m;
+ }
+
+ /**
+ * The order the router tries for a capability route: openvibe-sdk/placement ranks every provider whose
+ * capabilities cover the operation, by objective, price (the model rate card), measured latency and
+ * health (placement_state error rate + the breaker); hysteresis through the route's current placement.
+ * The stub is never a placement candidate (synthetic output must not win on price) — it is appended last.
+ */
+ async function poolCandidates(route, features, operation) {
+ const stats = await statsByProvider();
+ const offers = [];
+ const rateCards = [];
+ const seen = new Set();
+ const addOffer = async (p, model, caps) => {
+ const id = offerId(p.key, model);
+ if (seen.has(id)) return;
+ seen.add(id);
+ const creds = p.auth_mode === 'none' || Boolean(resolveSecret(p.secret_ref, env));
+ const h = await health(p.key);
+ const st = stats.get(p.key) || { p95: null, requests: 0, errors: 0 };
+ const errorRate = st.requests ? st.errors / st.requests : 0;
+ const status = p.status !== 'active' || !creds ? 'down' : h.state === 'open' ? 'down' : (h.state === 'half_open' || errorRate > 0.5) ? 'degraded' : 'up';
+ const free = isFreeProvider(p);
+ offers.push({
+ offer_id: id, kind: 'provider', provider: p.key, region: 'global',
+ trust: (p.metadata && p.metadata.local) || p.kind === 'stub' ? 'community' : 'first-party',
+ capabilities: caps, latency_ms: st.p95 ? { p95: st.p95 } : {}, health: { status },
+ pricing: free ? { model: 'prepaid' } : { model: 'metered', rate_card: id },
+ });
+ if (!free) {
+ const price = await priceFor(p.key, model);
+ rateCards.push({ id, provider: p.key, metric: 'tokens', unit_size: 1e6, unit_price_usd: (price.in + price.out) / 2, free_allowance: 0, reset_period: 'month' });
+ }
+ };
+ 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);
+ }
+ // 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.
+ for (const pin of route.pinned || []) {
+ if (!pin || !pin.provider || pin.provider === 'stub') continue;
+ const p = await registry.getProvider(pin.provider);
+ if (!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);
+ }
+ const c = route.constraints || {};
+ const requirements = {
+ kind: 'ai', mobility: 'request', latency_class: c.latency_class || 'interactive', objective: c.objective || 'balanced',
+ capabilities: features, units: Number(c.units) || 1, latency_op: operation,
+ ...(c.max_latency_ms != null ? { max_latency_ms: Number(c.max_latency_ms) } : {}),
+ ...(c.max_cost_usd != null ? { max_cost_usd: Number(c.max_cost_usd) } : {}),
+ ...(c.trust ? { trust: c.trust } : {}),
+ ...(c.authority ? { authority: c.authority } : {}),
+ };
+ const state = await getRouteState.get(route.key);
+ const current = state && state.current_provider ? offerId(state.current_provider, state.current_model) : null;
+ const result = placement.plan(requirements, offers, {
+ rateCards, now: clock.now(), current, minGain: config.placement.minGain,
+ weights: { cost: config.placement.costWeight, latency: config.placement.latencyWeight },
+ });
+ const byId = new Map(offers.map(o => [o.offer_id, o]));
+ const explain = {
+ objective: result.objective, reasons: result.reasons, selected: result.selected,
+ candidates: result.candidates.map((x) => {
+ const o = byId.get(x.id) || {};
+ return { provider: o.provider || splitOfferId(x.id).provider, model: splitOfferId(x.id).model, eligible: Boolean(x.eligible),
+ excluded_reason: x.excluded_because || null, cost: x.estimated_cost_usd == null ? null : x.estimated_cost_usd, latency: x.estimated_latency_ms == null ? null : x.estimated_latency_ms };
+ }),
+ };
+ const order = [];
+ if (result.selected) order.push(result.selected);
+ const rest = result.candidates.filter(x => x.id !== result.selected).sort((a, b) => {
+ const ea = a.eligible && Number.isFinite(a.estimated_cost_usd) ? 0 : 1;
+ const eb = b.eligible && Number.isFinite(b.estimated_cost_usd) ? 0 : 1;
+ return ea - eb || (a.score ?? Infinity) - (b.score ?? Infinity);
+ });
+ for (const x of rest) order.push(x.id);
+ return { order: order.map(splitOfferId), explain };
+ }
+
+ /** A pinned route: [primary, ...fallbacks], exactly as before. */
+ async function pinnedCandidates(route, features) {
const list = [route.primary, ...(route.fallbacks || [])].filter(c => c && c.provider);
const list = [route.primary, ...(route.fallbacks || [])].filter(c => c && c.provider);
- if (config.stubFallback && !list.some(c => c.provider === 'stub') && await registry.getProvider('stub')) list.push({ provider: 'stub', model: null, auto: true });
- return list;
+ const explain = { objective: null, reasons: ['pinned route: the primary, then its fallbacks'], selected: null, candidates: [] };
+ for (const c of list) {
+ const entry = await adapter(c.provider);
+ const p = entry && entry.record;
+ const why = await skipReason(p, entry && entry.adapter, features);
+ explain.candidates.push({ provider: c.provider, model: c.model || (p && p.default_model) || null, eligible: !why, excluded_reason: why, cost: null, latency: null });
+ }
+ return { order: list.map(c => ({ provider: c.provider, model: c.model || null })), explain };
+ }
+
+ async function candidates(route, features, operation) {
+ const { order, explain } = 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')) {
+ order.push({ provider: 'stub', model: null });
+ explain.candidates.push({ provider: 'stub', model: null, eligible: true, excluded_reason: null, cost: 0, latency: null });
+ }
+ return { order, explain };
}
}
async function skipReason(p, a, features) {
async function skipReason(p, a, features) {
@@ -191,7 +321,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -191,7 +321,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
if (req.image) features.push('vision');
if (req.image) features.push('vision');
if (req.json) features.push('json');
if (req.json) features.push('json');
const tried = [];
const tried = [];
- const list = await candidates(route);
+ const { order: list, explain } = await candidates(route, features, operation);
for (let i = 0; i < list.length; i++) {
for (let i = 0; i < list.length; i++) {
const c = list[i];
const c = list[i];
const entry = await adapter(c.provider);
const entry = await adapter(c.provider);
@@ -233,7 +363,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -233,7 +363,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
debug_response: ctx.debugRaw ? String(result.text || '').slice(0, 200000) : null,
debug_response: ctx.debugRaw ? String(result.text || '').slice(0, 200000) : null,
});
});
}
}
- return { result, provider: c.provider, providerKind: p.kind, model: result.model || model, fallbackUsed: fallback, usage, cost, latencyMs: latency, synthetic: Boolean(result.synthetic || a.synthetic), tried, startedAt: started };
+ explain.selected = c.provider;
+ return { result, provider: c.provider, providerKind: p.kind, model: result.model || model, fallbackUsed: fallback, usage, cost, latencyMs: latency, synthetic: Boolean(result.synthetic || a.synthetic), tried, startedAt: started, explain };
} catch (err) {
} catch (err) {
lastErr = err;
lastErr = err;
if (ctx.signal && ctx.signal.aborted) {
if (ctx.signal && ctx.signal.aborted) {
@@ -247,12 +378,15 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -247,12 +378,15 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
}
}
}
}
// A request the provider rejected as malformed is the caller's fault, not a sign the
// A request the provider rejected as malformed is the caller's fault, not a sign the
- // provider is unhealthy: it must not open the circuit other callers depend on.
- if (!callerFault(lastErr)) await recordFailure(c.provider, lastErr);
+ // provider is unhealthy; a 429 is capacity, not health either, and moves on without
+ // opening the circuit. Neither must take a shared provider away from other callers.
+ if (!callerFault(lastErr) && !rateLimited(lastErr)) await recordFailure(c.provider, lastErr);
tried.push({ provider: c.provider, error: String(lastErr && lastErr.message || lastErr).slice(0, 200) });
tried.push({ provider: c.provider, error: String(lastErr && lastErr.message || lastErr).slice(0, 200) });
log.warn(`[ai] ${operation} via ${c.provider}/${model || '-'} failed: ${lastErr && lastErr.message}`);
log.warn(`[ai] ${operation} via ${c.provider}/${model || '-'} failed: ${lastErr && lastErr.message}`);
}
}
- throw new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried });
+ explain.selected = null;
+ explain.reasons = [...(explain.reasons || []), 'no provider on this route could answer'];
+ throw new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried, explain });
}
}
return { adapter, health, resetHealth, execute, executeWithCredential, priceFor, costOf, stats, statFor, candidates };
return { adapter, health, resetHealth, execute, executeWithCredential, priceFor, costOf, stats, statFor, candidates };
@@ -156,16 +156,23 @@ function createRegistry(db, { clock = { now: () => Date.now() }, env = process.e
@@ -156,16 +156,23 @@ function createRegistry(db, { clock = { now: () => Date.now() }, env = process.e
return {
return {
key: r.key, version: r.version, status: r.status, primary: { provider: r.primary_provider, model: r.primary_model },
key: r.key, version: r.version, status: r.status, primary: { provider: r.primary_provider, model: r.primary_model },
fallbacks: parseJson(r.fallbacks, []), options: parseJson(r.options, {}), max_output_tokens: r.max_output_tokens,
fallbacks: parseJson(r.fallbacks, []), options: parseJson(r.options, {}), max_output_tokens: r.max_output_tokens,
- response_format: r.response_format, timeout_ms: r.timeout_ms, alias_of: r.alias_of, created_by: r.created_by, created_at: r.created_at,
+ response_format: r.response_format, timeout_ms: r.timeout_ms, alias_of: r.alias_of,
+ capability: r.capability || null, constraints: parseJson(r.constraints, {}), pinned: parseJson(r.pinned, []),
+ created_by: r.created_by, created_at: r.created_at,
};
};
}
}
- const routeContent = (r) => sha256({ p: r.primary, f: r.fallbacks, o: r.options, m: r.max_output_tokens, rf: r.response_format, t: r.timeout_ms, a: r.alias_of, s: r.status });
+ const routeContent = (r) => sha256({ p: r.primary, f: r.fallbacks, o: r.options, m: r.max_output_tokens, rf: r.response_format, t: r.timeout_ms, a: r.alias_of, s: r.status, c: r.capability || null, x: r.constraints || {}, pi: r.pinned || [] });
async function createRouteVersion(key, input, { actor = 'system', trace = null } = {}) {
async function createRouteVersion(key, input, { actor = 'system', trace = null } = {}) {
if (!KEY_RE.test(key || '')) throw new AiError(422, 'ai.invalid', 'route key must be lowercase dotted');
if (!KEY_RE.test(key || '')) throw new AiError(422, 'ai.invalid', 'route key must be lowercase dotted');
const prev = decodeRoute(await latestAny('routes', key));
const prev = decodeRoute(await latestAny('routes', key));
const r = {
const r = {
primary: input.primary || (prev && prev.primary),
primary: input.primary || (prev && prev.primary),
fallbacks: input.fallbacks !== undefined ? input.fallbacks : (prev ? prev.fallbacks : []),
fallbacks: input.fallbacks !== undefined ? input.fallbacks : (prev ? prev.fallbacks : []),
+ // A pool route names a capability; sending a version without one is a deliberate pin (an admin edit
+ // that states primary/fallbacks), so capability is taken from the input only, never inherited.
+ capability: input.capability !== undefined ? (input.capability || null) : (prev && prev.capability && input.primary === undefined ? prev.capability : null),
+ constraints: input.constraints !== undefined ? input.constraints : ((input.primary === undefined && prev) ? prev.constraints : {}),
+ pinned: input.pinned !== undefined ? input.pinned : ((input.primary === undefined && prev) ? prev.pinned : []),
options: input.options !== undefined ? input.options : (prev ? prev.options : {}),
options: input.options !== undefined ? input.options : (prev ? prev.options : {}),
max_output_tokens: input.max_output_tokens !== undefined ? input.max_output_tokens : (prev ? prev.max_output_tokens : null),
max_output_tokens: input.max_output_tokens !== undefined ? input.max_output_tokens : (prev ? prev.max_output_tokens : null),
response_format: input.response_format || (prev && prev.response_format) || 'text',
response_format: input.response_format || (prev && prev.response_format) || 'text',
@@ -173,21 +180,29 @@ function createRegistry(db, { clock = { now: () => Date.now() }, env = process.e
@@ -173,21 +180,29 @@ function createRegistry(db, { clock = { now: () => Date.now() }, env = process.e
alias_of: input.alias_of !== undefined ? input.alias_of : (prev ? prev.alias_of : null),
alias_of: input.alias_of !== undefined ? input.alias_of : (prev ? prev.alias_of : null),
status: input.status || 'active',
status: input.status || 'active',
};
};
- if (!r.alias_of) {
- if (!r.primary || !r.primary.provider) throw new AiError(422, 'ai.invalid', 'route needs primary.provider');
+ if (r.capability && !FEATURES.includes(r.capability)) throw new AiError(422, 'ai.invalid', `route capability must be one of ${FEATURES.join(', ')}`);
+ if (!r.alias_of && !r.capability) {
+ if (!r.primary || !r.primary.provider) throw new AiError(422, 'ai.invalid', 'route needs primary.provider or a capability');
for (const c of [r.primary, ...(r.fallbacks || [])]) {
for (const c of [r.primary, ...(r.fallbacks || [])]) {
if (!c || !await getProvider(c.provider)) throw new AiError(422, 'ai.invalid', `route references unknown provider ${c && c.provider}`);
if (!c || !await getProvider(c.provider)) throw new AiError(422, 'ai.invalid', `route references unknown provider ${c && c.provider}`);
}
}
- } else if (!await latestAny('routes', r.alias_of)) throw new AiError(422, 'ai.invalid', `alias target ${r.alias_of} does not exist`);
+ } else if (r.alias_of && !await latestAny('routes', r.alias_of)) throw new AiError(422, 'ai.invalid', `alias target ${r.alias_of} does not exist`);
+ for (const c of r.pinned || []) {
+ if (!c || !await getProvider(c.provider)) throw new AiError(422, 'ai.invalid', `pinned candidate references unknown provider ${c && c.provider}`);
+ }
if (!Array.isArray(r.fallbacks)) throw new AiError(422, 'ai.invalid', 'fallbacks must be an array');
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 (!['text', 'json'].includes(r.response_format)) throw new AiError(422, 'ai.invalid', 'response_format must be text or json');
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');
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);
const version = await nextVersion('routes', key);
- await db.prepare(`INSERT INTO routes (key, version, status, primary_provider, primary_model, fallbacks, options, max_output_tokens, response_format, timeout_ms, alias_of, created_by, created_at)
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
- .run(key, version, r.status, r.alias_of ? (r.primary && r.primary.provider) || 'stub' : r.primary.provider, r.primary ? r.primary.model || null : null,
- JSON.stringify(r.fallbacks || []), JSON.stringify(r.options || {}), r.max_output_tokens, r.response_format, r.timeout_ms, r.alias_of, actor, now());
- await audit(actor, 'route.version', 'route', key, { trace, metadata: { version, previous: prev ? prev.version : null, status: r.status, alias_of: r.alias_of } });
+ const primaryProvider = r.alias_of ? ((r.primary && r.primary.provider) || 'stub')
+ : (r.primary && r.primary.provider) || ((r.pinned && r.pinned[0] && r.pinned[0].provider) || 'stub');
+ await db.prepare(`INSERT INTO routes (key, version, status, primary_provider, primary_model, fallbacks, options, max_output_tokens, response_format, timeout_ms, alias_of, capability, constraints, pinned, created_by, created_at)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
+ .run(key, version, r.status, primaryProvider, (r.primary && r.primary.model) || null,
+ JSON.stringify(r.fallbacks || []), JSON.stringify(r.options || {}), r.max_output_tokens, r.response_format, r.timeout_ms, r.alias_of,
+ r.capability || null, JSON.stringify(r.constraints || {}), JSON.stringify(r.pinned || []), actor, now());
+ await audit(actor, 'route.version', 'route', key, { trace, metadata: { version, previous: prev ? prev.version : null, status: r.status, alias_of: r.alias_of, capability: r.capability || null } });
return await getRoute(key, version);
return await getRoute(key, version);
}
}
async function getRoute(key, version) {
async function getRoute(key, version) {
@@ -85,6 +85,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -85,6 +85,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
provider: r.provider_key, model: r.model_key, fallback_used: Boolean(r.fallback_used), run_id: r.id, cached_from: r.cached_from, synthetic: Boolean(r.synthetic),
provider: r.provider_key, model: r.model_key, fallback_used: Boolean(r.fallback_used), run_id: r.id, cached_from: r.cached_from, synthetic: Boolean(r.synthetic),
},
},
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 },
+ explain: parseJson(r.explain, null),
citations_count: r.citations_count,
citations_count: r.citations_count,
grounding: parseJson(r.grounding, null),
grounding: parseJson(r.grounding, null),
retry_of: r.retry_of,
retry_of: r.retry_of,
@@ -203,11 +204,12 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -203,11 +204,12 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
const hit = await cache.get(cacheKey, scope);
const hit = await cache.get(cacheKey, scope);
if (hit) {
if (hit) {
await db.tx(async () => {
await db.tx(async () => {
+ const cacheExplain = { objective: 'cache', reasons: ["the answer was already in this scope's cache; no provider was called"], selected: null, candidates: [] };
await db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, requester_type, requester_id, on_behalf_of, attribution,
await db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, requester_type, requester_id, on_behalf_of, attribution,
- source_service, target, target_key, input, input_hash, output, model_key, cache_key, cached_from, retry_of, idempotency_key, trace_id, request_id, options, created_at, started_at, finished_at, citations_count)
+ source_service, target, target_key, input, input_hash, output, model_key, cache_key, cached_from, retry_of, idempotency_key, trace_id, request_id, options, explain, created_at, started_at, finished_at, citations_count)
VALUES (@id, @workflow_key, @workflow_version, @template_key, @template_version, @route_key, @route_version, 'cached', @requester_type, @requester_id, @on_behalf_of, @attribution,
VALUES (@id, @workflow_key, @workflow_version, @template_key, @template_version, @route_key, @route_version, 'cached', @requester_type, @requester_id, @on_behalf_of, @attribution,
- @source_service, @target, @target_key, @input, @input_hash, @output, @model_key, @cache_key, @cached_from, @retry_of, @idempotency_key, @trace_id, @request_id, @options, @created_at, @created_at, @created_at, 0)`)
- .run({ ...base, output: JSON.stringify(hit.output), model_key: hit.model_key, cache_key: cacheKey, cached_from: hit.run_id });
+ @source_service, @target, @target_key, @input, @input_hash, @output, @model_key, @cache_key, @cached_from, @retry_of, @idempotency_key, @trace_id, @request_id, @options, @explain, @created_at, @created_at, @created_at, 0)`)
+ .run({ ...base, output: JSON.stringify(hit.output), model_key: hit.model_key, cache_key: cacheKey, cached_from: hit.run_id, explain: JSON.stringify(cacheExplain) });
const src = await citations(hit.run_id);
const src = await citations(hit.run_id);
if (src.length) await addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');
if (src.length) await 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.
// The reused output carries the grounding it was produced with.
@@ -301,9 +303,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -301,9 +303,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
// 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 = await db.tx(async () => {
const done = await db.tx(async () => {
const d = await db.prepare(`UPDATE runs SET status = 'succeeded', output = ?, grounding = ?, synthetic = ?, provider_key = ?, model_key = ?, fallback_used = ?, attempts = ?, tokens_in = ?, tokens_out = ?, cost_usd = ?,
const d = await 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), explain = ?, finished_at = ? WHERE id = ? AND status = 'running'`)
.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,
.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, r.explain ? JSON.stringify(r.explain) : null, iso(clock.now()), id);
if (!d.changes) return d;
if (!d.changes) return d;
if (r.citations.length) await addCitations(id, r.citations, 'workflow');
if (r.citations.length) await addCitations(id, r.citations, 'workflow');
await events.runChanged(await getRow(id));
await events.runChanged(await getRow(id));
@@ -323,9 +325,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -323,9 +325,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
const detail = interrupted ? 'the service shut down while this run was in progress' : cancelled ? 'cancelled' : (err instanceof AiError ? err.detail : 'internal error');
const detail = interrupted ? 'the service shut down while this run was in progress' : cancelled ? 'cancelled' : (err instanceof AiError ? err.detail : 'internal error');
if (!(err instanceof AiError) && !cancelled) log.error(`[runs] ${id} crashed: ${err && err.stack || err}`);
if (!(err instanceof AiError) && !cancelled) log.error(`[runs] ${id} crashed: ${err && err.stack || err}`);
const extra = err instanceof AiError && err.extra ? JSON.stringify(err.extra).slice(0, 2000) : null;
const extra = err instanceof AiError && err.extra ? JSON.stringify(err.extra).slice(0, 2000) : null;
+ const explain = err instanceof AiError && err.extra && err.extra.explain ? JSON.stringify(err.extra.explain) : null;
await db.tx(async () => {
await db.tx(async () => {
- const f = await db.prepare(`UPDATE runs SET status = ?, error_code = ?, error_detail = ?, attempts = ?, finished_at = ? WHERE id = ? AND status IN ('queued', 'running')`)
- .run(cancelled ? 'cancelled' : 'failed', code, extra ? `${detail} ${extra}` : detail, attempts, iso(clock.now()), id);
+ const f = await db.prepare(`UPDATE runs SET status = ?, error_code = ?, error_detail = ?, attempts = ?, explain = COALESCE(?, explain), finished_at = ? WHERE id = ? AND status IN ('queued', 'running')`)
+ .run(cancelled ? 'cancelled' : 'failed', code, extra ? `${detail} ${extra}` : detail, attempts, explain, iso(clock.now()), id);
if (f.changes) await events.runChanged(await getRow(id));
if (f.changes) await events.runChanged(await getRow(id));
});
});
const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
@@ -258,6 +258,7 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
@@ -258,6 +258,7 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
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,
route: last.route || null,
route: last.route || null,
+ explain: exec ? exec.explain || null : null,
};
};
}
}
@@ -3,11 +3,12 @@
@@ -3,11 +3,12 @@
* Boot-time seeding from configuration and code. Idempotent: every boot re-applies it.
* Boot-time seeding from configuration and code. Idempotent: every boot re-applies it.
*
*
* providers stub (always), shared (Live's shared key: AI_PROVIDER / AI_BASE_URL / AI_API_KEY /
* providers stub (always), shared (Live's shared key: AI_PROVIDER / AI_BASE_URL / AI_API_KEY /
- * AI_MODEL), fallback (AI_FALLBACK_*), http-seam (AI_HTTP_SEAM_URL), whisper (WHISPER_*)
+ * 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
* — records whose origin is 'seed' follow the environment; admin-edited ones are left alone
- * models the configured default and per-role models
- * routes live.<role> for Live's roles (AI_MODEL_<ROLE> overrides), live.stt, default.chat|json|
- * embedding, and the historical product route keys as explicit aliases of default.json
+ * 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
+ * product route keys as explicit aliases of default.json
* templates, workflows from server/workflows/{live,core,products}.js — a code change becomes a
* 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)
* new version unless an admin has versioned that key since (admin edits win)
* quotas AI_MAX_COST_USD_PER_DAY (global/day, Live's ai_max_cost_usd_per_day) and a per-service
* quotas AI_MAX_COST_USD_PER_DAY (global/day, Live's ai_max_cost_usd_per_day) and a per-service
@@ -50,10 +51,6 @@ function providerRecords(config, env) {
@@ -50,10 +51,6 @@ function providerRecords(config, env) {
sharedProviderRecord(config, env),
sharedProviderRecord(config, env),
{ key: 'whisper', display_name: 'whisper.cpp (local speech-to-text)', kind: 'whisper', status: 'active', auth_mode: 'none', default_model: 'whisper.cpp', capabilities: ['transcribe'], timeout_ms: 600000, priority: 20, metadata: { configured_by: 'WHISPER_*' } },
{ key: 'whisper', display_name: 'whisper.cpp (local speech-to-text)', kind: 'whisper', status: 'active', auth_mode: 'none', default_model: 'whisper.cpp', capabilities: ['transcribe'], timeout_ms: 600000, priority: 20, metadata: { configured_by: 'WHISPER_*' } },
];
];
- if (config.fallback.baseUrl) {
- const kind = config.fallback.kind === 'anthropic' ? 'anthropic' : 'openai';
- out.push({ key: 'fallback', display_name: 'Fallback provider', kind, status: 'active', base_url: config.fallback.baseUrl, auth_mode: resolveSecret(config.fallback.apiKeyRef, env) ? (kind === 'anthropic' ? 'x-api-key' : 'bearer') : 'none', secret_ref: config.fallback.apiKeyRef, default_model: config.fallback.model || null, capabilities: [], timeout_ms: 30000, priority: 50, metadata: { configured_by: 'AI_FALLBACK_*' } });
- }
if (config.localLlm && config.localLlm.url) {
if (config.localLlm && config.localLlm.url) {
// A local model server (llama.cpp, Ollama) speaking the OpenAI API: no key, never paid (WS-O task 5).
// A local model server (llama.cpp, Ollama) speaking the OpenAI API: no key, never paid (WS-O task 5).
out.push({ key: 'local', display_name: 'Local model (OpenAI-compatible server on this host)', kind: 'openai', status: 'active', base_url: config.localLlm.url, auth_mode: 'none', secret_ref: null, default_model: config.localLlm.model || 'local', capabilities: ['chat', 'generate', 'summarize', 'classify', 'extract'], timeout_ms: config.localLlm.timeoutMs, priority: 5, metadata: { configured_by: 'AI_LOCAL_LLM_URL', local: true, paid: false } });
out.push({ key: 'local', display_name: 'Local model (OpenAI-compatible server on this host)', kind: 'openai', status: 'active', base_url: config.localLlm.url, auth_mode: 'none', secret_ref: null, default_model: config.localLlm.model || 'local', capabilities: ['chat', 'generate', 'summarize', 'classify', 'extract'], timeout_ms: config.localLlm.timeoutMs, priority: 5, metadata: { configured_by: 'AI_LOCAL_LLM_URL', local: true, paid: false } });
@@ -74,29 +71,33 @@ async function seed({ registry, quotas, config, env = process.env, db }) {
@@ -74,29 +71,33 @@ async function seed({ registry, quotas, config, env = process.env, db }) {
}
}
}
}
const shared = await registry.getProvider('shared');
const shared = await registry.getProvider('shared');
- const hasFallback = Boolean(await registry.getProvider('fallback'));
- const fallbacks = hasFallback ? [{ provider: 'fallback', model: null }] : [];
// Models
// Models
- const models = new Set([shared.default_model, ...Object.values(config.shared.roleModels)]);
+ const models = new Set([shared.default_model]);
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' });
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('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' });
if (!await registry.getModel('whisper', 'whisper.cpp')) await registry.upsertModel({ provider_key: 'whisper', model_key: 'whisper.cpp', type: 'stt', supports: { json: false } }, { actor: 'seed' });
- // Routes
+ // 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.
+ const authority = shared ? (shared.default_model ? `shared:${shared.default_model}` : 'shared') : null;
+ 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) {
for (const role of live.ROLES) {
- await registry.seedVersioned('route', `live.${role}`, {
- primary: { provider: 'shared', model: config.shared.roleModels[role] || null }, fallbacks,
- options: {}, max_output_tokens: null, response_format: 'text', timeout_ms: ROLE_TIMEOUT[role], alias_of: null,
- });
+ 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 });
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 });
- await registry.seedVersioned('route', 'default.chat', { primary: { provider: 'shared', model: null }, fallbacks, options: {}, max_output_tokens: 800, response_format: 'text', timeout_ms: 30000, alias_of: null });
- await registry.seedVersioned('route', 'default.json', { primary: { provider: 'shared', model: null }, fallbacks, options: { temperature: 0.3 }, max_output_tokens: 2400, response_format: 'json', timeout_ms: 60000, alias_of: null });
+ 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', { 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 });
// media.analyze (WS-O task 5): the local model when there is one; the paid route is used only under a media.paid budget.
// 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 });
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: config.shared.roleModels.summary || null }, fallbacks, options: { temperature: 0.3 }, max_output_tokens: 350, response_format: 'text', timeout_ms: 60000, 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 });
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' });
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)
// Templates, then workflows (workflows reference templates)
new file mode 100644
new file mode 100644
@@ -0,0 +1,82 @@
@@ -0,0 +1,82 @@
+'use strict';
+// T6 placement (openvibe-sdk/placement, used by the provider router): ordering by price, latency and
+// objective; hard constraints exclude a candidate with a reason (capability, health, trust, ceiling);
+// an unpriced paid provider is never assumed free; hysteresis keeps the current placement unless a
+// candidate is clearly better; correctness keeps the named authority. Pure offers — no network.
+const assert = require('assert');
+const { plan } = require('openvibe-sdk/placement');
+const { suite } = require('./helpers');
+
+const t = suite('placement');
+const card = (id, provider, price) => ({ id, provider, metric: 'tokens', unit_size: 1e6, unit_price_usd: price, free_allowance: 0, reset_period: 'month' });
+const offer = (id, opts = {}) => ({
+ offer_id: id, kind: 'provider', region: 'global', trust: 'first-party', capabilities: ['chat'],
+ health: { status: 'up' }, pricing: { model: 'metered', rate_card: id }, ...opts,
+});
+const req = (over = {}) => ({ kind: 'ai', mobility: 'request', latency_class: 'interactive', objective: 'balanced', capabilities: ['chat'], units: 1, ...over });
+
+t.test('cheapest picks the lower-priced offer', () => {
+ const offers = [offer('pricey'), offer('cheap')];
+ const r = plan(req({ objective: 'cheapest' }), offers, { rateCards: [card('pricey', 'pricey', 30), card('cheap', 'cheap', 5)] });
+ assert.strictEqual(r.selected, 'cheap');
+ assert.deepStrictEqual(r.candidates.map((c) => [c.id, c.eligible]), [['pricey', true], ['cheap', true]]);
+});
+
+t.test('lowest-latency picks the faster offer', () => {
+ const offers = [offer('slow', { latency_ms: { p95: 900 } }), offer('fast', { latency_ms: { p95: 90 } })];
+ const r = plan(req({ objective: 'lowest-latency' }), offers, { rateCards: [card('slow', 'slow', 1), card('fast', 'fast', 1)] });
+ assert.strictEqual(r.selected, 'fast');
+});
+
+t.test('hard constraints exclude with a reason; the eligible one is chosen', () => {
+ const offers = [
+ offer('nocap', { capabilities: ['vision'] }),
+ offer('sick', { health: { status: 'down' } }),
+ offer('outsider', { trust: 'community' }),
+ offer('slow', { latency_ms: { p95: 5000 } }),
+ offer('good'),
+ ];
+ const r = plan(req({ trust: ['first-party'], max_latency_ms: 1000 }), offers, { rateCards: offers.map((o) => card(o.offer_id, o.offer_id, 1)) });
+ const why = Object.fromEntries(r.candidates.map((c) => [c.id, c.excluded_because]));
+ assert.match(why.nocap, /lacks chat/);
+ assert.match(why.sick, /health down/);
+ assert.match(why.outsider, /trust/);
+ assert.match(why.slow, /over 1000 ms/);
+ assert.strictEqual(r.selected, 'good');
+});
+
+t.test('an unpriced paid provider is excluded, never assumed free', () => {
+ const unpriced = { offer_id: 'free?', kind: 'provider', region: 'global', trust: 'first-party', capabilities: ['chat'], health: { status: 'up' }, pricing: { model: 'metered' } };
+ const offers = [offer('priced'), unpriced];
+ const r = plan(req({ objective: 'cheapest' }), offers, { rateCards: [card('priced', 'priced', 5)] });
+ const u = r.candidates.find((c) => c.id === 'free?');
+ assert.strictEqual(u.eligible, true, 'not a hard-constraint failure');
+ assert.ok(!Number.isFinite(u.estimated_cost_usd), 'no price known, so no finite cost estimate');
+ assert.strictEqual(r.selected, 'priced');
+});
+
+t.test('hysteresis: a slightly better candidate does not move the current placement', () => {
+ const offers = [offer('current'), offer('rival')];
+ const rateCards = [card('current', 'current', 10), card('rival', 'rival', 9)];
+ const stays = plan(req({ objective: 'cheapest' }), offers, { rateCards, current: 'current', minGain: 0.5 });
+ assert.strictEqual(stays.selected, 'current', stays.reasons.join('; '));
+ assert.ok(stays.reasons.some((x) => /stays on current/.test(x)));
+ const moves = plan(req({ objective: 'cheapest' }), offers, { rateCards, current: 'current', minGain: 0.05 });
+ assert.strictEqual(moves.selected, 'rival');
+});
+
+t.test('correctness keeps the named authority, whatever the price', () => {
+ const offers = [offer('authority'), offer('cheaper')];
+ const r = plan(req({ objective: 'correctness', authority: 'authority' }), offers, { rateCards: [card('authority', 'authority', 50), card('cheaper', 'cheaper', 1)] });
+ assert.strictEqual(r.selected, 'authority');
+ assert.ok(r.reasons.some((x) => /authority keeps critical work/.test(x)));
+});
+
+t.test('a candidate that is no longer eligible fails over immediately (no minGain hold)', () => {
+ const offers = [offer('current', { health: { status: 'down' } }), offer('other')];
+ const r = plan(req({ objective: 'cheapest' }), offers, { rateCards: [card('current', 'current', 1), card('other', 'other', 9)], current: 'current', minGain: 0.9 });
+ assert.strictEqual(r.selected, 'other');
+ assert.ok(r.reasons.some((x) => /no longer eligible/.test(x)));
+});
+
+t.run();
new file mode 100644
new file mode 100644
@@ -0,0 +1,199 @@
@@ -0,0 +1,199 @@
+'use strict';
+// T6 provider router (phase 1): a route names a capability and the router tries its pool in
+// openvibe-sdk/placement order, over the existing retry/breaker/log loop.
+// - forced failure on EVERY route from registry.listRoutes(): the primary fails and the pool falls
+// back to another member (logged), or the route answers the explicit 503 provider.unavailable for
+// a single-provider capability — never synthetic output;
+// - two seam providers on one capability: the cheaper is chosen; a circuit-open provider is excluded;
+// an upstream 429 shifts to the other without opening the circuit;
+// - explain{objective,reasons,selected,candidates} rides every run response and GET /runs/:id, with
+// the per-attempt request log still there.
+const assert = require('assert');
+const { boot, request, token, suite, seamServer, ALL } = require('./helpers');
+
+const t = suite('provider-router');
+const tok = token('live', ALL);
+let h;
+const seam = () => seamServer((body) => (body.operation === 'embed'
+ ? { vectors: [[0.1, 0.2]], model: 'seam-model', usage: { input: 1, output: 0 } }
+ : { text: 'a real answer', model: 'seam-model', usage: { input: 5, output: 2 } }));
+
+async function provider(body) {
+ const r = await request(h.base, 'POST', '/api/v1/providers', { tok, body: { kind: 'http', auth_mode: 'none', timeout_ms: 3000, ...body } });
+ assert.strictEqual(r.status, 201, r.text);
+}
+async function model(providerKey, modelKey, price) {
+ const r = await request(h.base, 'POST', '/api/v1/models', { tok, body: { provider_key: providerKey, model_key: modelKey, type: 'chat', cost: { in_per_mtok: price, out_per_mtok: price } } });
+ assert.strictEqual(r.status, 201, r.text);
+}
+async function route(key, body) {
+ const r = await request(h.base, 'POST', `/api/v1/routes/${key}/versions`, { tok, body });
+ assert.strictEqual(r.status, 201, r.text);
+ return r.body.route;
+}
+const CANDIDATE_KEYS = ['cost', 'eligible', 'excluded_reason', 'latency', 'model', 'provider'];
+
+t.test('forced failure on every route: a logged fallback, or an explicit 503 (never synthetic)', async () => {
+ const ok = await seamServer(() => ({ text: 'a real answer', model: 'seam-model', usage: { input: 5, output: 2 } }));
+ h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '50' } });
+ await provider({ key: 'dead', base_url: 'http://127.0.0.1:1/x', capabilities: ['chat', 'generate', 'summarize', 'extract', 'json'] });
+ await provider({ key: 'ok', base_url: ok.url, capabilities: ['chat', 'generate', 'summarize', 'extract', 'json'] });
+
+ const routes = new Map();
+ for (const r of await h.registry.listRoutes()) {
+ const rr = await h.registry.resolveRoute(r.key);
+ if (rr && !rr.disabled) routes.set(rr.key, rr);
+ }
+ assert.ok(routes.size >= 8, `routes discovered: ${[...routes.keys()].join(', ')}`);
+
+ const failed = [];
+ for (const rr of routes.values()) {
+ const op = rr.capability === 'transcribe' || rr.key === 'live.stt' ? 'transcribe' : rr.capability === 'embed' ? 'embed' : 'chat';
+ const req = op === 'embed' ? { input: ['x'], timeoutMs: 2000 }
+ : op === 'transcribe' ? { filePath: '/nope.mp4', language: 'en', timeoutMs: 2000 }
+ : { messages: [{ role: 'user', content: 'hi' }], timeoutMs: 2000 };
+ const logs = [];
+ let exec = null; let err = null;
+ try { exec = await h.pool.execute(rr, op, req, { routeKey: rr.key, routeVersion: rr.version, logRequest: (e) => logs.push(e) }); } catch (e) { err = e; }
+
+ const explain = exec ? exec.explain : (err && err.extra && err.extra.explain);
+ assert.ok(explain && Array.isArray(explain.candidates), `${rr.key}: explain present`);
+ for (const c of explain.candidates) assert.deepStrictEqual(Object.keys(c).sort(), CANDIDATE_KEYS, `${rr.key}: candidate shape`);
+ if (err) {
+ assert.strictEqual(err.status, 503, `${rr.key}: explicit 503 (${err.message})`);
+ assert.strictEqual(err.code, 'provider.unavailable', `${rr.key}: stable code`);
+ assert.ok(logs.some((e) => e.status !== 'ok'), `${rr.key}: the failure was logged`);
+ failed.push(rr.key);
+ } else {
+ assert.ok(!exec.synthetic, `${rr.key}: never a synthetic answer`);
+ assert.ok(exec.provider, `${rr.key}: a provider answered`);
+ if (rr.capability === 'chat') {
+ assert.ok(logs.some((e) => e.status === 'ok' && e.fallback === 1), `${rr.key}: the dead primary failed and the pool fell back`);
+ assert.strictEqual(exec.explain.selected, 'ok', `${rr.key}: explain names the winner`);
+ }
+ }
+ }
+ assert.deepStrictEqual(failed.sort(), ['default.embedding', 'live.stt', 'media.paid'], `failed routes: ${failed}`);
+ await h.stop(); await ok.close();
+});
+
+t.test('two providers on one capability: the cheaper is chosen; a circuit-open one is excluded', async () => {
+ const ok = await seam();
+ const flaky = await seamServer(() => ({ status: 500, body: { error: { message: 'boom' } } }));
+ h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '2', AI_BREAKER_COOLDOWN_MS: '600000', AI_PLACEMENT_MIN_GAIN: '0' } });
+ await provider({ key: 'cheap', base_url: ok.url, capabilities: ['classify'], default_model: 'cheap-m' });
+ await model('cheap', 'cheap-m', 1);
+ await provider({ key: 'pricey', base_url: ok.url, capabilities: ['classify'], default_model: 'pricey-m' });
+ await model('pricey', 'pricey-m', 20);
+ const r = await route('test.pool-cheap', { capability: 'classify', constraints: { objective: 'cheapest' }, timeout_ms: 3000 });
+ assert.strictEqual(r.capability, 'classify');
+
+ const exec = await h.pool.execute(r, 'classify', { messages: [{ role: 'user', content: 'x' }], timeoutMs: 3000 }, { routeKey: r.key, routeVersion: r.version });
+ assert.strictEqual(exec.provider, 'cheap', 'the cheaper provider is chosen');
+ assert.strictEqual(exec.explain.selected, 'cheap');
+ const byProvider = Object.fromEntries(exec.explain.candidates.map((c) => [c.provider, c]));
+ assert.ok(byProvider.cheap.cost < byProvider.pricey.cost, 'rate cards carry the real prices');
+
+ // A provider that keeps failing opens its circuit; placement then excludes it and the other answers.
+ await provider({ key: 'flaky', base_url: flaky.url, capabilities: ['extract'], default_model: 'flaky-m' });
+ await model('flaky', 'flaky-m', 1);
+ await provider({ key: 'good', base_url: ok.url, capabilities: ['extract'], default_model: 'good-m' });
+ await model('good', 'good-m', 20);
+ const rb = await route('test.pool-breaker', { capability: 'extract', constraints: { objective: 'cheapest' }, timeout_ms: 3000 });
+ const extract = { messages: [{ role: 'user', content: 'x' }], timeoutMs: 3000 };
+ for (let i = 0; i < 2; i++) {
+ const e = await h.pool.execute(rb, 'extract', extract, { routeKey: rb.key, routeVersion: rb.version });
+ assert.strictEqual(e.provider, 'good', 'the failing provider falls back to good');
+ assert.strictEqual(e.explain.selected, 'good');
+ }
+ assert.strictEqual((await h.pool.health('flaky')).state, 'open');
+ const e3 = await h.pool.execute(rb, 'extract', extract, { routeKey: rb.key, routeVersion: rb.version });
+ const flakyCand = e3.explain.candidates.find((c) => c.provider === 'flaky');
+ assert.strictEqual(flakyCand.eligible, false, 'the open circuit excludes it');
+ assert.match(flakyCand.excluded_reason, /health down/);
+ assert.strictEqual(e3.provider, 'good');
+ await ok.close(); await flaky.close();
+});
+
+t.test('an upstream 429 shifts to another provider without opening the circuit', async () => {
+ const ok = await seamServer(() => ({ text: 'steady answer', model: 'steady-m', usage: { input: 3, output: 1 } }));
+ const busy = await seamServer(() => ({ status: 429, body: { error: { message: 'rate limited' } } }));
+ await provider({ key: 'busy', base_url: busy.url, capabilities: ['enrich'], default_model: 'busy-m' });
+ await model('busy', 'busy-m', 0.5);
+ await provider({ key: 'steady', base_url: ok.url, capabilities: ['enrich'], default_model: 'steady-m' });
+ await model('steady', 'steady-m', 20);
+ const r = await route('test.pool-429', { capability: 'enrich', constraints: { objective: 'cheapest' }, timeout_ms: 3000 });
+
+ const exec = await h.pool.execute(r, 'enrich', { messages: [{ role: 'user', content: 'x' }], timeoutMs: 3000 }, { routeKey: r.key, routeVersion: r.version });
+ assert.strictEqual(exec.provider, 'steady', 'the 429 provider is skipped for this call');
+ assert.strictEqual(exec.explain.selected, 'steady');
+ assert.strictEqual((await h.pool.health('busy')).state, 'closed', 'a 429 is capacity, not ill health');
+ await ok.close(); await busy.close();
+});
+
+t.test('explain rides the run response and GET /runs/:id; the request log stays', async () => {
+ const fails = await seamServer(() => ({ status: 500, body: { error: { message: 'down' } } }));
+ const ok = await seamServer(() => ({ text: 'the fallback answer', model: 'p2-m', usage: { input: 4, output: 2 } }));
+ await provider({ key: 'p1', base_url: fails.url, capabilities: ['chat', 'generate'], default_model: 'p1-m' });
+ await model('p1', 'p1-m', 1);
+ await provider({ key: 'p2', base_url: ok.url, capabilities: ['chat', 'generate'], default_model: 'p2-m' });
+ await model('p2', 'p2-m', 20);
+ const r = await route('default.chat', { capability: 'chat', constraints: { objective: 'cheapest' }, timeout_ms: 3000, response_format: 'text' });
+ assert.strictEqual(r.capability, 'chat');
+
+ const res = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok, body: { prompt: 'hello', options: { cache: false } } });
+ assert.strictEqual(res.status, 201, res.text);
+ const run = res.body.run;
+ assert.strictEqual(run.status, 'succeeded');
+ assert.strictEqual(run.provenance.provider, 'p2');
+ assert.ok(run.explain, 'the run carries explain');
+ assert.strictEqual(run.explain.objective, 'cheapest');
+ assert.ok(Array.isArray(run.explain.reasons) && run.explain.reasons.length);
+ assert.strictEqual(run.explain.selected, 'p2');
+ for (const c of run.explain.candidates) assert.deepStrictEqual(Object.keys(c).sort(), CANDIDATE_KEYS);
+ const shared = run.explain.candidates.find((c) => c.provider === 'shared');
+ assert.ok(shared && shared.eligible === false && shared.excluded_reason, 'the credential-less shared provider is excluded with a reason');
+
+ const got = await request(h.base, 'GET', `/api/v1/runs/${run.id}`, { tok });
+ assert.deepStrictEqual(got.body.run.explain, run.explain, 'GET /runs/:id returns the same explain');
+ const reqs = got.body.requests;
+ assert.ok(reqs.some((x) => x.provider_key === 'p1' && x.status === 'error'), 'the failed primary is in the request log');
+ const fb = reqs.find((x) => x.provider_key === 'p2');
+ assert.strictEqual(fb.status, 'ok');
+ assert.strictEqual(fb.fallback, 1, 'the fallback attempt is marked');
+
+ // A run that nothing can answer still carries explain (selected null) and the explicit 503 code.
+ await request(h.base, 'POST', '/api/v1/providers/p2/disable', { tok });
+ const bad = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok, body: { prompt: 'nobody', options: { cache: false } } });
+ assert.strictEqual(bad.body.run.status, 'failed', bad.text);
+ assert.strictEqual(bad.body.run.error.code, 'provider.unavailable');
+ assert.ok(bad.body.run.explain && bad.body.run.explain.selected === null, 'a failed run still explains itself');
+ await fails.close(); await ok.close();
+});
+
+t.test('seeded text pools keep the shared provider as authority: a free local model is only the failover', async () => {
+ const { plan } = require('openvibe-sdk/placement');
+ const shared = await h.registry.getProvider('shared');
+ const authority = shared.default_model ? `shared:${shared.default_model}` : 'shared';
+ let checked = 0;
+ for (const r of await h.registry.listRoutes()) {
+ const rr = await h.registry.resolveRoute(r.key);
+ if (!rr || rr.disabled || rr.capability !== 'chat' || rr.created_by !== 'seed' || !/^(live\.|default\.(chat|json)$)/.test(rr.key)) continue;
+ checked++;
+ assert.strictEqual(rr.constraints.objective, 'correctness', `${rr.key}: correctness keeps the authority`);
+ assert.strictEqual(rr.constraints.authority, authority, `${rr.key}: the shared provider is the authority`);
+ }
+ assert.ok(checked >= 5, `seeded text pools checked: ${checked}`);
+ // The planner under those constraints: the paid shared offer beats the free local one while it is up, and
+ // the local one takes over the moment shared is down.
+ const offer = (id, status, pricing) => ({ offer_id: id, kind: 'provider', provider: id.split(':')[0], region: 'global', trust: 'first-party', capabilities: ['chat'], latency_ms: {}, health: { status }, pricing });
+ const req = { kind: 'ai', mobility: 'request', latency_class: 'interactive', objective: 'correctness', capabilities: ['chat'], units: 1, authority: 'shared:m' };
+ const cards = [{ id: 'shared:m', provider: 'shared', metric: 'tokens', unit_size: 1e6, unit_price_usd: 2, free_allowance: 0, reset_period: 'month' }];
+ const up = plan(req, [offer('shared:m', 'up', { model: 'metered', rate_card: 'shared:m' }), offer('local:l', 'up', { model: 'prepaid' })], { rateCards: cards });
+ assert.strictEqual(up.selected, 'shared:m', JSON.stringify(up.reasons));
+ const down = plan(req, [offer('shared:m', 'down', { model: 'metered', rate_card: 'shared:m' }), offer('local:l', 'up', { model: 'prepaid' })], { rateCards: cards });
+ assert.strictEqual(down.selected, 'local:l', JSON.stringify(down.reasons));
+});
+
+t.test('shutdown', async () => { await h.stop(); });
+t.run();