openvibers/openvibe.ai Commits

Make the owner's three economy switches act (#448)

OP OpenVibers committed 3 days ago · 92d15bcac4 · 12 files, +540 −60

* Make the owner's three economy switches act

* Merge remote-tracking branch 'origin/main' into feat/economy-switches

Changes

ddiff --git a/migrations/0073_economy_switches.sql b/migrations/0073_economy_switches.sql
nnew file mode 100644
iindex 0000000..7f07570
---- /dev/null
++++ b/migrations/0073_economy_switches.sql
@@@ -0,0 +1,12 @@
++-- phase: expand
++-- The owner's economy switches act (2026-11-10). Prompt caching is now gated by the owner's
++-- owner_economy.prompt_cache row, so the saving it reports needs the write side of the provider's
++-- cache as well as its reads: the adapters already return cache_write (server/providers/usage.js),
++-- and requests only kept tokens_cached. This column records the writes, so the panel can price a
++-- cache read against the input price and subtract the write premium from the provider's own
++-- cache-write card (the price book's cost_cache_write_per_mtok, migrations/0022).
++-- Expand-only: the column is nullable-free with a default, and no existing row changes meaning.
++
++ALTER TABLE requests ADD COLUMN IF NOT EXISTS tokens_cache_write bigint NOT NULL DEFAULT 0;
++
++CREATE INDEX IF NOT EXISTS idx_requests_created_cached ON requests(created_at) WHERE tokens_cached > 0;
ddiff --git a/server/batch/batcher.js b/server/batch/batcher.js
iindex 8074cca..3ae68ec 100644
---- a/server/batch/batcher.js
++++ b/server/batch/batcher.js
@@@ -28,7 +28,7 @@ const crypto = require('crypto');
const { AiError, providerSecret } = require('../util');
const { retailRate, accessOf } = require('../providers/model-catalog');
const { retailChargeNano } = require('../credits');
--const { providerConfig } = require('./providers');
++const { providerConfig, batchProviderFor } = require('./providers');
const { createOpenAiCompatibleAdapter } = require('./providers/openai-compatible');
const { createAnthropicAdapter } = require('./providers/anthropic');
const { createGeminiAdapter } = require('./providers/gemini');
@@@ -111,10 +111,10 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
/** The provider's adapter, built once and kept. */
async function adapterFor(providerKey) {
if (adapters.has(providerKey)) return adapters.get(providerKey);
-- const cfg = providerConfig(providerKey);
-- if (!cfg || !cfg.enabled) throw new AiError(422, 'batch.no_native_api', 'this provider has no verified batch API');
const row = await registry.getProvider(providerKey);
if (!row || row.status !== 'active') throw new AiError(422, 'batch.provider_inactive', 'this provider is not active');
++ const cfg = batchProviderFor(providerKey, row, env);
++ if (!cfg) throw new AiError(422, 'batch.no_native_api', 'this provider has no configured batch API');
const factory = ADAPTERS[cfg.adapter];
if (!factory) throw new AiError(422, 'batch.no_native_api', 'this provider has no adapter for its batch API');
const adapter = factory({
@@@ -220,10 +220,10 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
const capacity = require('./capacity'); // lazy: capacity requires this module for its cards
for (const offer of offers || []) {
if (!offer || !offer.provider || !offer.model) continue;
-- const cfg = providerConfig(offer.provider);
-- // Only a provider whose batch API is enabled and whose adapter speaks the OpenAI batch shape.
-- if (!cfg || !cfg.enabled || cfg.adapter !== 'openai-compatible') continue;
const provider = await registry.getProvider(offer.provider);
++ const cfg = batchProviderFor(offer.provider, provider, env);
++ // Only a provider whose batch API is configured for this record.
++ if (!cfg) continue;
if (!provider || provider.status !== 'active') continue;
const cards = await cardsFor(offer.provider, offer.model);
if (!cards || !cards.batch) continue;
@@@ -289,8 +289,9 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
*/
async function enqueue({ principal, provider, model, prompt, kind = 'service', customer = null, deadlineHours = null, discount = null, prefer = null, costCheck = null, requestId = null, maxTokens = 4096, holdCard = null, route = null, charge = null, servicePaid = false, maxCostNano = null }) {
if (settings.enabled === false) throw new AiError(503, 'batch.disabled', 'native batches are not open on this service');
-- const cfg = providerConfig(provider);
-- if (!cfg || !cfg.enabled) throw new AiError(422, 'batch.no_native_api', 'this provider has no verified batch API');
++ const providerRow = await registry.getProvider(provider);
++ const cfg = batchProviderFor(provider, providerRow, env);
++ if (!cfg) throw new AiError(422, 'batch.no_native_api', 'this provider has no configured batch API');
const cards = await cardsFor(provider, model);
if (!cards) throw new AiError(422, 'batch.no_price_card', 'this provider has no price card for that model');
const body = { workflow: 'ai.complete', options: { latency: 'batch' }, input: { messages: [{ role: 'user', content: prompt }] } };
@@@ -358,7 +359,7 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
AND deadline_at > ?::timestamptz ORDER BY created_at LIMIT ?`).all(providerKey, modelKey, iso(), Number(settings.maxItemsPerGroup));
if (!items || !items.length) return null;
const adapter = await adapterFor(providerKey);
-- const cfg = providerConfig(providerKey);
++ const cfg = batchProviderFor(providerKey, await registry.getProvider(providerKey), env);
const jobId = `pb_${crypto.randomBytes(10).toString('hex')}`;
const payload = items.map((i) => ({ customId: i.id, body: JSON.parse(i.request_body || '{}') }));
await db.prepare(`INSERT INTO provider_batches (id, provider, model, adapter, status, item_count, discount, created_at, updated_at)
ddiff --git a/server/batch/capacity.js b/server/batch/capacity.js
iindex 66de0ac..6f8d6c0 100644
---- a/server/batch/capacity.js
++++ b/server/batch/capacity.js
@@@ -26,7 +26,7 @@
*/
const { retailRate } = require('../providers/model-catalog');
const { insideWindow } = require('../pricebook/offpeak');
--const { enabledProviders } = require('./providers');
++const { PROVIDERS, batchProviderFor } = require('./providers');
const { scaleCard, toCard } = require('./batcher');
/** Half of a nano price, rounded down: the customer is never charged more than half. */
@@@ -173,10 +173,11 @@ async function batchCapacity(model, at, { registry = null, community = null, pri
// batch on a paid model then runs at any hour, instead of waiting for an off-peak window.
if (registry) {
const rows = await registry.listModels();
-- for (const [key, cfg] of Object.entries(enabledProviders())) {
-- if (Number(cfg.discount) > 0.5) continue; // the real price would not fit inside the charge
++ for (const [key, base] of Object.entries(PROVIDERS)) {
++ if (Number(base.discount) > 0.5) continue; // the real price would not fit inside the charge
const provider = await registry.getProvider(key);
-- if (!provider || provider.status !== 'active') continue;
++ const cfg = batchProviderFor(key, provider);
++ if (!cfg) continue;
const row = (rows || []).find((r) => String(r.provider_key) === key && String(r.model_key) === String(model.model_key) && r.status === 'active');
if (!row) continue;
const retail = toCard(retailRate(row));
ddiff --git a/server/batch/providers/anthropic.js b/server/batch/providers/anthropic.js
iindex a9737de..3b07df9 100644
---- a/server/batch/providers/anthropic.js
++++ b/server/batch/providers/anthropic.js
@@@ -28,6 +28,38 @@ const COMPLETION_WINDOW = '24h';
/** Anthropic needs a max_tokens on every message; the batcher asks for a bounded answer. */
const DEFAULT_MAX_TOKENS = 4096;
++/** One text content block, from a string or an array of OpenAI/Anthropic-style text parts. */
++function textOf(content) {
++ if (typeof content === 'string') return content;
++ if (!Array.isArray(content)) return String(content == null ? '' : content);
++ return content.map((p) => (typeof p === 'string' ? p : (p && (p.text || p.content)) || '')).filter(Boolean).join('\n');
++}
++
++/**
++ * The batcher builds an OpenAI Chat Completions body (server/workflows/engine.js buildBody). Anthropic's
++ * Message Batches take system separately and only user/assistant messages, so the same words are moved
++ * to the shape it documents: a system message becomes the system array, the rest keep their role order.
++ */
++function paramsOf(body = {}) {
++ const system = [];
++ const fromBody = body.system == null ? '' : (Array.isArray(body.system) ? body.system.map((b) => (typeof b === 'string' ? b : b && b.text || '')).join('\n\n') : String(body.system));
++ if (fromBody.trim()) system.push({ type: 'text', text: fromBody });
++ const messages = [];
++ for (const m of body.messages || []) {
++ if (!m || !m.role) continue;
++ const text = textOf(m.content);
++ if (m.role === 'system') { if (text.trim()) system.push({ type: 'text', text }); continue; }
++ messages.push({ role: m.role === 'assistant' ? 'assistant' : 'user', content: [{ type: 'text', text }] });
++ }
++ return {
++ model: body.model,
++ max_tokens: body.max_tokens || body.max_completion_tokens || DEFAULT_MAX_TOKENS,
++ messages,
++ ...(system.length ? { system } : {}),
++ ...(body.temperature != null ? { temperature: body.temperature } : {}),
++ };
++}
++
function createAnthropicAdapter({ baseUrl, apiKey, fetchImpl = globalThis.fetch, limits = {}, timeoutMs = 60000 } = {}) {
const base = String(baseUrl || '').replace(/\/+$/, '');
const maxItems = Number(limits.maxItems) > 0 ? Number(limits.maxItems) : 100000;
@@@ -69,7 +101,7 @@ function createAnthropicAdapter({ baseUrl, apiKey, fetchImpl = globalThis.fetch,
if (items.length > maxItems) throw new BatchProviderError(`this provider takes at most ${maxItems} requests in one batch`, { code: 'batch.too_many_items', status: 422 });
const bytes = Buffer.byteLength(JSON.stringify(items), 'utf8');
if (bytes > maxBytes) throw new BatchProviderError(`this provider's limit is ${maxBytes} bytes and this batch is ${bytes}`, { code: 'batch.file_too_large', status: 422 });
-- return { requests: items.map((i) => ({ custom_id: i.customId, params: { model: i.body.model, max_tokens: i.body.max_tokens || DEFAULT_MAX_TOKENS, messages: i.body.messages } })), bytes };
++ return { requests: items.map((i) => ({ custom_id: i.customId, params: paramsOf(i.body) })), bytes };
},
async submit({ items, completionWindow = COMPLETION_WINDOW, metadata = null } = {}) {
ddiff --git a/server/batch/providers/gemini.js b/server/batch/providers/gemini.js
iindex 237cf5a..1b0dbd0 100644
---- a/server/batch/providers/gemini.js
++++ b/server/batch/providers/gemini.js
@@@ -25,6 +25,31 @@ const BatchProviderError = require('./openai-compatible').BatchProviderError;
/** The window the provider promises for its answers. */
const COMPLETION_WINDOW = '24h';
++function textOf(content) {
++ if (typeof content === 'string') return content;
++ if (!Array.isArray(content)) return String(content == null ? '' : content);
++ return content.map((p) => (typeof p === 'string' ? p : (p && (p.text || p.content)) || '')).filter(Boolean).join('\n');
++}
++
++/** The OpenAI Chat Completions body the batcher builds -> Gemini's contents/systemInstruction shape. */
++function requestOf(body = {}) {
++ const fromBody = body.system == null ? '' : (Array.isArray(body.system) ? body.system.map((b) => (typeof b === 'string' ? b : b && b.text || '')).join('\n\n') : String(body.system));
++ const contents = [];
++ const system = [];
++ if (String(fromBody).trim()) system.push(fromBody);
++ for (const m of body.messages || []) {
++ if (!m || !m.role) continue;
++ const text = textOf(m.content);
++ if (m.role === 'system') { if (text.trim()) system.push(text); continue; }
++ contents.push({ role: m.role === 'assistant' ? 'model' : 'user', parts: [{ text }] });
++ }
++ return {
++ contents: contents.length ? contents : [{ role: 'user', parts: [{ text: '' }] }],
++ ...(system.length ? { systemInstruction: { parts: [{ text: system.join('\n\n') }] } } : {}),
++ ...(body.temperature != null ? { generationConfig: { temperature: body.temperature } } : {}),
++ };
++}
++
function createGeminiAdapter({ baseUrl, apiKey, fetchImpl = globalThis.fetch, limits = {}, timeoutMs = 60000 } = {}) {
const base = String(baseUrl || '').replace(/\/+$/, '');
const maxItems = Number(limits.maxItems) > 0 ? Number(limits.maxItems) : 10000;
@@@ -63,7 +88,7 @@ function createGeminiAdapter({ baseUrl, apiKey, fetchImpl = globalThis.fetch, li
buildFile(items) {
if (!Array.isArray(items) || !items.length) throw new BatchProviderError('a batch needs at least one request');
if (items.length > maxItems) throw new BatchProviderError(`this provider takes at most ${maxItems} requests in one batch`, { code: 'batch.too_many_items', status: 422 });
-- const requests = items.map((i) => ({ request: { contents: [{ role: 'user', parts: [{ text: i.body.messages.map((m) => m.content).join('\n') }] }] } }));
++ const requests = items.map((i) => ({ request: requestOf(i.body) }));
const bytes = Buffer.byteLength(JSON.stringify(requests), 'utf8');
if (bytes > maxBytes) throw new BatchProviderError(`this provider's limit is ${maxBytes} bytes and this batch is ${bytes}`, { code: 'batch.file_too_large', status: 422 });
return { requests, bytes };
ddiff --git a/server/batch/providers/index.js b/server/batch/providers/index.js
iindex 20aeb59..d32aa59 100644
---- a/server/batch/providers/index.js
++++ b/server/batch/providers/index.js
@@@ -23,6 +23,8 @@
* half and are therefore not eligible for a half-price customer batch, though the network's own
* batches may still use them.
*/
++const { providerSecret } = require('../../util');
++
const PROVIDERS = Object.freeze({
// OpenAI's own Batch API: a JSONL file uploaded with purpose 'batch', a batch created with
// completion_window '24h' and endpoint '/v1/chat/completions', and the answers in a file whose
@@@ -77,6 +79,7 @@ const PROVIDERS = Object.freeze({
discount: 0.5,
window: '24h',
enabled: false, // no production key yet: the adapter exists, the route does not
++ when_configured: true, // and it turns on by itself for a provider with a key (batchProviderFor)
limits: { maxItems: 100000, maxBytes: 256 * 1024 * 1024, oneModelPerFile: false },
source: 'https://docs.anthropic.com/en/docs/build-with-claude/batch-processing',
},
@@@ -88,15 +91,38 @@ const PROVIDERS = Object.freeze({
discount: 0.5,
window: '24h',
enabled: false,
++ when_configured: true,
limits: { maxItems: 10000, maxBytes: 100 * 1024 * 1024, oneModelPerFile: true },
source: 'https://ai.google.dev/gemini-api/docs/batch',
},
});
--/** The providers that are enabled for a native batch, and their configuration. */
--function enabledProviders() {
++/** Is this provider record actually configured (a key slot with a key, or a keyless local seam)? */
++function providerConfigured(provider, env = process.env) {
++ if (!provider) return false;
++ if (provider.auth_mode === 'none') return true;
++ return Boolean(providerSecret(provider, env));
++}
++
++/**
++ * The batch configuration usable for one provider record: a provider that is always on, or one
++ * written for a provider that is on and has a key (`when_configured`). Anthropic and Gemini ship
++ * their adapters this way: the route is off until the owner configures the provider.
++ */
++function batchProviderFor(key, provider = null, env = process.env) {
++ const cfg = PROVIDERS[String(key || '')] || null;
++ if (!cfg) return null;
++ if (cfg.enabled) return cfg;
++ if (cfg.when_configured && provider && provider.status === 'active' && providerConfigured(provider, env)) return cfg;
++ return null;
++}
++
++/** The providers that are enabled for a native batch, and their configuration. `configured` names the
++ * providers that are on and keyed, so a when_configured adapter can join them. */
++function enabledProviders({ configured = [] } = {}) {
++ const keys = configured instanceof Set ? configured : new Set(configured);
const out = {};
-- for (const [key, cfg] of Object.entries(PROVIDERS)) if (cfg.enabled) out[key] = cfg;
++ for (const [key, cfg] of Object.entries(PROVIDERS)) if (cfg.enabled || (cfg.when_configured && keys.has(key))) out[key] = cfg;
return out;
}
@@@ -105,4 +131,4 @@ function providerConfig(key) {
return PROVIDERS[String(key || '')] || null;
}
--module.exports = { PROVIDERS, enabledProviders, providerConfig };
++module.exports = { PROVIDERS, enabledProviders, providerConfig, providerConfigured, batchProviderFor };
ddiff --git a/server/index.js b/server/index.js
iindex cd11c37..b393162 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -88,7 +88,10 @@ 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, ownerNotify });
++ // The owner's inference configuration: one read for the router's economy switches and the runs'
++ // answer cache and batch decisions (server/owner-config.js).
++ const ownerConfig = require('./owner-config').createOwnerConfig({ db, registry, clock });
++ const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log, credentialFetch, machineModels: machines, community, bench, variantEvidence, adultStatus, offPeak, creditGrants, ownerNotify, ownerConfig });
// 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
@@@ -166,7 +169,7 @@ async function start({ searchFetch = null, config, db: givenDb = null, clock = {
// BA3: the runs and the batcher need each other (a batch-class run is created first, then queued), so
// the runs hold this facade and the batcher fills it in below, before any request can arrive.
const batchRef = {};
-- const runs = createRuns({ choices, db, registry, engine, cache, quotas, config, clock, log, userModules, credentials, preferences, tiers, projectSecrets, pool, runBudgets, credits: creditsRef, community, presets, receipts, abuse, webhooks, batch: batchRef,
++ const runs = createRuns({ choices, db, registry, engine, cache, quotas, config, clock, log, userModules, credentials, preferences, tiers, projectSecrets, pool, runBudgets, credits: creditsRef, community, presets, receipts, abuse, webhooks, batch: batchRef, ownerConfig,
// BA2: a run that waits for a provider's batch answer is not interrupted by a restart: it stays
// queued until the batcher finishes it (the provider is still working on it).
batchWaiting: async (runId) => {
ddiff --git a/server/owner-config.js b/server/owner-config.js
iindex efd4bed..48ac485 100644
---- a/server/owner-config.js
++++ b/server/owner-config.js
@@@ -22,6 +22,49 @@ const { isUncensored } = require('./providers/pool-providers');
const KINDS = ['chat', 'uncensored', 'embed', 'image', 'transcribe', 'batch'];
const STRATEGIES = ['auto', 'cheapest', 'fastest', 'fixed'];
++/**
++ * A kind with no row keeps the behaviour it had before the owner configured it: batches, prompt caching
++ * and the answer cache are on, the small model first and the output limit are off. That is what the
++ * columns' defaults say (migrations/0072); an owner row is only ever needed to turn a way of paying
++ * less OFF. The uncensored kind never serves a stored answer, so its answer cache reads off.
++ */
++const ECONOMY_DEFAULTS = Object.freeze({ batches: true, prompt_cache: true, small_first: false, output_limit: null, answer_cache: true });
++const economyDefaultsFor = (capability) => ({ ...ECONOMY_DEFAULTS, ...(capability === 'uncensored' ? { answer_cache: false } : {}) });
++
++/** The kind of work a workflow is, from its own steps (server/workflows). */
++function workflowKind(wf) {
++ const steps = (wf && Array.isArray(wf.steps)) ? wf.steps : [];
++ if (steps.length && steps.every((s) => s && s.kind === 'embed')) return 'embed';
++ if (steps.length && steps.every((s) => s && s.kind === 'transcribe')) return 'transcribe';
++ if (steps.length && steps.every((s) => s && (s.kind === 'image' || s.kind === 'picture'))) return 'image';
++ return 'chat';
++}
++
++/**
++ * The economy kind of one run, as the router names it (server/providers/index.js ownerCapability):
++ * uncensored is its own kind, an embedding/image/transcription workflow keeps its kind, a deferrable
++ * run of any other workflow is 'batch' work, and everything else is chat. Batches of a kind are
++ * switched on that capability's row.
++ */
++function capabilityOfRun(wf, { contentPolicy = 'standard', latencyClass = null } = {}) {
++ if (contentPolicy === 'uncensored') return 'uncensored';
++ const kind = workflowKind(wf);
++ if (kind !== 'chat') return kind;
++ if (latencyClass === 'batch') return 'batch';
++ return 'chat';
++}
++
++/** The SQL for one run's economy kind, given aliases for runs and (optionally) requests. */
++function capabilitySql(runAlias = 'r', requestAlias = null) {
++ const r = runAlias;
++ const q = requestAlias;
++ const uncensored = `COALESCE(${r}.options, '') LIKE '%"policy":"uncensored"%'`;
++ const batch = `${r}.latency_class = 'batch'`;
++ const embed = q ? `${q}.operation = 'embed'` : `(${r}.workflow_key = 'ai.embed' OR ${r}.route_key LIKE '%embed%')`;
++ const image = q ? `${q}.operation = 'image'` : `(${r}.workflow_key = 'ai.image' OR ${r}.workflow_key LIKE '%picture%' OR ${r}.route_key LIKE '%image%')`;
++ const transcribe = q ? `${q}.operation = 'transcribe'` : `(${r}.workflow_key LIKE '%transcribe%' OR ${r}.route_key LIKE '%transcribe%')`;
++ return `CASE WHEN ${uncensored} THEN 'uncensored' WHEN ${batch} THEN 'batch' WHEN ${embed} THEN 'embed' WHEN ${image} THEN 'image' WHEN ${transcribe} THEN 'transcribe' ELSE 'chat' END`;
++}
function createOwnerConfig({ db, registry, clock = { now: () => Date.now() } } = {}) {
const now = () => new Date(clock.now()).toISOString();
@@@ -119,18 +162,34 @@ function createOwnerConfig({ db, registry, clock = { now: () => Date.now() } } =
// ── Economy toggles ────────────────────────────────────────
async function getEconomy() {
const rows = await db.prepare('SELECT * FROM owner_economy ORDER BY capability').all();
-- return rows.map((r) => ({ capability: r.capability, batches: !!r.batches, prompt_cache: !!r.prompt_cache, small_first: !!r.small_first, output_limit: r.output_limit, answer_cache: !!r.answer_cache, updated_by: r.updated_by, updated_at: r.updated_at }));
++ const saved = new Map(rows.map((r) => [r.capability, r]));
++ return KINDS.map((capability) => {
++ const r = saved.get(capability);
++ return {
++ capability,
++ ...economyDefaultsFor(capability),
++ ...(r ? { batches: !!r.batches, prompt_cache: !!r.prompt_cache, small_first: !!r.small_first, output_limit: r.output_limit, answer_cache: capability === 'uncensored' ? false : !!r.answer_cache } : {}),
++ updated_by: r ? r.updated_by : null,
++ updated_at: r ? r.updated_at : null,
++ };
++ });
++ }
++ async function economyFor(capability) {
++ return (await getEconomy()).find((r) => r.capability === capability) || { capability, ...economyDefaultsFor(capability) };
}
async function setEconomy(input, { actor = 'system', trace = null } = {}) {
const capability = kind((input || {}).capability);
const num = (v) => (v == null || v === '' ? null : Number(v));
const output_limit = num((input || {}).output_limit);
if (output_limit != null && !(Number.isFinite(output_limit) && output_limit > 0)) throw new AiError(422, 'ai.invalid', 'output_limit is a positive token count');
++ // The uncensored kind never serves a stored answer (server/runs.js), so its switch is recorded
++ // off whatever is sent: a row must never read as if a reused adult answer were possible.
++ const answer_cache = capability === 'uncensored' ? false : (input || {}).answer_cache === true;
const t = now();
await db.prepare(`INSERT INTO owner_economy (capability, batches, prompt_cache, small_first, output_limit, answer_cache, updated_by, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(capability) DO UPDATE SET batches = excluded.batches, prompt_cache = excluded.prompt_cache, small_first = excluded.small_first, output_limit = excluded.output_limit, answer_cache = excluded.answer_cache, updated_by = excluded.updated_by, updated_at = excluded.updated_at`)
-- .run(capability, (input || {}).batches === true, (input || {}).prompt_cache === true, (input || {}).small_first === true, output_limit, (input || {}).answer_cache === true, actor, t);
-- await registry.audit(actor, 'owner_economy.update', 'owner_economy', capability, { trace, metadata: { batches: (input || {}).batches === true, prompt_cache: (input || {}).prompt_cache === true, small_first: (input || {}).small_first === true, output_limit, answer_cache: (input || {}).answer_cache === true } });
++ .run(capability, (input || {}).batches === true, (input || {}).prompt_cache === true, (input || {}).small_first === true, output_limit, answer_cache, actor, t);
++ await registry.audit(actor, 'owner_economy.update', 'owner_economy', capability, { trace, metadata: { batches: (input || {}).batches === true, prompt_cache: (input || {}).prompt_cache === true, small_first: (input || {}).small_first === true, output_limit, answer_cache } });
return (await getEconomy()).find((r) => r.capability === capability);
}
@@@ -165,15 +224,31 @@ function createOwnerConfig({ db, registry, clock = { now: () => Date.now() } } =
async function savings(capability) {
const since = new Date(clock.now() - 7 * 86400000).toISOString();
const sinceDay = new Date(clock.now() - 6 * 86400000).toISOString().slice(0, 10);
-- const req = await db.prepare(`SELECT COALESCE(SUM(tokens_cached), 0)::bigint AS cached, COALESCE(SUM(tokens_in), 0)::bigint AS tin, COALESCE(SUM(tokens_out), 0)::bigint AS tout, COUNT(*)::bigint AS calls
-- FROM requests WHERE created_at >= ?`).get(since).catch(() => ({ cached: 0, tin: 0, tout: 0, calls: 0 }));
-- const cachedRuns = await db.prepare(`SELECT COUNT(*)::bigint AS n FROM runs WHERE created_at >= ? AND status = 'cached'`).get(since).catch(() => ({ n: 0 }));
-- // Batch savings are measurable: a batch run's cost is at the batch price; the retail card is
-- // the model's list price. Where the model row has no price, the row reports the batch count
-- // so the number stays honest (no invented savings).
++ // ── Prompt cache: the cached reads really served, and what they cost against the input price.
++ // The price book's cache-write price is part of the arithmetic: a write costs its premium, and
++ // it is only a saving when the reads have paid for it back. A model with no card counts no
++ // dollars (the tokens are still shown, never invented money).
++ const promptRows = await db.prepare(`WITH x AS (
++ SELECT ${capabilitySql('r', 'rq')} AS capability, rq.tokens_cached AS cached, rq.tokens_cache_write AS written, rq.tokens_in AS tin,
++ rq.provider_key, rq.model_key
++ FROM requests rq LEFT JOIN runs r ON r.id = rq.run_id WHERE rq.created_at >= ?)
++ SELECT capability, COALESCE(SUM(cached), 0)::bigint AS cached, COALESCE(SUM(tin), 0)::bigint AS tin, COUNT(*)::bigint AS calls,
++ COALESCE(SUM(GREATEST(0,
++ COALESCE(cached, 0) * (COALESCE(m.cost_in_per_mtok, 0) - COALESCE(m.cost_cached_per_mtok, COALESCE(m.cost_in_per_mtok, 0) * 0.1, 0))
++ - COALESCE(written, 0) * (COALESCE(m.cost_cache_write_per_mtok, COALESCE(m.cost_in_per_mtok, 0)) - COALESCE(m.cost_in_per_mtok, 0))
++ )), 0) / 1000000 AS saved_usd
++ FROM x LEFT JOIN models m ON m.provider_key = x.provider_key AND m.model_key = x.model_key
++ WHERE capability = ? GROUP BY capability`).get(since, capability).catch(() => null);
++ // ── Answer cache: the reused runs, and the source run's cost that each one did not pay again.
++ const answerRow = await db.prepare(`SELECT COUNT(*)::bigint AS n, COALESCE(SUM(src.cost_usd), 0) AS saved_usd
++ FROM runs r JOIN runs src ON src.id = r.cached_from
++ WHERE r.created_at >= ? AND r.status = 'cached' AND ${capabilitySql('r')} = ?`).get(since, capability).catch(() => null);
++ // ── Batches: a batch run's cost is at the batch price; the retail card is the model's list
++ // price. Where the model row has no price, the row reports the batch count so the number stays
++ // honest (no invented savings).
const batchRuns = await db.prepare(`SELECT m.provider_key, m.model_key, r.cost_usd, m.cost_in_per_mtok, m.cost_out_per_mtok, r.tokens_in, r.tokens_out
FROM runs r LEFT JOIN models m ON m.provider_key = r.provider_key AND m.model_key = r.model_key
-- WHERE r.created_at >= ? AND r.options LIKE '%"batch"%' LIMIT 1000`).all(since).catch(() => []);
++ WHERE r.created_at >= ? AND r.latency_class = 'batch' AND ${capabilitySql('r')} = ? LIMIT 1000`).all(since, capability).catch(() => []);
let batchSaved = 0, batchPriced = 0;
for (const b of batchRuns) {
const retail = (Number(b.tokens_in || 0) / 1e6) * Number(b.cost_in_per_mtok || 0) + (Number(b.tokens_out || 0) / 1e6) * Number(b.cost_out_per_mtok || 0);
@@@ -181,13 +256,13 @@ function createOwnerConfig({ db, registry, clock = { now: () => Date.now() } } =
}
return {
capability, window: '7d', since: sinceDay,
-- prompt_cache: { cached_tokens: Number(req.cached || 0), calls: Number(req.calls || 0) },
-- answer_cache: { cached_runs: Number(cachedRuns.n || 0) },
-- batches: { runs: batchRuns.length, priced: batchPriced, saved_usd: batchSaved ? Number(batchSaved.toFixed(6)) : null },
++ prompt_cache: { cached_tokens: Number(promptRows && promptRows.cached || 0), calls: Number(promptRows && promptRows.calls || 0), saved_usd: promptRows ? Number(Number(promptRows.saved_usd || 0).toFixed(6)) : 0 },
++ answer_cache: { cached_runs: Number(answerRow && answerRow.n || 0), saved_usd: answerRow ? Number(Number(answerRow.saved_usd || 0).toFixed(6)) : 0 },
++ batches: { runs: batchRuns.length, priced: batchPriced, saved_usd: batchPriced ? Number(batchSaved.toFixed(6)) : null },
};
}
-- return { KINDS, STRATEGIES, getCaps, setCap, capHit, spendFor, getRouting, setRouting, routingFor, getEconomy, setEconomy, costs, savings };
++ return { KINDS, STRATEGIES, getCaps, setCap, capHit, spendFor, getRouting, setRouting, routingFor, getEconomy, economyFor, setEconomy, costs, savings };
}
--module.exports = { createOwnerConfig, KINDS, STRATEGIES };
++module.exports = { createOwnerConfig, KINDS, STRATEGIES, capabilityOfRun, workflowKind };
ddiff --git a/server/providers/index.js b/server/providers/index.js
iindex 2287eb2..21b1385 100644
---- a/server/providers/index.js
++++ b/server/providers/index.js
@@@ -220,7 +220,16 @@ 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, ownerNotify = null }) {
++/** The prompt-cache switch is off for this kind: no cache marker and no warm-cache bias may be sent. */
++function withoutPromptCache(req) {
++ if (!req || typeof req !== 'object') return req;
++ const out = { ...req };
++ if (Array.isArray(out.system)) out.system = out.system.map((b) => (b && b.cache ? { ...b, cache: false } : b));
++ delete out.cacheKey;
++ return out;
++}
++
++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, ownerConfig = 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 }
@@@ -235,7 +244,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
// The owner's inference configuration (server/owner-config.js): spend caps, routing strategy
// per capability (cheapest / fastest / fixed order with a ceiling and a denylist) and economy
// toggles. Empty by installation: every check below is a no-op until the owner writes a row.
-- const ownerConfig = (() => { try { return require('../owner-config').createOwnerConfig({ db, registry, clock }); } catch { return null; } })();
++ const ownerSettings = ownerConfig || (() => { try { return require('../owner-config').createOwnerConfig({ db, registry, clock }); } catch { return null; } })();
const ownerCapability = (operation, ctx = {}, req = null) => {
if ((ctx.contentPolicy || 'standard') === 'uncensored') return 'uncensored';
if (operation === 'embed' || operation === 'embedding') return 'embed';
@@@ -246,18 +255,18 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
};
/** Apply the owner's routing row to an already-ranked order: denylist, ceiling, spend caps,
* then cheapest / fastest / fixed ordering. Excluded offers keep their reason in explain. */
-- async function applyOwner(list, explain, { operation, ctx = {}, req = null }) {
-- if (!ownerConfig || !list || !list.length) return list;
++ async function applyOwner(list, explain, { operation, ctx = {}, req = null, economy = undefined }) {
++ if (!ownerSettings || !list || !list.length) return list;
const capability = ownerCapability(operation, ctx, req);
-- const routing = await ownerConfig.routingFor(capability).catch(() => null);
-- const economy = await ownerConfig.getEconomy().catch(() => []);
-- const eco = economy.find((r) => r.capability === capability) || null;
++ const routing = await ownerSettings.routingFor(capability).catch(() => null);
++ const economyRows = economy === undefined ? await ownerSettings.getEconomy().catch(() => []) : null;
++ const eco = economy !== undefined ? economy : (economyRows.find((r) => r.capability === capability) || null);
const denied = new Set((routing && routing.denylist) || []);
const ceiling = routing ? routing.ceiling : null;
const out = [];
for (const c of list) {
if (denied.has(c.provider)) { explain.candidates.push({ provider: c.provider, model: c.model || null, eligible: false, excluded_reason: `owner_denylist (${capability})`, cost: null, latency: null }); continue; }
-- const cap = await ownerConfig.capHit(c.provider).catch(() => null);
++ const cap = await ownerSettings.capHit(c.provider).catch(() => null);
if (cap && cap.over) { explain.candidates.push({ provider: c.provider, model: c.model || null, eligible: false, excluded_reason: `spend_cap (today $${Number(cap.spent_today).toFixed(4)} of $${cap.daily == null ? '∞' : cap.daily}, month $${Number(cap.spent_month).toFixed(4)} of $${cap.monthly == null ? '∞' : cap.monthly})`, cost: null, latency: null }); continue; }
if (ceiling != null) {
const price = await priceFor(c.provider, c.model).catch(() => null);
@@@ -1310,7 +1319,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
const latency = Date.now() - t0;
if (ctx.logRequest) {
ctx.logRequest({ ...base, model_key: result.model || model, status: 'ok', output_hash: sha256(result.json || result.text || ''), tokens_in: usage.input || 0, tokens_out: usage.output || 0,
-- tokens_cached: usage.cached || 0, tokens_estimated: usage.estimated ? 1 : 0, cost_usd: cost, latency_ms: latency });
++ tokens_cached: usage.cached || 0, tokens_cache_write: usage.cache_write || 0, tokens_estimated: usage.estimated ? 1 : 0, cost_usd: cost, latency_ms: latency });
}
return { result, provider: cred.providerKey, providerKind: cred.provider, model: result.model || model, fallbackUsed: false, usage, cost, latencyMs: latency, synthetic: false, tried: [],
startedAt: t0, explain: { ...explain, selected: `${cred.provider}/${result.model || model}`, openvibe: ownFacts } };
@@@ -1612,9 +1621,15 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
}
list = [...list].sort((a, b) => Number(free.get(b.provider)) - Number(free.get(a.provider)));
}
++ // The owner's economy row for this kind, read once: the prompt-cache switch decides whether a
++ // warm-cache bias or a cache-control marker may be sent at all, before any candidate is tried.
++ const capability = ownerCapability(operation, ctx, req);
++ const ownerEconomy = ownerSettings ? await ownerSettings.economyFor(capability).catch(() => null) : null;
++ const promptCacheOn = !ownerEconomy || ownerEconomy.prompt_cache !== false;
++ if (!promptCacheOn) req = withoutPromptCache(req);
// L36 Y7: pick for the task. The hard needs first (a model that cannot do the job is never tried), then
// fit and cost: a light job takes the cheapest capable model, a heavy one the catalogue's strength tier.
-- const warmDeployment = warm.getDeployment(warmSubject(ctx, route), leadingText(req));
++ const warmDeployment = promptCacheOn ? warm.getDeployment(warmSubject(ctx, route), leadingText(req)) : null;
const warmProvider = warmDeployment ? warmDeployment.provider : null;
// L36 Y3: with free_first on, the free tiers lead (the price decides); with it off, the strength tier
// leads and the free tiers are the failover, not the first choice.
@@@ -1693,7 +1708,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
// here, on the authoritative list, so an owner's uncensored first/second always holds.
// The floor (server/policy/floor.js) and the 18+ gate run before/after every attempt
// regardless of this order; there is no switch that skips either.
-- if (!exact) list = await applyOwner(list, explain, { operation, ctx, req });
++ if (!exact) list = await applyOwner(list, explain, { operation, ctx, req, economy: ownerEconomy });
// AI Exchange P7: an alias route (economy/fast/balanced/auto) is ranked by measured evidence when
// AI_ROUTING_EVIDENCE is on. planAi (openvibe-sdk/placement) is the planner: this only prepares its
// inputs, keeps the incumbent for N decisions and explains the answer. Private never leaves local
@@@ -1981,7 +1996,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
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 || ''),
-- tokens_in: usage.input || 0, tokens_out: usage.output || 0, tokens_cached: usage.cached || 0, tokens_estimated: usage.estimated ? 1 : 0,
++ tokens_in: usage.input || 0, tokens_out: usage.output || 0, tokens_cached: usage.cached || 0, tokens_cache_write: usage.cache_write || 0, tokens_estimated: usage.estimated ? 1 : 0,
cost_usd: cost, latency_ms: latency,
debug_prompt: ctx.debugRaw ? JSON.stringify({ system: req.system, messages: req.messages }).slice(0, 200000) : null,
debug_response: ctx.debugRaw ? String(result.text || '').slice(0, 200000) : null,
ddiff --git a/server/runs.js b/server/runs.js
iindex da0fe51..ef6bfd0 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -24,6 +24,7 @@ const { preferenceLines } = require('./user-modules');
const { preferredTarget } = require('./providers/task');
const latency = require('./latency');
const tier = require('./policy/tier');
++const { capabilityOfRun } = require('./owner-config');
const schemas = require('./schemas');
const events = require('./events');
const usageRecord = require('./usage-record');
@@@ -125,7 +126,10 @@ function routePinRequest(value, principal, wf, internal = false) {
return { provider, model };
}
--function createRuns({ choices = null, db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null, credentials = null, preferences: personPreferences = null, tiers = null, projectSecrets = null, pool = null, runBudgets = null, credits = null, community = null, presets = null, receipts = null, abuse = null, webhooks = null, batchWaiting = null, batch = null }) {
++function createRuns({ choices = null, db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null, credentials = null, preferences: personPreferences = null, tiers = null, projectSecrets = null, pool = null, runBudgets = null, credits = null, community = null, presets = null, receipts = null, abuse = null, webhooks = null, batchWaiting = null, batch = null, ownerConfig = null }) {
++ // The owner's economy switches (server/owner-config.js). An unconfigured kind keeps its old
++ // behaviour (ECONOMY_DEFAULTS), so this only ever turns a way of paying less off.
++ const economyConfig = ownerConfig || (() => { try { return require('./owner-config').createOwnerConfig({ db, registry, clock }); } catch { return null; } })();
const inflight = new Map(); // run id -> { controller, promise, release, caller }
// BA2: the runs that wait for a provider's own batch API: they are created at once and finish later.
const deferred = new Map(); // run id -> { controller, scope, ctx, input, hold, resolveDone }
@@@ -204,9 +208,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, error_class, debug_prompt, debug_response, created_at)
++ tokens_in, tokens_out, tokens_cached, tokens_cache_write, 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, @error_class, @debug_prompt, @debug_response, @created_at)`);
++ @tokens_in, @tokens_out, @tokens_cached, @tokens_cache_write, @tokens_estimated, @cost_usd, @latency_ms, @error, @error_class, @debug_prompt, @debug_response, @created_at)`);
// ── Create ─────────────────────────────────────────────
function validateRefs(body) {
@@@ -312,6 +316,14 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
let latencyClass = resolvedLatency.class; // BA3 may move a batch-class run to the background class
// A faster class than the workflow allows was asked for: it runs in its own class, and the run's options say so.
if (resolvedLatency.clamped) options.latency_clamped = resolvedLatency.clamped;
++ // The owner's economy row for this run's kind (chat, uncensored, embed, image, transcribe or
++ // batch). One read, used by the answer cache and by the batch decision below.
++ const runCapability = capabilityOfRun(wf, { contentPolicy, latencyClass });
++ const runEconomy = economyConfig ? await economyConfig.economyFor(runCapability).catch(() => null) : null;
++ const answerCacheOn = runCapability !== 'uncensored' && (!runEconomy || runEconomy.answer_cache !== false);
++ // The adult tier is never put on a provider batch in this release: its floor and 18+ answer
++ // checks run on the live path, and a batch answer would bypass them.
++ const batchesOn = runCapability !== 'uncensored' && (!runEconomy || runEconomy.batches !== false);
// Community capacity is opt-in per request: a person's machine sees the prompt, so only a caller that says so
// may be placed there. The OpenAI and Messages routes set body.community from x-openvibe-community.
let communityOptIn = body.community === true || options.community === true;
@@@ -482,7 +494,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
// and another model's answer under the same prompt would silently break the pin. A run that may use the
// person's own key is never cached either: their key must be used, and the answer is theirs, not the scope's.
// A community or adult-tier answer cannot be reused by a request without the same policy.
-- const scope = taskChoice || options.cache === false || credentialSubject !== null || projectProvider || pinned || selfKeys || communityOptIn || contentPolicy !== 'standard' || advisorPolicy.enabled ? null : cache.scopeFor(wf.cache_mode, ctx);
++ const scope = taskChoice || options.cache === false || credentialSubject !== null || projectProvider || pinned || selfKeys || communityOptIn || contentPolicy !== 'standard' || advisorPolicy.enabled || !answerCacheOn ? null : cache.scopeFor(wf.cache_mode, ctx);
let cacheKey = null;
if (scope) {
// Some workflows key the cache on less than their whole input (hooks.js CACHE_INPUT: a signed URL's signature).
@@@ -527,9 +539,11 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
// route has one whose real batch price fits the half-price card (the same money rule as capacity
// route (d)). The run is created at once; the provider's answer finishes it on the ordinary path.
// When no provider on the route can batch, the run runs in the background class at once and says so.
++ // The owner's batches switch for this run's kind is the gate (server/owner-config.js): off means
++ // the deferrable run still answers, at once, in the background class — it is never delayed.
let autoBatch = null;
let batchDeadline = null;
-- if (latencyClass === 'batch' && !(principal && principal.deferBatch)) {
++ if (latencyClass === 'batch' && !(principal && principal.deferBatch) && batchesOn) {
batchDeadline = batchDeadlineHours(options.deadline_hours);
// A run paid by somebody else's key or by a free allowance is not put on the network's batch
// provider; a free-only app run cannot be placed on paid capacity either.
@@@ -567,6 +581,15 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
...(options.latency_clamped ? { latency_clamped: options.latency_clamped } : {}),
...(autoBatch ? { deadline_hours: batchDeadline, batch: { provider: autoBatch.provider, model: autoBatch.model } } : {}),
});
++ } else if (latencyClass === 'batch' && !(principal && principal.deferBatch) && !batchesOn) {
++ // The owner turned batches off for this kind: the run is not delayed and not queued. It
++ // answers in the background class now, and its options say which switch clamped it.
++ latencyClass = 'background';
++ ctx.latencyClass = 'background';
++ delete ctx.discount;
++ options.latency_clamped = { asked: 'batch', used: 'background', economy: `batches off for ${runCapability}` };
++ base.latency_class = latencyClass;
++ base.options = JSON.stringify({ ...parseJson(base.options, {}), latency_clamped: options.latency_clamped });
}
// Queue caps, then quota: both before any provider call (a refused run is not counted).
@@@ -929,7 +952,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
await insReq.run({
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,
++ tokens_in: e.tokens_in || 0, tokens_out: e.tokens_out || 0, tokens_cached: e.tokens_cached || 0, tokens_cache_write: e.tokens_cache_write || 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, 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()),
@@@ -1119,7 +1142,7 @@ function createRuns({ choices = null, db, registry, engine, cache, quotas, confi
.run(cancelled ? 'cancelled' : 'failed', code, extra ? `${detail} ${extra}` : detail, attempts, explain, iso(clock.now()), id);
if (f.changes) await events.runChanged(await getRow(id));
});
-- const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(tokens_cached),0)::bigint tc, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
++ const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(tokens_cached),0)::bigint tc, COALESCE(SUM(tokens_cache_write),0)::bigint tw, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
const lastAttempt = logged.filter(e => e.status !== 'skipped').at(-1);
await writeReceipt({ row: { ...await getRow(id), cost_usd: Number(spent.c) || 0, provider_key: lastAttempt?.provider_key || null, model_key: lastAttempt?.model_key || null }, explain: parseJson(explain, null), usage: { input: Number(spent.ti), output: Number(spent.tout), cached: Number(spent.tc) } });
// Balance: nothing reached a provider (or it failed before the first token) releases the hold; an abort
@@@ -1392,7 +1415,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, error_class, 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_cache_write, 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/server/workflows/engine.js b/server/workflows/engine.js
iindex 0f8c354..b2ca49f 100644
---- a/server/workflows/engine.js
++++ b/server/workflows/engine.js
@@@ -122,6 +122,17 @@ function collectCited(value, set = new Set()) {
return set;
}
++/**
++ * The stable part of a prompt first, the parts that change after it: prompt caching matches the
++ * longest identical prefix, so a system block the caller marked stable (cache: true) goes before a
++ * block that changes every request. Only the order changes; every word stays.
++ */
++function stableFirst(blocks) {
++ const list = (Array.isArray(blocks) ? blocks : []).filter((b) => b && String(b.text || '').trim());
++ if (list.length < 2) return list;
++ return [...list.filter((b) => b.cache), ...list.filter((b) => !b.cache)];
++}
++
function normMessages(messages, user) {
const out = [];
for (const m of messages || []) {
@@@ -262,8 +273,8 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
async function passthroughStep(step, input, wf, run, ctx) {
const routeKey = step.route || `${step.route_prefix || 'live'}.${input.role || 'legacy'}`;
const route = await routeFor(routeKey);
-- const sys = typeof input.system === 'string' ? (input.system.trim() ? [{ text: input.system, cache: false }] : [])
-- : Array.isArray(input.system) ? input.system.filter(x => x && String(x.text || '').trim()).map(x => ({ text: String(x.text), cache: Boolean(x.cache) })) : [];
++ const sys = stableFirst(typeof input.system === 'string' ? (input.system.trim() ? [{ text: input.system, cache: false }] : [])
++ : Array.isArray(input.system) ? input.system.filter(x => x && String(x.text || '').trim()).map(x => ({ text: String(x.text), cache: Boolean(x.cache) })) : []);
const base = { ...input, messages: normMessages(input.messages, input.user) };
// Files on the turn (ai.file@1) become text or image content for this call only.
const turn = Array.isArray(input.files) && input.files.length
@@@ -378,8 +389,8 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
if (step.kind === 'passthrough') {
if (input.image) return null;
const route = await routeFor(step.route || `${step.route_prefix || 'live'}.${input.role || 'legacy'}`);
-- const sys = typeof input.system === 'string' ? (input.system.trim() ? [{ text: input.system, cache: false }] : [])
-- : Array.isArray(input.system) ? input.system.filter(x => x && String(x.text || '').trim()).map(x => ({ text: String(x.text), cache: Boolean(x.cache) })) : [];
++ const sys = stableFirst(typeof input.system === 'string' ? (input.system.trim() ? [{ text: input.system, cache: false }] : [])
++ : Array.isArray(input.system) ? input.system.filter(x => x && String(x.text || '').trim()).map(x => ({ text: String(x.text), cache: Boolean(x.cache) })) : []);
const messages = normMessages(input.messages, input.user);
const json = input.json ? { name: input.json.name || 'result', schema: input.json.schema, strict: input.json.strict !== false, description: input.json.description } : null;
const maxTokens = Math.max(1, Math.round(input.max_tokens || 400));
ddiff --git a/test/economy-switches.test.js b/test/economy-switches.test.js
nnew file mode 100644
iindex 0000000..5a4bf31
---- /dev/null
++++ b/test/economy-switches.test.js
@@@ -0,0 +1,256 @@
++'use strict';
++/**
++ * The owner's three economy switches act (the owner 2026-11-10):
++ *
++ * - batches: a deferrable run (options.latency 'batch') waits on a provider's batch only when the
++ * switch is on for its kind; off, it answers at once in the background class and says why;
++ * - prompt cache: off, no cache-control marker and no warm-cache bias is sent; on, the adapters'
++ * own markers are sent and the stable system block comes first (only the order changes);
++ * - answer cache: off, an identical private run is computed again; on, the same person's repeat is
++ * served, never another person's, and never the uncensored kind.
++ * The savings each switch reports come from what happened: cached tokens priced against the input
++ * and cache-write cards, reused runs priced at their source's cost, and batch runs against list.
++ */
++const assert = require('assert');
++const http = require('http');
++const { boot, request, token, suite, seamServer, ALL } = require('./helpers');
++const { createOwnerConfig } = require('../server/owner-config');
++
++const t = suite('economy-switches');
++const OWNER = token('owner', ['ai.provider.manage', 'ai.usage.read']);
++const USR = (n) => `usr_01J${String(n).padStart(23, '0')}`;
++const A = { type: 'user', id: USR(1) };
++const B = { type: 'user', id: USR(2) };
++const ADULT = USR(3);
++
++/** An Anthropic-shaped seam: it answers chat and says how many prompt tokens were cached. */
++async function anthropicSeam() {
++ return await seamServer((body) => {
++ const call = String(body.model || 'claude-sonnet-5-5');
++ const marked = Array.isArray(body.system) && body.system.some((b) => b && b.cache_control);
++ return {
++ id: 'msg_test', type: 'message', role: 'assistant', model: call,
++ content: [{ type: 'text', text: `answer ${Date.now()}` }],
++ usage: { input_tokens: 10, output_tokens: 5, cache_read_input_tokens: marked ? 100 : 0, cache_creation_input_tokens: marked ? 20 : 0 },
++ };
++ });
++}
++
++/** A boot with one Anthropic provider and a priced model; the route is default.chat. */
++async function bootAnthropic(seam, { env = {}, fetchImpl } = {}) {
++ const h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_FREE_PROVIDERS: 'seam', AI_FREE_MAX_OUTPUT_PRICE_USD_PER_MTOK: '100', ...env }, fetchImpl });
++ await h.registry.upsertProvider({ key: 'seam', kind: 'anthropic', status: 'active', base_url: seam.url, auth_mode: 'none', secret_ref: null, default_model: 'claude-sonnet-5-5', capabilities: ['chat', 'generate', 'complete'], timeout_ms: 5000 }, { actor: 'test' });
++ await h.registry.upsertModel({ provider_key: 'seam', model_key: 'claude-sonnet-5-5', type: 'chat', status: 'active', cost: { in_per_mtok: 10, cached_per_mtok: 1, cache_write_per_mtok: 12.5, out_per_mtok: 20, source: 'urn:openvibe:ai:test', verified_at: '2026-01-01' } }, { actor: 'test' });
++ await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: token('live', ALL), body: { primary: { provider: 'seam', model: 'claude-sonnet-5-5' }, fallbacks: [] } });
++ return h;
++}
++
++const generate = (h, tok, body) => request(h.base, 'POST', '/api/v1/runs?wait=10000', { tok, body: { workflow: 'ai.generate', input: { prompt: 'the same question', system: 'You answer plainly.' }, ...body } });
++
++t.test('prompt cache: on sends the cache marker and prices the saving with the write card; off sends neither', async () => {
++ const seam = await anthropicSeam();
++ const h = await bootAnthropic(seam);
++ try {
++ const oc = createOwnerConfig({ db: h.db, registry: h.registry });
++ await oc.setEconomy({ capability: 'chat', batches: true, prompt_cache: true, small_first: false, answer_cache: false }, { actor: 'test' });
++ const on = await generate(h, token('live', ALL), { input: { prompt: 'cache on', system: 'You answer plainly.' }, options: { cache: false } });
++ assert.equal(on.status, 201, on.text);
++ assert.ok(Array.isArray(seam.last.system), 'the Anthropic body carries system blocks');
++ assert.deepEqual(seam.last.system[0].cache_control, { type: 'ephemeral' }, 'the stable block carries cache_control');
++ assert.match(seam.last.system[0].text, /You answer plainly/, 'the stable words are first');
++
++ const cached = await h.db.prepare('SELECT tokens_cached, tokens_cache_write FROM requests WHERE run_id = ? AND status = ?').get(on.body.run.id, 'ok');
++ assert.equal(Number(cached.tokens_cached), 100, 'the cached read is recorded');
++ assert.equal(Number(cached.tokens_cache_write), 20, 'the cache write is recorded');
++ const saving = await oc.savings('chat');
++ assert.equal(saving.prompt_cache.cached_tokens, 100);
++ // 100 * (10 - 1) - 20 * (12.5 - 10) = 900 - 50 nano-free dollars per million = $0.00085.
++ assert.equal(saving.prompt_cache.saved_usd, 0.00085, `the write card is subtracted (${saving.prompt_cache.saved_usd})`);
++
++ await oc.setEconomy({ capability: 'chat', batches: true, prompt_cache: false, small_first: false, answer_cache: false }, { actor: 'test' });
++ const off = await generate(h, token('live', ALL), { input: { prompt: 'cache off', system: 'You answer plainly.' }, options: { cache: false } });
++ assert.equal(off.status, 201, off.text);
++ assert.equal(seam.last.system[0].cache_control, undefined, 'no cache-control marker when the switch is off');
++ assert.ok(seam.last.system[0].text, 'the same words are still there');
++ assert.equal((await oc.getEconomy()).find((r) => r.capability === 'chat').prompt_cache, false);
++ } finally { await h.stop(); await seam.close(); }
++});
++
++t.test('answer cache: same person on is reused, off is computed; never across people', async () => {
++ const seam = await anthropicSeam();
++ const h = await bootAnthropic(seam);
++ try {
++ const oc = createOwnerConfig({ db: h.db, registry: h.registry });
++ const tok = token('live', ALL);
++ await oc.setEconomy({ capability: 'chat', batches: true, prompt_cache: false, small_first: false, answer_cache: true }, { actor: 'test' });
++ const first = await generate(h, tok, { on_behalf_of: A, input: { prompt: 'reuse me', system: 'You answer plainly.' } });
++ assert.equal(first.body.run.status, 'succeeded', JSON.stringify(first.body.run.error));
++ const calls = seam.calls;
++ const again = await generate(h, tok, { on_behalf_of: A, input: { prompt: 'reuse me', system: 'You answer plainly.' } });
++ assert.equal(again.body.run.status, 'cached', 'the same person is served the stored answer');
++ assert.equal(seam.calls, calls, 'no provider call');
++ const other = await generate(h, tok, { on_behalf_of: B, input: { prompt: 'reuse me', system: 'You answer plainly.' } });
++ assert.equal(other.body.run.status, 'succeeded', "another person never reads A's entry");
++ assert.equal(seam.calls, calls + 1);
++ assert.notEqual(other.body.run.output.text, first.body.run.output.text);
++
++ await oc.setEconomy({ capability: 'chat', batches: true, prompt_cache: false, small_first: false, answer_cache: false }, { actor: 'test' });
++ const off = await generate(h, tok, { on_behalf_of: A, input: { prompt: 'reuse me', system: 'You answer plainly.' } });
++ assert.equal(off.body.run.status, 'succeeded', 'off computes the same person again');
++ assert.equal(seam.calls, calls + 2);
++ const savings = await oc.savings('chat');
++ assert.ok(savings.answer_cache.cached_runs >= 1, 'the reused run is counted');
++ assert.ok(savings.answer_cache.saved_usd > 0, 'the reused run is priced at its source cost');
++ } finally { await h.stop(); await seam.close(); }
++});
++
++t.test('the answer cache never serves the uncensored kind', async () => {
++ const seam = await seamServer((body) => ({ id: 'c1', object: 'chat.completion', model: body.model, choices: [{ index: 0, message: { role: 'assistant', content: 'adult answer' }, finish_reason: 'stop' }], usage: { prompt_tokens: 4, completion_tokens: 5 } }));
++ const fetchImpl = async (url, opts) => {
++ if (String(url).includes('/oauth/token')) return new Response(JSON.stringify({ access_token: 't', expires_in: 300 }), { status: 200, headers: { 'content-type': 'application/json' } });
++ if (String(url).includes('/internal/identity/adult')) return new Response(JSON.stringify({ subject_id: ADULT, adult: true, confirmed_at: '2026-11-06T00:00:00Z' }), { status: 200, headers: { 'content-type': 'application/json' } });
++ return globalThis.fetch(url, opts);
++ };
++ const h = await boot({ env: { OV_OAUTH_CLIENT_SECRET: 'test-secret', AI_STUB_FALLBACK: 'false', AI_FREE_PROVIDERS: 'adult', AI_FREE_MAX_OUTPUT_PRICE_USD_PER_MTOK: '100' }, fetchImpl });
++ try {
++ await h.registry.upsertProvider({ key: 'adult', kind: 'openai', status: 'active', base_url: seam.url, auth_mode: 'none', secret_ref: null, default_model: 'uncensored-1', capabilities: ['chat', 'generate'], metadata: { uncensored: true } }, { actor: 'test' });
++ await h.registry.upsertModel({ provider_key: 'adult', model_key: 'uncensored-1', type: 'chat', status: 'active', cost: { in_per_mtok: 1, out_per_mtok: 2, source: 'urn:openvibe:ai:test', verified_at: '2026-01-01' }, metadata: { uncensored: true } }, { actor: 'test' });
++ await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: token('live', ALL), body: { primary: { provider: 'adult', model: 'uncensored-1' }, fallbacks: [] } });
++ const oc = createOwnerConfig({ db: h.db, registry: h.registry });
++ // The uncensored policy answers chat/complete/messages only, and every one of those workflows
++ // has cache_mode none; the run-level contentPolicy guard in server/runs.js is the second lock.
++ const body = { messages: [{ role: 'user', content: 'hello there' }], on_behalf_of: { type: 'user', id: ADULT }, policy: 'uncensored' };
++ const one = await request(h.base, 'POST', '/v1/chat/completions', { tok: token('live', ALL), body });
++ assert.equal(one.status, 200, one.text);
++ const two = await request(h.base, 'POST', '/v1/chat/completions', { tok: token('live', ALL), body });
++ assert.equal(two.status, 200, two.text);
++ assert.equal(two.body.choices[0].message.content, 'adult answer');
++ assert.equal(seam.calls, 2, 'both adult runs called the provider; no answer cache');
++ assert.equal(require('../server/owner-config').capabilityOfRun({ steps: [{ kind: 'passthrough' }] }, { contentPolicy: 'uncensored' }), 'uncensored');
++ assert.equal((await oc.getEconomy()).find((r) => r.capability === 'uncensored').answer_cache, false, 'the row reads off');
++ await oc.setEconomy({ capability: 'uncensored', batches: true, prompt_cache: true, small_first: false, answer_cache: true }, { actor: 'test' });
++ assert.equal((await oc.getEconomy()).find((r) => r.capability === 'uncensored').answer_cache, false, 'it cannot be turned on');
++ assert.equal((await oc.savings('uncensored')).answer_cache.cached_runs, 0);
++ } finally { await h.stop(); await seam.close(); }
++});
++
++/** A stub provider that speaks OpenAI's batch API and answers the ordinary chat path too. */
++function batchStub() {
++ const state = { files: new Map(), batches: new Map(), seq: 0, calls: [] };
++ const server = http.createServer((req, res) => {
++ let raw = '';
++ req.on('data', (c) => { raw += c; });
++ req.on('end', () => {
++ const p = new URL(req.url, 'http://x').pathname;
++ let body = {};
++ try { body = raw ? JSON.parse(raw) : {}; } catch { body = {}; }
++ state.calls.push({ path: p, method: req.method, body });
++ const send = (status, out, type) => { res.writeHead(status, { 'Content-Type': type || 'application/json' }); res.end(typeof out === 'string' ? out : JSON.stringify(out)); };
++ if (p === '/v1/files' && req.method === 'POST') { const id = `file_${++state.seq}`; state.files.set(id, body.file || ''); return send(200, { id, object: 'file' }); }
++ if (p === '/v1/batches' && req.method === 'POST') { const id = `batch_${++state.seq}`; state.batches.set(id, { input_file_id: body.input_file_id, status: 'in_progress', output_file_id: null }); return send(200, { id, object: 'batch', status: 'in_progress' }); }
++ if (p.startsWith('/v1/batches/')) { const b = state.batches.get(p.split('/')[3]); return b ? send(200, { id: p.split('/')[3], object: 'batch', status: b.status, output_file_id: b.output_file_id }) : send(404, { error: { message: 'no batch' } }); }
++ if (p.startsWith('/v1/files/') && p.endsWith('/content')) return send(200, state.files.get(p.split('/')[3]) || '', 'application/x-ndjson');
++ if (p === '/v1/chat/completions') return send(200, { id: 'c1', object: 'chat.completion', model: body.model || 'm', choices: [{ index: 0, message: { role: 'assistant', content: 'ordinary path' }, finish_reason: 'stop' }], usage: { prompt_tokens: 5, completion_tokens: 5 } });
++ return send(404, { error: { message: 'not found' } });
++ });
++ });
++ return new Promise((resolve) => {
++ server.listen(0, '127.0.0.1', () => resolve({
++ url: `http://127.0.0.1:${server.address().port}`, state,
++ close: () => new Promise((r) => { server.closeAllConnections?.(); server.close(() => r()); }),
++ answer(id) {
++ const b = state.batches.get(id);
++ const lines = (state.files.get(b.input_file_id) || '').split('\n').filter(Boolean).map((l) => JSON.parse(l));
++ const out = lines.map((line) => JSON.stringify({ custom_id: line.custom_id, response: { status_code: 200, body: { id: 'cmpl', object: 'chat.completion', model: line.body.model, choices: [{ index: 0, message: { role: 'assistant', content: 'batch answer' }, finish_reason: 'stop' }], usage: { prompt_tokens: 100, completion_tokens: 50 } } }, error: null }));
++ const outId = `out_${++state.seq}`;
++ state.files.set(outId, out.join('\n'));
++ b.output_file_id = outId; b.status = 'completed';
++ },
++ }));
++ });
++}
++
++t.test('batches: on queues a deferrable run, off answers it now and says why; a call not marked deferrable never waits', async () => {
++ const stub = await batchStub();
++ const h = await boot({ env: { AI_BATCH_NATIVE_MAX_ITEMS: '2', AI_BATCH_NATIVE_MAX_AGE_MS: '60000', AI_STUB_FALLBACK: 'false', AI_FREE_PROVIDERS: 'shared' } });
++ try {
++ await h.registry.upsertProvider({ key: 'shared', status: 'active', base_url: `${stub.url}/v1`, auth_mode: 'none', secret_ref: null, provider_access: 'OWN_DEPLOYMENT', capabilities: ['chat', 'generate'] }, { actor: 'test' });
++ await h.registry.upsertModel({ provider_key: 'shared', model_key: 'gpt-4o-mini', type: 'chat', status: 'active', cost: { in_per_mtok: 1, out_per_mtok: 2, source: 'urn:openvibe:ai:test', verified_at: '2026-01-01' } }, { actor: 'test' });
++ await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: token('live', ALL), body: { primary: { provider: 'shared', model: 'gpt-4o-mini' }, fallbacks: [] } });
++ const oc = createOwnerConfig({ db: h.db, registry: h.registry });
++ const body = (extra = {}) => ({ workflow: 'ai.generate', input: { prompt: `take your time ${Math.random()}` }, options: { cache: false, ...extra } });
++ const tok = token('live', ALL);
++
++ await oc.setEconomy({ capability: 'batch', batches: true, prompt_cache: true, small_first: false, answer_cache: true }, { actor: 'test' });
++ const on = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok, body: body({ latency: 'batch' }) });
++ assert.equal(on.status, 202, on.text);
++ assert.equal(on.body.run.latency_class, 'batch');
++ assert.ok(await h.db.prepare('SELECT id FROM batch_queue WHERE run_id = ?').get(on.body.run.id), 'the run waits for a batch');
++
++ const immediate = await request(h.base, 'POST', '/api/v1/runs?wait=10000', { tok, body: body() });
++ assert.equal(immediate.status, 201, immediate.text);
++ assert.equal(immediate.body.run.status, 'succeeded', 'a call not marked deferrable answers at once');
++ assert.equal(immediate.body.run.latency_class, 'interactive');
++
++ await oc.setEconomy({ capability: 'batch', batches: false, prompt_cache: true, small_first: false, answer_cache: true }, { actor: 'test' });
++ const off = await request(h.base, 'POST', '/api/v1/runs?wait=10000', { tok, body: body({ latency: 'batch' }) });
++ assert.equal(off.status, 201, off.text);
++ assert.equal(off.body.run.status, 'succeeded', 'the switch off answers instead of delaying');
++ assert.equal(off.body.run.latency_class, 'background');
++ assert.deepEqual(off.body.run.latency_clamped, { asked: 'batch', used: 'background', economy: 'batches off for batch' });
++ assert.equal((await h.db.prepare('SELECT COUNT(*) AS n FROM batch_queue WHERE run_id = ?').get(off.body.run.id)).n, 0, 'nothing waits');
++ assert.ok(stub.state.calls.some((c) => c.path === '/v1/chat/completions'), 'the ordinary path answered');
++
++ // The queue that was opened before the switch moved is still answered by the provider's batch.
++ const queued = await h.db.prepare("SELECT id, provider, model FROM batch_queue WHERE status = 'queued'").all();
++ if (queued.length) {
++ await h.db.prepare("UPDATE batch_queue SET created_at = ? WHERE status = 'queued'").run(new Date(Date.now() - 60000).toISOString());
++ await h.batcher.submitOnce();
++ const id = stub.state.batches.keys().next().value;
++ if (id) { stub.answer(id); await h.batcher.pollOnce(); }
++ }
++ const done = await h.db.prepare('SELECT latency_class, status FROM runs WHERE id = ?').get(on.body.run.id);
++ assert.equal(done.status, 'succeeded');
++ assert.equal(done.latency_class, 'batch');
++ const saving = await oc.savings('batch');
++ assert.equal(saving.batches.runs, 1);
++ assert.ok(saving.batches.saved_usd > 0, 'the batch price is measured against list');
++ } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('a caller max_price still refuses before any provider call', async () => {
++ const seam = await anthropicSeam();
++ const h = await bootAnthropic(seam);
++ try {
++ const calls = seam.calls;
++ const r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: token('live', ALL), body: { workflow: 'ai.generate', input: { prompt: 'price me' }, preset: { name: 'cheap', policy: { max_price: 0.001 } } } });
++ assert.equal(r.status, 201, r.text);
++ assert.equal(r.body.run.status, 'failed');
++ assert.equal(r.body.run.error.code, 'ai.max_price_exceeded');
++ assert.equal(seam.calls, calls, 'no provider was called');
++ } finally { await h.stop(); await seam.close(); }
++});
++
++t.test('Anthropic and Gemini batches turn on only for a configured provider', async () => {
++ const { batchProviderFor, enabledProviders } = require('../server/batch/providers');
++ const keyed = { status: 'active', auth_mode: 'key', secret_ref: 'env:BATCH_TEST_KEY' };
++ assert.equal(batchProviderFor('anthropic', keyed, {}), null, 'no key means off');
++ assert.ok(batchProviderFor('anthropic', keyed, { BATCH_TEST_KEY: 'x' }), 'a keyed provider turns it on');
++ assert.ok(batchProviderFor('gemini', { ...keyed, auth_mode: 'none', secret_ref: null }, {}), 'a local seam counts as configured');
++ assert.equal(batchProviderFor('anthropic', { ...keyed, status: 'disabled' }, { BATCH_TEST_KEY: 'x' }), null, 'an inactive provider stays off');
++ assert.ok(enabledProviders({ configured: ['anthropic'] }).anthropic, 'the configured list joins the always-on providers');
++ assert.equal(enabledProviders().anthropic, undefined, 'without a configured provider it is not offered');
++ // The batcher's body is OpenAI-shaped; the adapter moves it to the provider's own shape.
++ const { createAnthropicAdapter } = require('../server/batch/providers/anthropic');
++ const { createGeminiAdapter } = require('../server/batch/providers/gemini');
++ const a = createAnthropicAdapter({ baseUrl: 'https://example.test/v1', apiKey: 'x' });
++ const ab = a.buildFile([{ customId: '1', body: { model: 'claude-sonnet-5-5', max_tokens: 10, messages: [{ role: 'system', content: 'rules' }, { role: 'user', content: 'hi' }] } }]);
++ assert.equal(ab.requests[0].params.system[0].text, 'rules');
++ assert.equal(ab.requests[0].params.messages[0].role, 'user');
++ const g = createGeminiAdapter({ baseUrl: 'https://example.test/v1', apiKey: 'x' });
++ const gb = g.buildFile([{ customId: '1', body: { model: 'gemini-3-flash', messages: [{ role: 'system', content: 'rules' }, { role: 'user', content: 'hi' }] } }]);
++ assert.equal(gb.requests[0].request.systemInstruction.parts[0].text, 'rules');
++ assert.equal(gb.requests[0].request.contents[0].parts[0].text, 'hi');
++});
++
++t.run();