openvibers/openvibe.ai Commits
providers: fail over an empty balance, rest visibly, and probe it hourly (#315)
Changes
@@@ -174,6 +174,10 @@ AI_MODEL=#AI_MEDIA_STORAGE_HOSTS=*.backblazeb2.com,*.r2.cloudflarestorage.com,edge.openvibe.media#AI_LOCAL_LLM_MODEL=qwen2.5-1.5b-instruct#AI_LOCAL_LLM_TIMEOUT_MS=180000++# The host's model is a last offer for short text only: never more than this many tokens, and one generation at++# a time across local and local2, so an outage upstream cannot starve the host.++#AI_LOCAL_LLM_MAX_TOKENS=512++#AI_LOCAL_LLM_CONCURRENCY=1# A second local chat model server (its own llama.cpp process and another model family): provider `local2`,# so a caller can pin two local offers, e.g. Games' AI-vs-AI matches. Same rules: no key, free, no JSON.#AI_LOCAL_LLM2_URL=http://127.0.0.1:8092/v1@@@ -252,8 +256,16 @@ AI_MAX_COST_USD_PER_DAY=0#AI_DEBUG_RAW_LOG=false#AI_BREAKER_FAILURES=3#AI_BREAKER_COOLDOWN_MS=60000++# A provider that answered "insufficient balance" (HTTP 402, or a 429/403 insufficient_quota) is out of money:++# its circuit opens at once for this long (10 minutes), shared by every request, and it is never retried.++#AI_BREAKER_EXHAUSTED_COOLDOWN_MS=600000#AI_PROVIDER_RETRIES=1#AI_PROVIDER_RETRY_DELAY_MS=800++# Hourly balance probe (DeepSeek's GET /user/balance) exposed as ai_provider_balance_usd{provider} on /metrics;++# the operator alert fires below $2. Off under a restore drill (AI_DRILL=1).++#AI_BALANCE_PROBE=1++#AI_BALANCE_PROBE_INTERVAL_MS=3600000++#AI_BALANCE_PROBE_TIMEOUT_MS=10000# ── Media inputs: only these https hosts are fetched (plus the Media internal origin) ──OV_MEDIA_URL=https://openvibe.media@@@ -161,9 +161,15 @@ schema errors), `workflow.not_found`, `idempotency.conflict`, `quota.exceeded` (`Retry-After`), `govern.exceeded` (429 + `Retry-After`: a tier budget) and `govern.unavailable` (503: Valkeyunreachable for a paid or staff run), `queue.full` (429 + `Retry-After`: the run would wait behind `AI_MAX_QUEUED_RUNS`runs, or `AI_MAX_QUEUED_RUNS_PER_CALLER` of this caller's), `capability.denied`, `capability.namespace_denied`, `token.*`. A run that cannot--produce a real answer ends `failed` with an explicit code — `provider.unavailable`,--`route.unavailable`, `fetch.refused`, `source.unavailable`, `input.insufficient`, `output.empty`,--`output.invalid`, `run.interrupted` — and never with filler content.++produce a real answer ends `failed` with an explicit code — `ai.resting` (503: every offer is out of balance,++paused or busy; the detail is the sentence products show, "AI is resting for a moment", with `Retry-After`),++`provider.unavailable`, `route.unavailable`, `fetch.refused`, `source.unavailable`, `input.insufficient`,++`output.empty`, `output.invalid`, `run.interrupted` — and never with filler content.++++Each active provider with a balance API (DeepSeek's `GET /user/balance`) is asked hourly; the last reading is++`ai_provider_balance_usd{provider}` on `/metrics` and OpenVibe.Host's `OpenVibeAiProviderBalanceLow` alert turns a++reading below $2 into an operator ticket. The key is only sent in the request's `Authorization` header: it is never++logged, returned or stored by the probe. `AI_BALANCE_PROBE=0` (or a restore drill, `AI_DRILL=1`) turns it off.The capability ids and the `ai` service manifest are released in openvibe-contracts v0.76.0 (thedrafts stay in `docs/capabilities-proposal/`); `server/auth.js` checks them with the contracts grant@@@ -505,7 +511,7 @@ definition, but never over an admin's version.| `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; NVIDIA Parakeet TDT 0.6B v3 (`parakeet-cli`) for English and 24 European languages when `PARAKEET_MODEL` is set, falling back to whisper | `WHISPER_*`, `PARAKEET_*`, `ASR_ENGINE` — 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`; last offer for short text, one request at a time | `AI_LOCAL_LLM_URL`, `AI_LOCAL_LLM_MODEL`, `AI_LOCAL_LLM_TIMEOUT_MS`, `AI_LOCAL_LLM_MAX_TOKENS` (512), `AI_LOCAL_LLM_CONCURRENCY` (1) || `openai`, key `local2` | A second local chat server, another llama.cpp process and model family: no key, never paid, no JSON or embeddings. Two local offers let a caller pin a match between them (Games' AI-vs-AI job) | `AI_LOCAL_LLM2_URL`, `AI_LOCAL_LLM2_MODEL`, `AI_LOCAL_LLM2_TIMEOUT_MS` |Routing: a route either names a `capability` and the router builds its pool from every provider that@@@ -513,7 +519,10 @@ can serve it, ordered by `openvibe-sdk/placement` (objective, price, measured lahysteresis through the route's current placement), or pins a `[primary, ...fallbacks]` list. Aprovider 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++failure. A provider's own "insufficient balance" answer (HTTP 402, or 429/403 `insufficient_quota`) is never++retried: its circuit opens at once for `AI_BREAKER_EXHAUSTED_COOLDOWN_MS` (10 minutes, shared by every request)++and the router fails over. When nothing can serve, the run answers `ai.resting`; 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 excludedreason).@@@ -772,6 +781,12 @@ pair. On the production host since 2026-11-06: Llama-3.2-1B-Instruct Q4_K_M (LlaHugging Face sha256 `6f85a640a97cf2bf5b8e764087b1e83da0fdb51d7c9fab7d0fece9385611df83`, 807,694,464 bytes),checked against that hash before install; it uses about 0.8 GB.++The local model is a **last offer for short text** (summaries, titles, moderation hints, a chat line), never++for code or long reasoning: routes with `max_output_tokens` over 800 or `outlast_provider_timeout` (an app or a++wiki page) never offer it. It answers with `AI_LOCAL_LLM_MAX_TOKENS` (512) at most, and one request at a time++(`AI_LOCAL_LLM_CONCURRENCY`, 1) across `local` and `local2`, so a paid outage cannot starve the streams sharing++the host. A run that it answered names `local`/`local2` as its provider and marks the receipt local.++Local embeddings are off unless `AI_LOCAL_EMBED_MODEL` names an embedding model the server at `AI_LOCAL_LLM_URL`answers at `/embeddings`. Then `local` advertises `embed`, the model is registered as an `embedding` model, and`default.embedding` (a capability pool) offers it beside the shared provider's `AI_EMBEDDING_MODEL`; `local` is@@@ -792,6 +807,8 @@ mix vectors from different providers (re-embed when the provider changes).| Every workflow has versioned, compiling input/output schemas, runs on the stub and validates | `test/workflows.test.js` || Edits are new versions; runs pin versions; lifecycle; admin versions survive reseeding | `test/versioning.test.js` || A degraded primary falls back and records it; breaker; skips; timeouts; explicit `provider.unavailable` | `test/fallback.test.js` |++| 402/insufficient-quota opens the long exhausted circuit, never retries, fails over, then `ai.resting` | `test/provider-exhausted.test.js` |++| Hourly DeepSeek balance gauge, no key in logs, off under a drill | `test/provider-balance.test.js` || Quotas refuse with 429 + Retry-After **before** any provider call; per-service, attribution, cost caps | `test/quota.test.js` || Free allowance: free up to the card's allowance then fully priced, reset per day/month (never for `none`), per subject, never past the cap under concurrent claims, absent at 0 | `test/free-allowance.test.js` || Tier budgets: enforced per subject and per project and reset per window, two concurrent runs never both take the last unit, the free allowance's numbers unchanged on the Valkey store, one reading per accounted run from govern's `onUsage`, Valkey unreachable (paid 503, free admitted), off by default | `test/govern-tiers.test.js` |nnew file mode 100644@@@ -0,0 +1,5 @@++-- phase: expand++-- A provider that answered "insufficient balance" is out of money, not flaky: it is opened for the long++-- exhausted cooldown (AI_BREAKER_EXHAUSTED_COOLDOWN_MS, 10 minutes) instead of the ordinary breaker cooldown.++-- exhausted_at records when that happened; a successful (or manually reset) probe clears it. Additive only.++ALTER TABLE provider_health ADD COLUMN exhausted_at bigint; -- epoch ms of the last "insufficient balance" answer@@@ -24,7 +24,7 @@ const pkg = require('../package.json');const { adminRoutes } = require('openvibe-shared/config');--function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env = process.env, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console, credentials = null, preferences = null, grants = null, vault = null, settings = null, benchEvents = null, runBudgets = null, memory = null, memoryImport = null, community = null, presets = null, receipts = null }) {++function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env = process.env, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console, credentials = null, preferences = null, grants = null, vault = null, settings = null, benchEvents = null, runBudgets = null, memory = null, memoryImport = null, community = null, presets = null, receipts = null, balance = null }) {const app = express();// /.well-known/agent.json (openvibe-shared/agent-layer, SEO-2): what this site serves, for agents.require('openvibe-shared/agent-layer').mount(app, { descriptor: require('./agent-descriptor') });@@@ -36,7 +36,7 @@ function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys// GET /metrics answers direct loopback callers only (Track O).const metrics = instrument(app, { service: 'ai', release: release.release });app.use(require('openvibe-sdk/service').telemetryMiddleware); // request telemetry: a no-op until server/index.js calls telemetry.init()-- registerAiGauges(metrics.registry, { runs, registry, pool });++ registerAiGauges(metrics.registry, { runs, registry, pool, balance });app.use(http.middleware());// Browser reads of the two public GETs (/stats, /release.json) only: an exact origin from// AI_CORS_ORIGINS is echoed back, never credentials, and OPTIONS answers their preflight.@@@ -190,6 +190,10 @@ function load(env = process.env) {model: String(env.AI_LOCAL_LLM_MODEL || '').trim(),embedModel: String(env.AI_LOCAL_EMBED_MODEL || '').trim(),timeoutMs: int(env.AI_LOCAL_LLM_TIMEOUT_MS, 180000),++ // Never let the local model starve the host: one generation at a time across every local provider,++ // and a short answer unless a route already asked for less (server/providers/index.js).++ maxTokens: Math.max(1, int(env.AI_LOCAL_LLM_MAX_TOKENS, 512)),++ concurrency: Math.max(1, int(env.AI_LOCAL_LLM_CONCURRENCY, 1)),},// A second local chat server (its own llama.cpp process, another model family): the provider `local2`.// Same rules as `local`: no key, never paid, no JSON. Two local offers let a caller pin a match@@@ -300,12 +304,24 @@ function load(env = process.env) {breaker: {failureThreshold: Math.max(1, int(env.AI_BREAKER_FAILURES, 3)),cooldownMs: int(env.AI_BREAKER_COOLDOWN_MS, 60000),++ // A provider that answered "insufficient balance" is out of money, not flaky: its circuit stays open this++ // long (10 minutes), shared by every request, before one probe is allowed to spend again.++ exhaustedCooldownMs: Math.max(60000, int(env.AI_BREAKER_EXHAUSTED_COOLDOWN_MS, 10 * 60000)),},// One retry on a transient failure (timeout, 429, 5xx, connection reset) before the// router moves on to the fallback — Live's rule.providerRetries: Math.max(0, int(env.AI_PROVIDER_RETRIES, 1)),retryDelayMs: Math.max(0, int(env.AI_PROVIDER_RETRY_DELAY_MS, 800)),++ // ── Hourly provider balance probe (server/providers/balance.js) ──++ // Each active provider with a balance API (DeepSeek today) is asked GET <origin>/user/balance and the reading++ // is exposed as ai_provider_balance_usd{provider} on GET /metrics. Off under a restore drill (AI_DRILL=1).++ balance: {++ enabled: bool(env.AI_BALANCE_PROBE, !bool(env.AI_DRILL, false)),++ intervalMs: Math.max(60000, int(env.AI_BALANCE_PROBE_INTERVAL_MS, 60 * 60 * 1000)),++ timeoutMs: Math.max(1000, int(env.AI_BALANCE_PROBE_TIMEOUT_MS, 10000)),++ },++// ── Media inputs: only these hosts are ever fetched (SSRF guard) ──media: {publicUrl: trimUrl(env.OV_MEDIA_URL || 'https://openvibe.media'),@@@ -91,6 +91,8 @@ function buildReceipt({ run, explain = null, usage = {}, price = null, privacy =deployment: (provider || (ok && ok.provider) || run.provider_key) || null,provider_access: (x.openvibe && x.openvibe.provider_access) || null,resolved_model: (x.openvibe && x.openvibe.resolved_model) || null,++ // The host's own model answered (local / local2): the receipt says so, beside the provider name.++ local: Boolean(x.provider_local),},price: {in_per_mtok: round(price && price.in),@@@ -122,6 +124,7 @@ function compact(receipt) {id: receipt.id,run_id: receipt.run_id,offer: offer.provider ? `${offer.provider}${offer.model ? `/${offer.model}` : ''}` : null,++ local: Boolean(offer.local),price: { in: receipt.price && receipt.price.in_per_mtok, out: receipt.price && receipt.price.out_per_mtok, cached: receipt.price && receipt.price.cached_per_mtok },tokens: { in: receipt.usage && receipt.usage.tokens_in, out: receipt.usage && receipt.usage.tokens_out, cached: receipt.usage && receipt.usage.tokens_cached },cost_usd: receipt.cost_usd,@@@ -67,6 +67,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now// Network owns the 18+ account flag; the adult tier checks it before placement.const adultStatus = require('./policy/adult-status').createAdultStatus({ config, env, fetchImpl, clock, log });const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log, credentialFetch, machineModels: machines, community, bench, variantEvidence, adultStatus });++ // The hourly provider balance probe (server/providers/balance.js): a gauge for the operator alert, started++ // only when the service listens (never in tests) and off under a restore drill.++ const balance = require('./providers/balance').createBalanceProbe({ registry, config, env, fetchImpl, clock, log });// T6 step 8: the SaaS tier budgets on openvibe-sdk/govern in Valkey (server/govern.js); off unless AI_GOVERN_TIERS=1.const tiers = createTiers({ config, valkey, clock, log });// §2.1.8: the provider free allowance per subject, priced off the pool's own rate cards (server/free-allowance.js);@@@ -143,7 +146,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.nowclientSecret: env.OV_OAUTH_CLIENT_SECRET, fetch: fetchImpl,}) : null;const benchEvents = createBenchEvents({ bench, accountData, accountSend, config, log, fetchImpl, clientSecret: env.OV_OAUTH_CLIENT_SECRET });-- const app = createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env, clock, fetchImpl, log, credentials, preferences, grants, vault, settings, benchEvents, runBudgets, memory, memoryImport, community, presets, receipts });++ const app = createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env, clock, fetchImpl, log, credentials, preferences, grants, vault, settings, benchEvents, runBudgets, memory, memoryImport, community, presets, receipts, balance });let jwks = null;let keyLoaded = Promise.resolve(null);if (!keys.pinned && config.nodeEnv !== 'test') {@@@ -168,6 +171,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.nowlet benchSubscription = null;let accountSubscription = null;if (listen) userModules.start();++ if (listen) balance.start();if (listen && config.memoryJobs && config.memoryJobs.enabled) memoryJobs.start();// Durable Credit settles: the same jobs rule as Memory (never in test mode, never under a provider drill; the// queue's runOnce checks the drill itself).@@@ -198,6 +202,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.nowasync function close() {clearInterval(housekeeping);++ balance.stop();if (benchSubscription) benchSubscription.stop();if (accountSubscription) accountSubscription.stop();userModules.stop();@@@ -88,7 +88,7 @@ function createAiReadiness({ db, registry, pool, keys, runs, release = null }) {}/** AI gauges on the openvibe-shared/metrics registry. */--function registerAiGauges(registry, { runs, registry: ai, pool }) {++function registerAiGauges(registry, { runs, registry: ai, pool, balance = null }) {registry.gauge({name: 'ai_runs', help: 'Runs waiting for a slot (queued) and holding one (running)', labelNames: ['state'],collect: () => {@@@ -106,6 +106,11 @@ function registerAiGauges(registry, { runs, registry: ai, pool }) {name: 'ai_provider_credentials_missing', help: 'Active providers whose secret reference does not resolve (1 = missing)', labelNames: ['provider'],collect: async () => (await providerView(ai, pool)).filter(p => p.credentials !== 'not_required').map(p => ({ labels: { provider: p.key }, value: p.credentials === 'missing' ? 1 : 0 })),});++ // The hourly balance probe's last reading (server/providers/balance.js), never a live call on scrape.++ registry.gauge({++ name: 'ai_provider_balance_usd', help: 'Provider account balance in USD, as last read by the hourly probe', labelNames: ['provider'],++ collect: () => (balance ? balance.values() : []).map(r => ({ labels: { provider: r.provider }, value: r.usd })),++ });}module.exports = { createAiReadiness, registerAiGauges };nnew file mode 100644@@@ -0,0 +1,116 @@++'use strict';++/**++ * Hourly provider balance probe (Track O): a provider that publishes its account balance is asked what is++ * left, so the operator hears about an empty account before every AI feature on the network fails.++ *++ * DeepSeek GET https://api.deepseek.com/user/balance++ * { "is_available": true, "balance_infos": [ { "currency": "USD", "total_balance": "0.05", ... } ] }++ *++ * The reading is exposed as ai_provider_balance_usd{provider} on GET /metrics (server/observability.js) and++ * the operator alert fires below $2 (OpenVibe.Host deploy/prometheus/openvibe-rules.yml). The key is read by++ * reference and only ever sent in the request's Authorization header: it is never logged, returned or stored.++ * Off when AI_BALANCE_PROBE=0 or under a restore drill (AI_DRILL=1; server/config.js balance.enabled).++ */++const { resolveSecret } = require('../util');++++const HOST = /(^|\.)api\.deepseek\.com$/i;++++/** DeepSeek's answer: the USD total when it reports one (the gauge is dollars), else null. */++function deepseekUsd(j) {++ const infos = Array.isArray(j && j.balance_infos) ? j.balance_infos : [];++ const usd = infos.find((x) => x && String(x.currency || '').toUpperCase() === 'USD');++ const n = usd ? Number(usd.total_balance) : NaN;++ return Number.isFinite(n) ? n : null;++}++++function createBalanceProbe({ registry, config, env = process.env, fetchImpl = globalThis.fetch, clock = { now: () => Date.now() }, log = console }) {++ const latest = new Map(); // provider key -> { usd, at, currency }++ let timer = null;++ let running = null;++++ /** A provider record with a balance API, or null: DeepSeek today. */++ function apiOf(p) {++ try {++ const u = new URL(String(p && p.base_url || ''));++ return HOST.test(u.hostname) ? { origin: u.origin, path: '/user/balance' } : null;++ } catch { return null; }++ }++++ async function read(p) {++ const api = apiOf(p);++ if (!api) return null;++ const key = resolveSecret(p.secret_ref, env);++ if (!key) return null;++ let res;++ try {++ res = await fetchImpl(`${api.origin}${api.path}`, {++ headers: { Authorization: `Bearer ${key}`, Accept: 'application/json' },++ signal: AbortSignal.timeout(config.balance.timeoutMs),++ });++ } catch (err) {++ throw new Error(`balance API unreachable: ${err && err.message || err}`);++ }++ if (!res.ok) throw new Error(`balance API answered ${res.status}`);++ const j = await res.json().catch(() => null);++ const usd = deepseekUsd(j);++ return usd == null ? null : { usd };++ }++++ /** Every configured provider with a balance API, then the research key when no provider record carries it. */++ async function targets() {++ const out = [];++ for (const p of await registry.listProviders()) {++ if (p.status !== 'active') continue;++ if (apiOf(p)) out.push(p);++ }++ return out;++ }++++ async function poll() {++ if (running) return running;++ running = (async () => {++ let found = 0;++ for (const p of await targets()) {++ found++;++ try {++ const r = await read(p);++ if (r) latest.set(p.key, { usd: r.usd, at: clock.now(), currency: 'USD' });++ } catch (err) {++ log.warn(`[ai] balance probe ${p.key}: ${String(err && err.message || err).slice(0, 200)}`);++ }++ }++ // The research key can be the only DeepSeek credential (AI_RESEARCH_DEEPSEEK_KEY_REF) when the shared++ // provider is something else: still expose its balance under the label `deepseek`.++ if (!found && config.research && config.research.deepseek && resolveSecret(config.research.deepseekKeyRef, env)) {++ try {++ const r = await read({ base_url: config.research.deepseekUrl, secret_ref: config.research.deepseekKeyRef });++ if (r) latest.set('deepseek', { usd: r.usd, at: clock.now(), currency: 'USD' });++ } catch (err) {++ log.warn(`[ai] balance probe deepseek: ${String(err && err.message || err).slice(0, 200)}`);++ }++ }++ return values();++ })().finally(() => { running = null; });++ return running;++ }++++ function values() { return [...latest.entries()].map(([provider, r]) => ({ provider, usd: r.usd, at: r.at })); }++++ function start() {++ if (timer || !config.balance.enabled) return;++ const tick = () => { poll().catch((err) => log.warn(`[ai] balance probe: ${err && err.message}`)); };++ const first = setTimeout(tick, 5000);++ if (first.unref) first.unref();++ timer = setInterval(tick, config.balance.intervalMs);++ if (timer.unref) timer.unref();++ }++++ function stop() {++ if (timer) clearInterval(timer);++ timer = null;++ }++++ return { start, stop, poll, values, apiOf };++}++++module.exports = { createBalanceProbe, deepseekUsd };@@@ -47,6 +47,25 @@ async function postJson(url, headers, body, { signal, timeoutMs = 30000, fetchImreturn j;}++/**++ * The provider's own answer that its account is out of money: DeepSeek answers HTTP 402 with++ * `{ "error": { "message": "Insufficient Balance" } }`; OpenAI-shaped gateways use 429/403 with++ * `insufficient_quota`, or name a credit balance. This is not the caller's fault and not a transient++ * failure: the provider is exhausted until someone tops it up (server/providers/index.js opens its++ * circuit for the long cooldown and never retries into it).++ */++const BALANCE_RE = /insufficient\s+(balance|quota|credit|credits|funds)|out\s+of\s+(balance|credit|credits|funds)|credit\s+balance\s+is\s+too\s+low|balance\s+(is\s+)?(too\s+)?low|exceeded\s+your\s+current\s+quota|billing\s+(hard\s+)?limit/i;++function insufficientBalance(err) {++ if (!err) return false;++ if (err.status === 402) return true;++ const body = err.body && typeof err.body === 'object' ? err.body : {};++ const inner = body.error && typeof body.error === 'object' ? body.error : body;++ const code = String((inner && (inner.code || inner.type)) || err.code || '');++ const message = String((inner && inner.message) || body.message || err.message || '');++ if ((err.status === 403 || err.status === 429) && /insufficient_quota|billing/i.test(`${code} ${message}`)) return true;++ return BALANCE_RE.test(`${code} ${message}`);++}++/** Transient failures get one more try on the same provider (Live's rule). */function retryable(e) {if (!e) return false;@@@ -85,4 +104,4 @@ function schemaInstruction(json) {}module.exports = {-- schemaInstruction, ProviderError, deadline, describeAbort, postJson, retryable, approxTokens, estimateUsage, sleep };++ schemaInstruction, ProviderError, deadline, describeAbort, postJson, retryable, approxTokens, estimateUsage, sleep, insufficientBalance };@@@ -18,12 +18,16 @@* 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.++ * after that one half-open probe decides whether it closes again. A provider's own "insufficient balance"++ * answer opens it at once for the long exhausted cooldown, is never retried, and the router fails over.++ * When nothing can serve because offers are exhausted or paused, the route answers 503 ai.resting with a++ * human sentence and Retry-After. The host's local model is a last offer for short text only, one request at++ * a time, labelled local on the receipt.*/const placement = require('openvibe-sdk/placement');const contracts = require('openvibe-contracts');const { AiError, resolveSecret, sha256 } = require('../util');--const { retryable, sleep, ProviderError, schemaInstruction } = require('./common');++const { retryable, sleep, ProviderError, schemaInstruction, insufficientBalance } = require('./common');const { createStubProvider } = require('./stub');const { createOpenAiProvider } = require('./openai');const { createResponsesProvider } = require('./responses');@@@ -54,7 +58,8 @@ async function policyAllows(registry, provider, modelId, policy = 'standard') {/** 4xx that the request itself caused (bad image, schema, max_tokens...), as opposed to auth, rate or availability. */function callerFault(err) {const st = err && err.status;-- return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 403, 404, 408, 429].includes(st);++ // 402 payment required is the provider's account, not the caller's request: it opens the circuit (exhausted).++ return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 402, 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. */@@@ -204,6 +209,11 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.noconst machines = machineModels || require('../machines').createMachineModels({ config, env, fetchImpl });const cache = new Map(); // key -> { stamp, adapter }const stats = {}; // key -> { calls, failures, ... } (in memory; tests + /api/ready)++ // One local generation at a time (AI_LOCAL_LLM_CONCURRENCY, default 1): the host's model shares its CPU++ // with live streams, so a paid outage must not turn into a local-model stampede.++ const localLimit = Math.max(1, (config.localLlm && config.localLlm.concurrency) || 1);++ let localActive = 0;++ const exhaustedCooldownMs = () => Math.max(60000, Number(config.breaker.exhaustedCooldownMs) || 10 * 60000);// L36 Y7: the provider that last answered this prompt, for five minutes (server/providers/warm-cache.js).const warm = createWarmCache({ now: () => clock.now() });// AI Exchange P7: measured per-deployment evidence for alias routing. The feature flag is read live (D11@@@ -265,26 +275,35 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no// ── Circuit breaker ────────────────────────────────────const getHealth = db.prepare('SELECT * FROM provider_health WHERE provider_key = ?');-- const putHealth = db.prepare(`INSERT INTO provider_health (provider_key, state, consecutive_failures, opened_at, last_error, last_success_at, last_failure_at)-- VALUES (@provider_key, @state, @consecutive_failures, @opened_at, @last_error, @last_success_at, @last_failure_at)++ const putHealth = db.prepare(`INSERT INTO provider_health (provider_key, state, consecutive_failures, opened_at, exhausted_at, last_error, last_success_at, last_failure_at)++ VALUES (@provider_key, @state, @consecutive_failures, @opened_at, @exhausted_at, @last_error, @last_success_at, @last_failure_at)ON CONFLICT(provider_key) DO UPDATE SET state = excluded.state, consecutive_failures = excluded.consecutive_failures, opened_at = excluded.opened_at,-- last_error = excluded.last_error, last_success_at = excluded.last_success_at, last_failure_at = excluded.last_failure_at`);++ exhausted_at = excluded.exhausted_at, last_error = excluded.last_error, last_success_at = excluded.last_success_at, last_failure_at = excluded.last_failure_at`);async function health(key) {-- const h = await getHealth.get(key) || { provider_key: key, state: 'closed', consecutive_failures: 0, opened_at: null, last_error: null, last_success_at: null, last_failure_at: null };-- if (h.state === 'open' && clock.now() - (h.opened_at || 0) >= config.breaker.cooldownMs) return { ...h, state: 'half_open' };++ const h = await getHealth.get(key) || { provider_key: key, state: 'closed', consecutive_failures: 0, opened_at: null, exhausted_at: null, last_error: null, last_success_at: null, last_failure_at: null };++ if (h.state === 'open') {++ // Out of balance: a long cooldown (10 minutes), so every request shares the same pause instead of each++ // one retrying a provider that cannot answer until someone tops it up.++ const cooldown = h.exhausted_at ? exhaustedCooldownMs() : config.breaker.cooldownMs;++ if (clock.now() - (h.opened_at || 0) >= cooldown) return { ...h, state: 'half_open' };++ }return h;}async function recordSuccess(key) {const h = await health(key);if (h.state !== 'closed') await registry.audit('system', 'provider.circuit_closed', 'provider', key, { metadata: { after_failures: h.consecutive_failures } });-- await putHealth.run({ ...h, provider_key: key, state: 'closed', consecutive_failures: 0, opened_at: null, last_success_at: clock.now() });++ await putHealth.run({ ...h, provider_key: key, state: 'closed', consecutive_failures: 0, opened_at: null, exhausted_at: null, last_success_at: clock.now() });}async function recordFailure(key, err) {const h = await health(key);const failures = (h.consecutive_failures || 0) + 1;-- const open = h.state === 'half_open' || failures >= config.breaker.failureThreshold;-- if (open && h.state !== 'open') await registry.audit('system', 'provider.circuit_opened', 'provider', key, { metadata: { failures, error: String(err && err.message || err).slice(0, 200) } });-- await putHealth.run({ ...h, provider_key: key, state: open ? 'open' : 'closed', consecutive_failures: failures, opened_at: open ? clock.now() : h.opened_at, last_error: String(err && err.message || err).slice(0, 500), last_failure_at: clock.now() });++ const exhausted = insufficientBalance(err);++ // A balance answer opens the circuit at once (it will not recover on its own), not after failureThreshold.++ const open = exhausted || h.state === 'half_open' || failures >= config.breaker.failureThreshold;++ if (open && h.state !== 'open') await registry.audit('system', 'provider.circuit_opened', 'provider', key, { metadata: { failures, exhausted, error: String(err && err.message || err).slice(0, 200) } });++ await putHealth.run({ ...h, provider_key: key, state: open ? 'open' : 'closed', consecutive_failures: failures,++ opened_at: open ? clock.now() : h.opened_at, exhausted_at: exhausted ? clock.now() : (h.exhausted_at || null),++ last_error: String(err && err.message || err).slice(0, 500), last_failure_at: clock.now() });}async function resetHealth(key) { await db.prepare('DELETE FROM provider_health WHERE provider_key = ?').run(key); }@@@ -326,6 +345,30 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.nofunction isLocalProvider(p) {return p.kind === 'whisper' || Boolean(p.metadata && p.metadata.local);}++ /** The host's text model (local / local2), not its local speech-to-text: the caps below are for text generation. */++ function isLocalModel(p) {++ return Boolean(p.metadata && p.metadata.local);++ }++ /**++ * What the host's local model is offered: short text only (summaries, titles, moderation hints, a chat line),++ * never code or long reasoning. Routes that write whole apps or wiki pages (max_output_tokens well over a chat++ * answer, or outlast_provider_timeout), and an OpenAI completion carrying tools (an agent's loop), are not for++ * it; embeddings are small and stay welcome, and speech-to-text is a different local provider.++ */++ function localCanServe(route, operation, req = null) {++ if (operation === 'embed' || operation === 'transcribe') return true;++ if (!['chat', 'generate', 'summarize', 'classify', 'extract', 'enrich', 'complete', 'messages'].includes(operation)) return false;++ // A request carrying tools is an agent's loop, not a short chat answer.++ if (operation === 'complete' && req && req.openai && Array.isArray(req.openai.tools) && req.openai.tools.length) return false;++ if (operation === 'messages' && req && Array.isArray(req.tools) && req.tools.length) return false;++ if (route && route.options && route.options.outlast_provider_timeout) return false;++ // The answer this request actually asked for decides (engine.js puts input.max_tokens on req); a route's own++ // cap is the fallback. Over 800 tokens is long reasoning, not a short text answer.++ const wanted = Number(req && req.maxTokens);++ if (Number.isFinite(wanted)) return wanted <= 800;++ const max = Number(route && route.max_output_tokens);++ return !Number.isFinite(max) || max <= 800;++ }/*** Billing profiles whose capacity is already paid for: a subscription, a free server or a person's own key.* They carry no per-token rate card and are offered as prepaid. The profile is the providers.billing_profile@@@ -704,7 +747,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.noif (p.status !== 'active') return 'disabled';if (p.auth_mode !== 'none' && !credentialNotRequired(p) && !resolveSecret(p.secret_ref, env)) return 'no_credentials';if (features.some(f => !a.supports(f))) return 'unsupported';-- if ((await health(p.key)).state === 'open') return 'circuit_open';++ const h = await health(p.key);++ if (h.state === 'open') return h.exhausted_at ? 'exhausted' : 'circuit_open';return null;}@@@ -937,6 +981,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.noconst preferState = prefer ? { provider: prefer.provider, model: prefer.model, honoured: false, why: '' } : null;const tried = [];let unreadable = null; // a candidate could not read the caller's media (whisper's ffmpeg): no other will either++ let resting = false; // a paid provider is out of balance, or every remaining offer is paused: answer ai.resting++// A streamed answer (/v1/chat/completions, stream: true) is the caller's once a delta has gone out: a failure// after that is final, never retried or handed to a fallback whose words would follow the first one's.let streamed = false;@@@ -1245,15 +1291,39 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no}const why = await skipReason(p, a, features);if (why) {++ if (why === 'exhausted' || why === 'circuit_open') resting = true;statFor(c.provider).skipped++;tried.push({ provider: c.provider, skipped: why });attempts.push({ key: 'openvibe', provider: c.provider, model, why: chosen || 'the router chose it', status: 'skipped', error: why });if (ctx.logRequest) ctx.logRequest({ ...base, status: 'skipped', skip_reason: why });continue;}++ // The host's local model is the last offer, and only for short text: never code or long reasoning, and++ // one request at a time (AI_LOCAL_LLM_CONCURRENCY) so a paid outage cannot starve the streams it shares++ // the machine with. Its answer is labelled local on the run's receipt (provider_key = local / local2).++ const local = isLocalModel(p);++ if (local && !localCanServe(route, operation, req)) {++ statFor(c.provider).skipped++;++ tried.push({ provider: c.provider, skipped: 'unsupported' });++ attempts.push({ key: 'openvibe', provider: c.provider, model, why: chosen || 'the router chose it', status: 'skipped', error: 'the local model is not offered this job' });++ if (ctx.logRequest) ctx.logRequest({ ...base, status: 'skipped', skip_reason: 'unsupported' });++ continue;++ }++ if (local && localActive >= localLimit) {++ resting = true;++ statFor(c.provider).skipped++;++ tried.push({ provider: c.provider, skipped: 'busy' });++ attempts.push({ key: 'openvibe', provider: c.provider, model, why: chosen || 'the router chose it', status: 'skipped', error: 'the local model is busy with another answer' });++ if (ctx.logRequest) ctx.logRequest({ ...base, status: 'skipped', skip_reason: 'busy' });++ continue;++ }// L36 Y7: the fallback chain stops after three providers.if (attemptsMade >= maxAttempts) break;attemptsMade += 1;++ // Claim the local slot synchronously, before any await, so two requests cannot both pass the check.++ if (local) localActive++;++ let lastErr = null;++ try {// An image generation outlasts the shared provider's chat timeout (AI_PROVIDER_TIMEOUT_MS): the route's wins. So does a// route that says so (options.outlast_provider_timeout: space.build writes a whole app, a minute or two). Otherwise// the provider's timeout stays strict, so a slow provider fails over quickly.@@@ -1262,7 +1332,6 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.noconst cap = outlast ? Math.max(own, Number(route.timeout_ms) || 0, operation === 'image' ? 600000 : 0) : (own || 600000);const timeoutMs = Math.max(1000, Math.min(req.timeoutMs || route.timeout_ms || own || DEFAULT_PROVIDER_TIMEOUT_MS, cap));const retries = config.providerRetries;-- let lastErr = null;for (let attempt = 0; attempt <= retries; attempt++) {if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');const started = clock.now();@@@ -1291,7 +1360,12 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no}// max_cost_usd, when it binds, comes back as the affordable output cap for this attempt.const callMax = routed && Number.isSafeInteger(routed.maxTokens) && routed.maxTokens >= 1 && routed.maxTokens !== req.maxTokens ? routed.maxTokens : null;-- const callReq = callMax == null ? req : { ...req, maxTokens: callMax };++ let callReq = callMax == null ? req : { ...req, maxTokens: callMax };++ // The host's model writes a short answer: its own cap, never more, whatever the route asked for.++ if (local && operation !== 'embed') {++ const localMax = Math.max(1, (config.localLlm && config.localLlm.maxTokens) || 512);++ callReq = { ...callReq, maxTokens: Math.min(Number(callReq.maxTokens) || localMax, localMax) };++ }try {const fn = a[operation];if (typeof fn !== 'function') throw new ProviderError(`${operation} not implemented`, { code: 'provider.unsupported' });@@@ -1335,7 +1409,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.nowarm.remember(warmSubject(ctx, route), leadingText(req), c.provider, model);explain.attempts = attempts;explain.openvibe = attemptFacts;-- return { result, provider: c.provider, providerKind: p.kind, model: result.model || model, fallbackUsed: fallback || selfAttempts.some(a => a.status !== 'skipped'), usage, cost, latencyMs: latency, ttftMs, synthetic: Boolean(result.synthetic || a.synthetic), tried, startedAt: started, explain };++ if (local) explain.provider_local = true; // the Exchange receipt labels the offer local (server/exchange/receipts.js)++ return { result, provider: c.provider, providerKind: p.kind, local, model: result.model || model, fallbackUsed: fallback || selfAttempts.some(a => a.status !== 'skipped'), usage, cost, latencyMs: latency, ttftMs, synthetic: Boolean(result.synthetic || a.synthetic), tried, startedAt: started, explain };} catch (err) {lastErr = err;if (ctx.signal && ctx.signal.aborted) {@@@ -1357,15 +1432,21 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.noif (streamed && err && err.code === 'policy.floor_refused') throw new AiError(403, 'policy.floor_refused', err.detail || err.message, { rule: err.rule, policy: contentPolicy });if (streamed) throw new AiError(502, 'provider.stream_broken', `the answer broke off after it began streaming: ${String(err && err.message || err).slice(0, 200)}`, { tried: [...tried, { provider: c.provider, error: String(err && err.message || err).slice(0, 200) }], explain });// A timed-out image may still be generated and billed upstream: never pay for it twice.-- if (attempt < retries && retryable(err) && !(operation === 'image' && err.code === 'provider.timeout')) { await sleep(config.retryDelayMs + Math.floor(Math.random() * Math.min(700, config.retryDelayMs)), ctx.signal).catch(() => {}); continue; }++ // An insufficient-balance answer is never retried: the same provider cannot answer until it is topped up.++ if (insufficientBalance(err)) resting = true;++ if (attempt < retries && !insufficientBalance(err) && retryable(err) && !(operation === 'image' && err.code === 'provider.timeout')) { await sleep(config.retryDelayMs + Math.floor(Math.random() * Math.min(700, config.retryDelayMs)), ctx.signal).catch(() => {}); continue; }break;}}++ } finally {++ if (local) localActive--;++ }// A request the provider rejected as malformed is the caller's fault, not a sign the// 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 (lastErr && lastErr.code === 'media.unreadable') unreadable = lastErr;-- if (!callerFault(lastErr) && !rateLimited(lastErr)) await recordFailure(c.provider, lastErr);++ // An insufficient-balance 429 is not ordinary capacity: it must open the long exhausted circuit.++ if (insufficientBalance(lastErr) || (!callerFault(lastErr) && !rateLimited(lastErr))) await recordFailure(c.provider, lastErr);tried.push({ provider: c.provider, error: String(lastErr && lastErr.message || lastErr).slice(0, 200) });attempts.push({ key: 'openvibe', provider: c.provider, model: lastErr && lastErr.model || model, why: chosen || (fallback ? 'the next network provider after an earlier attempt failed' : 'the first network offer for this route'), status: lastErr && lastErr.code === 'provider.timeout' ? 'timeout' : 'failed', error: String(lastErr && lastErr.message || lastErr).slice(0, 300) });log.warn(`[ai] ${operation} via ${c.provider}/${model || '-'} failed: ${lastErr && lastErr.message}`);@@@ -1379,6 +1460,10 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.noexplain.reasons = [...(explain.reasons || []), 'the media could not be read'];throw new AiError(422, 'media.unreadable', String(unreadable.message || 'the media could not be read').slice(0, 300), { tried, explain });}++ if (resting) {++ explain.reasons = [...(explain.reasons || []), 'a provider is out of balance or paused; AI is resting'];++ throw new AiError(503, 'ai.resting', 'AI is resting for a moment', { retry_after_seconds: Math.ceil(exhaustedCooldownMs() / 1000), tried, explain });++ }// The free allowance / anonymous Try it had capable offers, but every one was above its ceiling. That is// not an outage: the caller needs their own key, or their own token so their Balance can pay.if (freeCeilingSkipped && attemptsMade === 0) {@@@ -827,7 +827,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowdetail = `The free AI allowance for ${ctx.personSubject} is spent, and the calling service has no budget to cover this run. Add a key for the person, or wait for the allowance to reset.`;}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;++ // ai.resting's detail is the one sentence products show; its tries and Retry-After live in the problem body++ // and the explain, never appended to that sentence.++ const extra = err instanceof AiError && err.extra && err.code !== 'ai.resting' ? 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 = ?, explain = COALESCE(?, explain), finished_at = ? WHERE id = ? AND status IN ('queued', 'running')`)@@@ -31,14 +31,14 @@ const PROBLEMS = { name: 'api', internalCode: 'ai.internal', internalDetail: 'in/*** Any error -> problem+json (openvibe-sdk/service sendError). Only an AiError answers with its own status,-- * code and extra (extra.errors is the problem's errors list; a 429 with retry_after_seconds sets Retry-After).-- * Anything else is logged and answers 500 ai.internal: it is passed on as its stack alone, so a provider-- * error's upstream HTTP status never becomes AI's answer.++ * code and extra (extra.errors is the problem's errors list; extra.retry_after_seconds sets Retry-After, for a++ * 429 refusal and for a 503 ai.resting alike). Anything else is logged and answers 500 ai.internal: it is passed++ * on as its stack alone, so a provider error's upstream HTTP status never becomes AI's answer.*/function sendError(res, err, ctx, log = console) {const req = { ov: ctx };if (!(err instanceof AiError)) return svc.sendError(res, req, { stack: (err && err.stack) || String(err) }, log, PROBLEMS);-- if (err.status === 429 && err.extra && err.extra.retry_after_seconds && !res.headersSent) res.setHeader('Retry-After', String(err.extra.retry_after_seconds));++ if (err.extra && err.extra.retry_after_seconds && !res.headersSent) res.setHeader('Retry-After', String(err.extra.retry_after_seconds));return svc.sendError(res, req, err, log, PROBLEMS);}nnew file mode 100644@@@ -0,0 +1,66 @@++'use strict';++// The hourly balance probe: a provider with a balance API (DeepSeek) is asked what its account holds, the++// reading becomes ai_provider_balance_usd{provider}, and the key is only ever sent in Authorization — never++// logged or exposed. Off under a drill (AI_DRILL=1 / AI_BALANCE_PROBE=0).++const assert = require('assert');++const { createRegistry } = require('openvibe-shared/metrics');++const { createBalanceProbe, deepseekUsd } = require('../server/providers/balance');++const { registerAiGauges } = require('../server/observability');++const { load } = require('../server/config');++const { suite } = require('./helpers');++++const t = suite('provider-balance');++++const record = (key, base_url, secret_ref = 'env:DEEPSEEK_API_KEY') => ({ key, status: 'active', base_url, secret_ref, kind: 'openai', metadata: {} });++++t.test('DeepSeek balance: USD total read, gauge exposed, key never logged', async () => {++ const calls = [];++ const fetchImpl = async (url, opts) => {++ calls.push({ url, auth: opts.headers.Authorization });++ return { ok: true, status: 200, json: async () => ({ is_available: true, balance_infos: [{ currency: 'USD', total_balance: '0.05', granted_balance: '0.00', topped_up_balance: '0.05' }] }) };++ };++ const warned = [];++ const config = load({ NODE_ENV: 'test', AI_BALANCE_PROBE: '1' });++ const probe = createBalanceProbe({++ registry: { listProviders: async () => [record('shared', 'https://api.deepseek.com/v1'), record('other', 'https://api.openai.com/v1')] },++ config, env: { DEEPSEEK_API_KEY: 'sk-deepseek-secret' }, fetchImpl,++ clock: { now: () => 1762400000000 }, log: { warn: (m) => warned.push(m), log() {} },++ });++ const values = await probe.poll();++ assert.deepStrictEqual(values, [{ provider: 'shared', usd: 0.05, at: 1762400000000 }]);++ assert.strictEqual(calls.length, 1, 'only the provider whose host has a balance API is asked');++ assert.strictEqual(calls[0].url, 'https://api.deepseek.com/user/balance');++ assert.strictEqual(calls[0].auth, 'Bearer sk-deepseek-secret');++ assert.ok(!JSON.stringify(warned).includes('sk-deepseek-secret'), 'the key is never in a log line');++ assert.deepStrictEqual(deepseekUsd({ balance_infos: [{ currency: 'CNY', total_balance: '10' }] }), null, 'a non-USD balance is not read as dollars');++++ const metrics = createRegistry();++ registerAiGauges(metrics, {++ runs: { stats: () => ({ queued: 0, running: 0, max_concurrent: 1, max_queued: 10 }) },++ registry: { listProviders: async () => [], publicProvider: (p) => p }, pool: { health: async () => ({ state: 'closed' }) },++ balance: probe,++ });++ const text = await metrics.metricsAsync();++ assert.ok(/ai_provider_balance_usd\{provider="shared"\} 0\.05\n/.test(text), text);++});++++t.test('a failed probe keeps the last reading and does not fail the caller', async () => {++ let status = 200;++ const fetchImpl = async () => ({ ok: status === 200, status, json: async () => ({ balance_infos: [{ currency: 'USD', total_balance: '7.25' }] }) });++ const probe = createBalanceProbe({++ registry: { listProviders: async () => [record('shared', 'https://api.deepseek.com/v1')] },++ config: load({ NODE_ENV: 'test' }), env: { DEEPSEEK_API_KEY: 'k' }, fetchImpl,++ clock: { now: () => 1 }, log: { warn() {}, log() {} },++ });++ assert.strictEqual((await probe.poll())[0].usd, 7.25);++ status = 503;++ assert.strictEqual((await probe.poll())[0].usd, 7.25, 'the last good balance stays');++});++++t.test('off under a drill or when AI_BALANCE_PROBE=0', () => {++ assert.strictEqual(load({ NODE_ENV: 'test', AI_DRILL: '1' }).balance.enabled, false);++ assert.strictEqual(load({ NODE_ENV: 'test', AI_BALANCE_PROBE: '0' }).balance.enabled, false);++ assert.strictEqual(load({ NODE_ENV: 'test' }).balance.enabled, true);++});++++t.run();nnew file mode 100644@@@ -0,0 +1,141 @@++'use strict';++// A provider that answers "insufficient balance" is exhausted, not flaky: its circuit opens at once for the++// long cooldown (10 minutes by default), shared by every request; it is never retried into; the route fails++// over down its offer list. When nothing can answer, the problem is 503 ai.resting with a human sentence and++// Retry-After, not a bare provider.unavailable.++const assert = require('assert');++const { boot, request, token, suite, seamServer, ALL } = require('./helpers');++const { AiError, sendError } = require('../server/util');++++const t = suite('provider-exhausted');++const tok = token('live', ALL);++++function fakeRes() {++ const headers = {};++ return { headers, statusCode: 0, body: null, headersSent: false, setHeader(k, v) { headers[k.toLowerCase()] = v; }, getHeader(k) { return headers[k.toLowerCase()]; }, end(s) { this.body = JSON.parse(s); this.headersSent = true; } };++}++++t.test('402 Insufficient Balance opens the circuit at once, fails over, and never retries the same provider', async () => {++ const paid = await seamServer(() => ({ status: 402, body: { error: { message: 'Insufficient Balance', type: 'unknown_error', param: null, code: 'invalid_request_error' } } }));++ const local = await seamServer(() => ({ choices: [{ message: { content: 'a short local answer' } }], model: 'local-model', usage: { prompt_tokens: 4, completion_tokens: 3 } }));++ let now = 1762400000000;++ const clock = { now: () => now };++ const h = await boot({ clock, env: {++ AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '1', AI_BREAKER_FAILURES: '5', AI_BREAKER_EXHAUSTED_COOLDOWN_MS: '600000',++ AI_PROVIDER: 'openai', AI_BASE_URL: `${paid.url}/v1`, AI_API_KEY: 'test-key', AI_MODEL: 'paid-model',++ AI_LOCAL_LLM_URL: `${local.url}/v1`, AI_LOCAL_LLM_MODEL: 'local-model', AI_LOCAL_LLM_TIMEOUT_MS: '5000',++ } });++ const route = await h.registry.resolveRoute('default.chat');++ const logs = [];++ const ask = (signal) => h.pool.execute(route, 'chat', { messages: [{ role: 'user', content: 'summarise this' }], maxTokens: 60, timeoutMs: 3000 }, { routeKey: route.key, routeVersion: route.version, logRequest: (e) => logs.push(e) });++++ const first = await ask();++ assert.strictEqual(first.provider, 'local', 'the exhausted paid offer fell over to the local model');++ assert.strictEqual(first.local, true, 'the receipt labels a local answer');++ assert.strictEqual(first.result.text, 'a short local answer');++ assert.strictEqual(paid.calls, 1, 'the same provider is never retried after a balance answer (AI_PROVIDER_RETRIES=1)');++ assert.strictEqual(local.calls, 1);++ assert.strictEqual((await h.pool.health('shared')).state, 'open');++ assert.ok((await h.pool.health('shared')).exhausted_at, 'the health row says why it opened');++ assert.ok(logs.some((e) => e.provider_key === 'shared' && e.status === 'error'), 'the balance failure is logged');++++ // Every request shares the pause: the next one skips the exhausted provider entirely.++ const second = await ask();++ assert.strictEqual(second.provider, 'local');++ assert.strictEqual(paid.calls, 1, 'a circuit-open exhausted provider is not called again');++ assert.strictEqual((await h.pool.health('shared')).state, 'open');++++ // The fake clock passes the long cooldown: one half-open probe is allowed, and a second 402 opens it again.++ now += 600001;++ assert.strictEqual((await h.pool.health('shared')).state, 'half_open');++ const third = await ask();++ assert.strictEqual(third.provider, 'local');++ assert.strictEqual(paid.calls, 2, 'exactly one probe after the cooldown, never a retry');++ assert.strictEqual((await h.pool.health('shared')).state, 'open');++++ // With the local model failing too, nothing can serve: 503 ai.resting, with Retry-After on the problem.++ local.handler = () => ({ status: 500, body: { error: { message: 'local is down' } } });++ local.calls = 0;++ let err = null;++ try { await ask(); } catch (e) { err = e; }++ assert.ok(err, 'nothing served');++ assert.strictEqual(err.status, 503);++ assert.strictEqual(err.code, 'ai.resting');++ assert.strictEqual(err.detail, 'AI is resting for a moment');++ assert.strictEqual(err.extra.retry_after_seconds, 600);++ const res = fakeRes();++ sendError(res, err, { requestId: 'req_1' }, { log() {}, warn() {}, error() {} });++ assert.strictEqual(res.statusCode, 503);++ assert.strictEqual(res.headers['retry-after'], '600');++ assert.strictEqual(res.body.detail, 'AI is resting for a moment');++++ // The run API keeps that sentence clean (its tries/explain ride the explain, not the detail).++ const run = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok, body: { workflow: 'ai.generate', input: { prompt: 'hi' }, options: { cache: false } } });++ assert.strictEqual(run.body.run.status, 'failed', run.text);++ assert.strictEqual(run.body.run.error.code, 'ai.resting', run.text);++ assert.strictEqual(run.body.run.error.detail, 'AI is resting for a moment', run.text);++ await h.stop(); await paid.close(); await local.close();++});++++t.test('an OpenAI-style 429 insufficient_quota is exhausted too, not ordinary rate-limit capacity', async () => {++ const paid = await seamServer(() => ({ status: 429, body: { error: { message: 'You exceeded your current quota', type: 'insufficient_quota', code: 'insufficient_quota' } } }));++ const local = await seamServer(() => ({ choices: [{ message: { content: 'local' } }], model: 'local-model', usage: { prompt_tokens: 1, completion_tokens: 1 } }));++ const h = await boot({ env: {++ AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '1', AI_BREAKER_FAILURES: '5',++ AI_PROVIDER: 'openai', AI_BASE_URL: `${paid.url}/v1`, AI_API_KEY: 'test-key', AI_MODEL: 'paid-model',++ AI_LOCAL_LLM_URL: `${local.url}/v1`, AI_LOCAL_LLM_MODEL: 'local-model', AI_LOCAL_LLM_TIMEOUT_MS: '5000',++ } });++ const route = await h.registry.resolveRoute('default.chat');++ const out = await h.pool.execute(route, 'chat', { messages: [{ role: 'user', content: 'hi' }], maxTokens: 30, timeoutMs: 3000 }, { routeKey: route.key });++ assert.strictEqual(out.provider, 'local');++ assert.strictEqual(paid.calls, 1, 'not retried and not treated as ordinary 429 capacity');++ const health = await h.pool.health('shared');++ assert.strictEqual(health.state, 'open');++ assert.ok(health.exhausted_at);++ await h.stop(); await paid.close(); await local.close();++});++++t.test('the local model is short text only, capped, and one request at a time', async () => {++ const paid = await seamServer(() => ({ status: 500, body: { error: { message: 'paid is down' } } }));++ let localCalls = 0;++ const local = await seamServer(async (body) => {++ localCalls++;++ await new Promise((r) => setTimeout(r, 250));++ return { choices: [{ message: { content: 'short' } }], model: 'local-model', usage: { prompt_tokens: 3, completion_tokens: 2 } };++ });++ const h = await boot({ env: {++ AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '50',++ AI_PROVIDER: 'openai', AI_BASE_URL: `${paid.url}/v1`, AI_API_KEY: 'test-key', AI_MODEL: 'paid-model',++ AI_LOCAL_LLM_URL: `${local.url}/v1`, AI_LOCAL_LLM_MODEL: 'local-model', AI_LOCAL_LLM_TIMEOUT_MS: '5000',++ AI_LOCAL_LLM_MAX_TOKENS: '256', AI_LOCAL_LLM_CONCURRENCY: '1',++ } });++ const route = await h.registry.resolveRoute('default.chat');++ const req = { messages: [{ role: 'user', content: 'summarise' }], maxTokens: 600, timeoutMs: 3000 };++++ const one = await h.pool.execute(route, 'chat', req, { routeKey: route.key });++ assert.strictEqual(one.provider, 'local');++ assert.strictEqual(local.last.max_tokens, 256, 'the local answer is capped at AI_LOCAL_LLM_MAX_TOKENS, whatever the route asked');++++ // Two at once: one gets the local model, the other is told AI is resting instead of a second local generation.++ const results = await Promise.allSettled([++ h.pool.execute(route, 'chat', req, { routeKey: route.key }),++ h.pool.execute(route, 'chat', req, { routeKey: route.key }),++ ]);++ const ok = results.filter((r) => r.status === 'fulfilled');++ const bad = results.filter((r) => r.status === 'rejected');++ assert.strictEqual(ok.length, 1, 'one concurrent local request at a time');++ assert.strictEqual(bad.length, 1);++ assert.strictEqual(bad[0].reason.code, 'ai.resting', bad[0].reason.message);++ assert.strictEqual(localCalls, 2, 'the second request never reached the local model');++++ // Code and long reasoning are never sent to the host's model.++ const before = localCalls;++ const longRoute = { ...route, max_output_tokens: 4000 };++ let longErr = null;++ try { await h.pool.execute(longRoute, 'chat', { ...req, maxTokens: 4000 }, { routeKey: route.key }); } catch (e) { longErr = e; }++ assert.ok(longErr, 'nothing served the long route');++ assert.strictEqual(localCalls, before, 'the local model was not offered a long answer');++ await h.stop(); await paid.close(); await local.close();++});++++t.run();
@@ -174,6 +174,10 @@ AI_MODEL=
@@ -174,6 +174,10 @@ AI_MODEL=
#AI_MEDIA_STORAGE_HOSTS=*.backblazeb2.com,*.r2.cloudflarestorage.com,edge.openvibe.media
#AI_MEDIA_STORAGE_HOSTS=*.backblazeb2.com,*.r2.cloudflarestorage.com,edge.openvibe.media
#AI_LOCAL_LLM_MODEL=qwen2.5-1.5b-instruct
#AI_LOCAL_LLM_MODEL=qwen2.5-1.5b-instruct
#AI_LOCAL_LLM_TIMEOUT_MS=180000
#AI_LOCAL_LLM_TIMEOUT_MS=180000
+# The host's model is a last offer for short text only: never more than this many tokens, and one generation at
+# a time across local and local2, so an outage upstream cannot starve the host.
+#AI_LOCAL_LLM_MAX_TOKENS=512
+#AI_LOCAL_LLM_CONCURRENCY=1
# A second local chat model server (its own llama.cpp process and another model family): provider `local2`,
# A second local chat model server (its own llama.cpp process and another model family): provider `local2`,
# so a caller can pin two local offers, e.g. Games' AI-vs-AI matches. Same rules: no key, free, no JSON.
# so a caller can pin two local offers, e.g. Games' AI-vs-AI matches. Same rules: no key, free, no JSON.
#AI_LOCAL_LLM2_URL=http://127.0.0.1:8092/v1
#AI_LOCAL_LLM2_URL=http://127.0.0.1:8092/v1
@@ -252,8 +256,16 @@ AI_MAX_COST_USD_PER_DAY=0
@@ -252,8 +256,16 @@ AI_MAX_COST_USD_PER_DAY=0
#AI_DEBUG_RAW_LOG=false
#AI_DEBUG_RAW_LOG=false
#AI_BREAKER_FAILURES=3
#AI_BREAKER_FAILURES=3
#AI_BREAKER_COOLDOWN_MS=60000
#AI_BREAKER_COOLDOWN_MS=60000
+# A provider that answered "insufficient balance" (HTTP 402, or a 429/403 insufficient_quota) is out of money:
+# its circuit opens at once for this long (10 minutes), shared by every request, and it is never retried.
+#AI_BREAKER_EXHAUSTED_COOLDOWN_MS=600000
#AI_PROVIDER_RETRIES=1
#AI_PROVIDER_RETRIES=1
#AI_PROVIDER_RETRY_DELAY_MS=800
#AI_PROVIDER_RETRY_DELAY_MS=800
+# Hourly balance probe (DeepSeek's GET /user/balance) exposed as ai_provider_balance_usd{provider} on /metrics;
+# the operator alert fires below $2. Off under a restore drill (AI_DRILL=1).
+#AI_BALANCE_PROBE=1
+#AI_BALANCE_PROBE_INTERVAL_MS=3600000
+#AI_BALANCE_PROBE_TIMEOUT_MS=10000
# ── Media inputs: only these https hosts are fetched (plus the Media internal origin) ──
# ── Media inputs: only these https hosts are fetched (plus the Media internal origin) ──
OV_MEDIA_URL=https://openvibe.media
OV_MEDIA_URL=https://openvibe.media
@@ -161,9 +161,15 @@ schema errors), `workflow.not_found`, `idempotency.conflict`, `quota.exceeded` (
@@ -161,9 +161,15 @@ schema errors), `workflow.not_found`, `idempotency.conflict`, `quota.exceeded` (
`Retry-After`), `govern.exceeded` (429 + `Retry-After`: a tier budget) and `govern.unavailable` (503: Valkey
`Retry-After`), `govern.exceeded` (429 + `Retry-After`: a tier budget) and `govern.unavailable` (503: Valkey
unreachable for a paid or staff run), `queue.full` (429 + `Retry-After`: the run would wait behind `AI_MAX_QUEUED_RUNS`
unreachable for a paid or staff run), `queue.full` (429 + `Retry-After`: the run would wait behind `AI_MAX_QUEUED_RUNS`
runs, or `AI_MAX_QUEUED_RUNS_PER_CALLER` of this caller's), `capability.denied`, `capability.namespace_denied`, `token.*`. A run that cannot
runs, or `AI_MAX_QUEUED_RUNS_PER_CALLER` of this caller's), `capability.denied`, `capability.namespace_denied`, `token.*`. A run that cannot
-produce a real answer ends `failed` with an explicit code — `provider.unavailable`,
-`route.unavailable`, `fetch.refused`, `source.unavailable`, `input.insufficient`, `output.empty`,
-`output.invalid`, `run.interrupted` — and never with filler content.
+produce a real answer ends `failed` with an explicit code — `ai.resting` (503: every offer is out of balance,
+paused or busy; the detail is the sentence products show, "AI is resting for a moment", with `Retry-After`),
+`provider.unavailable`, `route.unavailable`, `fetch.refused`, `source.unavailable`, `input.insufficient`,
+`output.empty`, `output.invalid`, `run.interrupted` — and never with filler content.
+
+Each active provider with a balance API (DeepSeek's `GET /user/balance`) is asked hourly; the last reading is
+`ai_provider_balance_usd{provider}` on `/metrics` and OpenVibe.Host's `OpenVibeAiProviderBalanceLow` alert turns a
+reading below $2 into an operator ticket. The key is only sent in the request's `Authorization` header: it is never
+logged, returned or stored by the probe. `AI_BALANCE_PROBE=0` (or a restore drill, `AI_DRILL=1`) turns it off.
The capability ids and the `ai` service manifest are released in openvibe-contracts v0.76.0 (the
The capability ids and the `ai` service manifest are released in openvibe-contracts v0.76.0 (the
drafts stay in `docs/capabilities-proposal/`); `server/auth.js` checks them with the contracts grant
drafts stay in `docs/capabilities-proposal/`); `server/auth.js` checks them with the contracts grant
@@ -505,7 +511,7 @@ definition, but never over an admin's version.
@@ -505,7 +511,7 @@ definition, but never over an admin's version.
| `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; NVIDIA Parakeet TDT 0.6B v3 (`parakeet-cli`) for English and 24 European languages when `PARAKEET_MODEL` is set, falling back to whisper | `WHISPER_*`, `PARAKEET_*`, `ASR_ENGINE` — Live's names |
| `whisper` | whisper.cpp on this host, with Live's VAD, hallucination filter, multilingual model and live/batch lanes; NVIDIA Parakeet TDT 0.6B v3 (`parakeet-cli`) for English and 24 European languages when `PARAKEET_MODEL` is set, falling back to whisper | `WHISPER_*`, `PARAKEET_*`, `ASR_ENGINE` — 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`; last offer for short text, one request at a time | `AI_LOCAL_LLM_URL`, `AI_LOCAL_LLM_MODEL`, `AI_LOCAL_LLM_TIMEOUT_MS`, `AI_LOCAL_LLM_MAX_TOKENS` (512), `AI_LOCAL_LLM_CONCURRENCY` (1) |
| `openai`, key `local2` | A second local chat server, another llama.cpp process and model family: no key, never paid, no JSON or embeddings. Two local offers let a caller pin a match between them (Games' AI-vs-AI job) | `AI_LOCAL_LLM2_URL`, `AI_LOCAL_LLM2_MODEL`, `AI_LOCAL_LLM2_TIMEOUT_MS` |
| `openai`, key `local2` | A second local chat server, another llama.cpp process and model family: no key, never paid, no JSON or embeddings. Two local offers let a caller pin a match between them (Games' AI-vs-AI job) | `AI_LOCAL_LLM2_URL`, `AI_LOCAL_LLM2_MODEL`, `AI_LOCAL_LLM2_TIMEOUT_MS` |
Routing: a route either names a `capability` and the router builds its pool from every provider that
Routing: a route either names a `capability` and the router builds its pool from every provider that
@@ -513,7 +519,10 @@ can serve it, ordered by `openvibe-sdk/placement` (objective, price, measured la
@@ -513,7 +519,10 @@ can serve it, ordered by `openvibe-sdk/placement` (objective, price, measured la
hysteresis through the route's current placement), or pins a `[primary, ...fallbacks]` list. A
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
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
**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
+failure. A provider's own "insufficient balance" answer (HTTP 402, or 429/403 `insufficient_quota`) is never
+retried: its circuit opens at once for `AI_BREAKER_EXHAUSTED_COOLDOWN_MS` (10 minutes, shared by every request)
+and the router fails over. When nothing can serve, the run answers `ai.resting`; 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`
(`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
(objective, reasons, selected, and every candidate with its estimated cost, latency and excluded
reason).
reason).
@@ -772,6 +781,12 @@ pair. On the production host since 2026-11-06: Llama-3.2-1B-Instruct Q4_K_M (Lla
@@ -772,6 +781,12 @@ pair. On the production host since 2026-11-06: Llama-3.2-1B-Instruct Q4_K_M (Lla
Hugging Face sha256 `6f85a640a97cf2bf5b8e764087b1e83da0fdb51d7c9fab7d0fece9385611df83`, 807,694,464 bytes),
Hugging Face sha256 `6f85a640a97cf2bf5b8e764087b1e83da0fdb51d7c9fab7d0fece9385611df83`, 807,694,464 bytes),
checked against that hash before install; it uses about 0.8 GB.
checked against that hash before install; it uses about 0.8 GB.
+The local model is a **last offer for short text** (summaries, titles, moderation hints, a chat line), never
+for code or long reasoning: routes with `max_output_tokens` over 800 or `outlast_provider_timeout` (an app or a
+wiki page) never offer it. It answers with `AI_LOCAL_LLM_MAX_TOKENS` (512) at most, and one request at a time
+(`AI_LOCAL_LLM_CONCURRENCY`, 1) across `local` and `local2`, so a paid outage cannot starve the streams sharing
+the host. A run that it answered names `local`/`local2` as its provider and marks the receipt local.
+
Local embeddings are off unless `AI_LOCAL_EMBED_MODEL` names an embedding model the server at `AI_LOCAL_LLM_URL`
Local embeddings are off unless `AI_LOCAL_EMBED_MODEL` names an embedding model the server at `AI_LOCAL_LLM_URL`
answers at `/embeddings`. Then `local` advertises `embed`, the model is registered as an `embedding` model, and
answers at `/embeddings`. Then `local` advertises `embed`, the model is registered as an `embedding` model, and
`default.embedding` (a capability pool) offers it beside the shared provider's `AI_EMBEDDING_MODEL`; `local` is
`default.embedding` (a capability pool) offers it beside the shared provider's `AI_EMBEDDING_MODEL`; `local` is
@@ -792,6 +807,8 @@ mix vectors from different providers (re-embed when the provider changes).
@@ -792,6 +807,8 @@ mix vectors from different providers (re-embed when the provider changes).
| Every workflow has versioned, compiling input/output schemas, runs on the stub and validates | `test/workflows.test.js` |
| Every workflow has versioned, compiling input/output schemas, runs on the stub and validates | `test/workflows.test.js` |
| Edits are new versions; runs pin versions; lifecycle; admin versions survive reseeding | `test/versioning.test.js` |
| Edits are new versions; runs pin versions; lifecycle; admin versions survive reseeding | `test/versioning.test.js` |
| A degraded primary falls back and records it; breaker; skips; timeouts; explicit `provider.unavailable` | `test/fallback.test.js` |
| A degraded primary falls back and records it; breaker; skips; timeouts; explicit `provider.unavailable` | `test/fallback.test.js` |
+| 402/insufficient-quota opens the long exhausted circuit, never retries, fails over, then `ai.resting` | `test/provider-exhausted.test.js` |
+| Hourly DeepSeek balance gauge, no key in logs, off under a drill | `test/provider-balance.test.js` |
| Quotas refuse with 429 + Retry-After **before** any provider call; per-service, attribution, cost caps | `test/quota.test.js` |
| Quotas refuse with 429 + Retry-After **before** any provider call; per-service, attribution, cost caps | `test/quota.test.js` |
| Free allowance: free up to the card's allowance then fully priced, reset per day/month (never for `none`), per subject, never past the cap under concurrent claims, absent at 0 | `test/free-allowance.test.js` |
| Free allowance: free up to the card's allowance then fully priced, reset per day/month (never for `none`), per subject, never past the cap under concurrent claims, absent at 0 | `test/free-allowance.test.js` |
| Tier budgets: enforced per subject and per project and reset per window, two concurrent runs never both take the last unit, the free allowance's numbers unchanged on the Valkey store, one reading per accounted run from govern's `onUsage`, Valkey unreachable (paid 503, free admitted), off by default | `test/govern-tiers.test.js` |
| Tier budgets: enforced per subject and per project and reset per window, two concurrent runs never both take the last unit, the free allowance's numbers unchanged on the Valkey store, one reading per accounted run from govern's `onUsage`, Valkey unreachable (paid 503, free admitted), off by default | `test/govern-tiers.test.js` |
new file mode 100644
new file mode 100644
@@ -0,0 +1,5 @@
@@ -0,0 +1,5 @@
+-- phase: expand
+-- A provider that answered "insufficient balance" is out of money, not flaky: it is opened for the long
+-- exhausted cooldown (AI_BREAKER_EXHAUSTED_COOLDOWN_MS, 10 minutes) instead of the ordinary breaker cooldown.
+-- exhausted_at records when that happened; a successful (or manually reset) probe clears it. Additive only.
+ALTER TABLE provider_health ADD COLUMN exhausted_at bigint; -- epoch ms of the last "insufficient balance" answer
@@ -24,7 +24,7 @@ const pkg = require('../package.json');
@@ -24,7 +24,7 @@ const pkg = require('../package.json');
const { adminRoutes } = require('openvibe-shared/config');
const { adminRoutes } = require('openvibe-shared/config');
-function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env = process.env, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console, credentials = null, preferences = null, grants = null, vault = null, settings = null, benchEvents = null, runBudgets = null, memory = null, memoryImport = null, community = null, presets = null, receipts = null }) {
+function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env = process.env, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console, credentials = null, preferences = null, grants = null, vault = null, settings = null, benchEvents = null, runBudgets = null, memory = null, memoryImport = null, community = null, presets = null, receipts = null, balance = null }) {
const app = express();
const app = express();
// /.well-known/agent.json (openvibe-shared/agent-layer, SEO-2): what this site serves, for agents.
// /.well-known/agent.json (openvibe-shared/agent-layer, SEO-2): what this site serves, for agents.
require('openvibe-shared/agent-layer').mount(app, { descriptor: require('./agent-descriptor') });
require('openvibe-shared/agent-layer').mount(app, { descriptor: require('./agent-descriptor') });
@@ -36,7 +36,7 @@ function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys
@@ -36,7 +36,7 @@ function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys
// GET /metrics answers direct loopback callers only (Track O).
// GET /metrics answers direct loopback callers only (Track O).
const metrics = instrument(app, { service: 'ai', release: release.release });
const metrics = instrument(app, { service: 'ai', release: release.release });
app.use(require('openvibe-sdk/service').telemetryMiddleware); // request telemetry: a no-op until server/index.js calls telemetry.init()
app.use(require('openvibe-sdk/service').telemetryMiddleware); // request telemetry: a no-op until server/index.js calls telemetry.init()
- registerAiGauges(metrics.registry, { runs, registry, pool });
+ registerAiGauges(metrics.registry, { runs, registry, pool, balance });
app.use(http.middleware());
app.use(http.middleware());
// Browser reads of the two public GETs (/stats, /release.json) only: an exact origin from
// Browser reads of the two public GETs (/stats, /release.json) only: an exact origin from
// AI_CORS_ORIGINS is echoed back, never credentials, and OPTIONS answers their preflight.
// AI_CORS_ORIGINS is echoed back, never credentials, and OPTIONS answers their preflight.
@@ -190,6 +190,10 @@ function load(env = process.env) {
@@ -190,6 +190,10 @@ function load(env = process.env) {
model: String(env.AI_LOCAL_LLM_MODEL || '').trim(),
model: String(env.AI_LOCAL_LLM_MODEL || '').trim(),
embedModel: String(env.AI_LOCAL_EMBED_MODEL || '').trim(),
embedModel: String(env.AI_LOCAL_EMBED_MODEL || '').trim(),
timeoutMs: int(env.AI_LOCAL_LLM_TIMEOUT_MS, 180000),
timeoutMs: int(env.AI_LOCAL_LLM_TIMEOUT_MS, 180000),
+ // Never let the local model starve the host: one generation at a time across every local provider,
+ // and a short answer unless a route already asked for less (server/providers/index.js).
+ maxTokens: Math.max(1, int(env.AI_LOCAL_LLM_MAX_TOKENS, 512)),
+ concurrency: Math.max(1, int(env.AI_LOCAL_LLM_CONCURRENCY, 1)),
},
},
// A second local chat server (its own llama.cpp process, another model family): the provider `local2`.
// A second local chat server (its own llama.cpp process, another model family): the provider `local2`.
// Same rules as `local`: no key, never paid, no JSON. Two local offers let a caller pin a match
// Same rules as `local`: no key, never paid, no JSON. Two local offers let a caller pin a match
@@ -300,12 +304,24 @@ function load(env = process.env) {
@@ -300,12 +304,24 @@ function load(env = process.env) {
breaker: {
breaker: {
failureThreshold: Math.max(1, int(env.AI_BREAKER_FAILURES, 3)),
failureThreshold: Math.max(1, int(env.AI_BREAKER_FAILURES, 3)),
cooldownMs: int(env.AI_BREAKER_COOLDOWN_MS, 60000),
cooldownMs: int(env.AI_BREAKER_COOLDOWN_MS, 60000),
+ // A provider that answered "insufficient balance" is out of money, not flaky: its circuit stays open this
+ // long (10 minutes), shared by every request, before one probe is allowed to spend again.
+ exhaustedCooldownMs: Math.max(60000, int(env.AI_BREAKER_EXHAUSTED_COOLDOWN_MS, 10 * 60000)),
},
},
// One retry on a transient failure (timeout, 429, 5xx, connection reset) before the
// One retry on a transient failure (timeout, 429, 5xx, connection reset) before the
// router moves on to the fallback — Live's rule.
// router moves on to the fallback — Live's rule.
providerRetries: Math.max(0, int(env.AI_PROVIDER_RETRIES, 1)),
providerRetries: Math.max(0, int(env.AI_PROVIDER_RETRIES, 1)),
retryDelayMs: Math.max(0, int(env.AI_PROVIDER_RETRY_DELAY_MS, 800)),
retryDelayMs: Math.max(0, int(env.AI_PROVIDER_RETRY_DELAY_MS, 800)),
+ // ── Hourly provider balance probe (server/providers/balance.js) ──
+ // Each active provider with a balance API (DeepSeek today) is asked GET <origin>/user/balance and the reading
+ // is exposed as ai_provider_balance_usd{provider} on GET /metrics. Off under a restore drill (AI_DRILL=1).
+ balance: {
+ enabled: bool(env.AI_BALANCE_PROBE, !bool(env.AI_DRILL, false)),
+ intervalMs: Math.max(60000, int(env.AI_BALANCE_PROBE_INTERVAL_MS, 60 * 60 * 1000)),
+ timeoutMs: Math.max(1000, int(env.AI_BALANCE_PROBE_TIMEOUT_MS, 10000)),
+ },
+
// ── Media inputs: only these hosts are ever fetched (SSRF guard) ──
// ── Media inputs: only these hosts are ever fetched (SSRF guard) ──
media: {
media: {
publicUrl: trimUrl(env.OV_MEDIA_URL || 'https://openvibe.media'),
publicUrl: trimUrl(env.OV_MEDIA_URL || 'https://openvibe.media'),
@@ -91,6 +91,8 @@ function buildReceipt({ run, explain = null, usage = {}, price = null, privacy =
@@ -91,6 +91,8 @@ function buildReceipt({ run, explain = null, usage = {}, price = null, privacy =
deployment: (provider || (ok && ok.provider) || run.provider_key) || null,
deployment: (provider || (ok && ok.provider) || run.provider_key) || null,
provider_access: (x.openvibe && x.openvibe.provider_access) || null,
provider_access: (x.openvibe && x.openvibe.provider_access) || null,
resolved_model: (x.openvibe && x.openvibe.resolved_model) || null,
resolved_model: (x.openvibe && x.openvibe.resolved_model) || null,
+ // The host's own model answered (local / local2): the receipt says so, beside the provider name.
+ local: Boolean(x.provider_local),
},
},
price: {
price: {
in_per_mtok: round(price && price.in),
in_per_mtok: round(price && price.in),
@@ -122,6 +124,7 @@ function compact(receipt) {
@@ -122,6 +124,7 @@ function compact(receipt) {
id: receipt.id,
id: receipt.id,
run_id: receipt.run_id,
run_id: receipt.run_id,
offer: offer.provider ? `${offer.provider}${offer.model ? `/${offer.model}` : ''}` : null,
offer: offer.provider ? `${offer.provider}${offer.model ? `/${offer.model}` : ''}` : null,
+ local: Boolean(offer.local),
price: { in: receipt.price && receipt.price.in_per_mtok, out: receipt.price && receipt.price.out_per_mtok, cached: receipt.price && receipt.price.cached_per_mtok },
price: { in: receipt.price && receipt.price.in_per_mtok, out: receipt.price && receipt.price.out_per_mtok, cached: receipt.price && receipt.price.cached_per_mtok },
tokens: { in: receipt.usage && receipt.usage.tokens_in, out: receipt.usage && receipt.usage.tokens_out, cached: receipt.usage && receipt.usage.tokens_cached },
tokens: { in: receipt.usage && receipt.usage.tokens_in, out: receipt.usage && receipt.usage.tokens_out, cached: receipt.usage && receipt.usage.tokens_cached },
cost_usd: receipt.cost_usd,
cost_usd: receipt.cost_usd,
@@ -67,6 +67,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
@@ -67,6 +67,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
// Network owns the 18+ account flag; the adult tier checks it before placement.
// Network owns the 18+ account flag; the adult tier checks it before placement.
const adultStatus = require('./policy/adult-status').createAdultStatus({ config, env, fetchImpl, clock, log });
const adultStatus = require('./policy/adult-status').createAdultStatus({ config, env, fetchImpl, clock, log });
const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log, credentialFetch, machineModels: machines, community, bench, variantEvidence, adultStatus });
const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log, credentialFetch, machineModels: machines, community, bench, variantEvidence, adultStatus });
+ // The hourly provider balance probe (server/providers/balance.js): a gauge for the operator alert, started
+ // only when the service listens (never in tests) and off under a restore drill.
+ const balance = require('./providers/balance').createBalanceProbe({ registry, config, env, fetchImpl, clock, log });
// T6 step 8: the SaaS tier budgets on openvibe-sdk/govern in Valkey (server/govern.js); off unless AI_GOVERN_TIERS=1.
// T6 step 8: the SaaS tier budgets on openvibe-sdk/govern in Valkey (server/govern.js); off unless AI_GOVERN_TIERS=1.
const tiers = createTiers({ config, valkey, clock, log });
const tiers = createTiers({ config, valkey, clock, log });
// §2.1.8: the provider free allowance per subject, priced off the pool's own rate cards (server/free-allowance.js);
// §2.1.8: the provider free allowance per subject, priced off the pool's own rate cards (server/free-allowance.js);
@@ -143,7 +146,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
@@ -143,7 +146,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
clientSecret: env.OV_OAUTH_CLIENT_SECRET, fetch: fetchImpl,
clientSecret: env.OV_OAUTH_CLIENT_SECRET, fetch: fetchImpl,
}) : null;
}) : null;
const benchEvents = createBenchEvents({ bench, accountData, accountSend, config, log, fetchImpl, clientSecret: env.OV_OAUTH_CLIENT_SECRET });
const benchEvents = createBenchEvents({ bench, accountData, accountSend, config, log, fetchImpl, clientSecret: env.OV_OAUTH_CLIENT_SECRET });
- const app = createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env, clock, fetchImpl, log, credentials, preferences, grants, vault, settings, benchEvents, runBudgets, memory, memoryImport, community, presets, receipts });
+ const app = createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env, clock, fetchImpl, log, credentials, preferences, grants, vault, settings, benchEvents, runBudgets, memory, memoryImport, community, presets, receipts, balance });
let jwks = null;
let jwks = null;
let keyLoaded = Promise.resolve(null);
let keyLoaded = Promise.resolve(null);
if (!keys.pinned && config.nodeEnv !== 'test') {
if (!keys.pinned && config.nodeEnv !== 'test') {
@@ -168,6 +171,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
@@ -168,6 +171,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
let benchSubscription = null;
let benchSubscription = null;
let accountSubscription = null;
let accountSubscription = null;
if (listen) userModules.start();
if (listen) userModules.start();
+ if (listen) balance.start();
if (listen && config.memoryJobs && config.memoryJobs.enabled) memoryJobs.start();
if (listen && config.memoryJobs && config.memoryJobs.enabled) memoryJobs.start();
// Durable Credit settles: the same jobs rule as Memory (never in test mode, never under a provider drill; the
// Durable Credit settles: the same jobs rule as Memory (never in test mode, never under a provider drill; the
// queue's runOnce checks the drill itself).
// queue's runOnce checks the drill itself).
@@ -198,6 +202,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
@@ -198,6 +202,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
async function close() {
async function close() {
clearInterval(housekeeping);
clearInterval(housekeeping);
+ balance.stop();
if (benchSubscription) benchSubscription.stop();
if (benchSubscription) benchSubscription.stop();
if (accountSubscription) accountSubscription.stop();
if (accountSubscription) accountSubscription.stop();
userModules.stop();
userModules.stop();
@@ -88,7 +88,7 @@ function createAiReadiness({ db, registry, pool, keys, runs, release = null }) {
@@ -88,7 +88,7 @@ function createAiReadiness({ db, registry, pool, keys, runs, release = null }) {
}
}
/** AI gauges on the openvibe-shared/metrics registry. */
/** AI gauges on the openvibe-shared/metrics registry. */
-function registerAiGauges(registry, { runs, registry: ai, pool }) {
+function registerAiGauges(registry, { runs, registry: ai, pool, balance = null }) {
registry.gauge({
registry.gauge({
name: 'ai_runs', help: 'Runs waiting for a slot (queued) and holding one (running)', labelNames: ['state'],
name: 'ai_runs', help: 'Runs waiting for a slot (queued) and holding one (running)', labelNames: ['state'],
collect: () => {
collect: () => {
@@ -106,6 +106,11 @@ function registerAiGauges(registry, { runs, registry: ai, pool }) {
@@ -106,6 +106,11 @@ function registerAiGauges(registry, { runs, registry: ai, pool }) {
name: 'ai_provider_credentials_missing', help: 'Active providers whose secret reference does not resolve (1 = missing)', labelNames: ['provider'],
name: 'ai_provider_credentials_missing', help: 'Active providers whose secret reference does not resolve (1 = missing)', labelNames: ['provider'],
collect: async () => (await providerView(ai, pool)).filter(p => p.credentials !== 'not_required').map(p => ({ labels: { provider: p.key }, value: p.credentials === 'missing' ? 1 : 0 })),
collect: async () => (await providerView(ai, pool)).filter(p => p.credentials !== 'not_required').map(p => ({ labels: { provider: p.key }, value: p.credentials === 'missing' ? 1 : 0 })),
});
});
+ // The hourly balance probe's last reading (server/providers/balance.js), never a live call on scrape.
+ registry.gauge({
+ name: 'ai_provider_balance_usd', help: 'Provider account balance in USD, as last read by the hourly probe', labelNames: ['provider'],
+ collect: () => (balance ? balance.values() : []).map(r => ({ labels: { provider: r.provider }, value: r.usd })),
+ });
}
}
module.exports = { createAiReadiness, registerAiGauges };
module.exports = { createAiReadiness, registerAiGauges };
new file mode 100644
new file mode 100644
@@ -0,0 +1,116 @@
@@ -0,0 +1,116 @@
+'use strict';
+/**
+ * Hourly provider balance probe (Track O): a provider that publishes its account balance is asked what is
+ * left, so the operator hears about an empty account before every AI feature on the network fails.
+ *
+ * DeepSeek GET https://api.deepseek.com/user/balance
+ * { "is_available": true, "balance_infos": [ { "currency": "USD", "total_balance": "0.05", ... } ] }
+ *
+ * The reading is exposed as ai_provider_balance_usd{provider} on GET /metrics (server/observability.js) and
+ * the operator alert fires below $2 (OpenVibe.Host deploy/prometheus/openvibe-rules.yml). The key is read by
+ * reference and only ever sent in the request's Authorization header: it is never logged, returned or stored.
+ * Off when AI_BALANCE_PROBE=0 or under a restore drill (AI_DRILL=1; server/config.js balance.enabled).
+ */
+const { resolveSecret } = require('../util');
+
+const HOST = /(^|\.)api\.deepseek\.com$/i;
+
+/** DeepSeek's answer: the USD total when it reports one (the gauge is dollars), else null. */
+function deepseekUsd(j) {
+ const infos = Array.isArray(j && j.balance_infos) ? j.balance_infos : [];
+ const usd = infos.find((x) => x && String(x.currency || '').toUpperCase() === 'USD');
+ const n = usd ? Number(usd.total_balance) : NaN;
+ return Number.isFinite(n) ? n : null;
+}
+
+function createBalanceProbe({ registry, config, env = process.env, fetchImpl = globalThis.fetch, clock = { now: () => Date.now() }, log = console }) {
+ const latest = new Map(); // provider key -> { usd, at, currency }
+ let timer = null;
+ let running = null;
+
+ /** A provider record with a balance API, or null: DeepSeek today. */
+ function apiOf(p) {
+ try {
+ const u = new URL(String(p && p.base_url || ''));
+ return HOST.test(u.hostname) ? { origin: u.origin, path: '/user/balance' } : null;
+ } catch { return null; }
+ }
+
+ async function read(p) {
+ const api = apiOf(p);
+ if (!api) return null;
+ const key = resolveSecret(p.secret_ref, env);
+ if (!key) return null;
+ let res;
+ try {
+ res = await fetchImpl(`${api.origin}${api.path}`, {
+ headers: { Authorization: `Bearer ${key}`, Accept: 'application/json' },
+ signal: AbortSignal.timeout(config.balance.timeoutMs),
+ });
+ } catch (err) {
+ throw new Error(`balance API unreachable: ${err && err.message || err}`);
+ }
+ if (!res.ok) throw new Error(`balance API answered ${res.status}`);
+ const j = await res.json().catch(() => null);
+ const usd = deepseekUsd(j);
+ return usd == null ? null : { usd };
+ }
+
+ /** Every configured provider with a balance API, then the research key when no provider record carries it. */
+ async function targets() {
+ const out = [];
+ for (const p of await registry.listProviders()) {
+ if (p.status !== 'active') continue;
+ if (apiOf(p)) out.push(p);
+ }
+ return out;
+ }
+
+ async function poll() {
+ if (running) return running;
+ running = (async () => {
+ let found = 0;
+ for (const p of await targets()) {
+ found++;
+ try {
+ const r = await read(p);
+ if (r) latest.set(p.key, { usd: r.usd, at: clock.now(), currency: 'USD' });
+ } catch (err) {
+ log.warn(`[ai] balance probe ${p.key}: ${String(err && err.message || err).slice(0, 200)}`);
+ }
+ }
+ // The research key can be the only DeepSeek credential (AI_RESEARCH_DEEPSEEK_KEY_REF) when the shared
+ // provider is something else: still expose its balance under the label `deepseek`.
+ if (!found && config.research && config.research.deepseek && resolveSecret(config.research.deepseekKeyRef, env)) {
+ try {
+ const r = await read({ base_url: config.research.deepseekUrl, secret_ref: config.research.deepseekKeyRef });
+ if (r) latest.set('deepseek', { usd: r.usd, at: clock.now(), currency: 'USD' });
+ } catch (err) {
+ log.warn(`[ai] balance probe deepseek: ${String(err && err.message || err).slice(0, 200)}`);
+ }
+ }
+ return values();
+ })().finally(() => { running = null; });
+ return running;
+ }
+
+ function values() { return [...latest.entries()].map(([provider, r]) => ({ provider, usd: r.usd, at: r.at })); }
+
+ function start() {
+ if (timer || !config.balance.enabled) return;
+ const tick = () => { poll().catch((err) => log.warn(`[ai] balance probe: ${err && err.message}`)); };
+ const first = setTimeout(tick, 5000);
+ if (first.unref) first.unref();
+ timer = setInterval(tick, config.balance.intervalMs);
+ if (timer.unref) timer.unref();
+ }
+
+ function stop() {
+ if (timer) clearInterval(timer);
+ timer = null;
+ }
+
+ return { start, stop, poll, values, apiOf };
+}
+
+module.exports = { createBalanceProbe, deepseekUsd };
@@ -47,6 +47,25 @@ async function postJson(url, headers, body, { signal, timeoutMs = 30000, fetchIm
@@ -47,6 +47,25 @@ async function postJson(url, headers, body, { signal, timeoutMs = 30000, fetchIm
return j;
return j;
}
}
+/**
+ * The provider's own answer that its account is out of money: DeepSeek answers HTTP 402 with
+ * `{ "error": { "message": "Insufficient Balance" } }`; OpenAI-shaped gateways use 429/403 with
+ * `insufficient_quota`, or name a credit balance. This is not the caller's fault and not a transient
+ * failure: the provider is exhausted until someone tops it up (server/providers/index.js opens its
+ * circuit for the long cooldown and never retries into it).
+ */
+const BALANCE_RE = /insufficient\s+(balance|quota|credit|credits|funds)|out\s+of\s+(balance|credit|credits|funds)|credit\s+balance\s+is\s+too\s+low|balance\s+(is\s+)?(too\s+)?low|exceeded\s+your\s+current\s+quota|billing\s+(hard\s+)?limit/i;
+function insufficientBalance(err) {
+ if (!err) return false;
+ if (err.status === 402) return true;
+ const body = err.body && typeof err.body === 'object' ? err.body : {};
+ const inner = body.error && typeof body.error === 'object' ? body.error : body;
+ const code = String((inner && (inner.code || inner.type)) || err.code || '');
+ const message = String((inner && inner.message) || body.message || err.message || '');
+ if ((err.status === 403 || err.status === 429) && /insufficient_quota|billing/i.test(`${code} ${message}`)) return true;
+ return BALANCE_RE.test(`${code} ${message}`);
+}
+
/** Transient failures get one more try on the same provider (Live's rule). */
/** Transient failures get one more try on the same provider (Live's rule). */
function retryable(e) {
function retryable(e) {
if (!e) return false;
if (!e) return false;
@@ -85,4 +104,4 @@ function schemaInstruction(json) {
@@ -85,4 +104,4 @@ function schemaInstruction(json) {
}
}
module.exports = {
module.exports = {
- schemaInstruction, ProviderError, deadline, describeAbort, postJson, retryable, approxTokens, estimateUsage, sleep };
+ schemaInstruction, ProviderError, deadline, describeAbort, postJson, retryable, approxTokens, estimateUsage, sleep, insufficientBalance };
@@ -18,12 +18,16 @@
@@ -18,12 +18,16 @@
* reasons, selected, candidates with their excluded reasons) rides every run response.
* 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. A provider's own "insufficient balance"
+ * answer opens it at once for the long exhausted cooldown, is never retried, and the router fails over.
+ * When nothing can serve because offers are exhausted or paused, the route answers 503 ai.resting with a
+ * human sentence and Retry-After. The host's local model is a last offer for short text only, one request at
+ * a time, labelled local on the receipt.
*/
*/
const placement = require('openvibe-sdk/placement');
const placement = require('openvibe-sdk/placement');
const contracts = require('openvibe-contracts');
const contracts = require('openvibe-contracts');
const { AiError, resolveSecret, sha256 } = require('../util');
const { AiError, resolveSecret, sha256 } = require('../util');
-const { retryable, sleep, ProviderError, schemaInstruction } = require('./common');
+const { retryable, sleep, ProviderError, schemaInstruction, insufficientBalance } = require('./common');
const { createStubProvider } = require('./stub');
const { createStubProvider } = require('./stub');
const { createOpenAiProvider } = require('./openai');
const { createOpenAiProvider } = require('./openai');
const { createResponsesProvider } = require('./responses');
const { createResponsesProvider } = require('./responses');
@@ -54,7 +58,8 @@ async function policyAllows(registry, provider, modelId, policy = 'standard') {
@@ -54,7 +58,8 @@ async function policyAllows(registry, provider, modelId, policy = 'standard') {
/** 4xx that the request itself caused (bad image, schema, max_tokens...), as opposed to auth, rate or availability. */
/** 4xx that the request itself caused (bad image, schema, max_tokens...), as opposed to auth, rate or availability. */
function callerFault(err) {
function callerFault(err) {
const st = err && err.status;
const st = err && err.status;
- return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 403, 404, 408, 429].includes(st);
+ // 402 payment required is the provider's account, not the caller's request: it opens the circuit (exhausted).
+ return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 402, 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. */
/** An upstream 429 is rate-limit capacity, not provider ill health: it shifts to the next candidate but must not open the circuit. */
@@ -204,6 +209,11 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -204,6 +209,11 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
const machines = machineModels || require('../machines').createMachineModels({ config, env, fetchImpl });
const machines = machineModels || require('../machines').createMachineModels({ config, env, fetchImpl });
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)
+ // One local generation at a time (AI_LOCAL_LLM_CONCURRENCY, default 1): the host's model shares its CPU
+ // with live streams, so a paid outage must not turn into a local-model stampede.
+ const localLimit = Math.max(1, (config.localLlm && config.localLlm.concurrency) || 1);
+ let localActive = 0;
+ const exhaustedCooldownMs = () => Math.max(60000, Number(config.breaker.exhaustedCooldownMs) || 10 * 60000);
// L36 Y7: the provider that last answered this prompt, for five minutes (server/providers/warm-cache.js).
// L36 Y7: the provider that last answered this prompt, for five minutes (server/providers/warm-cache.js).
const warm = createWarmCache({ now: () => clock.now() });
const warm = createWarmCache({ now: () => clock.now() });
// AI Exchange P7: measured per-deployment evidence for alias routing. The feature flag is read live (D11
// AI Exchange P7: measured per-deployment evidence for alias routing. The feature flag is read live (D11
@@ -265,26 +275,35 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -265,26 +275,35 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
// ── Circuit breaker ────────────────────────────────────
// ── Circuit breaker ────────────────────────────────────
const getHealth = db.prepare('SELECT * FROM provider_health WHERE provider_key = ?');
const getHealth = db.prepare('SELECT * FROM provider_health WHERE provider_key = ?');
- const putHealth = db.prepare(`INSERT INTO provider_health (provider_key, state, consecutive_failures, opened_at, last_error, last_success_at, last_failure_at)
- VALUES (@provider_key, @state, @consecutive_failures, @opened_at, @last_error, @last_success_at, @last_failure_at)
+ const putHealth = db.prepare(`INSERT INTO provider_health (provider_key, state, consecutive_failures, opened_at, exhausted_at, last_error, last_success_at, last_failure_at)
+ VALUES (@provider_key, @state, @consecutive_failures, @opened_at, @exhausted_at, @last_error, @last_success_at, @last_failure_at)
ON CONFLICT(provider_key) DO UPDATE SET state = excluded.state, consecutive_failures = excluded.consecutive_failures, opened_at = excluded.opened_at,
ON CONFLICT(provider_key) DO UPDATE SET state = excluded.state, consecutive_failures = excluded.consecutive_failures, opened_at = excluded.opened_at,
- last_error = excluded.last_error, last_success_at = excluded.last_success_at, last_failure_at = excluded.last_failure_at`);
+ exhausted_at = excluded.exhausted_at, last_error = excluded.last_error, last_success_at = excluded.last_success_at, last_failure_at = excluded.last_failure_at`);
async function health(key) {
async function health(key) {
- const h = await getHealth.get(key) || { provider_key: key, state: 'closed', consecutive_failures: 0, opened_at: null, last_error: null, last_success_at: null, last_failure_at: null };
- if (h.state === 'open' && clock.now() - (h.opened_at || 0) >= config.breaker.cooldownMs) return { ...h, state: 'half_open' };
+ const h = await getHealth.get(key) || { provider_key: key, state: 'closed', consecutive_failures: 0, opened_at: null, exhausted_at: null, last_error: null, last_success_at: null, last_failure_at: null };
+ if (h.state === 'open') {
+ // Out of balance: a long cooldown (10 minutes), so every request shares the same pause instead of each
+ // one retrying a provider that cannot answer until someone tops it up.
+ const cooldown = h.exhausted_at ? exhaustedCooldownMs() : config.breaker.cooldownMs;
+ if (clock.now() - (h.opened_at || 0) >= cooldown) return { ...h, state: 'half_open' };
+ }
return h;
return h;
}
}
async function recordSuccess(key) {
async function recordSuccess(key) {
const h = await health(key);
const h = await health(key);
if (h.state !== 'closed') await registry.audit('system', 'provider.circuit_closed', 'provider', key, { metadata: { after_failures: h.consecutive_failures } });
if (h.state !== 'closed') await registry.audit('system', 'provider.circuit_closed', 'provider', key, { metadata: { after_failures: h.consecutive_failures } });
- await putHealth.run({ ...h, provider_key: key, state: 'closed', consecutive_failures: 0, opened_at: null, last_success_at: clock.now() });
+ await putHealth.run({ ...h, provider_key: key, state: 'closed', consecutive_failures: 0, opened_at: null, exhausted_at: null, last_success_at: clock.now() });
}
}
async function recordFailure(key, err) {
async function recordFailure(key, err) {
const h = await health(key);
const h = await health(key);
const failures = (h.consecutive_failures || 0) + 1;
const failures = (h.consecutive_failures || 0) + 1;
- const open = h.state === 'half_open' || failures >= config.breaker.failureThreshold;
- if (open && h.state !== 'open') await registry.audit('system', 'provider.circuit_opened', 'provider', key, { metadata: { failures, error: String(err && err.message || err).slice(0, 200) } });
- await putHealth.run({ ...h, provider_key: key, state: open ? 'open' : 'closed', consecutive_failures: failures, opened_at: open ? clock.now() : h.opened_at, last_error: String(err && err.message || err).slice(0, 500), last_failure_at: clock.now() });
+ const exhausted = insufficientBalance(err);
+ // A balance answer opens the circuit at once (it will not recover on its own), not after failureThreshold.
+ const open = exhausted || h.state === 'half_open' || failures >= config.breaker.failureThreshold;
+ if (open && h.state !== 'open') await registry.audit('system', 'provider.circuit_opened', 'provider', key, { metadata: { failures, exhausted, error: String(err && err.message || err).slice(0, 200) } });
+ await putHealth.run({ ...h, provider_key: key, state: open ? 'open' : 'closed', consecutive_failures: failures,
+ opened_at: open ? clock.now() : h.opened_at, exhausted_at: exhausted ? clock.now() : (h.exhausted_at || null),
+ last_error: String(err && err.message || err).slice(0, 500), last_failure_at: clock.now() });
}
}
async function resetHealth(key) { await db.prepare('DELETE FROM provider_health WHERE provider_key = ?').run(key); }
async function resetHealth(key) { await db.prepare('DELETE FROM provider_health WHERE provider_key = ?').run(key); }
@@ -326,6 +345,30 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -326,6 +345,30 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
function isLocalProvider(p) {
function isLocalProvider(p) {
return p.kind === 'whisper' || Boolean(p.metadata && p.metadata.local);
return p.kind === 'whisper' || Boolean(p.metadata && p.metadata.local);
}
}
+ /** The host's text model (local / local2), not its local speech-to-text: the caps below are for text generation. */
+ function isLocalModel(p) {
+ return Boolean(p.metadata && p.metadata.local);
+ }
+ /**
+ * What the host's local model is offered: short text only (summaries, titles, moderation hints, a chat line),
+ * never code or long reasoning. Routes that write whole apps or wiki pages (max_output_tokens well over a chat
+ * answer, or outlast_provider_timeout), and an OpenAI completion carrying tools (an agent's loop), are not for
+ * it; embeddings are small and stay welcome, and speech-to-text is a different local provider.
+ */
+ function localCanServe(route, operation, req = null) {
+ if (operation === 'embed' || operation === 'transcribe') return true;
+ if (!['chat', 'generate', 'summarize', 'classify', 'extract', 'enrich', 'complete', 'messages'].includes(operation)) return false;
+ // A request carrying tools is an agent's loop, not a short chat answer.
+ if (operation === 'complete' && req && req.openai && Array.isArray(req.openai.tools) && req.openai.tools.length) return false;
+ if (operation === 'messages' && req && Array.isArray(req.tools) && req.tools.length) return false;
+ if (route && route.options && route.options.outlast_provider_timeout) return false;
+ // The answer this request actually asked for decides (engine.js puts input.max_tokens on req); a route's own
+ // cap is the fallback. Over 800 tokens is long reasoning, not a short text answer.
+ const wanted = Number(req && req.maxTokens);
+ if (Number.isFinite(wanted)) return wanted <= 800;
+ const max = Number(route && route.max_output_tokens);
+ return !Number.isFinite(max) || max <= 800;
+ }
/**
/**
* Billing profiles whose capacity is already paid for: a subscription, a free server or a person's own key.
* Billing profiles whose capacity is already paid for: a subscription, a free server or a person's own key.
* They carry no per-token rate card and are offered as prepaid. The profile is the providers.billing_profile
* They carry no per-token rate card and are offered as prepaid. The profile is the providers.billing_profile
@@ -704,7 +747,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -704,7 +747,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
if (p.status !== 'active') return 'disabled';
if (p.status !== 'active') return 'disabled';
if (p.auth_mode !== 'none' && !credentialNotRequired(p) && !resolveSecret(p.secret_ref, env)) return 'no_credentials';
if (p.auth_mode !== 'none' && !credentialNotRequired(p) && !resolveSecret(p.secret_ref, env)) return 'no_credentials';
if (features.some(f => !a.supports(f))) return 'unsupported';
if (features.some(f => !a.supports(f))) return 'unsupported';
- if ((await health(p.key)).state === 'open') return 'circuit_open';
+ const h = await health(p.key);
+ if (h.state === 'open') return h.exhausted_at ? 'exhausted' : 'circuit_open';
return null;
return null;
}
}
@@ -937,6 +981,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -937,6 +981,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
const preferState = prefer ? { provider: prefer.provider, model: prefer.model, honoured: false, why: '' } : null;
const preferState = prefer ? { provider: prefer.provider, model: prefer.model, honoured: false, why: '' } : null;
const tried = [];
const tried = [];
let unreadable = null; // a candidate could not read the caller's media (whisper's ffmpeg): no other will either
let unreadable = null; // a candidate could not read the caller's media (whisper's ffmpeg): no other will either
+ let resting = false; // a paid provider is out of balance, or every remaining offer is paused: answer ai.resting
+
// A streamed answer (/v1/chat/completions, stream: true) is the caller's once a delta has gone out: a failure
// A streamed answer (/v1/chat/completions, stream: true) is the caller's once a delta has gone out: a failure
// after that is final, never retried or handed to a fallback whose words would follow the first one's.
// after that is final, never retried or handed to a fallback whose words would follow the first one's.
let streamed = false;
let streamed = false;
@@ -1245,15 +1291,39 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -1245,15 +1291,39 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
}
}
const why = await skipReason(p, a, features);
const why = await skipReason(p, a, features);
if (why) {
if (why) {
+ if (why === 'exhausted' || why === 'circuit_open') resting = true;
statFor(c.provider).skipped++;
statFor(c.provider).skipped++;
tried.push({ provider: c.provider, skipped: why });
tried.push({ provider: c.provider, skipped: why });
attempts.push({ key: 'openvibe', provider: c.provider, model, why: chosen || 'the router chose it', status: 'skipped', error: why });
attempts.push({ key: 'openvibe', provider: c.provider, model, why: chosen || 'the router chose it', status: 'skipped', error: why });
if (ctx.logRequest) ctx.logRequest({ ...base, status: 'skipped', skip_reason: why });
if (ctx.logRequest) ctx.logRequest({ ...base, status: 'skipped', skip_reason: why });
continue;
continue;
}
}
+ // The host's local model is the last offer, and only for short text: never code or long reasoning, and
+ // one request at a time (AI_LOCAL_LLM_CONCURRENCY) so a paid outage cannot starve the streams it shares
+ // the machine with. Its answer is labelled local on the run's receipt (provider_key = local / local2).
+ const local = isLocalModel(p);
+ if (local && !localCanServe(route, operation, req)) {
+ statFor(c.provider).skipped++;
+ tried.push({ provider: c.provider, skipped: 'unsupported' });
+ attempts.push({ key: 'openvibe', provider: c.provider, model, why: chosen || 'the router chose it', status: 'skipped', error: 'the local model is not offered this job' });
+ if (ctx.logRequest) ctx.logRequest({ ...base, status: 'skipped', skip_reason: 'unsupported' });
+ continue;
+ }
+ if (local && localActive >= localLimit) {
+ resting = true;
+ statFor(c.provider).skipped++;
+ tried.push({ provider: c.provider, skipped: 'busy' });
+ attempts.push({ key: 'openvibe', provider: c.provider, model, why: chosen || 'the router chose it', status: 'skipped', error: 'the local model is busy with another answer' });
+ if (ctx.logRequest) ctx.logRequest({ ...base, status: 'skipped', skip_reason: 'busy' });
+ continue;
+ }
// L36 Y7: the fallback chain stops after three providers.
// L36 Y7: the fallback chain stops after three providers.
if (attemptsMade >= maxAttempts) break;
if (attemptsMade >= maxAttempts) break;
attemptsMade += 1;
attemptsMade += 1;
+ // Claim the local slot synchronously, before any await, so two requests cannot both pass the check.
+ if (local) localActive++;
+ let lastErr = null;
+ try {
// An image generation outlasts the shared provider's chat timeout (AI_PROVIDER_TIMEOUT_MS): the route's wins. So does a
// An image generation outlasts the shared provider's chat timeout (AI_PROVIDER_TIMEOUT_MS): the route's wins. So does a
// route that says so (options.outlast_provider_timeout: space.build writes a whole app, a minute or two). Otherwise
// route that says so (options.outlast_provider_timeout: space.build writes a whole app, a minute or two). Otherwise
// the provider's timeout stays strict, so a slow provider fails over quickly.
// the provider's timeout stays strict, so a slow provider fails over quickly.
@@ -1262,7 +1332,6 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -1262,7 +1332,6 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
const cap = outlast ? Math.max(own, Number(route.timeout_ms) || 0, operation === 'image' ? 600000 : 0) : (own || 600000);
const cap = outlast ? Math.max(own, Number(route.timeout_ms) || 0, operation === 'image' ? 600000 : 0) : (own || 600000);
const timeoutMs = Math.max(1000, Math.min(req.timeoutMs || route.timeout_ms || own || DEFAULT_PROVIDER_TIMEOUT_MS, cap));
const timeoutMs = Math.max(1000, Math.min(req.timeoutMs || route.timeout_ms || own || DEFAULT_PROVIDER_TIMEOUT_MS, cap));
const retries = config.providerRetries;
const retries = config.providerRetries;
- let lastErr = null;
for (let attempt = 0; attempt <= retries; attempt++) {
for (let attempt = 0; attempt <= retries; attempt++) {
if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
const started = clock.now();
const started = clock.now();
@@ -1291,7 +1360,12 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -1291,7 +1360,12 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
}
}
// max_cost_usd, when it binds, comes back as the affordable output cap for this attempt.
// max_cost_usd, when it binds, comes back as the affordable output cap for this attempt.
const callMax = routed && Number.isSafeInteger(routed.maxTokens) && routed.maxTokens >= 1 && routed.maxTokens !== req.maxTokens ? routed.maxTokens : null;
const callMax = routed && Number.isSafeInteger(routed.maxTokens) && routed.maxTokens >= 1 && routed.maxTokens !== req.maxTokens ? routed.maxTokens : null;
- const callReq = callMax == null ? req : { ...req, maxTokens: callMax };
+ let callReq = callMax == null ? req : { ...req, maxTokens: callMax };
+ // The host's model writes a short answer: its own cap, never more, whatever the route asked for.
+ if (local && operation !== 'embed') {
+ const localMax = Math.max(1, (config.localLlm && config.localLlm.maxTokens) || 512);
+ callReq = { ...callReq, maxTokens: Math.min(Number(callReq.maxTokens) || localMax, localMax) };
+ }
try {
try {
const fn = a[operation];
const fn = a[operation];
if (typeof fn !== 'function') throw new ProviderError(`${operation} not implemented`, { code: 'provider.unsupported' });
if (typeof fn !== 'function') throw new ProviderError(`${operation} not implemented`, { code: 'provider.unsupported' });
@@ -1335,7 +1409,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -1335,7 +1409,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
warm.remember(warmSubject(ctx, route), leadingText(req), c.provider, model);
warm.remember(warmSubject(ctx, route), leadingText(req), c.provider, model);
explain.attempts = attempts;
explain.attempts = attempts;
explain.openvibe = attemptFacts;
explain.openvibe = attemptFacts;
- return { result, provider: c.provider, providerKind: p.kind, model: result.model || model, fallbackUsed: fallback || selfAttempts.some(a => a.status !== 'skipped'), usage, cost, latencyMs: latency, ttftMs, synthetic: Boolean(result.synthetic || a.synthetic), tried, startedAt: started, explain };
+ if (local) explain.provider_local = true; // the Exchange receipt labels the offer local (server/exchange/receipts.js)
+ return { result, provider: c.provider, providerKind: p.kind, local, model: result.model || model, fallbackUsed: fallback || selfAttempts.some(a => a.status !== 'skipped'), usage, cost, latencyMs: latency, ttftMs, 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) {
@@ -1357,15 +1432,21 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -1357,15 +1432,21 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
if (streamed && err && err.code === 'policy.floor_refused') throw new AiError(403, 'policy.floor_refused', err.detail || err.message, { rule: err.rule, policy: contentPolicy });
if (streamed && err && err.code === 'policy.floor_refused') throw new AiError(403, 'policy.floor_refused', err.detail || err.message, { rule: err.rule, policy: contentPolicy });
if (streamed) throw new AiError(502, 'provider.stream_broken', `the answer broke off after it began streaming: ${String(err && err.message || err).slice(0, 200)}`, { tried: [...tried, { provider: c.provider, error: String(err && err.message || err).slice(0, 200) }], explain });
if (streamed) throw new AiError(502, 'provider.stream_broken', `the answer broke off after it began streaming: ${String(err && err.message || err).slice(0, 200)}`, { tried: [...tried, { provider: c.provider, error: String(err && err.message || err).slice(0, 200) }], explain });
// A timed-out image may still be generated and billed upstream: never pay for it twice.
// A timed-out image may still be generated and billed upstream: never pay for it twice.
- if (attempt < retries && retryable(err) && !(operation === 'image' && err.code === 'provider.timeout')) { await sleep(config.retryDelayMs + Math.floor(Math.random() * Math.min(700, config.retryDelayMs)), ctx.signal).catch(() => {}); continue; }
+ // An insufficient-balance answer is never retried: the same provider cannot answer until it is topped up.
+ if (insufficientBalance(err)) resting = true;
+ if (attempt < retries && !insufficientBalance(err) && retryable(err) && !(operation === 'image' && err.code === 'provider.timeout')) { await sleep(config.retryDelayMs + Math.floor(Math.random() * Math.min(700, config.retryDelayMs)), ctx.signal).catch(() => {}); continue; }
break;
break;
}
}
}
}
+ } finally {
+ if (local) localActive--;
+ }
// 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; a 429 is capacity, not health either, and moves on without
// 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.
// opening the circuit. Neither must take a shared provider away from other callers.
if (lastErr && lastErr.code === 'media.unreadable') unreadable = lastErr;
if (lastErr && lastErr.code === 'media.unreadable') unreadable = lastErr;
- if (!callerFault(lastErr) && !rateLimited(lastErr)) await recordFailure(c.provider, lastErr);
+ // An insufficient-balance 429 is not ordinary capacity: it must open the long exhausted circuit.
+ if (insufficientBalance(lastErr) || (!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) });
attempts.push({ key: 'openvibe', provider: c.provider, model: lastErr && lastErr.model || model, why: chosen || (fallback ? 'the next network provider after an earlier attempt failed' : 'the first network offer for this route'), status: lastErr && lastErr.code === 'provider.timeout' ? 'timeout' : 'failed', error: String(lastErr && lastErr.message || lastErr).slice(0, 300) });
attempts.push({ key: 'openvibe', provider: c.provider, model: lastErr && lastErr.model || model, why: chosen || (fallback ? 'the next network provider after an earlier attempt failed' : 'the first network offer for this route'), status: lastErr && lastErr.code === 'provider.timeout' ? 'timeout' : 'failed', error: String(lastErr && lastErr.message || lastErr).slice(0, 300) });
log.warn(`[ai] ${operation} via ${c.provider}/${model || '-'} failed: ${lastErr && lastErr.message}`);
log.warn(`[ai] ${operation} via ${c.provider}/${model || '-'} failed: ${lastErr && lastErr.message}`);
@@ -1379,6 +1460,10 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -1379,6 +1460,10 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
explain.reasons = [...(explain.reasons || []), 'the media could not be read'];
explain.reasons = [...(explain.reasons || []), 'the media could not be read'];
throw new AiError(422, 'media.unreadable', String(unreadable.message || 'the media could not be read').slice(0, 300), { tried, explain });
throw new AiError(422, 'media.unreadable', String(unreadable.message || 'the media could not be read').slice(0, 300), { tried, explain });
}
}
+ if (resting) {
+ explain.reasons = [...(explain.reasons || []), 'a provider is out of balance or paused; AI is resting'];
+ throw new AiError(503, 'ai.resting', 'AI is resting for a moment', { retry_after_seconds: Math.ceil(exhaustedCooldownMs() / 1000), tried, explain });
+ }
// The free allowance / anonymous Try it had capable offers, but every one was above its ceiling. That is
// The free allowance / anonymous Try it had capable offers, but every one was above its ceiling. That is
// not an outage: the caller needs their own key, or their own token so their Balance can pay.
// not an outage: the caller needs their own key, or their own token so their Balance can pay.
if (freeCeilingSkipped && attemptsMade === 0) {
if (freeCeilingSkipped && attemptsMade === 0) {
@@ -827,7 +827,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -827,7 +827,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
detail = `The free AI allowance for ${ctx.personSubject} is spent, and the calling service has no budget to cover this run. Add a key for the person, or wait for the allowance to reset.`;
detail = `The free AI allowance for ${ctx.personSubject} is spent, and the calling service has no budget to cover this run. Add a key for the person, or wait for the allowance to reset.`;
}
}
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;
+ // ai.resting's detail is the one sentence products show; its tries and Retry-After live in the problem body
+ // and the explain, never appended to that sentence.
+ const extra = err instanceof AiError && err.extra && err.code !== 'ai.resting' ? JSON.stringify(err.extra).slice(0, 2000) : null;
const explain = err instanceof AiError && err.extra && err.extra.explain ? JSON.stringify(err.extra.explain) : 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 = ?, explain = COALESCE(?, explain), finished_at = ? WHERE id = ? AND status IN ('queued', 'running')`)
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')`)
@@ -31,14 +31,14 @@ const PROBLEMS = { name: 'api', internalCode: 'ai.internal', internalDetail: 'in
@@ -31,14 +31,14 @@ const PROBLEMS = { name: 'api', internalCode: 'ai.internal', internalDetail: 'in
/**
/**
* Any error -> problem+json (openvibe-sdk/service sendError). Only an AiError answers with its own status,
* Any error -> problem+json (openvibe-sdk/service sendError). Only an AiError answers with its own status,
- * code and extra (extra.errors is the problem's errors list; a 429 with retry_after_seconds sets Retry-After).
- * Anything else is logged and answers 500 ai.internal: it is passed on as its stack alone, so a provider
- * error's upstream HTTP status never becomes AI's answer.
+ * code and extra (extra.errors is the problem's errors list; extra.retry_after_seconds sets Retry-After, for a
+ * 429 refusal and for a 503 ai.resting alike). Anything else is logged and answers 500 ai.internal: it is passed
+ * on as its stack alone, so a provider error's upstream HTTP status never becomes AI's answer.
*/
*/
function sendError(res, err, ctx, log = console) {
function sendError(res, err, ctx, log = console) {
const req = { ov: ctx };
const req = { ov: ctx };
if (!(err instanceof AiError)) return svc.sendError(res, req, { stack: (err && err.stack) || String(err) }, log, PROBLEMS);
if (!(err instanceof AiError)) return svc.sendError(res, req, { stack: (err && err.stack) || String(err) }, log, PROBLEMS);
- if (err.status === 429 && err.extra && err.extra.retry_after_seconds && !res.headersSent) res.setHeader('Retry-After', String(err.extra.retry_after_seconds));
+ if (err.extra && err.extra.retry_after_seconds && !res.headersSent) res.setHeader('Retry-After', String(err.extra.retry_after_seconds));
return svc.sendError(res, req, err, log, PROBLEMS);
return svc.sendError(res, req, err, log, PROBLEMS);
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,66 @@
@@ -0,0 +1,66 @@
+'use strict';
+// The hourly balance probe: a provider with a balance API (DeepSeek) is asked what its account holds, the
+// reading becomes ai_provider_balance_usd{provider}, and the key is only ever sent in Authorization — never
+// logged or exposed. Off under a drill (AI_DRILL=1 / AI_BALANCE_PROBE=0).
+const assert = require('assert');
+const { createRegistry } = require('openvibe-shared/metrics');
+const { createBalanceProbe, deepseekUsd } = require('../server/providers/balance');
+const { registerAiGauges } = require('../server/observability');
+const { load } = require('../server/config');
+const { suite } = require('./helpers');
+
+const t = suite('provider-balance');
+
+const record = (key, base_url, secret_ref = 'env:DEEPSEEK_API_KEY') => ({ key, status: 'active', base_url, secret_ref, kind: 'openai', metadata: {} });
+
+t.test('DeepSeek balance: USD total read, gauge exposed, key never logged', async () => {
+ const calls = [];
+ const fetchImpl = async (url, opts) => {
+ calls.push({ url, auth: opts.headers.Authorization });
+ return { ok: true, status: 200, json: async () => ({ is_available: true, balance_infos: [{ currency: 'USD', total_balance: '0.05', granted_balance: '0.00', topped_up_balance: '0.05' }] }) };
+ };
+ const warned = [];
+ const config = load({ NODE_ENV: 'test', AI_BALANCE_PROBE: '1' });
+ const probe = createBalanceProbe({
+ registry: { listProviders: async () => [record('shared', 'https://api.deepseek.com/v1'), record('other', 'https://api.openai.com/v1')] },
+ config, env: { DEEPSEEK_API_KEY: 'sk-deepseek-secret' }, fetchImpl,
+ clock: { now: () => 1762400000000 }, log: { warn: (m) => warned.push(m), log() {} },
+ });
+ const values = await probe.poll();
+ assert.deepStrictEqual(values, [{ provider: 'shared', usd: 0.05, at: 1762400000000 }]);
+ assert.strictEqual(calls.length, 1, 'only the provider whose host has a balance API is asked');
+ assert.strictEqual(calls[0].url, 'https://api.deepseek.com/user/balance');
+ assert.strictEqual(calls[0].auth, 'Bearer sk-deepseek-secret');
+ assert.ok(!JSON.stringify(warned).includes('sk-deepseek-secret'), 'the key is never in a log line');
+ assert.deepStrictEqual(deepseekUsd({ balance_infos: [{ currency: 'CNY', total_balance: '10' }] }), null, 'a non-USD balance is not read as dollars');
+
+ const metrics = createRegistry();
+ registerAiGauges(metrics, {
+ runs: { stats: () => ({ queued: 0, running: 0, max_concurrent: 1, max_queued: 10 }) },
+ registry: { listProviders: async () => [], publicProvider: (p) => p }, pool: { health: async () => ({ state: 'closed' }) },
+ balance: probe,
+ });
+ const text = await metrics.metricsAsync();
+ assert.ok(/ai_provider_balance_usd\{provider="shared"\} 0\.05\n/.test(text), text);
+});
+
+t.test('a failed probe keeps the last reading and does not fail the caller', async () => {
+ let status = 200;
+ const fetchImpl = async () => ({ ok: status === 200, status, json: async () => ({ balance_infos: [{ currency: 'USD', total_balance: '7.25' }] }) });
+ const probe = createBalanceProbe({
+ registry: { listProviders: async () => [record('shared', 'https://api.deepseek.com/v1')] },
+ config: load({ NODE_ENV: 'test' }), env: { DEEPSEEK_API_KEY: 'k' }, fetchImpl,
+ clock: { now: () => 1 }, log: { warn() {}, log() {} },
+ });
+ assert.strictEqual((await probe.poll())[0].usd, 7.25);
+ status = 503;
+ assert.strictEqual((await probe.poll())[0].usd, 7.25, 'the last good balance stays');
+});
+
+t.test('off under a drill or when AI_BALANCE_PROBE=0', () => {
+ assert.strictEqual(load({ NODE_ENV: 'test', AI_DRILL: '1' }).balance.enabled, false);
+ assert.strictEqual(load({ NODE_ENV: 'test', AI_BALANCE_PROBE: '0' }).balance.enabled, false);
+ assert.strictEqual(load({ NODE_ENV: 'test' }).balance.enabled, true);
+});
+
+t.run();
new file mode 100644
new file mode 100644
@@ -0,0 +1,141 @@
@@ -0,0 +1,141 @@
+'use strict';
+// A provider that answers "insufficient balance" is exhausted, not flaky: its circuit opens at once for the
+// long cooldown (10 minutes by default), shared by every request; it is never retried into; the route fails
+// over down its offer list. When nothing can answer, the problem is 503 ai.resting with a human sentence and
+// Retry-After, not a bare provider.unavailable.
+const assert = require('assert');
+const { boot, request, token, suite, seamServer, ALL } = require('./helpers');
+const { AiError, sendError } = require('../server/util');
+
+const t = suite('provider-exhausted');
+const tok = token('live', ALL);
+
+function fakeRes() {
+ const headers = {};
+ return { headers, statusCode: 0, body: null, headersSent: false, setHeader(k, v) { headers[k.toLowerCase()] = v; }, getHeader(k) { return headers[k.toLowerCase()]; }, end(s) { this.body = JSON.parse(s); this.headersSent = true; } };
+}
+
+t.test('402 Insufficient Balance opens the circuit at once, fails over, and never retries the same provider', async () => {
+ const paid = await seamServer(() => ({ status: 402, body: { error: { message: 'Insufficient Balance', type: 'unknown_error', param: null, code: 'invalid_request_error' } } }));
+ const local = await seamServer(() => ({ choices: [{ message: { content: 'a short local answer' } }], model: 'local-model', usage: { prompt_tokens: 4, completion_tokens: 3 } }));
+ let now = 1762400000000;
+ const clock = { now: () => now };
+ const h = await boot({ clock, env: {
+ AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '1', AI_BREAKER_FAILURES: '5', AI_BREAKER_EXHAUSTED_COOLDOWN_MS: '600000',
+ AI_PROVIDER: 'openai', AI_BASE_URL: `${paid.url}/v1`, AI_API_KEY: 'test-key', AI_MODEL: 'paid-model',
+ AI_LOCAL_LLM_URL: `${local.url}/v1`, AI_LOCAL_LLM_MODEL: 'local-model', AI_LOCAL_LLM_TIMEOUT_MS: '5000',
+ } });
+ const route = await h.registry.resolveRoute('default.chat');
+ const logs = [];
+ const ask = (signal) => h.pool.execute(route, 'chat', { messages: [{ role: 'user', content: 'summarise this' }], maxTokens: 60, timeoutMs: 3000 }, { routeKey: route.key, routeVersion: route.version, logRequest: (e) => logs.push(e) });
+
+ const first = await ask();
+ assert.strictEqual(first.provider, 'local', 'the exhausted paid offer fell over to the local model');
+ assert.strictEqual(first.local, true, 'the receipt labels a local answer');
+ assert.strictEqual(first.result.text, 'a short local answer');
+ assert.strictEqual(paid.calls, 1, 'the same provider is never retried after a balance answer (AI_PROVIDER_RETRIES=1)');
+ assert.strictEqual(local.calls, 1);
+ assert.strictEqual((await h.pool.health('shared')).state, 'open');
+ assert.ok((await h.pool.health('shared')).exhausted_at, 'the health row says why it opened');
+ assert.ok(logs.some((e) => e.provider_key === 'shared' && e.status === 'error'), 'the balance failure is logged');
+
+ // Every request shares the pause: the next one skips the exhausted provider entirely.
+ const second = await ask();
+ assert.strictEqual(second.provider, 'local');
+ assert.strictEqual(paid.calls, 1, 'a circuit-open exhausted provider is not called again');
+ assert.strictEqual((await h.pool.health('shared')).state, 'open');
+
+ // The fake clock passes the long cooldown: one half-open probe is allowed, and a second 402 opens it again.
+ now += 600001;
+ assert.strictEqual((await h.pool.health('shared')).state, 'half_open');
+ const third = await ask();
+ assert.strictEqual(third.provider, 'local');
+ assert.strictEqual(paid.calls, 2, 'exactly one probe after the cooldown, never a retry');
+ assert.strictEqual((await h.pool.health('shared')).state, 'open');
+
+ // With the local model failing too, nothing can serve: 503 ai.resting, with Retry-After on the problem.
+ local.handler = () => ({ status: 500, body: { error: { message: 'local is down' } } });
+ local.calls = 0;
+ let err = null;
+ try { await ask(); } catch (e) { err = e; }
+ assert.ok(err, 'nothing served');
+ assert.strictEqual(err.status, 503);
+ assert.strictEqual(err.code, 'ai.resting');
+ assert.strictEqual(err.detail, 'AI is resting for a moment');
+ assert.strictEqual(err.extra.retry_after_seconds, 600);
+ const res = fakeRes();
+ sendError(res, err, { requestId: 'req_1' }, { log() {}, warn() {}, error() {} });
+ assert.strictEqual(res.statusCode, 503);
+ assert.strictEqual(res.headers['retry-after'], '600');
+ assert.strictEqual(res.body.detail, 'AI is resting for a moment');
+
+ // The run API keeps that sentence clean (its tries/explain ride the explain, not the detail).
+ const run = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok, body: { workflow: 'ai.generate', input: { prompt: 'hi' }, options: { cache: false } } });
+ assert.strictEqual(run.body.run.status, 'failed', run.text);
+ assert.strictEqual(run.body.run.error.code, 'ai.resting', run.text);
+ assert.strictEqual(run.body.run.error.detail, 'AI is resting for a moment', run.text);
+ await h.stop(); await paid.close(); await local.close();
+});
+
+t.test('an OpenAI-style 429 insufficient_quota is exhausted too, not ordinary rate-limit capacity', async () => {
+ const paid = await seamServer(() => ({ status: 429, body: { error: { message: 'You exceeded your current quota', type: 'insufficient_quota', code: 'insufficient_quota' } } }));
+ const local = await seamServer(() => ({ choices: [{ message: { content: 'local' } }], model: 'local-model', usage: { prompt_tokens: 1, completion_tokens: 1 } }));
+ const h = await boot({ env: {
+ AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '1', AI_BREAKER_FAILURES: '5',
+ AI_PROVIDER: 'openai', AI_BASE_URL: `${paid.url}/v1`, AI_API_KEY: 'test-key', AI_MODEL: 'paid-model',
+ AI_LOCAL_LLM_URL: `${local.url}/v1`, AI_LOCAL_LLM_MODEL: 'local-model', AI_LOCAL_LLM_TIMEOUT_MS: '5000',
+ } });
+ const route = await h.registry.resolveRoute('default.chat');
+ const out = await h.pool.execute(route, 'chat', { messages: [{ role: 'user', content: 'hi' }], maxTokens: 30, timeoutMs: 3000 }, { routeKey: route.key });
+ assert.strictEqual(out.provider, 'local');
+ assert.strictEqual(paid.calls, 1, 'not retried and not treated as ordinary 429 capacity');
+ const health = await h.pool.health('shared');
+ assert.strictEqual(health.state, 'open');
+ assert.ok(health.exhausted_at);
+ await h.stop(); await paid.close(); await local.close();
+});
+
+t.test('the local model is short text only, capped, and one request at a time', async () => {
+ const paid = await seamServer(() => ({ status: 500, body: { error: { message: 'paid is down' } } }));
+ let localCalls = 0;
+ const local = await seamServer(async (body) => {
+ localCalls++;
+ await new Promise((r) => setTimeout(r, 250));
+ return { choices: [{ message: { content: 'short' } }], model: 'local-model', usage: { prompt_tokens: 3, completion_tokens: 2 } };
+ });
+ const h = await boot({ env: {
+ AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '50',
+ AI_PROVIDER: 'openai', AI_BASE_URL: `${paid.url}/v1`, AI_API_KEY: 'test-key', AI_MODEL: 'paid-model',
+ AI_LOCAL_LLM_URL: `${local.url}/v1`, AI_LOCAL_LLM_MODEL: 'local-model', AI_LOCAL_LLM_TIMEOUT_MS: '5000',
+ AI_LOCAL_LLM_MAX_TOKENS: '256', AI_LOCAL_LLM_CONCURRENCY: '1',
+ } });
+ const route = await h.registry.resolveRoute('default.chat');
+ const req = { messages: [{ role: 'user', content: 'summarise' }], maxTokens: 600, timeoutMs: 3000 };
+
+ const one = await h.pool.execute(route, 'chat', req, { routeKey: route.key });
+ assert.strictEqual(one.provider, 'local');
+ assert.strictEqual(local.last.max_tokens, 256, 'the local answer is capped at AI_LOCAL_LLM_MAX_TOKENS, whatever the route asked');
+
+ // Two at once: one gets the local model, the other is told AI is resting instead of a second local generation.
+ const results = await Promise.allSettled([
+ h.pool.execute(route, 'chat', req, { routeKey: route.key }),
+ h.pool.execute(route, 'chat', req, { routeKey: route.key }),
+ ]);
+ const ok = results.filter((r) => r.status === 'fulfilled');
+ const bad = results.filter((r) => r.status === 'rejected');
+ assert.strictEqual(ok.length, 1, 'one concurrent local request at a time');
+ assert.strictEqual(bad.length, 1);
+ assert.strictEqual(bad[0].reason.code, 'ai.resting', bad[0].reason.message);
+ assert.strictEqual(localCalls, 2, 'the second request never reached the local model');
+
+ // Code and long reasoning are never sent to the host's model.
+ const before = localCalls;
+ const longRoute = { ...route, max_output_tokens: 4000 };
+ let longErr = null;
+ try { await h.pool.execute(longRoute, 'chat', { ...req, maxTokens: 4000 }, { routeKey: route.key }); } catch (e) { longErr = e; }
+ assert.ok(longErr, 'nothing served the long route');
+ assert.strictEqual(localCalls, before, 'the local model was not offered a long answer');
+ await h.stop(); await paid.close(); await local.close();
+});
+
+t.run();