openvibers/openvibe.ai Commits

Provider failure classes, per-offer circuits and measured catalog status (#420)

OP OpenVibers committed 3 days ago · c9d03a0278 · 19 files, +928 −86

* Classify provider failures and add per-offer circuits

* Do not invent a status_since for an up offer

* Describe the per-offer breaker in the pool header

* Claim the owner alert before sending it

* Record the failure class for a person's own key too

* A half-open offer is degraded; the open gauge is 1 only while open

* Say the open gauge counts open circuits only

* Merge remote-tracking branch 'origin/main' into feat/provider-circuits-1109

* Renumber the circuit migration after main took 0057

* Keep the market fixture test in step with the price book's sources

* Merge main and preserve provider circuit changes

* Sequence provider circuits migration after effort defaults

* Merge current Exchange main into provider circuits

* Merge traffic share main and sequence circuits migration

* Merge remote-tracking branch 'origin/main' into job/ship2-ai411

Changes

ddiff --git a/.env.example b/.env.example
iindex 7623de5..d23ecb7 100644
---- a/.env.example
++++ b/.env.example
@@@ -289,10 +289,17 @@ AI_MAX_COST_USD_PER_DAY=0
# Raw prompt/response logging: opt-in debugging only, and only for runs whose caller sets options.debug
#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
++# The pause after a run of transient failures (timeouts, 5xx, a broken stream): 30 seconds, doubling on every
++# failed probe up to 10 minutes. Per provider+model (migrations/0060), so another offer still answers.
++#AI_BREAKER_BACKOFF_MS=30000
++#AI_BREAKER_MAX_BACKOFF_MS=600000
++# A provider that answered "insufficient balance" (HTTP 402, or a 429/403 insufficient_quota) or refused its key
++# (401/403) opens that offer's circuit at once. This long after it opened (15 minutes) the next real request may
++# probe it once; it stays open until a probe succeeds or the provider's key changes.
++#AI_BREAKER_EXHAUSTED_COOLDOWN_MS=900000
++# The owner's Network subject (usr_…): one notification per open balance/auth circuit, through the Network
++# notification push. Unset means no push; the console, the catalogue and the metrics still say what happened.
++#AI_OWNER_SUBJECT=
#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;
ddiff --git a/README.md b/README.md
iindex 7190607..905e023 100644
---- a/README.md
++++ b/README.md
@@@ -159,7 +159,8 @@ have no CORS access. Service callers do not need CORS.
`/api/ready` (openvibe-shared/ready) is 503 when the database, the active workflows or the Network
key fail; provider configuration (an active provider without its credentials, or with an open
circuit) degrades it. `/metrics` (openvibe-shared/metrics) carries golden signals by route template,
--`ai_runs{state="queued"|"running"}` and `ai_provider_circuit{provider,state}`.
++`ai_runs{state="queued"|"running"}`, `ai_provider_circuit{provider,state}` and the measured
++`ai_provider_circuit_open{provider,model,class}` and `ai_provider_failures_total{provider,class}`.
Each HTTP request is aggregated by route template, method and status class, never by raw URL; `/shared`
files, static assets, health, readiness and `/metrics` are not counted. Once a minute one `http.request`
@@@ -519,7 +520,7 @@ never a value.
| Group | Table(s) |
|---|---|
--| Providers | `providers` (+ `provider_health`: circuit breaker) — secrets only as `env:NAME` references |
++| Providers | `providers`, `provider_circuits` (per provider+model breaker), `provider_failure_counts`, `provider_health` (roll-up) — secrets only as `env:NAME` references |
| Models | `models` (type, limits, cost per Mtok, JSON/tools/streaming/vision support) |
| Routing profiles | `routes` — versioned; primary + fallbacks, options, output limit, format, timeout, explicit aliases |
| Prompt templates | `templates` — versioned; input/output schema, system/user prompt, default route, owner, visibility, `draft\|active\|deprecated\|archived` |
@@@ -554,9 +555,15 @@ 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
provider that is disabled, missing credentials, circuit-open or unable to do the operation is
**skipped and logged**; every call has a strict timeout, cancellation and one retry on a transient
--failure. A 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
++failure. A failure is classified as `balance` (402, `insufficient_quota`, "credit balance is too low",
++OpenRouter's 402 credits), `auth` (401/403), `rate_limit` (429), `transient` (timeout, 5xx, broken
++stream) or `request` (400/413/422 and context too long). The circuit is per provider+model
++(`provider_circuits`, migrations/0060, survives a restart): balance and auth open it at once and are
++never retried; a run of transient failures opens it for 30 s doubling to 10 min; rate limits and
++malformed requests never open it. A balance/auth circuit may probe the next real request after
++`AI_BREAKER_EXHAUSTED_COOLDOWN_MS` (15 minutes) and closes on a success or when the provider's key
++changes. The catalogue reports `up`, `degraded` (over 5% of at least 10 requests in the last hour) or
++`down`, with `status_reason`, `status_since` and `status_class`. 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 excluded
@@@ -965,6 +972,7 @@ mix vectors from different providers (re-embed when the provider changes).
| 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` |
++| Every provider error shape maps to balance/auth/rate_limit/transient/request; per-offer circuits, 30 s doubling backoff, persistence, key reset, catalogue status and the owner alert | `test/provider-circuits.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` |
ddiff --git a/migrations/0065_provider_circuits.sql b/migrations/0065_provider_circuits.sql
nnew file mode 100644
iindex 0000000..dd9360c
---- /dev/null
++++ b/migrations/0065_provider_circuits.sql
@@@ -0,0 +1,37 @@
++-- phase: expand
++-- Per provider+model circuits (server/providers/index.js). A provider can fail one offer while another
++-- offer on the same provider still answers, so the breaker is keyed by the offer, not only by the provider.
++-- The row survives a restart: state, the class that opened it, when a half-open probe may run, the
++-- transient backoff that doubles to the ceiling, the owner's-key fingerprint (a SHA-256, never the key)
++-- and the one owner notification sent for this open circuit.
++CREATE TABLE IF NOT EXISTS provider_circuits (
++ provider_key text COLLATE "C" NOT NULL,
++ model_key text COLLATE "C" NOT NULL DEFAULT '',
++ state text COLLATE "C" NOT NULL DEFAULT 'closed', -- closed | open | half_open
++ failure_class text COLLATE "C", -- balance | auth | rate_limit | transient | request (the class that opened it, or the last one)
++ consecutive_failures bigint NOT NULL DEFAULT 0,
++ opened_at bigint, -- epoch ms the current open state began
++ retry_at bigint, -- epoch ms a half-open probe may run
++ backoff_ms bigint NOT NULL DEFAULT 0, -- the current transient backoff (0 for balance/auth)
++ key_stamp text COLLATE "C", -- sha256(provider key): a new key resets the circuits
++ last_error text COLLATE "C", -- the provider's own short error, never a prompt or answer
++ last_success_at bigint,
++ last_failure_at bigint,
++ notified_at bigint, -- one owner alert per open balance/auth circuit
++ PRIMARY KEY (provider_key, model_key)
++);
++CREATE INDEX IF NOT EXISTS idx_provider_circuits_state ON provider_circuits(state);
++
++-- Why failures happened, by provider and class, for ai_provider_failures_total. Kept here so the
++-- counter survives a restart; a request log row also carries its error_class (migrations/0060).
++CREATE TABLE IF NOT EXISTS provider_failure_counts (
++ provider_key text COLLATE "C" NOT NULL,
++ failure_class text COLLATE "C" NOT NULL,
++ count bigint NOT NULL DEFAULT 0,
++ updated_at bigint,
++ PRIMARY KEY (provider_key, failure_class)
++);
++
++-- The class of each failed attempt (server/providers/failure.js): balance, auth, rate_limit, transient or
++-- request. The error message is still stored in `error`; this is the machine-readable class, no body.
++ALTER TABLE requests ADD COLUMN IF NOT EXISTS error_class text COLLATE "C";
ddiff --git a/server/config.js b/server/config.js
iindex 2d96444..740f758 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -129,6 +129,10 @@ function load(env = process.env) {
ssoAudience: String(env.AI_SSO_AUDIENCE || 'openvibe.network').trim(),
},
++ // The owner's Network subject (usr_…), for the one alert sent when a provider's balance or key
++ // fails. Unset means no push is attempted; the console and the metrics still say what happened.
++ ownerSubject: /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(env.AI_OWNER_SUBJECT || '')) ? String(env.AI_OWNER_SUBJECT) : null,
++
// ── The shared provider (Live's "shared key") ──
// AI_ENABLED is Live's ai_enabled master switch. Unset means "on when a key or a
// self-hosted base URL is configured".
@@@ -359,6 +363,10 @@ function load(env = process.env) {
// 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, 15 * 60000)),
++ // A run of transient failures opens the offer for this long, doubling on every failed probe to the
++ // ceiling. The first pause is 30 seconds and the offer is never away for more than 10 minutes.
++ transientBackoffMs: Math.max(1000, int(env.AI_BREAKER_BACKOFF_MS, 30000)),
++ transientMaxMs: Math.max(30000, int(env.AI_BREAKER_MAX_BACKOFF_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.
ddiff --git a/server/console/views.js b/server/console/views.js
iindex 95ff753..2e395f8 100644
---- a/server/console/views.js
++++ b/server/console/views.js
@@@ -133,12 +133,14 @@ function healthText(h, { breaker = {}, now = Date.now() } = {}) {
const st = (h && h.state) || 'closed';
if (st === 'closed') return { text: h && h.consecutive_failures ? `Working (${h.consecutive_failures} error${h.consecutive_failures === 1 ? '' : 's'} in a row so far)` : 'Working', cls: 'ok' };
if (st === 'half_open') return { text: 'Testing again: the next request tries it once', cls: 'warn' };
-- const cooldown = h.exhausted_at ? Number(breaker.exhaustedCooldownMs || 600000) : Number(breaker.cooldownMs || 60000);
-- const at = h.opened_at ? new Date(Number(h.opened_at) + cooldown) : null;
++ const cooldown = h.exhausted_at ? Number(breaker.exhaustedCooldownMs || 900000) : Number(breaker.cooldownMs || 30000);
++ const at = h.retry_at ? new Date(Number(h.retry_at)) : h.opened_at ? new Date(Number(h.opened_at) + cooldown) : null;
const hhmm = at && Number.isFinite(at.getTime()) ? `${at.toISOString().slice(11, 16)} UTC` : 'soon';
-- const why = h.exhausted_at ? 'the provider says the account is out of balance' : `${h.consecutive_failures || 1} error${(h.consecutive_failures || 1) === 1 ? '' : 's'} in a row`;
++ const why = h.status_reason || (h.failure_class === 'balance' || h.exhausted_at ? 'the provider says the account is out of credit'
++ : h.failure_class === 'auth' ? 'the key was refused'
++ : `${h.consecutive_failures || 1} error${(h.consecutive_failures || 1) === 1 ? '' : 's'} in a row`);
void now;
-- return { text: `Paused after ${why}; tries again at ${hhmm}`, cls: 'bad' };
++ return { text: `Paused: ${why}; tries again at ${hhmm}`, cls: 'bad' };
}
/** Where a provider's key comes from, in words: entered here, the server's settings file, not set or not needed. */
function keyText(p, hint) {
ddiff --git a/server/exchange/catalog.js b/server/exchange/catalog.js
iindex 5794f99..1f6769f 100644
---- a/server/exchange/catalog.js
++++ b/server/exchange/catalog.js
@@@ -17,6 +17,7 @@ const upstream = require('../providers/upstream');
const observed = require('./observed');
const privacy = require('./privacy');
const { allowsOffer } = require('../free-tier');
++const { offerStatus } = require('../providers/failure');
const FAMILY_BY_PREFIX = [
['deepseek', 'DeepSeek'], ['qwen', 'Qwen'], ['qwq', 'Qwen'], ['kimi', 'Kimi'], ['moonshot', 'Kimi'],
@@@ -147,17 +148,35 @@ function createCatalog({ db, registry, pool, env = process.env, clock = { now: (
const providers = await registry.listProviders();
const byKey = new Map(providers.map((p) => [p.key, p]));
const fidelityMap = fidelity ? await fidelity.read() : new Map();
-- // The router's breaker state per provider (server/providers/index.js): an offer whose provider answered
-- // "insufficient balance" is exhausted until a call succeeds again, and the catalog says so rather than
-- // promising free capacity that cannot answer.
-- const healthRows = await db.prepare('SELECT provider_key, state, exhausted_at FROM provider_health').all().catch(() => []);
++ // The router's per-offer circuit (server/providers/index.js, migrations/0060) is the authority; a
++ // provider-level provider_health row is still read for a state written before it (or by a test).
++ // Measured failures are folded in below: over 5% of at least 10 requests in the last hour is degraded.
++ const circuits = await db.prepare('SELECT * FROM provider_circuits').all().catch(() => []);
++ const circuitBy = new Map((circuits || []).map((c) => [`${c.provider_key}/${c.model_key || ''}`, c]));
++ const healthRows = await db.prepare('SELECT provider_key, state, opened_at, exhausted_at FROM provider_health').all().catch(() => []);
const healthByKey = new Map((healthRows || []).map((h) => [h.provider_key, h]));
-- const statusOf = (key) => {
-- const h = healthByKey.get(key);
-- if (!h || h.state === 'closed') return 'up';
-- if (h.state === 'open' && h.exhausted_at) return 'exhausted';
-- if (h.state === 'open') return 'down';
-- return 'degraded';
++ const hourAgo = new Date(now - 3600000).toISOString();
++ const measuredRows = await db.prepare(`SELECT provider_key, model_key,
++ COUNT(*)::bigint AS requests,
++ COUNT(*) FILTER (WHERE status IN ('error', 'timeout'))::bigint AS failures,
++ MIN(created_at) FILTER (WHERE status IN ('error', 'timeout')) AS first_error,
++ MIN(created_at) FILTER (WHERE error_class = 'balance') AS first_balance,
++ MIN(created_at) FILTER (WHERE error_class = 'auth') AS first_auth,
++ MAX(created_at) FILTER (WHERE status = 'ok') AS last_ok
++ FROM requests WHERE created_at >= ? AND status IN ('ok', 'error', 'timeout')
++ GROUP BY provider_key, model_key`).all(hourAgo).catch(() => []);
++ const measuredBy = new Map((measuredRows || []).map((r) => [`${r.provider_key}/${r.model_key || ''}`, r]));
++ const statusFor = (providerKey, modelKey) => {
++ const key = `${providerKey}/${modelKey || ''}`;
++ let circuit = circuitBy.get(key) || circuitBy.get(`${providerKey}/`) || null;
++ if (!circuit) {
++ const legacy = healthByKey.get(providerKey);
++ if (legacy && legacy.state && legacy.state !== 'closed') {
++ circuit = { state: legacy.state, failure_class: legacy.exhausted_at ? 'balance' : null,
++ opened_at: legacy.opened_at, status_reason: legacy.exhausted_at ? 'the provider says the account is out of credit' : null };
++ }
++ }
++ return offerStatus({ circuit, measured: measuredBy.get(key) || measuredBy.get(`${providerKey}/`) || null });
};
const upstreamByKey = new Map();
for (const p of providers) upstreamByKey.set(p.key, upstream.upstreamOf(p));
@@@ -195,6 +214,7 @@ function createCatalog({ db, registry, pool, env = process.env, clock = { now: (
const checkedAt = laterOf(logged.get(`${row.provider_key}/${row.model_key}`), cards ? cards.out.verified_at : null);
// The refusal rate of this offer, as the refusal job measured it.
const refusal = refusalBy.get(`${row.provider_key}/${row.model_key}`) || null;
++ const health = statusFor(provider.key, row.model_key);
const offer = {
provider: up.key,
provider_name: up.name,
@@@ -209,8 +229,11 @@ function createCatalog({ db, registry, pool, env = process.env, clock = { now: (
// capacity with no per-token card): both its provider is a named AI_FREE_PROVIDERS source and its
// output price is at or below AI_FREE_MAX_OUTPUT_PRICE_USD_PER_MTOK.
free_tier: allowsOffer({ providerKey: provider.key, providerRecord: provider, outputPrice: price ? price.out : null }, freeUse),
-- // up | degraded | down | exhausted (out of balance), from the router's breaker.
-- status: statusOf(provider.key),
++ // up | degraded | down, measured from this offer's circuit and the last hour of requests.
++ status: health.status,
++ status_reason: health.status_reason,
++ status_since: health.status_since,
++ status_class: health.status_class,
// Rounded to the micro-dollar: a derived cached price was printed as 0.020000000000000004.
on_our_gpus: provider.key.startsWith('gpu.'),
fixed_price: row.metadata?.owner_fixed_price === true,
ddiff --git a/server/index.js b/server/index.js
iindex 00b2d6d..ded705f 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -32,7 +32,7 @@ const { AiError } = require('./util');
const { gracefulStop, telemetry } = require('openvibe-sdk/service');
const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
--async function start({ searchFetch = null, config, db: givenDb = null, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, listen = true, credentialFetch = null, valkey = null, research = null, machineModels = null, variantEvidence = null, community: givenCommunity = null, communityEndpoints: givenEndpoints = null, webhooks: givenWebhooks = null, abuse: givenAbuse = null, memoryTransport = null, memoryLookup = null } = {}) {
++async function start({ searchFetch = null, config, db: givenDb = null, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, listen = true, credentialFetch = null, valkey = null, research = null, machineModels = null, variantEvidence = null, community: givenCommunity = null, communityEndpoints: givenEndpoints = null, webhooks: givenWebhooks = null, abuse: givenAbuse = null, memoryTransport = null, memoryLookup = null, ownerNotify = null } = {}) {
config = config || load(env);
schemas.configure(config.schemaCache);
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
@@@ -88,7 +88,9 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
const offPeak = require('./pricebook/offpeak').createOffPeak({ db, clock });
const creditGrants = require('./providers/credit-grants').createCreditGrants({ db, env, fetchImpl, clock });
const sessionCookies = require('./session-cookies').createSessionCookies({ db, config, clock });
-- const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log, credentialFetch, machineModels: machines, community, bench, variantEvidence, adultStatus, offPeak, creditGrants });
++ const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log, credentialFetch, machineModels: machines, community, bench, variantEvidence, adultStatus, offPeak, creditGrants, ownerNotify });
++ // An open balance/auth circuit that was persisted by an earlier process is announced once (notified_at dedupes).
++ if (listen && config.nodeEnv !== 'test') pool.notifyOpenCircuits().catch((err) => log.warn(`[ai] owner alerts: ${err && err.message}`));
// 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 });
ddiff --git a/server/observability.js b/server/observability.js
iindex c27e083..fe06a19 100644
---- a/server/observability.js
++++ b/server/observability.js
@@@ -102,6 +102,33 @@ function registerAiGauges(registry, { runs, registry: ai, pool, balance = null,
name: 'ai_provider_circuit', help: 'Circuit-breaker state of each active provider (1 = the current state)', labelNames: ['provider', 'state'],
collect: async () => (await providerView(ai, pool)).flatMap(p => CIRCUIT_STATES.map(state => ({ labels: { provider: p.key, state }, value: p.circuit === state ? 1 : 0 }))),
});
++ // The measured per-offer breaker (migrations/0060): 1 while the offer's circuit is open, 0 while it serves
++ // or waits for a half-open probe. The class is why it opened: balance, auth or transient.
++ registry.gauge({
++ name: 'ai_provider_circuit_open', help: 'Open circuit per provider and model (1 = open)', labelNames: ['provider', 'model', 'class'],
++ collect: async () => {
++ if (!db) return [];
++ const rows = await db.prepare('SELECT provider_key, model_key, state, failure_class FROM provider_circuits').all().catch(() => []);
++ const seen = new Set(rows.map((r) => `${r.provider_key}/${r.model_key || ''}`));
++ const legacy = await db.prepare('SELECT provider_key, state, exhausted_at FROM provider_health').all().catch(() => []);
++ return [
++ ...rows.map((r) => ({ labels: { provider: r.provider_key, model: r.model_key || '', class: r.failure_class || '' }, value: r.state === 'open' ? 1 : 0 })),
++ ...(legacy || []).filter((h) => !seen.has(`${h.provider_key}/`)).map((h) => ({
++ labels: { provider: h.provider_key, model: '', class: h.state === 'closed' ? '' : (h.exhausted_at ? 'balance' : 'transient') },
++ value: h.state === 'open' ? 1 : 0,
++ })),
++ ];
++ },
++ });
++ // Why attempts failed, by provider and class, counted in the database so a restart keeps the total.
++ registry.gauge({
++ name: 'ai_provider_failures_total', help: 'Failed provider attempts by provider and failure class (balance, auth, rate_limit, transient, request)', labelNames: ['provider', 'class'],
++ collect: async () => {
++ if (!db) return [];
++ const rows = await db.prepare('SELECT provider_key, failure_class, count FROM provider_failure_counts').all().catch(() => []);
++ return (rows || []).map((r) => ({ labels: { provider: r.provider_key, class: r.failure_class }, value: Number(r.count) || 0 }));
++ },
++ });
registry.gauge({
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 })),
ddiff --git a/server/openapi.js b/server/openapi.js
iindex 8c19667..1e30415 100644
---- a/server/openapi.js
++++ b/server/openapi.js
@@@ -447,6 +447,10 @@ const exchangeOffer = {
access: str({ enum: ['OWN_DEPLOYMENT', 'AUTHORIZED_WHOLESALE', 'UNVERIFIED'] }),
free: { type: 'boolean', description: 'Prepaid capacity with no per-token price (a free server, a subscription, a local machine).' },
free_tier: { type: 'boolean', description: 'The free allowance and the anonymous Try it may use this offer: its provider is one of AI_FREE_PROVIDERS (default deepseek,local,community; local covers local2 and whisper.cpp) and its output price is at or below AI_FREE_MAX_OUTPUT_PRICE_USD_PER_MTOK (default 1.5). Distinct from free: a priced DeepSeek Flash offer is free_tier true while a dearer priced offer is false.' },
++ status: { type: 'string', enum: ['up', 'degraded', 'down'], description: 'Measured, never guessed: down while this offer\'s circuit is open; degraded when more than 5% of at least 10 requests in the last hour failed; up otherwise.' },
++ status_reason: nullable('string', { description: 'Plain words for a degraded or down offer: "the provider says the account is out of credit", "the key was refused" or "too many errors in the last hour"' }),
++ status_since: nullable('string', { description: 'When the current status began (ISO 8601)' }),
++ status_class: nullable('string', { enum: ['balance', 'auth', 'rate_limit', 'transient', 'request'], description: 'The failure class behind a degraded or down status' }),
price: nullable('object', { properties: { in: { type: 'number' }, out: { type: 'number' }, cached: { type: 'number' }, per: str(), currency: str() } }),
price_source: nullable('string', { description: 'The price page the card was read from, when recorded' }),
price_checked_at: nullable('string', { description: 'The date someone checked the price, when recorded' }),
@@@ -526,7 +530,7 @@ route('put', '/agent-role-defaults/{role}', {
});
route('get', '/exchange/catalog', {
operationId: 'exchangeCatalog', summary: 'Every model and every offer that serves it', tags: ['exchange'], security: [],
-- description: 'Public. The live offers: price in, out and cached per million tokens, measured p50/p95, uptime, requests, and the provider data policy. Speed and uptime are read from the last seven days of real runs (time to the first token where a receipt recorded one, else the answering attempt\'s total); cached answers are counted apart and excluded from speed, uptime and requests. Uptime is succeeded / (succeeded + failed); null when no data, never 0. A prepaid offer has no price (free: true); free_tier says whether the free allowance and the anonymous Try it may use an offer at all (a named AI_FREE_PROVIDERS source under the output price ceiling). Models carry a kind (chat, embedding, stt, image); chat lists must use kind "chat" and never mix embeddings or speech-to-text.',
++ description: 'Public. The live offers: measured status (up/degraded/down, with a plain reason and since when), price in, out and cached per million tokens, measured p50/p95, uptime, requests, and the provider data policy. Speed and uptime are read from the last seven days of real runs (time to the first token where a receipt recorded one, else the answering attempt\'s total); cached answers are counted apart and excluded from speed, uptime and requests. Uptime is succeeded / (succeeded + failed); null when no data, never 0. A prepaid offer has no price (free: true); free_tier says whether the free allowance and the anonymous Try it may use an offer at all (a named AI_FREE_PROVIDERS source under the output price ceiling). Models carry a kind (chat, embedding, stt, image); chat lists must use kind "chat" and never mix embeddings or speech-to-text.',
parameters: [query('kind', str({ enum: ['chat', 'embedding', 'stt', 'image'] }), 'Only models of this kind')],
responses: { 200: json('The models, grouped by model, with their offers', { type: 'object', required: ['models'], properties: {
generated_at: str({ format: 'date-time' }), window_days: integer(), count: integer(),
ddiff --git a/server/ops.js b/server/ops.js
iindex 1c88dff..935f883 100644
---- a/server/ops.js
++++ b/server/ops.js
@@@ -21,7 +21,12 @@ function createOps({ db, registry, pool, quotas, cache, runs }) {
const today = new Date().toISOString().slice(0, 10);
const spend = await db.prepare('SELECT COALESCE(SUM(cost_usd),0) cost, COALESCE(SUM(requests),0)::bigint requests FROM usage_daily WHERE day = ?').get(today);
return {
-- providers: (await Promise.all((await registry.listProviders()).map(async p => ({ key: p.key, kind: p.kind, status: p.status, credentials: (await providerView(p)).credentials, health: (await pool.health(p.key)).state })))),
++ providers: (await Promise.all((await registry.listProviders()).map(async p => {
++ const h = await pool.health(p.key);
++ return { key: p.key, kind: p.kind, status: p.status, credentials: (await providerView(p)).credentials,
++ health: h.state, failure_class: h.failure_class || null, status_reason: h.status_reason || null,
++ status_since: h.status_since || null, circuits: h.circuits || [] };
++ }))),
runs: counts, today: { day: today, requests: spend.requests, cost_usd: spend.cost },
workflows: (await registry.listWorkflows()).length, templates: (await registry.listTemplates()).length, routes: (await registry.listRoutes()).length,
inflight: runs.inflight.size, quotas: await quotas.counters(),
ddiff --git a/server/providers/common.js b/server/providers/common.js
iindex defa843..d3b5142 100644
---- a/server/providers/common.js
++++ b/server/providers/common.js
@@@ -54,7 +54,7 @@ async function postJson(url, headers, body, { signal, timeoutMs = 30000, fetchIm
* 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;
++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;
ddiff --git a/server/providers/failure.js b/server/providers/failure.js
nnew file mode 100644
iindex 0000000..8bd83e0
---- /dev/null
++++ b/server/providers/failure.js
@@@ -0,0 +1,114 @@
++'use strict';
++/**
++ * Why a provider attempt failed, in one of five classes, and what the catalogue says about it.
++ *
++ * balance the provider's account has no money left: DeepSeek's 402 "Insufficient Balance",
++ * OpenAI's 429 `insufficient_quota`, Anthropic's 400 "credit balance is too low",
++ * OpenRouter's 402 "Insufficient credits", or any provider naming a credit/billing limit
++ * auth the key was refused: HTTP 401, or a 403 that is not a balance answer
++ * rate_limit the provider's capacity is used up (HTTP 429, RESOURCE_EXHAUSTED); shifts to another offer
++ * for this request but never opens a circuit
++ * transient a timeout, a connection failure, a 5xx or a broken stream; a run of these opens a circuit
++ * for a backoff that starts at 30 seconds and doubles to 10 minutes
++ * request the provider rejected the request itself (400, 404, 413, 422, context too long, a bad
++ * image or schema); the caller's fault and never a reason to open a circuit
++ *
++ * The class is pure and derived from the status and the provider's own error code/message. The body is
++ * never stored: only the class (and the provider's short message) rides the request log and the circuit.
++ */
++const { insufficientBalance, retryable } = require('./common');
++
++const CLASSES = Object.freeze(['balance', 'auth', 'rate_limit', 'transient', 'request']);
++
++/** The provider's error code and message, wherever its shape puts them (OpenAI, Anthropic, Gemini, OpenRouter). */
++function detailOf(err) {
++ const body = err && 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 || inner.status)) || (err && err.code) || '');
++ const message = String((inner && inner.message) || body.message || (err && err.message) || '');
++ const status = Number.isInteger(err && err.status) ? err.status : null;
++ return { code, message, status, text: `${code} ${message}` };
++}
++
++/** The one class this failure belongs to, or null for a cancellation (not a provider failure). */
++function classifyFailure(err) {
++ if (!err) return { class: 'transient', code: '', status: null };
++ if (err.code === 'provider.cancelled') return null;
++ const { code, message, status, text } = detailOf(err);
++
++ // Balance first: several providers say it with a 403 or 429, which would otherwise read as auth or
++ // ordinary capacity. The helper also knows DeepSeek's 402 and OpenAI's `insufficient_quota`.
++ if (insufficientBalance(err)) return { class: 'balance', code, status, message };
++
++ if (status === 401 || status === 403) return { class: 'auth', code, status, message };
++ // A streamed error can lose the HTTP status; the provider's own words still say what happened.
++ if (/invalid\s+(api[ _-]?key|x-api-key)|incorrect\s+api\s+key|authentication[ _-]?error|unauthorized|permission[ _-]?denied/i.test(text)
++ && !/quota|balance|credit/i.test(text)) return { class: 'auth', code, status, message };
++
++ if (status === 429 || /rate[ _-]?limit|too many requests|resource[ _-]?exhausted/i.test(text)) {
++ // A 429 that names an empty account was caught above; everything else is capacity for this request.
++ return { class: 'rate_limit', code, status, message };
++ }
++
++ if (status === 408 || status === 425 || (status != null && status >= 500)) return { class: 'transient', code, status, message };
++ if (err.code === 'provider.timeout' || err.code === 'provider.unreachable' || err.code === 'provider.stream_broken') return { class: 'transient', code, status, message };
++ if (/fetch failed|ECONNRESET|ETIMEDOUT|EAI_AGAIN|socket hang up|connection (failed|reset)|no response within/i.test(text)) return { class: 'transient', code, status, message };
++
++ if (status != null && status >= 400) return { class: 'request', code, status, message };
++ if (err.code === 'provider.output' || err.code === 'provider.unsupported' || err.code === 'media.unreadable' || err.code === 'policy.floor_refused') return { class: 'request', code, status, message };
++ if (/context (length|window)|too long|maximum context|invalid (request|schema|image)|could not (decode|read)|refus/i.test(text)) return { class: 'request', code, status, message };
++ return { class: retryable(err) ? 'transient' : 'request', code, status, message };
++}
++
++/** The plain sentence the catalogue and the console show for a class. */
++function reasonForClass(failureClass) {
++ if (failureClass === 'balance') return 'the provider says the account is out of credit';
++ if (failureClass === 'auth') return 'the key was refused';
++ return 'too many errors in the last hour';
++}
++
++const iso = (ms) => (Number.isFinite(Number(ms)) && Number(ms) > 0 ? new Date(Number(ms)).toISOString() : null);
++
++/**
++ * A circuit's status, before the measured error rate is folded in. An open circuit is down; a
++ * half-open one is degraded, because the next real request may probe it and it is not yet serving
++ * normally (the same mapping the placement states use).
++ */
++function circuitStatus(circuit) {
++ if (!circuit || (circuit.state !== 'open' && circuit.state !== 'half_open')) return null;
++ const failureClass = circuit.failure_class || (circuit.exhausted_at ? 'balance' : null);
++ return {
++ status: circuit.state === 'open' ? 'down' : 'degraded',
++ status_reason: circuit.status_reason || reasonForClass(failureClass),
++ status_since: iso(circuit.opened_at || circuit.status_since),
++ status_class: failureClass,
++ };
++}
++
++/**
++ * The measured status of one offer:
++ * down its circuit is open (or waiting for a half-open probe)
++ * degraded more than 5% of the last hour's at least 10 requests failed
++ * up anything else
++ * plus status_reason in plain words and status_since (ISO), never an invented time.
++ *
++ * `measured` is { requests, failures, first_error, first_balance, first_auth, last_ok } from the request log.
++ */
++function offerStatus({ circuit = null, measured = null } = {}) {
++ const down = circuitStatus(circuit);
++ if (down) return down;
++ const requests = Number(measured && measured.requests) || 0;
++ const failures = Number(measured && measured.failures) || 0;
++ if (requests >= 10 && failures / requests > 0.05) {
++ const failureClass = measured.first_balance ? 'balance' : measured.first_auth ? 'auth' : null;
++ return {
++ status: 'degraded',
++ status_reason: reasonForClass(failureClass),
++ status_since: measured.first_error || null,
++ status_class: failureClass,
++ };
++ }
++ return { status: 'up', status_reason: null, status_since: null, status_class: null };
++}
++
++module.exports = { CLASSES, classifyFailure, detailOf, reasonForClass, circuitStatus, offerStatus, iso };
ddiff --git a/server/providers/index.js b/server/providers/index.js
iindex 1848378..116f26b 100644
---- a/server/providers/index.js
++++ b/server/providers/index.js
@@@ -17,9 +17,11 @@
* through placement_state.current); or it pins `[primary, ...fallbacks]`. `explain` (objective,
* reasons, selected, candidates with their excluded reasons) rides every run response.
*
-- * Circuit breaker per provider: `failureThreshold` consecutive failures open it for `cooldownMs`;
-- * after that one half-open probe decides whether it closes again. A provider's own "insufficient balance"
-- * answer opens it at once for the long exhausted cooldown, is never retried, and the router fails over.
++ * Circuit breaker per provider+model (server/providers/failure.js): a run of transient failures opens
++ * that offer for 30 s doubling to 10 min; a balance or auth answer opens it at once for the long
++ * exhausted cooldown and is never retried; the next real request after the cooldown is the one probe.
++ * Another model on the same provider keeps answering. The state is stored (provider_circuits, 0060),
++ * so a restart keeps it, and a changed provider key resets it.
* 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.
@@@ -28,6 +30,8 @@ const placement = require('openvibe-sdk/placement');
const contracts = require('openvibe-contracts');
const { AiError, providerSecret, sha256 } = require('../util');
const { retryable, sleep, ProviderError, schemaInstruction, insufficientBalance } = require('./common');
++const { classifyFailure, reasonForClass, circuitStatus, offerStatus, iso } = require('./failure');
++const { createOwnerNotifier } = require('./owner-notify');
const { sourcesOf, ceilingOf, allowsProvider, allowsCommunity } = require('../free-tier');
const { createStubProvider } = require('./stub');
const { createOpenAiProvider } = require('./openai');
@@@ -216,7 +220,7 @@ function selfModelFor(cred, operation, req, ctx, prefs) {
return { model, kind, why };
}
--function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, credentialFetch = null, machineModels = null, community = null, bench = null, variantEvidence = null, adultStatus = null, offPeak = null, creditGrants = null }) {
++function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, credentialFetch = null, machineModels = null, community = null, bench = null, variantEvidence = null, adultStatus = null, offPeak = null, creditGrants = null, ownerNotify = null }) {
// A person's own machine as their model (credential provider node, server/machines.js).
const machines = machineModels || require('../machines').createMachineModels({ config, env, fetchImpl });
const cache = new Map(); // key -> { stamp, adapter }
@@@ -285,39 +289,271 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
return { record: p, adapter: a };
}
-- // ── Circuit breaker ────────────────────────────────────
++ // ── Circuit breaker (per provider + model) ─────────────
++ // A provider can fail one offer while another offer on the same provider still answers, so the
++ // breaker is keyed by (provider, model). provider_health stays as the provider-level roll-up for
++ // the console, readiness and older callers; the per-offer rows are the authority.
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, 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,
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, 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 getCircuit = db.prepare('SELECT * FROM provider_circuits WHERE provider_key = ? AND model_key = ?');
++ const listCircuits = db.prepare('SELECT * FROM provider_circuits WHERE provider_key = ? ORDER BY model_key');
++ const putCircuit = db.prepare(`INSERT INTO provider_circuits
++ (provider_key, model_key, state, failure_class, consecutive_failures, opened_at, retry_at, backoff_ms, key_stamp, last_error, last_success_at, last_failure_at, notified_at)
++ VALUES (@provider_key, @model_key, @state, @failure_class, @consecutive_failures, @opened_at, @retry_at, @backoff_ms, @key_stamp, @last_error, @last_success_at, @last_failure_at, @notified_at)
++ ON CONFLICT(provider_key, model_key) DO UPDATE SET
++ state = excluded.state, failure_class = excluded.failure_class, consecutive_failures = excluded.consecutive_failures,
++ opened_at = excluded.opened_at, retry_at = excluded.retry_at, backoff_ms = excluded.backoff_ms, key_stamp = excluded.key_stamp,
++ last_error = excluded.last_error, last_success_at = excluded.last_success_at, last_failure_at = excluded.last_failure_at, notified_at = excluded.notified_at`);
++ const putFailureCount = db.prepare(`INSERT INTO provider_failure_counts (provider_key, failure_class, count, updated_at)
++ VALUES (@provider_key, @failure_class, 1, @updated_at)
++ ON CONFLICT(provider_key, failure_class) DO UPDATE SET count = provider_failure_counts.count + excluded.count, updated_at = excluded.updated_at`);
++ const notifyOwner = ownerNotify || createOwnerNotifier({ config, env, fetchImpl, log });
++
++ /** The fingerprint of the provider's current key: a SHA-256, never the key itself. */
++ function stampFor(p) {
++ if (!p || p.auth_mode === 'none') return null;
++ const secret = providerSecret(p, env);
++ return secret ? sha256(secret) : null;
++ }
++ async function keyStamp(key) { return stampFor(await registry.getProvider(key)); }
++
++ /**
++ * A key change closes every circuit on its provider: a new key may answer where the old one was
++ * refused. The provider record's secret_ref alone is not enough (an env value can change under the
++ * same reference), so the resolved key's fingerprint is compared. Nothing about the key is stored.
++ */
++ async function resetForNewKey(key, stamp) {
++ const rows = await listCircuits.all(key);
++ if (!rows.length) return false;
++ if (rows.every((r) => (r.key_stamp || null) === (stamp || null))) return false;
++ await db.prepare('DELETE FROM provider_circuits WHERE provider_key = ?').run(key);
++ await db.prepare('DELETE FROM provider_health WHERE provider_key = ?').run(key);
++ await registry.audit('system', 'provider.circuit_reset', 'provider', key, { metadata: { reason: 'the provider key changed' } });
++ return true;
++ }
++
++ const CLOSED = Object.freeze({ state: 'closed', consecutive_failures: 0, opened_at: null, retry_at: null, backoff_ms: 0, failure_class: null, last_error: null, last_success_at: null, last_failure_at: null, notified_at: null });
++
++ /** One circuit row as the router and the admin API show it; an open row past retry_at is a half-open probe. */
++ function circuitView(row) {
++ if (!row) return { ...CLOSED, status: 'up', status_reason: null, status_since: null };
++ const state = row.state === 'open' && Number(row.retry_at || 0) <= clock.now() ? 'half_open' : row.state;
++ const failureClass = row.failure_class || (row.exhausted_at ? 'balance' : null);
++ const down = state !== 'closed';
++ return {
++ provider_key: row.provider_key, model_key: row.model_key || '',
++ state, failure_class: failureClass, consecutive_failures: Number(row.consecutive_failures) || 0,
++ opened_at: row.opened_at == null ? null : Number(row.opened_at),
++ retry_at: row.retry_at == null ? null : Number(row.retry_at),
++ backoff_ms: Number(row.backoff_ms) || 0,
++ exhausted_at: failureClass === 'balance' ? (row.opened_at == null ? null : Number(row.opened_at)) : null,
++ last_error: row.last_error || null,
++ last_success_at: row.last_success_at == null ? null : Number(row.last_success_at),
++ last_failure_at: row.last_failure_at == null ? null : Number(row.last_failure_at),
++ notified_at: row.notified_at == null ? null : Number(row.notified_at),
++ status: down ? 'down' : 'up',
++ status_reason: down ? reasonForClass(failureClass) : null,
++ status_since: down ? iso(row.opened_at) : null,
++ };
++ }
++
++ /** A legacy provider_health row (a direct insert, or a state from before 0060), still honoured. */
++ function legacyView(h, model = undefined) {
++ const failureClass = h.exhausted_at ? 'balance' : null;
++ let state = h.state;
++ if (state === 'open') {
const cooldown = h.exhausted_at ? exhaustedCooldownMs() : config.breaker.cooldownMs;
-- if (clock.now() - (h.opened_at || 0) >= cooldown) return { ...h, state: 'half_open' };
++ if (clock.now() - (h.opened_at || 0) >= cooldown) state = 'half_open';
}
-- return h;
++ const down = state !== 'closed';
++ return {
++ ...CLOSED, ...h, provider_key: h.provider_key, ...(model === undefined ? {} : { model_key: String(model || '') }),
++ state, failure_class: failureClass, consecutive_failures: Number(h.consecutive_failures) || 0,
++ opened_at: h.opened_at == null ? null : Number(h.opened_at),
++ exhausted_at: h.exhausted_at == null ? null : Number(h.exhausted_at),
++ status: down ? 'down' : 'up',
++ status_reason: down ? reasonForClass(failureClass) : null,
++ status_since: down ? iso(h.opened_at) : null,
++ };
}
-- 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, exhausted_at: null, last_success_at: clock.now() });
++
++ /** The provider-level roll-up of its offers, written where readiness and older callers read it. */
++ async function syncProviderHealth(key) {
++ const rows = await listCircuits.all(key);
++ if (!rows.length) { await db.prepare('DELETE FROM provider_health WHERE provider_key = ?').run(key); return; }
++ const open = rows.filter((r) => r.state === 'open');
++ const half = rows.filter((r) => r.state === 'half_open');
++ const state = open.length ? 'open' : half.length ? 'half_open' : 'closed';
++ const pick = open[0] || half[0] || rows[0];
++ const exhausted = rows.find((r) => r.failure_class === 'balance' && r.state !== 'closed');
++ const latest = rows.reduce((a, b) => (Number(b.last_failure_at || 0) > Number(a.last_failure_at || 0) ? b : a), rows[0]);
++ const success = Math.max(...rows.map((r) => Number(r.last_success_at) || 0)) || null;
++ const failure = Math.max(...rows.map((r) => Number(r.last_failure_at) || 0)) || null;
++ await putHealth.run({
++ provider_key: key, state, consecutive_failures: Math.max(...rows.map((r) => Number(r.consecutive_failures) || 0)),
++ opened_at: pick.opened_at == null ? null : Number(pick.opened_at),
++ exhausted_at: exhausted && exhausted.opened_at != null ? Number(exhausted.opened_at) : null,
++ last_error: latest.last_error || null, last_success_at: success, last_failure_at: failure,
++ });
}
-- async function recordFailure(key, err) {
-- const h = await health(key);
-- const failures = (h.consecutive_failures || 0) + 1;
-- 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() });
++
++ /**
++ * The health of one offer (model given) or of the provider as a whole (model omitted).
++ * A provider-level answer carries every offer's circuit in `circuits`.
++ */
++ async function health(key, model = undefined, { stamp } = {}) {
++ const current = stamp !== undefined ? stamp : await keyStamp(key);
++ await resetForNewKey(key, current);
++ if (model !== undefined) {
++ const row = await getCircuit.get(key, String(model || ''));
++ if (row) return circuitView(row);
++ // Once a provider has per-offer rows, a model without one has no failure of its own: another
++ // model's open circuit must not take it away. provider_health is only a legacy fallback.
++ if ((await listCircuits.all(key)).length) {
++ return { ...CLOSED, provider_key: key, model_key: String(model || ''), status: 'up', status_reason: null, status_since: null };
++ }
++ const legacy = await getHealth.get(key);
++ return legacy && legacy.state !== 'closed' ? legacyView(legacy, model) : { ...CLOSED, provider_key: key, model_key: String(model || ''), status: 'up', status_reason: null, status_since: null };
++ }
++ const rows = await listCircuits.all(key);
++ if (rows.length) {
++ const views = rows.map(circuitView);
++ const open = views.find((v) => v.state === 'open');
++ const half = views.find((v) => v.state === 'half_open');
++ const pick = open || half || views[0];
++ return {
++ ...pick, provider_key: key, model_key: undefined,
++ state: open ? 'open' : half ? 'half_open' : 'closed',
++ consecutive_failures: Math.max(...views.map((v) => v.consecutive_failures || 0)),
++ last_error: views.reduce((a, b) => (Number(b.last_failure_at || 0) > Number(a.last_failure_at || 0) ? b : a), views[0]).last_error,
++ last_success_at: Math.max(...views.map((v) => Number(v.last_success_at) || 0)) || null,
++ last_failure_at: Math.max(...views.map((v) => Number(v.last_failure_at) || 0)) || null,
++ circuits: views,
++ };
++ }
++ const legacy = await getHealth.get(key);
++ return legacy ? legacyView(legacy) : { ...CLOSED, provider_key: key, status: 'up', status_reason: null, status_since: null, circuits: [] };
++ }
++
++ async function recordSuccess(key, model = '', { stamp } = {}) {
++ const modelKey = String(model || '');
++ const row = await getCircuit.get(key, modelKey);
++ const wasOpen = row && row.state && row.state !== 'closed';
++ if (wasOpen) await registry.audit('system', 'provider.circuit_closed', 'provider', key, { metadata: { after_failures: Number(row.consecutive_failures) || 0, model: modelKey } });
++ const current = stamp !== undefined ? stamp : await keyStamp(key);
++ const prev = row || {};
++ await putCircuit.run({
++ provider_key: key, model_key: modelKey, state: 'closed', failure_class: null, consecutive_failures: 0,
++ opened_at: null, retry_at: null, backoff_ms: 0, key_stamp: current,
++ last_error: prev.last_error || null, last_success_at: clock.now(), last_failure_at: prev.last_failure_at || null, notified_at: null,
++ });
++ await syncProviderHealth(key);
++ }
++
++ /**
++ * Record one failed attempt. Every class is counted (ai_provider_failures_total); only balance, auth
++ * and a run of transient failures touch the circuit. rate_limit and request failures move this
++ * request on without taking the offer away from anyone else.
++ */
++ async function recordFailure(key, model = '', err, { stamp } = {}) {
++ const failure = classifyFailure(err);
++ if (!failure) return null;
++ const now = clock.now();
++ // Our own policy refusing an answer is not the provider failing: it produced one.
++ const counts = err && err.code !== 'policy.floor_refused';
++ if (counts) await putFailureCount.run({ provider_key: key, failure_class: failure.class, updated_at: now });
++ if (!['balance', 'auth', 'transient'].includes(failure.class)) return { ...failure, opened: false };
++
++ const modelKey = String(model || '');
++ const current = stamp !== undefined ? stamp : await keyStamp(key);
++ await resetForNewKey(key, current);
++ const row = await getCircuit.get(key, modelKey) || { state: 'closed' };
++ const failures = (Number(row.consecutive_failures) || 0) + 1;
++ const wasOpen = row.state === 'open' || row.state === 'half_open';
++ // A failure after retry_at is the half-open probe failing: the pause starts again from now.
++ const probing = row.state === 'half_open' || (row.state === 'open' && Number(row.retry_at || 0) <= now);
++ let state = 'closed';
++ let openedAt = null;
++ let retryAt = null;
++ let backoff = 0;
++ let notified = row.notified_at == null ? null : Number(row.notified_at);
++
++ if (failure.class === 'balance' || failure.class === 'auth') {
++ state = 'open';
++ openedAt = wasOpen && !probing && row.opened_at ? Number(row.opened_at) : now;
++ retryAt = openedAt + exhaustedCooldownMs();
++ if (!wasOpen) notified = null;
++ } else if (wasOpen || failures >= config.breaker.failureThreshold) {
++ state = 'open';
++ const initial = config.breaker.transientBackoffMs;
++ const max = config.breaker.transientMaxMs;
++ backoff = wasOpen ? Math.min(max, Math.max(initial, (Number(row.backoff_ms) || initial) * 2)) : initial;
++ openedAt = wasOpen && !probing && row.opened_at ? Number(row.opened_at) : now;
++ retryAt = now + backoff;
++ }
++
++ if (state === 'open' && !wasOpen) {
++ await registry.audit('system', 'provider.circuit_opened', 'provider', key, {
++ metadata: { failures, failure_class: failure.class, model: modelKey, error: String(failure.message || err && err.message || err).slice(0, 200) },
++ });
++ }
++ const saved = {
++ provider_key: key, model_key: modelKey, state,
++ failure_class: failure.class, consecutive_failures: failures,
++ opened_at: openedAt, retry_at: retryAt, backoff_ms: backoff, key_stamp: current,
++ last_error: String(failure.message || err && err.message || err || '').slice(0, 500),
++ last_success_at: row.last_success_at == null ? null : Number(row.last_success_at),
++ last_failure_at: now, notified_at: notified,
++ };
++ await putCircuit.run(saved);
++ await syncProviderHealth(key);
++ if (state === 'open' && (failure.class === 'balance' || failure.class === 'auth') && !notified) {
++ await announceOnce(key, modelKey, failure.class, now);
++ }
++ return { ...failure, opened: state === 'open', state, opened_at: openedAt, retry_at: retryAt, backoff_ms: backoff };
++ }
++
++ /**
++ * Announce an open circuit at most once: the database row is claimed before the push, so two
++ * requests that fail together cannot both send. A failed push releases the claim for a later try.
++ */
++ async function announceOnce(key, model, failureClass, at = clock.now()) {
++ const claimed = await db.prepare('UPDATE provider_circuits SET notified_at = ? WHERE provider_key = ? AND model_key = ? AND notified_at IS NULL')
++ .run(at, key, String(model || ''));
++ if (!claimed || !claimed.changes) return false;
++ if (await alertOwner(key, model, failureClass)) return true;
++ await db.prepare('UPDATE provider_circuits SET notified_at = NULL WHERE provider_key = ? AND model_key = ? AND notified_at = ?')
++ .run(key, String(model || ''), at);
++ return false;
++ }
++
++ /** One owner message per open balance/auth circuit, with the provider's own class, never a request body. */
++ async function alertOwner(key, model, failureClass) {
++ const offer = model ? `${key}/${model}` : key;
++ const title = failureClass === 'balance' ? 'A provider is out of credit' : 'A provider key was refused';
++ const message = failureClass === 'balance'
++ ? `${offer} says its account is out of credit. OpenVibe.AI is using the other offers until it is topped up.`
++ : `${offer} refused its key. OpenVibe.AI is using the other offers until the key is replaced.`;
++ return await notifyOwner({ title, message });
++ }
++
++ /** At boot: an open balance/auth circuit that was never announced (a crash before the push) is announced once. */
++ async function notifyOpenCircuits() {
++ const rows = await db.prepare("SELECT * FROM provider_circuits WHERE state = 'open' AND failure_class IN ('balance', 'auth') AND notified_at IS NULL").all();
++ let sent = 0;
++ for (const row of rows) {
++ if (await announceOnce(row.provider_key, row.model_key, row.failure_class)) sent++;
++ }
++ return sent;
++ }
++
++ async function resetHealth(key) {
++ await db.prepare('DELETE FROM provider_circuits WHERE provider_key = ?').run(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); }
// ── Pricing: platform.rate-card@1 per metric (server/providers/rate-cards.js; models row -> AI_PRICING_JSON -> flat) ──
/** The { in, cached, out } rate cards for a provider + model, or null for a provider that is never billed. */
@@@ -694,7 +930,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
const entry = await adapter(c.provider);
const p = entry && entry.record;
const modelId = c.model || (p && p.default_model) || null;
-- const why = await skipReason(p, entry && entry.adapter, features);
++ const why = await skipReason(p, entry && entry.adapter, features, modelId);
const offered = appOffered(ctx, p);
const access = accessOffered(p);
const policyOk = p ? await policyAllows(registry, p, modelId, ctx.contentPolicy || 'standard') : false;
@@@ -856,13 +1092,17 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
return { order, ranked: view, reasons, keepOrder };
}
-- async function skipReason(p, a, features) {
++ async function skipReason(p, a, features, model = undefined) {
if (!p || !a) return 'unknown_provider';
if (p.status !== 'active') return 'disabled';
if (p.auth_mode !== 'none' && !credentialNotRequired(p) && !providerSecret(p, env)) return 'no_credentials';
if (features.some(f => !a.supports(f))) return 'unsupported';
-- const h = await health(p.key);
-- if (h.state === 'open') return h.exhausted_at ? 'exhausted' : 'circuit_open';
++ const h = await health(p.key, model, { stamp: stampFor(p) });
++ if (h.state === 'open') {
++ if (h.failure_class === 'balance' || h.exhausted_at) return 'exhausted';
++ if (h.failure_class === 'auth') return 'key_refused';
++ return 'circuit_open';
++ }
return null;
}
@@@ -918,7 +1158,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
attempts: [{ key: 'your machine', provider: 'link', model: result.model, why: 'your own machine', status: 'ok', error: null }],
openvibe: ownFacts } };
} catch (err) {
-- if (ctx.logRequest) ctx.logRequest({ ...base, status: 'error', latency_ms: Date.now() - t0, error: String(err && err.message || err).slice(0, 500) });
++ if (ctx.logRequest) ctx.logRequest({ ...base, status: 'error', latency_ms: Date.now() - t0, error: String(err && err.message || err).slice(0, 500), error_class: classifyFailure(err)?.class || null });
if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
throw err;
}
@@@ -987,7 +1227,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
} catch (err) {
lastErr = err;
if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
-- if (ctx.logRequest) ctx.logRequest({ ...base, status: err && err.code === 'provider.timeout' ? 'timeout' : 'error', latency_ms: Date.now() - t0, error: scrub(err && err.message || err).slice(0, 500) });
++ if (ctx.logRequest) ctx.logRequest({ ...base, status: err && err.code === 'provider.timeout' ? 'timeout' : 'error', latency_ms: Date.now() - t0, error: scrub(err && err.message || err).slice(0, 500), error_class: classifyFailure(err)?.class || null });
// L36 Y8: the provider refused the key (401/403). The key becomes failing and the event says so.
if (err && (err.status === 401 || err.status === 403) && typeof ctx.onCredentialFailed === 'function') {
await ctx.onCredentialFailed({ owner: cred.owner || null, subject: cred.subject || null, provider: cred.provider, providerKey: cred.providerKey, error: scrub(err.message) });
@@@ -1239,7 +1479,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
if (price && Number(price.out) > freeCeiling) { why.push('over_price_ceiling'); freeCeilingSkipped = true; }
else {
if (!accessOffered(p)) why.push(`provider_access ${modelCatalog.accessOf(p)}`);
-- const skip = await skipReason(p, entry && entry.adapter, features);
++ const skip = await skipReason(p, entry && entry.adapter, features, offer.model);
if (skip) why.push(skip);
}
}
@@@ -1491,9 +1731,9 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
continue;
}
}
-- const why = await skipReason(p, a, features);
++ const why = await skipReason(p, a, features, model);
if (why) {
-- if (why === 'exhausted' || why === 'circuit_open') resting = true;
++ if (why === 'exhausted' || why === 'circuit_open' || why === 'key_refused') 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 });
@@@ -1625,7 +1865,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
: await costOnTier({ providerKey: c.provider, model: result.model || model, usage, tier: ranOn, p });
const cost = priced.cost;
statFor(c.provider).ok++;
-- await recordSuccess(c.provider);
++ await recordSuccess(c.provider, model);
if (ctx.logRequest) {
ctx.logRequest({
...base, model_key: result.model || model, status: 'ok', output_hash: sha256(result.json || result.text || result.vectors || result.segments || result.images || ''),
@@@ -1660,7 +1900,8 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
}
await recordEvidence({ provider: c.provider, model, ok: false, timeout: Boolean(err && err.code === 'provider.timeout'), latencyMs: Date.now() - t0 });
statFor(c.provider).failures++;
-- if (ctx.logRequest) ctx.logRequest({ ...base, status: err && err.code === 'provider.timeout' ? 'timeout' : 'error', latency_ms: Date.now() - t0, error: String(err && err.message || err).slice(0, 500) });
++ const failure = classifyFailure(err);
++ if (ctx.logRequest) ctx.logRequest({ ...base, status: err && err.code === 'provider.timeout' ? 'timeout' : 'error', latency_ms: Date.now() - t0, error: String(err && err.message || err).slice(0, 500), error_class: failure ? failure.class : null });
// A stream that broke off still delivers what the upstream had already said (the guard's held tail
// included), unless that tail itself breaks the floor.
if (streamed && floorGuard) {
@@@ -1673,8 +1914,9 @@ 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) 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.
-- // An insufficient-balance answer is never retried: the same provider cannot answer until it is topped up.
-- if (insufficientBalance(err)) resting = true;
++ // An insufficient-balance or refused-key answer is never retried: the offer cannot answer
++ // until it is topped up or the key is replaced.
++ if (failure && (failure.class === 'balance' || failure.class === 'auth')) resting = true;
// BA1: the provider refused the tier we asked for (a Flex tier with no capacity, or a
// model that does not serve it). The same run goes once more on the standard tier: the
// refusal is counted and logged above, and it produced no tokens, so nothing is billed
@@@ -1694,14 +1936,12 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
} 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;
-- // 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) });
++ // classifyFailure decides what a failure means: balance/auth open at once, a run of transient
++ // failures opens for a doubling backoff, a 429 or a malformed request never opens a circuit.
++ const failedAs = await recordFailure(c.provider, model, lastErr);
++ tried.push({ provider: c.provider, error: String(lastErr && lastErr.message || lastErr).slice(0, 200), error_class: failedAs && failedAs.class });
++ 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), error_class: failedAs && failedAs.class });
log.warn(`[ai] ${operation} via ${c.provider}/${model || '-'} failed: ${lastErr && lastErr.message}`);
}
explain.selected = null;
@@@ -1737,7 +1977,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
throw new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried, explain });
}
-- return { adapter, health, resetHealth, execute, executeWithCredential, offers, benchView, priceFor, costOf, rateCardsFor, providerStates, stats, statFor, candidates, machines, routingEvidence };
++ return { adapter, health, resetHealth, notifyOpenCircuits, execute, executeWithCredential, offers, benchView, priceFor, costOf, rateCardsFor, providerStates, stats, statFor, candidates, machines, routingEvidence };
}
module.exports = { createProviderPool, createFloorGuard, policyAllows, machineMessages, machineTools };
ddiff --git a/server/providers/owner-notify.js b/server/providers/owner-notify.js
nnew file mode 100644
iindex 0000000..682ecc8
---- /dev/null
++++ b/server/providers/owner-notify.js
@@@ -0,0 +1,64 @@
++'use strict';
++/**
++ * One owner alert when a provider's account runs out of credit or its key is refused.
++ *
++ * This is the same Network notification path the rest of the platform uses
++ * (POST /internal/notifications/push, network.notifications.push): the SDK resolves the owner's
++ * Network user from `AI_OWNER_SUBJECT` and pushes one GENERIC notification. It is sent once per open
++ * balance/auth circuit, never per request; the circuit row keeps notified_at so a restart does not
++ * repeat it. Without the owner subject (or the service credentials) the alert is skipped, and the
++ * console and the Prometheus gauges still show what happened.
++ */
++const { serviceAuth } = require('openvibe-contracts');
++const { createNotificationsClient } = require('openvibe-sdk/notifications');
++
++function createOwnerNotifier({ config, env = process.env, fetchImpl = globalThis.fetch, log = console }) {
++ const subject = config.ownerSubject;
++ const base = String(config.networkInternalUrl || '').replace(/\/+$/, '');
++ const clientSecret = String(env.OV_OAUTH_CLIENT_SECRET || '');
++ const clientId = (config.console && config.console.clientId) || 'ai';
++ if (!subject || !base || !clientSecret) {
++ return async () => false;
++ }
++
++ const tokens = serviceAuth.createTokenClient({
++ tokenUrl: `${base}/oauth/token`, clientId, clientSecret,
++ audience: 'openvibe.network', scope: 'identity.subject.resolve network.notifications.push', fetchImpl,
++ });
++
++ async function json({ path, method = 'GET', json: body, query } = {}) {
++ const url = new URL(base + path);
++ for (const [k, v] of Object.entries(query || {})) if (v != null) url.searchParams.set(k, String(v));
++ const headers = { Accept: 'application/json', ...(await tokens.authHeaders()) };
++ if (body !== undefined) headers['Content-Type'] = 'application/json';
++ const res = await fetchImpl(url, {
++ method, headers, body: body === undefined ? undefined : JSON.stringify(body),
++ signal: AbortSignal.timeout(8000),
++ });
++ const out = await res.json().catch(() => null);
++ if (!res.ok) {
++ const err = new Error(`Network ${method} ${path} ${res.status}`);
++ err.status = res.status;
++ throw err;
++ }
++ return out;
++ }
++
++ const notifications = createNotificationsClient({ json }, { baseUrl: base });
++
++ return async function notifyOwner({ title, message, url }) {
++ try {
++ const out = await notifications.push({
++ subjectId: subject, type: 'GENERIC', title, message,
++ service: 'ai', url: url || `${String(config.exchangeUrl || 'https://openvibe.exchange').replace(/\/+$/, '')}/console/providers`,
++ category: 'system',
++ });
++ return Boolean(out && out.sent);
++ } catch (err) {
++ log.warn(`[ai] owner alert not sent: ${String(err && err.message || err).slice(0, 200)}`);
++ return false;
++ }
++ };
++}
++
++module.exports = { createOwnerNotifier };
ddiff --git a/server/runs.js b/server/runs.js
iindex 174b421..e6aaa2f 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -188,9 +188,9 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
}
const insReq = db.prepare(`INSERT INTO requests (run_id, seq, operation, provider_key, model_key, route_key, route_version, status, skip_reason, fallback, prompt_hash, input_hash, output_hash,
-- tokens_in, tokens_out, tokens_cached, tokens_estimated, cost_usd, latency_ms, error, debug_prompt, debug_response, created_at)
++ tokens_in, tokens_out, tokens_cached, tokens_estimated, cost_usd, latency_ms, error, error_class, debug_prompt, debug_response, created_at)
VALUES (@run_id, @seq, @operation, @provider_key, @model_key, @route_key, @route_version, @status, @skip_reason, @fallback, @prompt_hash, @input_hash, @output_hash,
-- @tokens_in, @tokens_out, @tokens_cached, @tokens_estimated, @cost_usd, @latency_ms, @error, @debug_prompt, @debug_response, @created_at)`);
++ @tokens_in, @tokens_out, @tokens_cached, @tokens_estimated, @cost_usd, @latency_ms, @error, @error_class, @debug_prompt, @debug_response, @created_at)`);
// ── Create ─────────────────────────────────────────────
function validateRefs(body) {
@@@ -896,7 +896,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
run_id: id, seq: seq++, operation: e.operation, provider_key: e.provider_key, model_key: e.model_key || null, route_key: e.route_key || null, route_version: e.route_version || null,
status: e.status, skip_reason: e.skip_reason || null, fallback: e.fallback || 0, prompt_hash: e.prompt_hash || null, input_hash: row.input_hash, output_hash: e.output_hash || null,
tokens_in: e.tokens_in || 0, tokens_out: e.tokens_out || 0, tokens_cached: e.tokens_cached || 0, tokens_estimated: e.tokens_estimated || 0, cost_usd: e.cost_usd || 0,
-- latency_ms: e.latency_ms == null ? null : e.latency_ms, error: e.error || null,
++ latency_ms: e.latency_ms == null ? null : e.latency_ms, error: e.error || null, error_class: e.error_class || null,
debug_prompt: config.debugRawLog && opts.debug ? e.debug_prompt || null : null, debug_response: config.debugRawLog && opts.debug ? e.debug_response || null : null,
created_at: iso(clock.now()),
});
@@@ -1211,7 +1211,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
}
async function requestsFor(runId) {
-- return await db.prepare('SELECT id, seq, operation, provider_key, model_key, route_key, route_version, status, skip_reason, fallback, prompt_hash, input_hash, output_hash, tokens_in, tokens_out, tokens_cached, tokens_estimated, cost_usd, latency_ms, error, created_at FROM requests WHERE run_id = ? ORDER BY seq').all(runId);
++ return await db.prepare('SELECT id, seq, operation, provider_key, model_key, route_key, route_version, status, skip_reason, fallback, prompt_hash, input_hash, output_hash, tokens_in, tokens_out, tokens_cached, tokens_estimated, cost_usd, latency_ms, error, error_class, created_at FROM requests WHERE run_id = ? ORDER BY seq').all(runId);
}
/** Boot: runs interrupted by a restart are failed explicitly (retry is one call away). */
ddiff --git a/test/exchange-catalog-kinds.test.js b/test/exchange-catalog-kinds.test.js
iindex f72f890..ff66989 100644
---- a/test/exchange-catalog-kinds.test.js
++++ b/test/exchange-catalog-kinds.test.js
@@@ -93,7 +93,7 @@ t.test('chat, embeddings, speech-to-text and image are separate kinds, and ?kind
assert.strictEqual(bad.body.code, 'exchange.kind_invalid');
});
--t.test('an out-of-balance provider is exhausted, and its free offer is not free now', async () => {
++t.test('an out-of-balance provider is down, with a plain reason, and its free offer is not free now', async () => {
// A catalog of its own (the route's answer is cached for 30 s), over the same database and registry.
const { createCatalog } = require('../server/exchange/catalog');
const catalog = createCatalog({ db: h.db, registry: h.registry, pool: h.pool, env: h.env, config: h.config });
@@@ -111,7 +111,10 @@ t.test('an out-of-balance provider is exhausted, and its free offer is not free
const after = await catalog.models({ days: 7 });
const flash = flashOf(after);
const offer = flash.offers.find((o) => o.provider === 'deepseek');
-- assert.strictEqual(offer.status, 'exhausted');
++ assert.strictEqual(offer.status, 'down');
++ assert.strictEqual(offer.status_reason, 'the provider says the account is out of credit');
++ assert.strictEqual(offer.status_class, 'balance');
++ assert.ok(offer.status_since, 'the catalogue says since when');
assert.strictEqual(offer.free_tier, true, 'still a free-tier source by the rules');
assert.strictEqual(offer.free_now, false, 'but not free right now');
assert.strictEqual(flash.free_now, false);
ddiff --git a/test/fixtures/provider-failures.json b/test/fixtures/provider-failures.json
nnew file mode 100644
iindex 0000000..9d71adf
---- /dev/null
++++ b/test/fixtures/provider-failures.json
@@@ -0,0 +1,58 @@
++{
++ "note": "Recorded error shapes of the providers AI routes to (their own docs and production answers, checked 2026-11-09). Tests replay them; no test reaches the network. `class` is what server/providers/failure.js must say.",
++ "deepseek_balance": {
++ "class": "balance",
++ "status": 402,
++ "body": { "error": { "message": "Insufficient Balance", "type": "unknown_error", "param": null, "code": "invalid_request_error" } }
++ },
++ "openai_quota": {
++ "class": "balance",
++ "status": 429,
++ "body": { "error": { "message": "You exceeded your current quota, please check your plan and billing details.", "type": "insufficient_quota", "code": "insufficient_quota" } }
++ },
++ "anthropic_credit": {
++ "class": "balance",
++ "status": 400,
++ "body": { "type": "error", "error": { "type": "invalid_request_error", "message": "Your credit balance is too low to access the Anthropic API. Please go to Plans & Billing to upgrade or purchase credits." } }
++ },
++ "openrouter_credits": {
++ "class": "balance",
++ "status": 402,
++ "body": { "error": { "code": 402, "message": "Insufficient credits" } }
++ },
++ "openai_auth": {
++ "class": "auth",
++ "status": 401,
++ "body": { "error": { "message": "Incorrect API key provided: sk-test****", "type": "invalid_request_error", "code": "invalid_api_key" } }
++ },
++ "anthropic_auth": {
++ "class": "auth",
++ "status": 401,
++ "body": { "type": "error", "error": { "type": "authentication_error", "message": "invalid x-api-key" } }
++ },
++ "openrouter_rate": {
++ "class": "rate_limit",
++ "status": 429,
++ "body": { "error": { "code": 429, "message": "Rate limit exceeded" } }
++ },
++ "gemini_quota": {
++ "class": "rate_limit",
++ "status": 429,
++ "body": { "error": { "code": 429, "status": "RESOURCE_EXHAUSTED", "message": "Quota exceeded for quota metric 'Generate Content API requests per minute' and limit 'GenerateContentRequestsPerMinutePerProjectPerRegion' of service 'generativelanguage.googleapis.com'." } }
++ },
++ "deepseek_transient": {
++ "class": "transient",
++ "status": 503,
++ "body": { "error": { "message": "Service temporarily unavailable" } }
++ },
++ "context_too_long": {
++ "class": "request",
++ "status": 400,
++ "body": { "error": { "message": "This model's maximum context length is 8192 tokens", "type": "invalid_request_error", "code": "context_length_exceeded" } }
++ },
++ "bad_image": {
++ "class": "request",
++ "status": 400,
++ "body": { "error": { "message": "Invalid image data", "type": "invalid_request_error", "code": "invalid_image" } }
++ }
++}
ddiff --git a/test/helpers.js b/test/helpers.js
iindex ed30413..96084dd 100644
---- a/test/helpers.js
++++ b/test/helpers.js
@@@ -45,7 +45,7 @@ function tmpDir() {
// A boot on a dir used before is a restart on the same database: its database stays open.
const restartable = new Map();
--async function boot({ searchFetch, env = {}, dir, clock, fetchImpl, credentialFetch, research, machineModels, log, variantEvidence, community, communityEndpoints, webhooks, memoryTransport, memoryLookup } = {}) {
++async function boot({ searchFetch, env = {}, dir, clock, fetchImpl, credentialFetch, research, machineModels, log, variantEvidence, community, communityEndpoints, webhooks, memoryTransport, memoryLookup, ownerNotify } = {}) {
const reuse = dir !== undefined;
if (dir === undefined) dir = tmpDir();
const fullEnv = {
@@@ -62,7 +62,7 @@ async function boot({ searchFetch, env = {}, dir, clock, fetchImpl, credentialFe
// An injected endpoint store needs the migrated database, so a factory is called with it before the app boots.
const endpointStore = typeof communityEndpoints === 'function' ? communityEndpoints(testdb.db) : communityEndpoints;
const webhookStore = typeof webhooks === 'function' ? webhooks(testdb.db) : webhooks;
-- const h = await start({ searchFetch, config, db: testdb.db, env: fullEnv, log: log || silent, clock, fetchImpl, credentialFetch, research: researcher, machineModels, variantEvidence, community, communityEndpoints: endpointStore, webhooks: webhookStore, memoryTransport, memoryLookup });
++ const h = await start({ searchFetch, config, db: testdb.db, env: fullEnv, log: log || silent, clock, fetchImpl, credentialFetch, research: researcher, machineModels, variantEvidence, community, communityEndpoints: endpointStore, webhooks: webhookStore, memoryTransport, memoryLookup, ownerNotify });
const base = `http://127.0.0.1:${h.server.address().port}`;
return { ...h, base, dir, env: fullEnv, testdb, stop: async () => { await h.close(); if (!reuse) await testdb.close(); } };
}
ddiff --git a/test/provider-circuits.test.js b/test/provider-circuits.test.js
nnew file mode 100644
iindex 0000000..e08eb6d
---- /dev/null
++++ b/test/provider-circuits.test.js
@@@ -0,0 +1,240 @@
++'use strict';
++// Provider failure classes and the per provider+model circuit (migrations/0060).
++//
++// The recorded shapes of every provider's error answer are replayed against a local seam (no test
++// reaches the network). A balance or auth answer opens that offer's circuit at once; a run of
++// transient failures opens it for 30 s doubling to 10 min; rate limits and malformed requests never
++// open it. The state survives a restart, a key change resets it, the catalogue says up/degraded/down
++// with a plain reason and since when, and the owner hears about a balance/auth circuit once.
++const assert = require('assert');
++const { boot, request, token, suite, seamServer, ALL, tmpDir } = require('./helpers');
++const { ProviderError } = require('../server/providers/common');
++const { classifyFailure } = require('../server/providers/failure');
++const { createCatalog } = require('../server/exchange/catalog');
++const fixtures = require('./fixtures/provider-failures.json');
++
++const t = suite('provider-circuits');
++const admin = token('live', ALL);
++
++async function provider(h, body) {
++ const r = await request(h.base, 'POST', '/api/v1/providers', { tok: admin, body: { kind: 'http', auth_mode: 'none', timeout_ms: 5000, capabilities: ['chat', 'generate', 'json'], ...body } });
++ assert.strictEqual(r.status, 201, r.text);
++}
++async function model(h, providerKey, modelKey) {
++ const r = await request(h.base, 'POST', '/api/v1/models', { tok: admin, body: { provider_key: providerKey, model_key: modelKey, type: 'chat', cost: { in_per_mtok: 1, out_per_mtok: 1 } } });
++ assert.strictEqual(r.status, 201, r.text);
++}
++async function route(h, primary, fallbacks = []) {
++ const r = await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: admin, body: { primary, fallbacks, timeout_ms: 5000 } });
++ assert.strictEqual(r.status, 201, r.text);
++ return r.body.route;
++}
++const ask = (h, r, logs = []) => h.pool.execute(r, 'chat', { messages: [{ role: 'user', content: 'hello' }], maxTokens: 30, timeoutMs: 3000 }, { routeKey: r.key, routeVersion: r.version, logRequest: (e) => logs.push(e) });
++
++t.test('every recorded provider error shape maps to the right class', async () => {
++ for (const [name, fx] of Object.entries(fixtures)) {
++ if (!fx || !fx.class) continue;
++ const err = new ProviderError(String((fx.body && fx.body.error && fx.body.error.message) || name), { status: fx.status, body: fx.body, code: 'provider.http' });
++ assert.strictEqual(classifyFailure(err).class, fx.class, `${name} is ${fx.class}, got ${JSON.stringify(classifyFailure(err))}`);
++ }
++ const timeout = new ProviderError('no response within 5s', { code: 'provider.timeout' });
++ assert.strictEqual(classifyFailure(timeout).class, 'transient', 'a timeout is transient');
++ const unreachable = new ProviderError('connection failed: fetch failed', { code: 'provider.unreachable' });
++ assert.strictEqual(classifyFailure(unreachable).class, 'transient', 'a connection failure is transient');
++ assert.strictEqual(classifyFailure(new ProviderError('cancelled', { code: 'provider.cancelled' })), null, 'a cancellation is not a provider failure');
++ assert.strictEqual(classifyFailure(new Error('the backend fell over')).class, 'request', 'an unshaped non-retryable error is a request error');
++});
++
++t.test('balance opens only that offer: the next model answers, the owner hears once, the catalogue says down', async () => {
++ const seam = await seamServer((body) => {
++ if (body.model === 'model-a') return { status: 402, body: fixtures.deepseek_balance.body };
++ return { text: 'from model-b', model: 'model-b', usage: { input: 3, output: 2 } };
++ });
++ let now = 1762400000000;
++ const clock = { now: () => now };
++ const heard = [];
++ let h = await boot({ clock, ownerNotify: async (n) => { heard.push(n); return true; }, env: {
++ AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '5', AI_BREAKER_EXHAUSTED_COOLDOWN_MS: '900000',
++ } });
++ try {
++ await provider(h, { key: 'seam', base_url: seam.url });
++ await model(h, 'seam', 'model-a');
++ await model(h, 'seam', 'model-b');
++ const r = await route(h, { provider: 'seam', model: 'model-a' }, [{ provider: 'seam', model: 'model-b' }]);
++ const logs = [];
++ const first = await ask(h, r, logs);
++ assert.strictEqual(first.provider, 'seam');
++ assert.strictEqual(first.model, 'model-b', 'the same provider, the next offer, answers');
++ assert.strictEqual(seam.calls, 2, 'one call to each offer: the balance offer is not retried');
++ assert.strictEqual(seam.last.model, 'model-b');
++ const log = logs.find((e) => e.provider_key === 'seam' && e.model_key === 'model-a');
++ assert.ok(log, 'the failed offer is logged');
++ assert.strictEqual(log.error_class, 'balance', 'the request log records the class, not the body');
++
++ const a = await h.pool.health('seam', 'model-a');
++ const b = await h.pool.health('seam', 'model-b');
++ assert.strictEqual(a.state, 'open');
++ assert.strictEqual(a.failure_class, 'balance');
++ assert.ok(a.exhausted_at, 'a balance circuit is exhausted');
++ assert.strictEqual(a.status_reason, 'the provider says the account is out of credit');
++ assert.strictEqual(b.state, 'closed', 'the other model on the same provider is untouched');
++ assert.strictEqual((await h.pool.health('seam')).state, 'open', 'the provider roll-up is open');
++
++ const again = await ask(h, r);
++ assert.strictEqual(again.provider, 'seam');
++ assert.strictEqual(again.model, 'model-b');
++ assert.strictEqual(seam.calls, 3, 'the open offer is skipped on the next request');
++ const skipped = again.explain.attempts.find((x) => x.model === 'model-a' && x.status === 'skipped');
++ assert.strictEqual(skipped.error, 'exhausted', 'the skip says why');
++ assert.strictEqual(heard.length, 1, 'one owner message, not one per request');
++ assert.match(heard[0].message, /out of credit/);
++
++ const adminView = await request(h.base, 'GET', '/api/v1/providers/seam', { tok: admin });
++ assert.strictEqual(adminView.status, 200, adminView.text);
++ const health = adminView.body.provider.health;
++ assert.strictEqual(health.status_reason, 'the provider says the account is out of credit');
++ const circuit = health.circuits.find((c) => c.model_key === 'model-a');
++ assert.strictEqual(circuit.state, 'open');
++
++ const catalog = await request(h.base, 'GET', '/api/v1/exchange/catalog');
++ assert.strictEqual(catalog.status, 200, catalog.text);
++ const offer = catalog.body.models.find((m) => m.slug === 'seam/model-a').offers.find((o) => o.provider === 'seam');
++ assert.strictEqual(offer.status, 'down');
++ assert.strictEqual(offer.status_reason, 'the provider says the account is out of credit');
++ assert.strictEqual(offer.status_class, 'balance');
++ assert.ok(offer.status_since, 'the catalogue says since when');
++ const healthy = catalog.body.models.find((m) => m.slug === 'seam/model-b').offers.find((o) => o.provider === 'seam');
++ assert.strictEqual(healthy.status, 'up');
++
++ const metrics = await request(h.base, 'GET', '/metrics');
++ assert.ok(/ai_provider_circuit_open\{provider="seam",model="model-a",class="balance"\} 1\n/.test(metrics.text), metrics.text);
++ assert.ok(/ai_provider_circuit_open\{provider="seam",model="model-b",class=""\} 0\n/.test(metrics.text), metrics.text);
++ assert.ok(/ai_provider_failures_total\{provider="seam",class="balance"\} 1\n/.test(metrics.text), metrics.text);
++
++ // The 15-minute probe: still open before it, half-open after, and a second 402 restarts the pause.
++ now += 899999;
++ assert.strictEqual((await h.pool.health('seam', 'model-a')).state, 'open');
++ now += 2;
++ assert.strictEqual((await h.pool.health('seam', 'model-a')).state, 'half_open');
++ const probe = await ask(h, r);
++ assert.strictEqual(probe.model, 'model-b', 'the probe failed and the same request moved to the next offer');
++ assert.strictEqual(seam.calls, 5, 'the probe reached the half-open offer once, then the next offer answered');
++ assert.strictEqual((await h.pool.health('seam', 'model-a')).state, 'open', 'the failed probe opens it again');
++ assert.strictEqual(heard.length, 1, 'a reopened circuit is the same open episode: no second alert');
++ } finally { await h.stop(); await seam.close(); }
++});
++
++t.test('a run of transient failures opens for 30 s, doubles on a failed probe, and closes on success', async () => {
++ let failing = true;
++ const seam = await seamServer(() => failing ? { status: 503, body: { error: { message: 'overloaded' } } } : { text: 'back', model: 'm', usage: { input: 2, output: 1 } });
++ let now = 1762400000000;
++ const clock = { now: () => now };
++ const h = await boot({ clock, env: {
++ AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '3',
++ AI_BREAKER_BACKOFF_MS: '30000', AI_BREAKER_MAX_BACKOFF_MS: '60000',
++ } });
++ try {
++ await provider(h, { key: 'seam', base_url: seam.url });
++ await model(h, 'seam', 'm');
++ const r = await route(h, { provider: 'seam', model: 'm' });
++ for (let i = 0; i < 2; i++) await ask(h, r).catch(() => {});
++ assert.strictEqual((await h.pool.health('seam', 'm')).state, 'closed', 'two failures are below the threshold');
++ await ask(h, r).catch(() => {});
++ let health = await h.pool.health('seam', 'm');
++ assert.strictEqual(health.state, 'open');
++ assert.strictEqual(health.failure_class, 'transient');
++ assert.strictEqual(health.backoff_ms, 30000, 'the first pause is 30 seconds');
++ assert.strictEqual(health.retry_at - health.opened_at, 30000);
++
++ now += 30000;
++ assert.strictEqual((await h.pool.health('seam', 'm')).state, 'half_open', 'one real request may probe');
++ await ask(h, r).catch(() => {});
++ health = await h.pool.health('seam', 'm');
++ assert.strictEqual(health.state, 'open');
++ assert.strictEqual(health.backoff_ms, 60000, 'a failed probe doubles the pause');
++
++ now += 60000;
++ failing = false;
++ const ok = await ask(h, r);
++ assert.strictEqual(ok.provider, 'seam');
++ health = await h.pool.health('seam', 'm');
++ assert.strictEqual(health.state, 'closed', 'a successful probe closes it');
++ assert.strictEqual(health.consecutive_failures, 0);
++ } finally { await h.stop(); await seam.close(); }
++});
++
++t.test('rate limits and malformed requests never open a circuit; the class is logged', async () => {
++ const seam = await seamServer((body) => {
++ const fx = fixtures[body.model];
++ return fx ? { status: fx.status, body: fx.body } : { text: 'ok', model: body.model, usage: { input: 1, output: 1 } };
++ });
++ const h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', AI_BREAKER_FAILURES: '5' } });
++ try {
++ await provider(h, { key: 'seam', base_url: seam.url });
++ for (const name of ['deepseek_balance', 'anthropic_credit', 'openai_auth', 'anthropic_auth', 'openrouter_rate', 'gemini_quota', 'deepseek_transient', 'context_too_long', 'bad_image']) {
++ await model(h, 'seam', name);
++ const r = await route(h, { provider: 'seam', model: name });
++ const logs = [];
++ await ask(h, r, logs).catch(() => {});
++ const health = await h.pool.health('seam', name);
++ const expected = fixtures[name].class;
++ assert.strictEqual(logs.find((e) => e.status === 'error').error_class, expected, `${name} logged as ${expected}`);
++ if (expected === 'rate_limit' || expected === 'request' || expected === 'transient') {
++ assert.strictEqual(health.state, 'closed', `${name} must not open the circuit`);
++ } else {
++ assert.strictEqual(health.state, 'open', `${name} opens at once`);
++ }
++ }
++ const counts = await h.db.prepare('SELECT failure_class, SUM(count)::bigint AS n FROM provider_failure_counts GROUP BY failure_class').all();
++ const by = Object.fromEntries(counts.map((x) => [x.failure_class, Number(x.n)]));
++ assert.ok(by.balance >= 2 && by.auth >= 2 && by.rate_limit >= 2 && by.request >= 2 && by.transient >= 1, JSON.stringify(by));
++ } finally { await h.stop(); await seam.close(); }
++});
++
++t.test('the catalogue measures degraded from the last hour, never below 10 requests', async () => {
++ const h = await boot({ env: { AI_STUB_FALLBACK: 'false' } });
++ try {
++ await provider(h, { key: 'seam', base_url: 'http://127.0.0.1:1/x' });
++ await model(h, 'seam', 'm');
++ const catalog = createCatalog({ db: h.db, registry: h.registry, pool: h.pool, env: h.env, config: h.config });
++ const offerOf = async () => (await catalog.models({ days: 7 })).models.find((m) => m.slug === 'seam/m').offers.find((o) => o.provider === 'seam');
++ assert.strictEqual((await offerOf()).status, 'up');
++ const now = Date.now();
++ const ins = h.db.prepare('INSERT INTO requests (operation, provider_key, model_key, status, error_class, created_at) VALUES (?, ?, ?, ?, ?, ?)');
++ for (let i = 0; i < 9; i++) await ins.run('chat', 'seam', 'm', 'ok', null, new Date(now - 300000).toISOString());
++ await ins.run('chat', 'seam', 'm', 'error', 'transient', new Date(now - 120000).toISOString());
++ let offer = await offerOf();
++ assert.strictEqual(offer.status, 'degraded', '1 of 10 is over 5%');
++ assert.strictEqual(offer.status_reason, 'too many errors in the last hour');
++ assert.ok(offer.status_since, 'the first error of the window');
++ await h.db.prepare("DELETE FROM requests WHERE provider_key='seam' AND model_key='m' AND status='ok' AND created_at < ?").run(new Date(now - 1000).toISOString());
++ offer = await offerOf();
++ assert.strictEqual(offer.status, 'up', '9 requests is below the measured floor');
++ } finally { await h.stop(); }
++});
++
++t.test('a restart keeps the circuit; a new provider key clears it', async () => {
++ const seam = await seamServer(() => ({ status: 401, body: fixtures.openai_auth.body }));
++ const dir = tmpDir();
++ const env = { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0', TEST_KEY_A: 'key-aaa', TEST_KEY_B: 'key-bbb' };
++ let h = await boot({ dir, env });
++ try {
++ await provider(h, { key: 'keyed', base_url: seam.url, auth_mode: 'bearer', secret_ref: 'env:TEST_KEY_A' });
++ await model(h, 'keyed', 'm');
++ const r = await route(h, { provider: 'keyed', model: 'm' });
++ await ask(h, r).catch(() => {});
++ assert.strictEqual((await h.pool.health('keyed', 'm')).state, 'open');
++ assert.strictEqual((await h.pool.health('keyed', 'm')).failure_class, 'auth');
++ assert.strictEqual((await h.pool.health('keyed', 'm')).status_reason, 'the key was refused');
++ await h.stop();
++ h = await boot({ dir, env });
++ assert.strictEqual((await h.pool.health('keyed', 'm')).state, 'open', 'the circuit survived the restart');
++ const counts = await h.db.prepare("SELECT count FROM provider_failure_counts WHERE provider_key='keyed' AND failure_class='auth'").get();
++ assert.ok(Number(counts.count) >= 1, 'the failure count survived too');
++ const changed = await request(h.base, 'PATCH', '/api/v1/providers/keyed', { tok: admin, body: { secret_ref: 'env:TEST_KEY_B' } });
++ assert.strictEqual(changed.status, 200, changed.text);
++ assert.strictEqual((await h.pool.health('keyed', 'm')).state, 'closed', 'a new key closes the old key\'s circuit');
++ } finally { await h.stop(); await seam.close(); }
++});
++
++t.run();