openvibers/openvibe.ai Commits

Native provider batch APIs behind one batcher (BA2) (#360)

OP OpenVibers committed 5 days ago · e35cb0bdda · 16 files, +2076 −17

* Native provider batch APIs behind one batcher (BA2)

The providers' own batch APIs, behind one queue. An item whose run has
latency_class 'batch' waits in batch_queue, is grouped by provider and model,
and goes out as one file and one batch when the group is full or old. One
worker polls the batches the provider is working on (the claim is the
database's and expires on its own, so several workers are safe), writes every
answer onto its own run and finishes the run on the ordinary path: the same
settle, the same receipt, the same events.

- server/batch/providers/: the adapters. openai-compatible (OpenAI, DashScope,
  Moonshot, DeepInfra), anthropic (Message Batches) and gemini. Each one says
  its discount, its window and its limits, and only a provider whose batch API
  was read on its own documentation page is enabled.
- server/batch/batcher.js: the queue, the submitter, the poller and the
  settlement. The cost written on a run is the provider's own batch price (its
  list price times its own discount), in nano-USD. A deadline the provider
  cannot meet re-routes the item to the ordinary path, and the item says so.
- server/batch/capacity.js: route kind (d), a provider's native batch whose
  real price is at most half the retail card, so a customer batch on a paid
  model runs at any hour. The settle check still holds.
- The service grant ai.batches.create: a first-party service may open a batch
  of the network's own, billed to the network's account, with the run's
  attribution naming the service. The grant is local; the Contracts entry
  comes later.
- migrations/0045_native_batches.sql: the two tables of the queue, the
  widened owner check on batches, and runs.latency_class (ADD COLUMN IF NOT
  EXISTS, so BA1's 0044 and this one coexist).

* The batch limits and the doc pages, read on the providers' own pages

DashScope's own page says 50 000 requests per file and 500 MB, not the
10 000 and 100 MB that were carried over from an earlier reading, and its
page has moved. Moonshot's batch guide has moved to the Kimi platform, and
DeepInfra's batch reference is one page of its own documentation. The
registry now cites the page each provider's shape was read on.

* Batch runs that survive a restart, and the money check before the work (BA2)

A waiting batch run lives in the database, not in this process. A provider batch takes up to 24 hours
and the service is deployed several times a day, so:

- recoverInterrupted() leaves a run alone while its batch_queue item is waiting or submitted (it stays
  queued); drain() releases it without aborting it;
- completeBatch() works when the run is not in the Map: it rebuilds what execute() needs from the
  database - the run row, the run's input and the plan the batcher stored with the item (the principal
  snapshot, the card, the route) - takes the tier hold and the quota reservation then, and finishes on
  the ordinary path. The Map stays only as a fast path;
- an item that was being settled when the service stopped goes back into the queue.

The money: the customer's Balance is checked before the provider is given the work. A run whose estimate
does not fit is refused with the usual 402, so the network never owes a provider for a run nobody can
pay for. The hold itself is taken and settled when the answer arrives: the charge happens on completion,
and a batch that waits hours does not hold the customer's Balance in the meantime.

The service grant stays a local rule, and a token without ai.batches.create is refused even when it
carries every other AI capability.

Changes

ddiff --git a/README.md b/README.md
iindex 5a98e86..8ab0795 100644
---- a/README.md
++++ b/README.md
@@@ -75,6 +75,11 @@ routing preferences; only the person's own token or AI-site session) and `ai.app
registered since openvibe-contracts 0.109.0). A service token's `ns`
claim limits which workflow namespaces it may run; an app token is fixed to its project's.
++A first-party service may also open a batch of the network's own with the `ai.batches.create` grant
++(see [Batches](#batches)): the batch is billed to the network's account, there is no Balance
++session for it, and the run's attribution names the service. The grant is checked locally by the
++batch service; the Contracts capability entry comes later (BA3).
++
Called elsewhere, as the service principal `ai`: `events.event.publish` (Events, the outbox relay) and
`events.subscription.manage` (Events, creating AI's own `bench.rating.changed` subscription), and
`network.modules.read` / `network.modules.write` for `ai.preferences` and `ai.usage_summary`
@@@ -184,6 +189,65 @@ no `ns` runs nothing. Every caller's Network grant names its namespaces (Live: `
`network.site_copy`, `media.analyze`). The fallback for service tokens without `ns` (`AI_NS_FALLBACK`)
and its rollback lever (`AI_NS_REQUIRED`) were retired on 2026-09-28 (shims C-22 and C-23).
++## Batches
++
++Many prompts in, the answers later. A batch is half the retail price for the customer, and it runs
++where the money works out: on OpenVibe's own capacity and free providers, on a community offer that
++asks at most half, on a paid provider inside a price-book off-peak window, or on **a provider's own
++batch API** when that provider's real batch price is at most half the retail card. The customer
++batch therefore runs on a paid model at any hour, not only in an off-peak window.
++
++The providers whose own batch API this service speaks, with the discount, the window and the limits
++read on each provider's own documentation page (a provider whose batch API could not be read is not
++listed and is never called):
++
++| provider | adapter | batch price | window | limits |
++|---|---|---|---|---|
++| `shared` (OpenAI) | OpenAI-compatible | half the list price | 24 h | 50 000 requests, 200 MB, one model per file |
++| `qwen` (DashScope) | OpenAI-compatible | half the real-time price | 24 h | 50 000 requests, 500 MB, one model per file |
++| `kimi` (Moonshot) | OpenAI-compatible | 60% of the list price | 24 h | 10 000 requests, 100 MB, one model per file |
++| `deepinfra` | OpenAI-compatible | 80% of the list price | 24 h | 10 000 requests, 100 MB, one model per file |
++| `anthropic` | Message Batches | half the list price | 24 h | 100 000 requests, 256 MB (adapter written, not enabled) |
++| `gemini` | Gemini Batch API | half the list price | 24 h | 10 000 requests (adapter written, **not enabled**: the endpoint could not be read on the provider's own page) |
++
++Only the providers whose batch price is at most half the retail card (`shared`, `qwen`) may carry a
++**customer** batch: the customer is charged half, so the network's real cost has to fit inside that
++half. Kimi and DeepInfra are below half and are therefore not eligible for a customer batch, though
++the network's own batches may still use them.
++
++How a native batch runs: an item whose run has `latency_class = 'batch'` waits in the queue
++(`batch_queue`), grouped by provider and model. A group goes to the provider when it is full
++(`AI_BATCH_NATIVE_MAX_ITEMS`, default 1000) or old (`AI_BATCH_NATIVE_MAX_AGE_MS`, default 5
++minutes). One worker polls the batches the provider is working on (the claim is the database's and
++expires on its own, so several workers are safe), writes every answer onto its own run (the status,
++the output, the tokens and the cost **at the provider's batch price**) and finishes the run on the
++ordinary path: the same settle, the same receipt, the same events. A deadline the provider cannot
++meet re-routes the item to the ordinary (background or standard) path, and the item says so.
++
++The money: the cost written on a run is what the provider really charged for the tier it really ran
++on (its own list price times its own batch discount), computed in nano-USD. The customer is charged
++half the retail card through Billing like every other request; this queue is a queue and a status,
++never a second ledger.
++
++**The balance is checked when the batch is submitted, and the money is taken when the answer
++arrives.** A run whose estimate does not fit the customer's Balance (or the project's budget) is
++refused before the provider is given the work, so the network never owes a provider for a run nobody
++can pay for; the item carries the reason. The hold itself is taken and settled when the answer comes
++back, so a batch that waits hours does not hold the customer's Balance in the meantime.
++
++**A waiting batch run is durable.** It waits in the database, not in the service's memory: the
++service is deployed several times a day, and a restart leaves the waiting runs queued (the provider
++is still working on them) instead of failing them. The batcher finishes them when the answer is
++there, with the ordinary settle, the ordinary receipt and the ordinary events.
++
++**Service batches.** A first-party service principal whose token carries the `ai.batches.create`
++grant may open a batch of the network's own: it is billed to the network's account (there is no
++Balance session for it) and the run's attribution names the service. The grant is checked locally by
++the batch service (`server/batch/service.js`), because the pinned `openvibe-contracts` build does
++not know that capability id yet: the Contracts capability entry comes later (BA3). Until then the
++route's own guard still asks for `ai.run.create` (or an app's `ai.app.run`) as well.
++
++
## Model catalog, exact ids and the disclosure
Behind `AI_MODEL_CATALOG` (off by default; with it off both `GET /v1/models` endpoints answer the legacy virtual
ddiff --git a/docs/batches.md b/docs/batches.md
nnew file mode 100644
iindex 0000000..cbc50d3
---- /dev/null
++++ b/docs/batches.md
@@@ -0,0 +1,58 @@
++## Batches
++
++Many prompts in, the answers later. A batch is half the retail price for the customer, and it runs
++where the money works out: on OpenVibe's own capacity and free providers, on a community offer that
++asks at most half, on a paid provider inside a price-book off-peak window, or on **a provider's own
++batch API** when that provider's real batch price is at most half the retail card. The customer
++batch therefore runs on a paid model at any hour, not only in an off-peak window.
++
++The providers whose own batch API this service speaks, with the discount, the window and the limits
++read on each provider's own documentation page (a provider whose batch API could not be read is not
++listed and is never called):
++
++| provider | adapter | batch price | window | limits |
++|---|---|---|---|---|
++| `shared` (OpenAI) | OpenAI-compatible | half the list price | 24 h | 50 000 requests, 200 MB, one model per file |
++| `qwen` (DashScope) | OpenAI-compatible | half the real-time price | 24 h | 50 000 requests, 500 MB, one model per file |
++| `kimi` (Moonshot) | OpenAI-compatible | 60% of the list price | 24 h | 10 000 requests, 100 MB, one model per file |
++| `deepinfra` | OpenAI-compatible | 80% of the list price | 24 h | 10 000 requests, 100 MB, one model per file |
++| `anthropic` | Message Batches | half the list price | 24 h | 100 000 requests, 256 MB (adapter written, not enabled) |
++| `gemini` | Gemini Batch API | half the list price | 24 h | 10 000 requests (adapter written, **not enabled**: the endpoint could not be read on the provider's own page) |
++
++Only the providers whose batch price is at most half the retail card (`shared`, `qwen`) may carry a
++**customer** batch: the customer is charged half, so the network's real cost has to fit inside that
++half. Kimi and DeepInfra are below half and are therefore not eligible for a customer batch, though
++the network's own batches may still use them.
++
++How a native batch runs: an item whose run has `latency_class = 'batch'` waits in the queue
++(`batch_queue`), grouped by provider and model. A group goes to the provider when it is full
++(`AI_BATCH_NATIVE_MAX_ITEMS`, default 1000) or old (`AI_BATCH_NATIVE_MAX_AGE_MS`, default 5
++minutes). One worker polls the batches the provider is working on (the claim is the database's and
++expires on its own, so several workers are safe), writes every answer onto its own run (the status,
++the output, the tokens and the cost **at the provider's batch price**) and finishes the run on the
++ordinary path: the same settle, the same receipt, the same events. A deadline the provider cannot
++meet re-routes the item to the ordinary (background or standard) path, and the item says so.
++
++The money: the cost written on a run is what the provider really charged for the tier it really ran
++on (its own list price times its own batch discount), computed in nano-USD. The customer is charged
++half the retail card through Billing like every other request; this queue is a queue and a status,
++never a second ledger.
++
++**The balance is checked when the batch is submitted, and the money is taken when the answer
++arrives.** A run whose estimate does not fit the customer's Balance (or the project's budget) is
++refused before the provider is given the work, so the network never owes a provider for a run nobody
++can pay for; the item carries the reason. The hold itself is taken and settled when the answer comes
++back, so a batch that waits hours does not hold the customer's Balance in the meantime.
++
++**A waiting batch run is durable.** It waits in the database, not in the service's memory: the
++service is deployed several times a day, and a restart leaves the waiting runs queued (the provider
++is still working on them) instead of failing them. The batcher finishes them when the answer is
++there, with the ordinary settle, the ordinary receipt and the ordinary events.
++
++**Service batches.** A first-party service principal whose token carries the `ai.batches.create`
++grant may open a batch of the network's own: it is billed to the network's account (there is no
++Balance session for it) and the run's attribution names the service. The grant is checked locally by
++the batch service (`server/batch/service.js`), because the pinned `openvibe-contracts` build does
++not know that capability id yet: the Contracts capability entry comes later (BA3). Until then the
++route's own guard still asks for `ai.run.create` (or an app's `ai.app.run`) as well.
++
ddiff --git a/migrations/0045_native_batches.sql b/migrations/0045_native_batches.sql
nnew file mode 100644
iindex 0000000..d65341d
---- /dev/null
++++ b/migrations/0045_native_batches.sql
@@@ -0,0 +1,76 @@
++-- phase: expand
++-- 0045: the native batch queue (BA2).
++--
++-- The providers' own batch APIs need somewhere to wait. This adds the two tables the native batcher
++-- uses: the provider batches it has submitted (one row per batch the provider is working on) and the
++-- queue of the items that wait for an answer (one row per request, with the run that waits for it).
++--
++-- The rule that shaped the tables: this is a QUEUE and a STATUS, never a second ledger. The money a
++-- batch costs is the money of the ordinary runs (Billing holds and settles them), and the price of a
++-- provider's batch tier is read from the registry's own card, never written here.
++--
++-- Nothing is removed and nothing is narrowed. The batches table's owner check only ever grows: a
++-- first-party service may now open a batch of its own (billed to the network), which it could not do
++-- before because there was no owner type for it.
++
++-- The run's latency class. BA1 adds the same column in 0044; both are ADD COLUMN IF NOT EXISTS, so
++-- whichever migration lands first wins and the other is a no-op.
++ALTER TABLE runs ADD COLUMN IF NOT EXISTS latency_class text COLLATE "C" NOT NULL DEFAULT 'interactive';
++
++-- A batch of the network's own: the owner may be a service.
++ALTER TABLE batches DROP CONSTRAINT IF EXISTS batches_owner_type_check;
++ALTER TABLE batches ADD CONSTRAINT batches_owner_type_check CHECK (owner_type IN ('user', 'app', 'service'));
++
++-- One row per batch the provider is working on.
++CREATE TABLE IF NOT EXISTS provider_batches (
++ id text COLLATE "C" PRIMARY KEY, -- pb_<id>
++ provider text COLLATE "C" NOT NULL, -- the provider whose batch API took it
++ model text COLLATE "C" NOT NULL,
++ adapter text COLLATE "C" NOT NULL, -- which adapter spoke to it
++ provider_batch_id text COLLATE "C", -- the provider's own id for the batch
++ completion_window text COLLATE "C", -- the window the provider promised ('24h')
++ status text COLLATE "C" NOT NULL DEFAULT 'submitting'
++ CHECK (status IN ('submitting', 'submitted', 'completed', 'failed', 'expired')),
++ item_count integer NOT NULL DEFAULT 0,
++ bytes bigint NOT NULL DEFAULT 0,
++ discount numeric NOT NULL DEFAULT 1, -- the provider's batch price, as a fraction of its list price
++ claimed_until timestamptz, -- while this is in the future a worker owns the poll
++ claimed_by text COLLATE "C",
++ error_code text COLLATE "C",
++ error_detail text COLLATE "C",
++ created_at timestamptz NOT NULL,
++ updated_at timestamptz NOT NULL,
++ submitted_at timestamptz,
++ finished_at timestamptz
++);
++CREATE INDEX IF NOT EXISTS provider_batches_poll_idx ON provider_batches (status, submitted_at);
++
++-- One row per request that waits for an answer.
++CREATE TABLE IF NOT EXISTS batch_queue (
++ id text COLLATE "C" PRIMARY KEY, -- bni_<id>: the custom_id the provider answers with
++ run_id text COLLATE "C", -- the run that waits for the answer
++ kind text COLLATE "C" NOT NULL DEFAULT 'service'
++ CHECK (kind IN ('customer', 'service')),
++ batch_id text COLLATE "C", -- the customer's batch (batches.id), for a customer item
++ idx integer, -- the customer's item index in that batch
++ provider text COLLATE "C" NOT NULL,
++ model text COLLATE "C" NOT NULL,
++ prompt text COLLATE "C" NOT NULL,
++ request_body text COLLATE "C" NOT NULL, -- the provider's own request body (JSON)
++ hold_card text COLLATE "C", -- the retail card the customer is charged on (JSON)
++ plan text COLLATE "C", -- what a restarted service needs to finish the run (JSON)
++ status text COLLATE "C" NOT NULL DEFAULT 'queued'
++ CHECK (status IN ('queued', 'submitted', 'settling', 'succeeded', 'failed', 'expired', 'rerouted')),
++ job_id text COLLATE "C" REFERENCES provider_batches (id) ON DELETE SET NULL,
++ answer text COLLATE "C",
++ usage text COLLATE "C",
++ cost_nano_usd bigint NOT NULL DEFAULT 0, -- what the provider really charged, in nano-USD
++ error_code text COLLATE "C",
++ error_detail text COLLATE "C",
++ deadline_at timestamptz NOT NULL,
++ created_at timestamptz NOT NULL,
++ updated_at timestamptz NOT NULL
++);
++CREATE INDEX IF NOT EXISTS batch_queue_group_idx ON batch_queue (provider, model, status, created_at);
++CREATE INDEX IF NOT EXISTS batch_queue_job_idx ON batch_queue (job_id, status);
++CREATE INDEX IF NOT EXISTS batch_queue_run_idx ON batch_queue (run_id);
ddiff --git a/server/api/batches.js b/server/api/batches.js
iindex a2a09a9..78cc566 100644
---- a/server/api/batches.js
++++ b/server/api/batches.js
@@@ -21,6 +21,9 @@ function batchesRouter({ batches, auth, config, clock = { now: () => Date.now()
const r = express.Router();
// The same guards the run routes use: a person's own token, a service token with ai.run.create, or
// a developer app with ai.app.run.
++ // BA2: a first-party service may also open a batch of the network's own when its token carries the
++ // ai.batches.create grant. The grant is checked by the batch service (server/batch/service.js), which
++ // knows it as a local rule: the contracts do not know that capability id yet (BA3 adds it).
const create = auth.appOrCap(CAPS.appRun, CAPS.runCreate);
const read = auth.appOrCap(CAPS.appRun, CAPS.runRead, CAPS.runCreate);
const wrap = (fn) => async (req, res) => { try { const out = await fn(req, res); if (out !== undefined) res.json(out); } catch (err) { sendError(res, err, req.ov, log); } };
ddiff --git a/server/batch/batcher.js b/server/batch/batcher.js
nnew file mode 100644
iindex 0000000..88d796b
---- /dev/null
++++ b/server/batch/batcher.js
@@@ -0,0 +1,517 @@
++'use strict';
++/**
++ * The native batcher: the providers' own batch APIs, behind one queue (BA2).
++ *
++ * What it does:
++ *
++ * - An item whose run has latency_class 'batch' waits here, in the batch_queue table. The run itself
++ * is created at once (the customer can see it, and its class says 'batch'), but it holds no execution
++ * slot and no quota until the provider answers.
++ * - Items are grouped by provider + model. A group is submitted to the provider's own batch API when it
++ * reaches a size limit (config.batches.native.maxItemsPerGroup) or an age limit (maxAgeMs): a small
++ * group waits, so the provider's file is worth sending.
++ * - One worker polls the batches the provider is working on. The claim is taken from the database and
++ * expires on its own, so several workers are safe and a dead worker blocks nothing.
++ * - When the answers arrive, each one is written onto its own run (the status, the output, the tokens and
++ * the cost at the provider's batch price) and the run finishes on the ordinary path: the same settle,
++ * the same receipt, the same events. A customer watching the run sees nothing new.
++ * - A deadline the provider cannot meet re-routes the item: the run goes back to the ordinary
++ * (background or standard) path, and the item says so.
++ *
++ * The money rule: the cost written on the run is what the provider really charges for the tier it really
++ * ran on, which is its own list price multiplied by its own batch discount (server/batch/providers). When a
++ * provider's batch price is not known, the item never runs there: the network never claims a saving it
++ * cannot see. This queue is not a ledger: the money is the ordinary runs' money, held and settled by
++ * Billing.
++ */
++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 { createOpenAiCompatibleAdapter } = require('./providers/openai-compatible');
++const { createAnthropicAdapter } = require('./providers/anthropic');
++const { createGeminiAdapter } = require('./providers/gemini');
++
++const ADAPTERS = {
++ 'openai-compatible': createOpenAiCompatibleAdapter,
++ anthropic: createAnthropicAdapter,
++ gemini: createGeminiAdapter,
++};
++
++/** The nano-USD a decimal amount of dollars is, without a float anywhere near the money. */
++const NANO_PER_USD = 1000000000n;
++
++/**
++ * The provider's own price for its batch tier: the list price (the registry's card) with every rate
++ * multiplied by the provider's batch discount. The discount is a fixed fraction of a million, so the
++ * multiplication is exact in BigInt and never a float.
++ */
++function scaleCard(card, discount) {
++ if (!card) return null;
++ const d = BigInt(Math.round(Number(discount) * 1000000));
++ const scale = (v) => (v == null ? null : (BigInt(String(v)) * d / 1000000n).toString());
++ const perM = card.nano_usd_per_m || {};
++ return {
++ version: `${card.version || 'retail'}:batch-${String(discount)}`,
++ nano_usd_per_m: {
++ input: scale(perM.input),
++ cached_read: scale(perM.cached_read),
++ cache_write: scale(perM.cache_write),
++ output: scale(perM.output),
++ },
++ per_request_nano_usd: scale(card.per_request_nano_usd == null ? '0' : card.per_request_nano_usd),
++ };
++}
++
++/** The retail card in the shape the Balance ledger counts (the shape retailOf() produces). */
++function toCard(rate) {
++ if (!rate) return null;
++ const put = (v) => (v == null ? null : String(v));
++ return {
++ version: String(rate.version || 'retail'),
++ nano_usd_per_m: {
++ input: put(rate.input_nano_usd_per_m),
++ cached_read: put(rate.cached_read_nano_usd_per_m),
++ cache_write: put(rate.cache_write_nano_usd_per_m),
++ output: put(rate.output_nano_usd_per_m),
++ },
++ per_request_nano_usd: put(rate.per_request_nano_usd == null ? '0' : rate.per_request_nano_usd),
++ };
++}
++
++function createBatcher({ db, runs, registry, credits = null, config = {}, clock = { now: () => Date.now() }, log = console, env = process.env, fetchImpl = globalThis.fetch, batches = null } = {}) {
++ // The batch service is attached after the batcher is built (the service needs the batcher too).
++ const native = (config.batches && config.batches.native) || {};
++ const settings = {
++ enabled: true,
++ // How many requests one provider batch carries (the provider's own limit is the ceiling; this is
++ // the network's own, so a file is never bigger than it needs to be).
++ maxItemsPerGroup: 1000,
++ // How long a group waits before it is submitted anyway: a small group still has to go out.
++ maxAgeMs: 5 * 60 * 1000,
++ // How often the worker submits and polls.
++ submitMs: 15000,
++ pollMs: 60000,
++ // How long a claimed provider batch stays claimed (a dead worker must block nothing).
++ claimTtlMs: 5 * 60 * 1000,
++ // The provider's own timeout for one call.
++ timeoutMs: 60000,
++ // The deadline a customer batch item gets when it names none.
++ deadlineHours: 24,
++ ...native,
++ };
++ const iso = () => new Date(clock.now()).toISOString();
++ const who = `batcher:${crypto.randomBytes(4).toString('hex')}`;
++ const adapters = new Map();
++ let service = null;
++ let running = false;
++ let timer = null;
++
++ /** 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 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({
++ baseUrl: row.base_url,
++ apiKey: providerSecret(row, env),
++ fetchImpl,
++ limits: cfg.limits,
++ timeoutMs: settings.timeoutMs,
++ });
++ adapters.set(providerKey, adapter);
++ return adapter;
++ }
++
++ /**
++ * The provider's own prices for one model: the list price (the registry's card), the batch price (the
++ * list price times the provider's own discount) and the access the provider has (which decides
++ * whether a customer's money may be held for it at all).
++ */
++ async function cardsFor(providerKey, modelKey) {
++ const cfg = providerConfig(providerKey);
++ const rows = await registry.listModels();
++ const row = (rows || []).find((r) => String(r.provider_key) === providerKey && String(r.model_key) === modelKey);
++ if (!row) return null;
++ const retail = toCard(retailRate(row));
++ if (!retail) return null;
++ const provider = await registry.getProvider(providerKey);
++ return {
++ row,
++ retail,
++ batch: scaleCard(retail, cfg ? cfg.discount : 1),
++ discount: cfg ? cfg.discount : 1,
++ access: provider ? accessOf(provider) : 'UNVERIFIED',
++ };
++ }
++
++ /**
++ * The plan of a waiting run: what a restarted service needs to finish it. It holds the principal
++ * snapshot (only the fields create() reads, never a token), the run's input, the card the run is
++ * charged at, and the route it went to.
++ */
++ function planOf({ principal, input, card, maxTokens, discount, maxCostNano, requestId, route, charge, servicePaid }) {
++ return {
++ principal: {
++ kind: principal.kind || null,
++ sub: principal.sub || null,
++ subject: principal.subject || null,
++ onBehalfOf: principal.onBehalfOf || null,
++ jti: principal.jti || null,
++ appId: principal.appId || null,
++ projectId: principal.projectId || null,
++ projectKey: principal.projectKey || null,
++ env: principal.env || null,
++ cap: Array.isArray(principal.cap) ? principal.cap : null,
++ ns: Array.isArray(principal.ns) ? principal.ns : null,
++ servicePaid: Boolean(servicePaid),
++ },
++ input: input || null,
++ card: card || null,
++ maxTokens: Number(maxTokens) || 4096,
++ discount: discount || null,
++ maxCostNano: maxCostNano == null ? null : String(maxCostNano),
++ requestId: requestId || null,
++ route: route || null,
++ charge: charge || null,
++ };
++ }
++
++ /** Who pays for the run, and at what discount: the payer's own view of it. */
++ function payerCtx(principal, { discount, maxCostNano, requestId }) {
++ const app = principal.kind === 'app';
++ const person = !app && principal.subject && principal.subject.type === 'user' ? String(principal.subject.id) : null;
++ return {
++ ...(app ? { appId: principal.appId || null, projectId: principal.projectId || null } : {}),
++ onBehalfOf: person,
++ personPays: Boolean(person),
++ discount: discount || null,
++ maxCostNano: maxCostNano == null ? null : String(maxCostNano),
++ runId: requestId || null,
++ };
++ }
++
++ /**
++ * Put one item in the queue. The run is created at once (so the customer sees it, and its latency
++ * class says 'batch'), but it waits for the provider instead of running.
++ */
++ 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 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 }] } };
++ const input = body.input;
++ // The money check, before the provider is given the work: a customer whose Balance cannot cover
++ // the run at the card it will be charged is refused here, so the network never owes a provider
++ // for a run nobody can pay for. The money itself is taken when the answer arrives.
++ if (credits && holdCard) {
++ const verdict = await credits.probe({
++ requestId,
++ ctx: payerCtx(principal, { discount, maxCostNano, requestId }),
++ input,
++ card: holdCard,
++ provider,
++ model,
++ access: cards.access,
++ maxTokens,
++ });
++ if (!verdict.ok) throw new AiError(402, String(verdict.reason || 'billing.insufficient_funds'), String(verdict.detail || 'Not enough Balance'));
++ }
++ // The principal carries the flag: only the batcher sets it, so no caller can hold a run open.
++ const created = await runs.create(body, { ...principal, deferBatch: true, servicePaid: Boolean(servicePaid) }, { requestId, discount, prefer, costCheck });
++ const id = `bni_${crypto.randomBytes(10).toString('hex')}`;
++ const hours = Number(deadlineHours) > 0 ? Number(deadlineHours) : Number(settings.deadlineHours) || 24;
++ const deadline = new Date(clock.now() + hours * 3600 * 1000).toISOString();
++ const providerBody = { model, messages: [{ role: 'user', content: prompt }], max_tokens: maxTokens };
++ const plan = planOf({ principal, input, card: holdCard, maxTokens, discount, maxCostNano, requestId, route, charge, servicePaid });
++ await db.prepare(`INSERT INTO batch_queue (id, run_id, kind, batch_id, idx, provider, model, prompt, request_body, hold_card, plan, status, deadline_at, created_at, updated_at)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?)`)
++ .run(id, created.run.id, kind, customer ? customer.batchId : null, customer ? customer.idx : null,
++ provider, model, prompt, JSON.stringify(providerBody),
++ holdCard ? JSON.stringify(holdCard) : null, JSON.stringify(plan), deadline, iso(), iso());
++ return { id, runId: created.run.id, cards };
++ }
++
++ /** The groups that are ready to be sent: a full group, or an old one. */
++ async function submitOnce() {
++ if (running) return { skipped: 'running' };
++ running = true;
++ try {
++ const groups = await db.prepare(`SELECT provider, model, COUNT(*) AS n, MIN(created_at) AS oldest
++ FROM batch_queue WHERE status = 'queued' GROUP BY provider, model`).all();
++ const sent = [];
++ for (const group of groups || []) {
++ const age = clock.now() - new Date(group.oldest).getTime();
++ if (Number(group.n) < Number(settings.maxItemsPerGroup) && age < Number(settings.maxAgeMs)) continue;
++ const job = await submitGroup(group.provider, group.model);
++ if (job) sent.push(job);
++ }
++ return { sent: sent.length, jobs: sent };
++ } catch (err) {
++ log.warn('[batch] the submitter stopped early: ' + String((err && err.message) || err));
++ return { error: String((err && err.code) || 'error') };
++ } finally {
++ running = false;
++ }
++ }
++
++ /** Send one group to the provider's own batch API, as one file and one batch. */
++ async function submitGroup(providerKey, modelKey) {
++ const items = await db.prepare(`SELECT * FROM batch_queue WHERE provider = ? AND model = ? AND status = 'queued'
++ ORDER BY created_at LIMIT ?`).all(providerKey, modelKey, Number(settings.maxItemsPerGroup));
++ if (!items || !items.length) return null;
++ const adapter = await adapterFor(providerKey);
++ const cfg = providerConfig(providerKey);
++ 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)
++ VALUES (?, ?, ?, ?, 'submitting', ?, ?, ?, ?)`)
++ .run(jobId, providerKey, modelKey, cfg.adapter, items.length, cfg.discount, iso(), iso());
++ let submitted;
++ try {
++ submitted = await adapter.submit({ items: payload, metadata: { display_name: jobId } });
++ } catch (err) {
++ // The provider refused or could not be reached: the items go back to the queue and are tried
++ // again later. Nothing is charged and nothing is lost.
++ await db.prepare(`UPDATE provider_batches SET status = 'failed', error_code = ?, error_detail = ?, finished_at = ?, updated_at = ? WHERE id = ?`)
++ .run(String((err && err.code) || 'batch.provider_error').slice(0, 80), String((err && err.message) || '').slice(0, 300), iso(), iso(), jobId);
++ for (const item of items) {
++ await db.prepare(`UPDATE batch_queue SET status = 'queued', updated_at = ? WHERE id = ? AND status = 'queued'`).run(iso(), item.id);
++ }
++ log.warn(`[batch] the provider refused a batch: ${String((err && err.message) || err)}`);
++ return null;
++ }
++ await db.prepare(`UPDATE provider_batches SET status = 'submitted', provider_batch_id = ?, bytes = ?, completion_window = ?, submitted_at = ?, updated_at = ?, claimed_until = NULL, claimed_by = NULL WHERE id = ?`)
++ .run(String(submitted.batchId), String(submitted.bytes || 0), adapter.completionWindow || '24h', iso(), iso(), jobId);
++ const placeholders = items.map(() => '?').join(',');
++ await db.prepare(`UPDATE batch_queue SET status = 'submitted', job_id = ?, updated_at = ? WHERE id IN (${placeholders})`)
++ .run(jobId, iso(), ...items.map((i) => i.id));
++ return { job: jobId, provider: providerKey, model: modelKey, items: items.length };
++ }
++
++ /** Claim one submitted provider batch: the claim is the database's, so two workers never share one. */
++ async function claimJob() {
++ const until = new Date(clock.now() + Math.max(1000, Number(settings.claimTtlMs) || 0)).toISOString();
++ const at = iso();
++ const row = await db.prepare(`UPDATE provider_batches SET claimed_until = ?, claimed_by = ?, updated_at = ?
++ WHERE id = (
++ SELECT id FROM provider_batches
++ WHERE status = 'submitted' AND (claimed_until IS NULL OR claimed_until <= ?::timestamptz)
++ ORDER BY submitted_at LIMIT 1 FOR UPDATE SKIP LOCKED)
++ RETURNING *`).get(until, who, at, at).catch(() => null);
++ return row || null;
++ }
++
++ /** Give the claim back, so another worker (or this one later) may poll this batch. */
++ async function releaseJob(job) {
++ await db.prepare('UPDATE provider_batches SET claimed_until = NULL, claimed_by = NULL, updated_at = ? WHERE id = ?').run(iso(), job.id);
++ }
++
++ /** One pass of the poller: take one submitted batch, ask the provider about it, settle what is ready. */
++ async function pollOnce() {
++ if (running) return { skipped: 'running' };
++ running = true;
++ try {
++ // An item that was being settled when the service stopped goes back into the queue: the
++ // provider's answer is still there, and the run must not wait for ever.
++ await db.prepare("UPDATE batch_queue SET status = 'submitted', updated_at = ? WHERE status = 'settling' AND updated_at <= ?")
++ .run(iso(), new Date(clock.now() - Math.max(1000, Number(settings.claimTtlMs) || 0)).toISOString());
++ const job = await claimJob();
++ if (!job) return { skipped: 'no batch' };
++ const adapter = await adapterFor(job.provider);
++ const state = await adapter.poll({ batchId: job.provider_batch_id });
++ if (state.status === 'completed') {
++ await finishJob(job, adapter, state);
++ return { job: job.id, status: 'completed' };
++ }
++ if (state.status === 'failed' || state.status === 'expired' || state.status === 'cancelled') {
++ await failJob(job, state);
++ return { job: job.id, status: state.status };
++ }
++ // The provider is still working. When the deadline of the items has passed, the provider cannot
++ // meet it: the items go back to the ordinary path, and say so.
++ if (await overdue(job)) {
++ await rerouteJob(job, 'the provider could not meet the deadline');
++ return { job: job.id, status: 'rerouted' };
++ }
++ await releaseJob(job);
++ return { job: job.id, status: state.status };
++ } catch (err) {
++ log.warn('[batch] the poller stopped early: ' + String((err && err.message) || err));
++ return { error: String((err && err.code) || 'error') };
++ } finally {
++ running = false;
++ }
++ }
++
++ /** The items of a batch whose deadline has passed while the provider was still working. */
++ async function overdue(job) {
++ const row = await db.prepare(`SELECT COUNT(*) AS n FROM batch_queue
++ WHERE job_id = ? AND status = 'submitted' AND deadline_at <= ?::timestamptz`).get(job.id, iso());
++ return Boolean(row && Number(row.n) > 0);
++ }
++
++ /** The provider answered: every item is written onto its run and the run finishes on the ordinary path. */
++ async function finishJob(job, adapter, state) {
++ const items = await db.prepare(`SELECT * FROM batch_queue WHERE job_id = ? AND status = 'submitted' ORDER BY created_at`).all(job.id);
++ const results = [];
++ if (state.outputFileId) {
++ const text = await adapter.download({ fileId: state.outputFileId });
++ for (const row of adapter.parseResults(text, { raw: state.raw })) results.push(row);
++ }
++ // The provider's error file: the requests it refused. A partial failure is a failure of that one
++ // item only: the other answers are still written onto their runs.
++ if (state.errorFileId) {
++ const text = await adapter.download({ fileId: state.errorFileId });
++ for (const row of adapter.parseResults(text, { raw: state.raw })) results.push(row);
++ }
++ const byId = new Map();
++ for (const row of results) byId.set(String(row.custom_id), row);
++ for (const item of items || []) {
++ const res = byId.get(String(item.id)) || null;
++ await settleItem(item, res);
++ }
++ await db.prepare(`UPDATE provider_batches SET status = 'completed', finished_at = ?, updated_at = ?, claimed_until = NULL, claimed_by = NULL WHERE id = ?`)
++ .run(iso(), iso(), job.id);
++ }
++
++ /** The provider gave up on the batch: every item fails, and the customer is charged nothing. */
++ async function failJob(job, state) {
++ const items = await db.prepare(`SELECT * FROM batch_queue WHERE job_id = ? AND status = 'submitted'`).all(job.id);
++ for (const item of items || []) {
++ await settleItem(item, { ok: false, error: { code: 'batch.provider_failed', detail: 'the provider did not answer this request' } });
++ }
++ await db.prepare(`UPDATE provider_batches SET status = 'failed', error_code = ?, error_detail = ?, finished_at = ?, updated_at = ?, claimed_until = NULL, claimed_by = NULL WHERE id = ?`)
++ .run(String(state.status || 'failed').slice(0, 40), 'the provider did not finish the batch', iso(), iso(), job.id);
++ }
++
++ /**
++ * Write one answer onto its run and fire the run's normal completion.
++ *
++ * The cost written on the run is the provider's BATCH price (its list price times its own batch
++ * discount), because that is what the provider really charged for the tier it really ran on.
++ */
++ async function settleItem(item, res) {
++ // Only an item that is still waiting is settled: two pollers can never settle the same one.
++ const claimed = await db.prepare(`UPDATE batch_queue SET status = 'settling', updated_at = ? WHERE id = ? AND status = 'submitted' RETURNING id`)
++ .get(iso(), item.id).catch(() => null);
++ if (!claimed) return false;
++ const cards = await cardsFor(item.provider, item.model);
++ // The plan the item was queued with: what a restarted service needs to finish this run.
++ const plan = item.plan ? JSON.parse(item.plan) : {};
++ const usage = (res && res.ok && res.usage) || { input: 0, output: 0, cached: 0, cache_write: 0 };
++ const net = Math.max(0, Number(usage.input || 0) - Number(usage.cached || 0) - Number(usage.cache_write || 0));
++ const nano = cards && cards.batch
++ ? retailChargeNano(cards.batch, { input: net, cached: Number(usage.cached || 0), cacheWrite: Number(usage.cache_write || 0), output: Number(usage.output || 0) })
++ : 0n;
++ const costUsd = Number(nano) / Number(NANO_PER_USD);
++ if (res && res.ok) {
++ await runs.completeBatch(item.run_id, {
++ ok: true,
++ output: res.output || { text: '' },
++ usage: { input: Number(usage.input || 0), output: Number(usage.output || 0), cached: Number(usage.cached || 0), cache_write: Number(usage.cache_write || 0) },
++ cost: costUsd,
++ provider: item.provider,
++ model: item.model,
++ holdRetail: item.hold_card ? JSON.parse(item.hold_card) : (cards ? cards.retail : null),
++ access: cards ? cards.access : 'UNVERIFIED',
++ explain: { objective: 'batch', reasons: [`answered by the provider's own batch API (${item.provider})`], selected: `batch:${item.provider}`, candidates: [], attempts: [] },
++ serviceTier: 'batch',
++ maxTokens: plan.maxTokens || 4096,
++ }, { plan, costCheck: await checkFor(item, plan) });
++ await db.prepare(`UPDATE batch_queue SET status = 'succeeded', answer = ?, usage = ?, cost_nano_usd = ?, updated_at = ? WHERE id = ?`)
++ .run(String((res.output && res.output.text) || '').slice(0, 20000), JSON.stringify(usage), String(nano), iso(), item.id);
++ } else {
++ const err = (res && res.error) || { code: 'batch.no_answer', detail: 'the provider did not answer this request' };
++ await runs.completeBatch(item.run_id, { ok: false, error: { code: String(err.code || 'batch.item_failed').slice(0, 80), detail: String(err.detail || '').slice(0, 400) } }, { plan });
++ await db.prepare(`UPDATE batch_queue SET status = 'failed', error_code = ?, error_detail = ?, updated_at = ? WHERE id = ?`)
++ .run(String(err.code || 'batch.item_failed').slice(0, 80), String(err.detail || '').slice(0, 400), iso(), item.id);
++ }
++ await writeBack(item);
++ return true;
++ }
++
++ /**
++ * The deadline passed and the provider cannot meet it: the item goes back to the ordinary path. The
++ * run answers through the pool as any other run does, and the item says it was re-routed.
++ */
++ async function rerouteJob(job, reason) {
++ const items = await db.prepare(`SELECT * FROM batch_queue WHERE job_id = ? AND status = 'submitted'`).all(job.id);
++ for (const item of items || []) {
++ const claimed = await db.prepare(`UPDATE batch_queue SET status = 'settling', updated_at = ? WHERE id = ? AND status = 'submitted' RETURNING id`)
++ .get(iso(), item.id).catch(() => null);
++ if (!claimed) continue;
++ await runs.completeBatch(item.run_id, { reroute: true }, { plan: item.plan ? JSON.parse(item.plan) : {} });
++ await db.prepare(`UPDATE batch_queue SET status = 'rerouted', error_code = 'batch.rerouted', error_detail = ?, updated_at = ? WHERE id = ?`)
++ .run(String(reason || '').slice(0, 300), iso(), item.id);
++ await writeBack(item);
++ }
++ await db.prepare(`UPDATE provider_batches SET status = 'expired', error_code = 'batch.rerouted', error_detail = ?, finished_at = ?, updated_at = ?, claimed_until = NULL, claimed_by = NULL WHERE id = ?`)
++ .run(String(reason || '').slice(0, 300), iso(), iso(), job.id);
++ }
++
++ /**
++ * The settle check of a customer item, rebuilt after a restart: the batch service knows the rule (the
++ * route's own price against what the customer paid, and the loss budget), and the item's row and the
++ * plan carry what the rule needs.
++ */
++ async function checkFor(item, plan) {
++ if (!service || !plan.route || !plan.charge || !item.batch_id) return null;
++ const batch = await db.prepare('SELECT * FROM batches WHERE id = ?').get(item.batch_id).catch(() => null);
++ if (!batch) return null;
++ const row = await db.prepare('SELECT * FROM batch_items WHERE batch_id = ? AND idx = ?').get(item.batch_id, item.idx).catch(() => null);
++ if (!row) return null;
++ return service.settleCheckFor({ item: row, route: plan.route, charge: plan.charge, batch });
++ }
++
++ /** Write the item's outcome back onto the customer's batch item, and close the batch when it is done. */
++ async function writeBack(item) {
++ if (!item.batch_id || item.idx == null) return;
++ const row = await db.prepare(`SELECT status, answer, usage, cost_nano_usd, error_code, error_detail FROM batch_queue WHERE id = ?`).get(item.id);
++ if (!row) return;
++ const status = row.status === 'succeeded' ? 'succeeded' : (row.status === 'rerouted' ? 'waiting' : 'failed');
++ await db.prepare(`UPDATE batch_items SET status = ?, run_id = ?, answer = ?, cost_nano_usd = ?, charged_nano_usd = ?, error_code = ?, error_detail = ?,
++ capacity = CASE WHEN ? = 'rerouted' THEN 'rerouted' ELSE capacity END, claimed_until = NULL, claimed_by = NULL, updated_at = ?
++ WHERE batch_id = ? AND idx = ?`)
++ .run(status, item.run_id || null, row.answer || null, row.cost_nano_usd || '0', row.cost_nano_usd || '0',
++ row.error_code || null, row.error_detail || null, row.status, iso(), item.batch_id, item.idx);
++ if (service && typeof service.finishBatch === 'function') {
++ const batch = await db.prepare('SELECT * FROM batches WHERE id = ?').get(item.batch_id).catch(() => null);
++ if (batch) await service.finishBatch(batch);
++ }
++ }
++
++ /** One pass of the whole worker: submit what is ready, then settle what the provider answered. */
++ async function runOnce() {
++ const submitted = await submitOnce();
++ const polled = await pollOnce();
++ return { submitted, polled };
++ }
++
++ function start() {
++ if (timer) return timer;
++ const every = Math.max(5000, Number(settings.pollMs) || 60000);
++ timer = setInterval(() => { runOnce().catch(() => {}); }, every);
++ timer.unref?.();
++ return timer;
++ }
++
++ function stop() {
++ if (timer) { clearInterval(timer); timer = null; }
++ }
++
++ /** The batch service, so that a customer's batch is closed when its items are done. */
++ function attach(theService) { service = theService; return api; }
++
++ const api = { enqueue, submitOnce, pollOnce, runOnce, start, stop, settings, cardsFor, scaleCard, attach };
++ return api;
++}
++
++module.exports = { createBatcher, scaleCard, toCard, ADAPTERS };
ddiff --git a/server/batch/capacity.js b/server/batch/capacity.js
iindex 75e8573..66de0ac 100644
---- a/server/batch/capacity.js
++++ b/server/batch/capacity.js
@@@ -26,6 +26,8 @@
*/
const { retailRate } = require('../providers/model-catalog');
const { insideWindow } = require('../pricebook/offpeak');
++const { enabledProviders } = require('./providers');
++const { scaleCard, toCard } = require('./batcher');
/** Half of a nano price, rounded down: the customer is never charged more than half. */
const halfOf = (v) => (v == null ? null : (BigInt(v) / 2n).toString());
@@@ -165,6 +167,33 @@ async function batchCapacity(model, at, { registry = null, community = null, pri
}
}
++ // (d) A provider's own batch API, when its real batch price is at most half the retail card. The
++ // customer is charged half the card, so the provider's own batch price has to fit inside that half:
++ // OpenAI and DashScope are the providers whose batch tier is half their list price today. A customer
++ // 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
++ const provider = await registry.getProvider(key);
++ if (!provider || provider.status !== 'active') 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));
++ if (!retail) continue;
++ const cost = scaleCard(retail, cfg.discount);
++ if (!cost || !within(cost.nano_usd_per_m, charge)) continue;
++ routes.push({
++ provider: key,
++ model: model.model_key,
++ source: 'native-batch',
++ cost: cost.nano_usd_per_m,
++ batch_discount: cfg.discount,
++ window: cfg.window,
++ });
++ }
++ }
++
return { charge, routes };
}
ddiff --git a/server/batch/providers/anthropic.js b/server/batch/providers/anthropic.js
nnew file mode 100644
iindex 0000000..a9737de
---- /dev/null
++++ b/server/batch/providers/anthropic.js
@@@ -0,0 +1,128 @@
++'use strict';
++/**
++ * Anthropic's Message Batches (BA2).
++ *
++ * The shape, as documented on Anthropic's own batch page:
++ *
++ * POST {base}/messages/batches the requests go in the BODY (there is no file here):
++ * { requests: [ { custom_id, params: { model, max_tokens, messages } } ] }
++ * GET {base}/messages/batches/{id} processing_status, request_counts, results_url
++ * GET {results_url} the answers, one JSON line per request
++ *
++ * A line of the answers is either
++ *
++ * { custom_id, result: { type: 'succeeded', message: { content: [ { type: 'text', text } ] } } }
++ * { custom_id, result: { type: 'error', error: { type, message } } }
++ *
++ * The discount is half the list price, the window is 24 hours, and one batch carries at most 100 000
++ * requests or 256 MB. Those three numbers are read on Anthropic's documentation page, not guessed.
++ *
++ * The provider is not enabled yet (no production key), so nothing here runs in production yet: the
++ * adapter is written so that turning the provider on is a one-line change in server/batch/providers.
++ */
++const BatchProviderError = require('./openai-compatible').BatchProviderError;
++
++/** The window Anthropic promises for its answers. */
++const COMPLETION_WINDOW = '24h';
++
++/** Anthropic needs a max_tokens on every message; the batcher asks for a bounded answer. */
++const DEFAULT_MAX_TOKENS = 4096;
++
++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;
++ const maxBytes = Number(limits.maxBytes) > 0 ? Number(limits.maxBytes) : 256 * 1024 * 1024;
++
++ async function call(method, path, { body = null, raw = false } = {}) {
++ if (!base) throw new BatchProviderError('this provider has no base URL for its batch API');
++ const headers = { Accept: 'application/json', 'anthropic-version': '2023-06-01' };
++ if (apiKey) headers['x-api-key'] = apiKey;
++ if (body !== null) headers['Content-Type'] = 'application/json';
++ let res;
++ try {
++ res = await fetchImpl(`${base}${path}`, {
++ method, headers,
++ ...(body !== null ? { body: JSON.stringify(body) } : {}),
++ signal: AbortSignal.timeout(timeoutMs),
++ });
++ } catch (err) {
++ throw new BatchProviderError(`the provider's batch API is unreachable: ${err && err.message}`, { code: 'batch.provider_unreachable', status: 503 });
++ }
++ const text = await res.text().catch(() => '');
++ if (!res.ok) {
++ let detail = text.slice(0, 300);
++ try { const j = JSON.parse(text); detail = (j && j.error && (j.error.message || j.error.code)) || detail; } catch { /* keep the text */ }
++ throw new BatchProviderError(`the provider's batch API refused the request (${res.status}): ${detail}`, { code: 'batch.provider_refused', status: 502 });
++ }
++ if (raw) return text;
++ try { return JSON.parse(text || '{}'); } catch { return {}; }
++ }
++
++ return {
++ kind: 'anthropic',
++ limits: { maxItems, maxBytes, oneModelPerFile: false },
++ completionWindow: COMPLETION_WINDOW,
++
++ /** The requests, in the body: there is no file here. */
++ 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 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 };
++ },
++
++ async submit({ items, completionWindow = COMPLETION_WINDOW, metadata = null } = {}) {
++ const file = this.buildFile(items);
++ const created = await call('POST', '/messages/batches', { body: { requests: file.requests, ...(metadata ? { metadata } : {}) } });
++ const batchId = created && created.id;
++ if (!batchId) throw new BatchProviderError('the provider did not name the batch');
++ return { batchId, bytes: file.bytes, items: items.length };
++ },
++
++ async poll({ batchId } = {}) {
++ const b = await call('GET', `/messages/batches/${encodeURIComponent(batchId)}`);
++ const status = String(b.processing_status || 'in_progress');
++ return { status, outputFileId: b.results_url || null, errorFileId: null, counts: b.request_counts || null, raw: b };
++ },
++
++ /** The answers, as the raw JSONL the provider wrote (the results URL is absolute). */
++ async download({ fileId } = {}) {
++ if (!fileId) return '';
++ if (/^https?:\/\//.test(fileId)) {
++ const res = await fetchImpl(fileId, { headers: { Accept: 'application/json', 'x-api-key': apiKey || '' }, signal: AbortSignal.timeout(timeoutMs) });
++ if (!res.ok) throw new BatchProviderError(`the provider's results file could not be read (${res.status})`, { code: 'batch.provider_refused', status: 502 });
++ return res.text();
++ }
++ return call('GET', `/messages/batches/${encodeURIComponent(fileId)}`, { raw: true });
++ },
++
++ /** The answers, in the shape every adapter returns. */
++ parseResults(text) {
++ const out = [];
++ for (const line of String(text || '').split('\n')) {
++ if (!line.trim()) continue;
++ let row;
++ try { row = JSON.parse(line); } catch { continue; }
++ const result = row.result || {};
++ if (result.type === 'succeeded') {
++ const content = (result.message && result.message.content) || [];
++ const answer = content.filter((c) => c && c.type === 'text').map((c) => c.text).join('');
++ const usage = result.message && result.message.usage;
++ out.push({
++ custom_id: row.custom_id,
++ ok: true,
++ output: { text: answer },
++ usage: { input: (usage && usage.input_tokens) || 0, output: (usage && usage.output_tokens) || 0, cached: 0, cache_write: 0 },
++ });
++ } else {
++ const err = result.error || {};
++ out.push({ custom_id: row.custom_id, ok: false, error: { code: String(err.type || 'batch.item_failed'), detail: String(err.message || 'the provider failed this request') } });
++ }
++ }
++ return out;
++ },
++ };
++}
++
++module.exports = { createAnthropicAdapter, COMPLETION_WINDOW };
ddiff --git a/server/batch/providers/gemini.js b/server/batch/providers/gemini.js
nnew file mode 100644
iindex 0000000..237cf5a
---- /dev/null
++++ b/server/batch/providers/gemini.js
@@@ -0,0 +1,123 @@
++'use strict';
++/**
++ * The Gemini Batch API (BA2).
++ *
++ * NOT VERIFIED. Gemini's batch documentation page is served to a browser (it redirects into AI Studio),
++ * so the endpoint below could not be read on the provider's own page while this was written. The
++ * adapter is therefore written but NOT enabled: server/batch/providers/index.js keeps it disabled and
++ * the network never sends a file to an endpoint nobody read. When the page can be read, the shape
++ * below is the one to check, and the provider is enabled only if it matches.
++ *
++ * The shape the adapter speaks (the Gemini API's batch mode):
++ *
++ * POST {base}/models/{model}:batchGenerateContent the requests go in the body (inline) or in a
++ * file the File API uploaded, and the answer names
++ * the batch (batches/...).
++ * GET {base}/{batch name} the state of the batch and, when it is done,
++ * the inlined answers.
++ *
++ * The discount on the provider's own page is half the list price, the window is 24 hours, and one
++ * batch carries at most 10 000 requests. Those numbers are what the registry records; the endpoint
++ * itself is what could not be read.
++ */
++const BatchProviderError = require('./openai-compatible').BatchProviderError;
++
++/** The window the provider promises for its answers. */
++const COMPLETION_WINDOW = '24h';
++
++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;
++ const maxBytes = Number(limits.maxBytes) > 0 ? Number(limits.maxBytes) : 100 * 1024 * 1024;
++
++ async function call(method, path, { body = null } = {}) {
++ if (!base) throw new BatchProviderError('this provider has no base URL for its batch API');
++ const headers = { Accept: 'application/json' };
++ if (apiKey) headers['x-goog-api-key'] = apiKey;
++ if (body !== null) headers['Content-Type'] = 'application/json';
++ let res;
++ try {
++ res = await fetchImpl(`${base}${path}`, {
++ method, headers,
++ ...(body !== null ? { body: JSON.stringify(body) } : {}),
++ signal: AbortSignal.timeout(timeoutMs),
++ });
++ } catch (err) {
++ throw new BatchProviderError(`the provider's batch API is unreachable: ${err && err.message}`, { code: 'batch.provider_unreachable', status: 503 });
++ }
++ const text = await res.text().catch(() => '');
++ if (!res.ok) {
++ let detail = text.slice(0, 300);
++ try { const j = JSON.parse(text); detail = (j && j.error && j.error.message) || detail; } catch { /* keep the text */ }
++ throw new BatchProviderError(`the provider's batch API refused the request (${res.status}): ${detail}`, { code: 'batch.provider_refused', status: 502 });
++ }
++ try { return JSON.parse(text || '{}'); } catch { return {}; }
++ }
++
++ return {
++ kind: 'gemini',
++ limits: { maxItems, maxBytes, oneModelPerFile: true },
++ completionWindow: COMPLETION_WINDOW,
++
++ /** The requests, inline. */
++ 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 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 };
++ },
++
++ async submit({ items, completionWindow = COMPLETION_WINDOW, metadata = null } = {}) {
++ const file = this.buildFile(items);
++ const model = String(items[0].body.model || '').replace(/^models\//, '');
++ if (!model) throw new BatchProviderError('this provider needs one model for the batch');
++ const created = await call('POST', `/models/${encodeURIComponent(model)}:batchGenerateContent`, {
++ body: { batch: { display_name: (metadata && metadata.display_name) || 'openvibe-batch', input_config: { requests: { requests: file.requests } } } },
++ });
++ const name = created && created.name;
++ if (!name) throw new BatchProviderError('the provider did not name the batch');
++ return { batchId: name, bytes: file.bytes, items: items.length };
++ },
++
++ async poll({ batchId } = {}) {
++ const b = await call('GET', `/${String(batchId || '').replace(/^\//, '')}`);
++ const state = String(b.state || 'BATCH_STATE_UNSPECIFIED');
++ const done = /SUCCEEDED/i.test(state);
++ const failed = /FAILED|CANCELLED|EXPIRED/i.test(state);
++ return { status: done ? 'completed' : (failed ? 'failed' : 'in_progress'), outputFileId: done ? 'inline' : null, errorFileId: null, raw: b };
++ },
++
++ async download({ fileId } = {}) {
++ // The answers come inline in the batch's own answer, not in a separate file.
++ return fileId === 'inline' ? '' : '';
++ },
++
++ /** The answers, in the shape every adapter returns. */
++ parseResults(text, { raw = null } = {}) {
++ const out = [];
++ const inlined = (raw && raw.response && raw.response.inlinedResponses && raw.response.inlinedResponses.inlinedResponses) || [];
++ for (const row of inlined) {
++ const custom = (row.request && row.request.customId) || row.custom_id;
++ const candidate = (row.response && row.response.candidates && row.response.candidates[0]) || null;
++ const parts = (candidate && candidate.content && candidate.content.parts) || [];
++ out.push({
++ custom_id: custom,
++ ok: Boolean(candidate),
++ output: { text: parts.map((p) => p.text || '').join('') },
++ usage: { input: 0, output: 0, cached: 0, cache_write: 0 },
++ });
++ }
++ for (const line of String(text || '').split('\n')) {
++ if (!line.trim()) continue;
++ let row;
++ try { row = JSON.parse(line); } catch { continue; }
++ out.push({ custom_id: row.custom_id, ok: Boolean(row.response), output: { text: String((row.response && row.response.text) || '') }, usage: { input: 0, output: 0, cached: 0, cache_write: 0 } });
++ }
++ return out;
++ },
++ };
++}
++
++module.exports = { createGeminiAdapter, COMPLETION_WINDOW };
ddiff --git a/server/batch/providers/index.js b/server/batch/providers/index.js
nnew file mode 100644
iindex 0000000..20aeb59
---- /dev/null
++++ b/server/batch/providers/index.js
@@@ -0,0 +1,108 @@
++'use strict';
++/**
++ * The providers whose own batch API this service speaks (BA2).
++ *
++ * A provider is listed here only when its batch API was read on its own documentation page and the
++ * endpoint, the shape of the file and the discount were confirmed there. A provider without a
++ * verified batch API is not listed: the network never claims a saving it cannot see, and it never
++ * sends a file to an endpoint it has not read.
++ *
++ * Every entry says three things, and the code uses all three:
++ *
++ * discount the fraction of the provider's OWN list price that its batch tier costs. The batch
++ * price of a model is its retail card multiplied by this, and it is what the run is
++ * recorded at (server/batch/batcher.js). OpenAI's Batch API is half the list price;
++ * Moonshot's is 60% of it; DeepInfra's is 80%.
++ * window how long the provider promises to have the answers ready ('24h' for all of them).
++ * limits what the provider accepts: how many requests in one file, how large that file may
++ * be, and whether one file may carry more than one model.
++ *
++ * The half-price customer batch (server/batch/capacity.js) may only use a provider whose discount is
++ * at most 0.5: the customer is charged half the retail card, so the network's real cost has to fit
++ * inside that half. That is OpenAI and DashScope today; Kimi (0.6) and DeepInfra (0.8) are below
++ * half and are therefore not eligible for a half-price customer batch, though the network's own
++ * batches may still use them.
++ */
++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
++ // id the batch names. Half the list price.
++ shared: {
++ adapter: 'openai-compatible',
++ label: 'OpenAI Batch API',
++ discount: 0.5,
++ window: '24h',
++ enabled: true,
++ limits: { maxItems: 50000, maxBytes: 200 * 1024 * 1024, oneModelPerFile: true },
++ source: 'https://platform.openai.com/docs/guides/batch',
++ },
++ // DashScope's OpenAI-compatible batch (file input): the same file-and-batch shape as OpenAI's, at
++ // half the price of real-time inference. Its own page says: up to 50 000 requests per file and a
++ // maximum file size of 500 MB, one model per line's body, completion window '24h'.
++ qwen: {
++ adapter: 'openai-compatible',
++ label: 'DashScope Batch (file input)',
++ discount: 0.5,
++ window: '24h',
++ enabled: true,
++ limits: { maxItems: 50000, maxBytes: 500 * 1024 * 1024, oneModelPerFile: true },
++ source: 'https://www.alibabacloud.com/help/en/model-studio/user-guide/batch-inference',
++ },
++ // Moonshot's Batch API: a JSONL file of at most 100 MB, one model per file, billed at 60% of the
++ // standard price (a 40% discount).
++ kimi: {
++ adapter: 'openai-compatible',
++ label: 'Moonshot Batch API',
++ discount: 0.6,
++ window: '24h',
++ enabled: true,
++ limits: { maxItems: 10000, maxBytes: 100 * 1024 * 1024, oneModelPerFile: true },
++ source: 'https://platform.kimi.ai/docs/guide/use-batch-api.md',
++ },
++ // DeepInfra's OpenAI-compatible Batch API: 20% cheaper than real time, answered within 24 hours.
++ deepinfra: {
++ adapter: 'openai-compatible',
++ label: 'DeepInfra Batch API',
++ discount: 0.8,
++ window: '24h',
++ enabled: true,
++ limits: { maxItems: 10000, maxBytes: 100 * 1024 * 1024, oneModelPerFile: true },
++ source: 'https://deepinfra.com/docs/batch/batch-endpoints',
++ },
++ // Anthropic's Message Batches: the requests go in the body (no file), the answers come back as
++ // a JSONL file at the results URL the batch names. Half the list price.
++ anthropic: {
++ adapter: 'anthropic',
++ label: 'Anthropic Message Batches',
++ discount: 0.5,
++ window: '24h',
++ enabled: false, // no production key yet: the adapter exists, the route does not
++ limits: { maxItems: 100000, maxBytes: 256 * 1024 * 1024, oneModelPerFile: false },
++ source: 'https://docs.anthropic.com/en/docs/build-with-claude/batch-processing',
++ },
++ // The Gemini Batch API: inline requests, answered later, at half the price. There is no
++ // production key for it yet, so the route is not enabled.
++ gemini: {
++ adapter: 'gemini',
++ label: 'Gemini Batch API',
++ discount: 0.5,
++ window: '24h',
++ enabled: false,
++ 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() {
++ const out = {};
++ for (const [key, cfg] of Object.entries(PROVIDERS)) if (cfg.enabled) out[key] = cfg;
++ return out;
++}
++
++/** One provider's configuration, or null when the provider has no (verified) batch API. */
++function providerConfig(key) {
++ return PROVIDERS[String(key || '')] || null;
++}
++
++module.exports = { PROVIDERS, enabledProviders, providerConfig };
ddiff --git a/server/batch/providers/openai-compatible.js b/server/batch/providers/openai-compatible.js
nnew file mode 100644
iindex 0000000..0ab2464
---- /dev/null
++++ b/server/batch/providers/openai-compatible.js
@@@ -0,0 +1,212 @@
++'use strict';
++/**
++ * The OpenAI-compatible batch API, as one adapter (BA2).
++ *
++ * The shape is the one OpenAI's Batch API documents, and the one DashScope, Moonshot and DeepInfra
++ * copy for their OpenAI-compatible endpoints:
++ *
++ * 1. upload a JSONL file POST {base}/files purpose "batch", one request per line;
++ * 2. create a batch POST {base}/batches input_file_id, endpoint, completion_window;
++ * 3. poll the batch GET {base}/batches/{id} status, output_file_id, error_file_id;
++ * 4. read the answers GET {base}/files/{id}/content one JSON line per request.
++ *
++ * A line of the file is an ordinary Chat Completions request with the caller's own id:
++ *
++ * {"custom_id":"bni_...","method":"POST","url":"/v1/chat/completions","body":{"model":"...","messages":[...]}}
++ *
++ * The provider answers with the same custom_id, which is how an answer finds the item it belongs to
++ * (server/batch/batcher.js). Nothing here touches the database or the money: the adapter only moves
++ * bytes and says what its tier costs.
++ */
++
++/** The endpoint the batch posts to: the OpenAI-compatible chat completions path. */
++const CHAT_ENDPOINT = '/v1/chat/completions';
++
++/** The window the provider promises for its answers. */
++const COMPLETION_WINDOW = '24h';
++
++class BatchProviderError extends Error {
++ constructor(message, { code = 'batch.provider_error', status = 502 } = {}) {
++ super(message);
++ this.code = code;
++ this.status = status;
++ }
++}
++
++/**
++ * One JSONL line per item, in the order the items were given.
++ *
++ * The custom_id is the item's own id, so the answers can be matched back without guessing. The
++ * shape of the request inside the line is the Chat Completions body the provider expects.
++ */
++function buildLines(items) {
++ return items.map((item) => JSON.stringify({
++ custom_id: item.customId,
++ method: 'POST',
++ url: item.url || CHAT_ENDPOINT,
++ body: item.body,
++ }));
++}
++
++/** The size of the file, in bytes, as the provider will count it. */
++function byteLength(lines) {
++ // One newline per line, as a JSONL file has.
++ return lines.reduce((n, line) => n + Buffer.byteLength(line, 'utf8') + 1, 0);
++}
++
++/**
++ * The adapter.
++ *
++ * baseUrl the provider's OpenAI-compatible base (with the version segment)
++ * apiKey the provider's key (never logged, never returned)
++ * fetchImpl the fetch to use (the tests pass a stub server)
++ * limits what the provider accepts (items, bytes, one model per file)
++ */
++function createOpenAiCompatibleAdapter({ baseUrl, apiKey, fetchImpl = globalThis.fetch, limits = {}, timeoutMs = 60000 } = {}) {
++ const base = String(baseUrl || '').replace(/\/+$/, '');
++ const maxItems = Number(limits.maxItems) > 0 ? Number(limits.maxItems) : 10000;
++ const maxBytes = Number(limits.maxBytes) > 0 ? Number(limits.maxBytes) : 100 * 1024 * 1024;
++ const oneModelPerFile = limits.oneModelPerFile !== false;
++
++ async function call(method, path, { body = null, raw = false } = {}) {
++ if (!base) throw new BatchProviderError('this provider has no base URL for its batch API');
++ const headers = { Accept: 'application/json' };
++ if (apiKey) headers.Authorization = `Bearer ${apiKey}`;
++ if (body !== null) headers['Content-Type'] = 'application/json';
++ let res;
++ try {
++ res = await fetchImpl(`${base}${path}`, {
++ method,
++ headers,
++ ...(body !== null ? { body: JSON.stringify(body) } : {}),
++ signal: AbortSignal.timeout(timeoutMs),
++ });
++ } catch (err) {
++ throw new BatchProviderError(`the provider's batch API is unreachable: ${err && err.message}`, { code: 'batch.provider_unreachable', status: 503 });
++ }
++ const text = await res.text().catch(() => '');
++ if (!res.ok) {
++ // The provider's own error, without the key and without the body of the request.
++ let detail = text.slice(0, 300);
++ try { const j = JSON.parse(text); detail = (j && (j.error && (j.error.message || j.error.code) || j.message)) || detail; } catch { /* keep the text */ }
++ throw new BatchProviderError(`the provider's batch API refused the request (${res.status}): ${detail}`, { code: 'batch.provider_refused', status: 502 });
++ }
++ if (raw) return text;
++ try { return JSON.parse(text || '{}'); } catch { return {}; }
++ }
++
++ return {
++ kind: 'openai-compatible',
++ limits: { maxItems, maxBytes, oneModelPerFile },
++ completionWindow: COMPLETION_WINDOW,
++
++ /**
++ * The file the provider will read: the lines, and their size.
++ * The size limit is the provider's own: a file over it is refused there, so it is refused here
++ * with the same rule (and never uploaded).
++ */
++ buildFile(items) {
++ if (!Array.isArray(items) || !items.length) throw new BatchProviderError('a batch file needs at least one request');
++ if (items.length > maxItems) throw new BatchProviderError(`this provider takes at most ${maxItems} requests in one file`, { code: 'batch.too_many_items', status: 422 });
++ if (oneModelPerFile) {
++ const models = new Set(items.map((i) => i.body && i.body.model).filter(Boolean));
++ if (models.size > 1) throw new BatchProviderError('this provider takes one model per file', { code: 'batch.mixed_models', status: 422 });
++ }
++ const lines = buildLines(items);
++ const bytes = byteLength(lines);
++ if (bytes > maxBytes) throw new BatchProviderError(`this provider's file limit is ${maxBytes} bytes and this file is ${bytes}`, { code: 'batch.file_too_large', status: 422 });
++ return { lines, bytes };
++ },
++
++ /** Upload the file and create the batch. → { batchId } */
++ async submit({ items, completionWindow = COMPLETION_WINDOW, metadata = null } = {}) {
++ const file = this.buildFile(items);
++ const uploaded = await call('POST', '/files', {
++ body: { purpose: 'batch', file: file.lines.join('\n') },
++ });
++ const fileId = uploaded && uploaded.id;
++ if (!fileId) throw new BatchProviderError('the provider did not name the uploaded file');
++ const created = await call('POST', '/batches', {
++ body: {
++ input_file_id: fileId,
++ endpoint: CHAT_ENDPOINT,
++ completion_window: completionWindow,
++ ...(metadata ? { metadata } : {}),
++ },
++ });
++ const batchId = created && created.id;
++ if (!batchId) throw new BatchProviderError('the provider did not name the batch');
++ return { batchId, fileId, bytes: file.bytes, items: items.length };
++ },
++
++ /** The state of the batch: { status, outputFileId, errorFileId } */
++ async poll({ batchId } = {}) {
++ const b = await call('GET', `/batches/${encodeURIComponent(batchId)}`);
++ return {
++ status: String(b.status || 'in_progress'),
++ outputFileId: b.output_file_id || null,
++ errorFileId: b.error_file_id || null,
++ raw: b,
++ };
++ },
++
++ /** The answers, as the raw JSONL the provider wrote. */
++ async download({ fileId } = {}) {
++ if (!fileId) return '';
++ return call('GET', `/files/${encodeURIComponent(fileId)}/content`, { raw: true });
++ },
++
++ /**
++ * The answers, in the shape every adapter returns: one entry per request, in the provider's own
++ * order, each one carrying the custom_id it was asked under (so the answer finds its item) and,
++ * when the request failed, the provider's own error.
++ *
++ * A line of OpenAI's output file is
++ * { id, custom_id, response: { status_code, body: { choices, usage } }, error: null }
++ * and a line of its ERROR file is
++ * { id, custom_id, response: null, error: { code, message } }
++ */
++ parseResults(text) {
++ const out = [];
++ for (const line of String(text || '').split('\n')) {
++ if (!line.trim()) continue;
++ let row;
++ try { row = JSON.parse(line); } catch { continue; }
++ const body = row.response && row.response.body;
++ const err = row.error || null;
++ if (!body || err) {
++ out.push({
++ custom_id: row.custom_id,
++ ok: false,
++ error: {
++ code: String((err && (err.code || err.type)) || 'batch.item_failed'),
++ detail: String((err && (err.message || err.detail)) || 'the provider failed this request').slice(0, 500),
++ },
++ });
++ continue;
++ }
++ const choice = (body.choices && body.choices[0]) || null;
++ const message = choice && (choice.message || null);
++ const text = (message && typeof message.content === 'string')
++ ? message.content
++ : (choice && typeof choice.text === 'string' ? choice.text : '');
++ const usage = body.usage || {};
++ const details = usage.prompt_tokens_details || {};
++ out.push({
++ custom_id: row.custom_id,
++ ok: true,
++ output: { text },
++ usage: {
++ input: Number(usage.prompt_tokens || 0),
++ output: Number(usage.completion_tokens || 0),
++ cached: Number(details.cached_tokens || 0),
++ cache_write: 0,
++ },
++ });
++ }
++ return out;
++ },
++ };
++}
++
++module.exports = { createOpenAiCompatibleAdapter, BatchProviderError, CHAT_ENDPOINT, COMPLETION_WINDOW };
ddiff --git a/server/batch/service.js b/server/batch/service.js
iindex aad0c9d..bc950a3 100644
---- a/server/batch/service.js
++++ b/server/batch/service.js
@@@ -32,13 +32,23 @@
*/
const crypto = require('crypto');
const { AiError } = require('../util');
++const { hasCap } = require('../auth');
++const { retailRate } = require('../providers/model-catalog');
++const { toCard } = require('./batcher');
const { batchCapacity, chargeOf } = require('./capacity');
const { retailChargeNano } = require('../credits');
const DEFAULT_MAX_ITEMS = 1000;
const EXPIRED_REASON = 'No low-cost capacity came free before your deadline. You were not charged.';
--function createBatchService({ db, runs, registry = null, community = null, priceBook = null, config = {}, clock = { now: () => Date.now() }, log = console, waitMs = 120000 }) {
++/**
++ * The grant a first-party service needs to open a batch of the network's own (BA2). It is checked here,
++ * not by the contracts: the capability entry comes later (BA3), and a pinned contracts build refuses an
++ * id it does not know. The grant is a local rule: the token must carry it, exactly or through its family.
++ */
++const BATCH_CAP = 'ai.batches.create';
++
++function createBatchService({ db, runs, registry = null, community = null, priceBook = null, batcher = null, config = {}, clock = { now: () => Date.now() }, log = console, waitMs = 120000 }) {
const settings = {
enabled: true,
// The cap: how many prompts one batch may carry.
@@@ -79,6 +89,17 @@ function createBatchService({ db, runs, registry = null, community = null, price
if (!id) throw new AiError(403, 'batch.no_payer', 'a batch is billed to an account; this token names none');
return { type: 'app', id: String(id), projectId: project ? String(project) : null };
}
++ // A service token's principal carries no 'kind': its subject names the type (svc:<id>).
++ if ((p.kind || (p.subject && p.subject.type)) === 'service') {
++ // A first-party service with the ai.batches.create grant may open a batch of the network's own:
++ // it is billed to the network's account, and the run's attribution names the service.
++ if (!hasCap(p.claims || { cap: p.cap }, BATCH_CAP)) {
++ throw new AiError(403, 'batch.no_grant', `a service may only open a batch with the ${BATCH_CAP} grant`);
++ }
++ const id = p.subject && p.subject.id;
++ if (!id) throw new AiError(403, 'batch.no_payer', 'a batch is billed to an account; this token names none');
++ return { type: 'service', id: String(id), projectId: null };
++ }
throw new AiError(403, 'batch.no_payer', 'a batch is billed to an account; a service token with no account of its own may not open one');
}
@@@ -87,6 +108,14 @@ function createBatchService({ db, runs, registry = null, community = null, price
if (batch.owner_type === 'user') {
return { kind: 'user', sub: `usr:${batch.owner_id}`, subject: { type: 'user', id: batch.owner_id }, onBehalfOf: null };
}
++ if (batch.owner_type === 'service') {
++ // A service's batch is billed to the network's own account: there is no Balance session, and the
++ // run's attribution names the service that asked.
++ return {
++ kind: 'service', sub: `svc:${batch.owner_id}`, subject: { type: 'service', id: batch.owner_id },
++ onBehalfOf: null, service: true,
++ };
++ }
return {
kind: 'app', sub: `app:${batch.owner_id}`, subject: { type: 'app', id: batch.owner_id },
appId: batch.owner_id, projectId: batch.project_id, projectKey: batch.project_id, env: 'production',
@@@ -108,11 +137,15 @@ function createBatchService({ db, runs, registry = null, community = null, price
return null;
}
-- /** The registry row of the model a batch asks for; the default chat route's model when it names none. */
-- async function resolveModel(batch) {
++ /**
++ * The registry row of the model a batch item asks for; the default chat route's model when it names none.
++ * The model is the item's own (batch_items.model): a batch carries many prompts and each of them may name
++ * its own model.
++ */
++ async function resolveModel(batch, item = null) {
if (!registry) return null;
const rows = await registry.listModels();
-- const wanted = String(batch.model || '').trim();
++ const wanted = String((item && item.model) || batch.model || '').trim();
if (wanted) {
const found = (rows || []).find((r) => String(r.model_key) === wanted
|| `${r.provider_key}/${r.model_key}` === wanted
@@@ -249,6 +282,9 @@ function createBatchService({ db, runs, registry = null, community = null, price
};
}
++ /** The settle check of a batch item: the batcher rebuilds it after a restart (the plan carries the route and the charge). */
++ function settleCheckFor({ item, route, charge, batch }) { return settleCheck({ item, route, charge, batch }); }
++
/**
* The settle check: the run settles only if the route it really used costs at most what the customer
* was charged. A route that costs more is refused, and the item keeps waiting.
@@@ -285,6 +321,51 @@ function createBatchService({ db, runs, registry = null, community = null, price
};
}
++ /**
++ * Put one prompt of the batch on the provider's own batch API. The run is created at once (the customer
++ * sees it) and waits for the answer; the batcher settles it when the provider answers.
++ */
++ async function enqueueNative(batch, item, route, charge, model) {
++ try {
++ const principal = principalFor(batch);
++ const check = settleCheck({ item, route, charge, batch });
++ // The card the customer is charged: the retail card of the model they asked for. The hold takes
++ // the half of it, exactly as the ordinary path does.
++ const retail = model ? toCard(retailRate(model)) : null;
++ const queued = await batcher.enqueue({
++ principal,
++ provider: route.provider,
++ model: route.model,
++ prompt: item.prompt,
++ kind: 'customer',
++ customer: { batchId: batch.id, idx: item.idx },
++ deadlineHours: Math.max(1, Math.round((new Date(item.deadline_at).getTime() - clock.now()) / 3600000)) || Number(settings.deadlineHours) || 24,
++ discount: String(settings.discount),
++ prefer: { provider: route.provider, model: route.model },
++ costCheck: check,
++ requestId: `batch-${batch.id}-${item.idx}`,
++ holdCard: retail,
++ route,
++ charge,
++ servicePaid: principal.kind === 'service',
++ });
++ await db.prepare(`UPDATE batch_items SET status = 'running', run_id = ?, route_provider = ?, route_model = ?, claimed_until = NULL, claimed_by = NULL, updated_at = ?
++ WHERE batch_id = ? AND idx = ?`)
++ .run(queued.runId || null, route.provider, route.model, iso(), batch.id, item.idx);
++ return { ok: true };
++ } catch (err) {
++ log.warn('[batch] an item could not be queued for the provider\'s batch API: ' + String((err && err.message) || err));
++ const code = String((err && err.code) || 'batch.queue_failed');
++ // BA2: a customer whose Balance cannot cover the run is refused here, before the provider is
++ // given the work: the item fails with the reason and the network owes nobody anything.
++ if (err && err.status === 402) {
++ await db.prepare("UPDATE batch_items SET status = 'failed', error_code = ?, error_detail = ?, claimed_until = NULL, claimed_by = NULL, updated_at = ? WHERE batch_id = ? AND idx = ?")
++ .run(code, String((err && err.detail) || 'Not enough Balance').slice(0, 300), iso(), batch.id, item.idx);
++ }
++ return { ok: false, code };
++ }
++ }
++
/** Run one prompt of the batch: an ordinary run, billed with the half-price card. */
async function runItem(batch, item, route, charge, regular = false) {
const principal = principalFor(batch);
@@@ -362,7 +443,7 @@ function createBatchService({ db, runs, registry = null, community = null, price
if (!item) return { skipped: 'no item' };
const batch = await db.prepare('SELECT * FROM batches WHERE id = ?').get(item.batch_id);
if (!batch) { await release(item, 'failed'); return { skipped: 'no batch' }; }
-- const model = await resolveModel(batch);
++ const model = await resolveModel(batch, item);
const charge = model ? chargeOf(model) : null;
const deadlinePassed = new Date(item.deadline_at).getTime() <= clock.now();
const capacity = model && charge
@@@ -375,6 +456,12 @@ function createBatchService({ db, runs, registry = null, community = null, price
});
if (routes.length) {
const route = routes[0];
++ // (d) A provider's own batch API: the item waits for the provider's answer instead of being
++ // answered now. The batcher writes the answer onto the run and the run finishes as usual.
++ if (route.source === 'native-batch' && batcher) {
++ const res = await enqueueNative(batch, item, route, capacity.charge, model);
++ return { batch: batch.id, item: item.idx, route: route.source, queued: res.ok, error: res.code };
++ }
const res = await runItem(batch, item, route, capacity.charge);
if (res.ok) await finishBatch(batch);
else await release(item, 'waiting');
@@@ -430,7 +517,7 @@ function createBatchService({ db, runs, registry = null, community = null, price
if (timer) { clearInterval(timer); timer = null; }
}
-- return { submit, get, list, runOnce, start, stop, settings, maxItems, EXPIRED_REASON, resolveModel };
++ return { submit, get, list, runOnce, start, stop, settings, maxItems, EXPIRED_REASON, resolveModel, finishBatch, settleCheckFor };
}
module.exports = { createBatchService, DEFAULT_MAX_ITEMS, EXPIRED_REASON };
ddiff --git a/server/config.js b/server/config.js
iindex 6ea673a..6e80ba0 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -387,6 +387,15 @@ function load(env = process.env) {
lossBudgetUsdPerDay: Math.max(0, Number(env.AI_BATCH_LOSS_BUDGET_USD_PER_DAY || 0) || 0),
tickMs: Math.max(5000, int(env.AI_BATCH_TICK_MS, 60 * 1000)),
claimTtlMs: Math.max(60000, int(env.AI_BATCH_CLAIM_TTL_MS, 15 * 60 * 1000)),
++ // The native batcher (BA2): the providers' own batch APIs.
++ native: {
++ // How many requests one provider batch carries.
++ maxItemsPerGroup: Math.max(1, int(env.AI_BATCH_NATIVE_MAX_ITEMS, 1000)),
++ // How long a group waits before it is submitted anyway (milliseconds).
++ maxAgeMs: Math.max(1000, int(env.AI_BATCH_NATIVE_MAX_AGE_MS, 5 * 60 * 1000)),
++ pollMs: Math.max(5000, int(env.AI_BATCH_NATIVE_POLL_MS, 60 * 1000)),
++ claimTtlMs: Math.max(60000, int(env.AI_BATCH_NATIVE_CLAIM_TTL_MS, 5 * 60 * 1000)),
++ },
},
// ── The living price book (server/pricebook/) ──
// The weekly job that reads every catalogued provider's own price listing and keeps the price the router
ddiff --git a/server/credits.js b/server/credits.js
iindex eda5648..9413435 100644
---- a/server/credits.js
++++ b/server/credits.js
@@@ -321,6 +321,42 @@ function createSession({ manager, requestId, payer, projectId, credentialId = nu
return { maxTokens: plan.maxTokens };
}
++ /**
++ * BA2: the hold of a batch run. The provider's batch API answers later, so no provider attempt
++ * happens before the answer and there is nothing for onRoute to hold on. The hold is taken here
++ * instead, on the same card the ordinary path would have used: the retail card, halved when the
++ * request is a batch. A provider the network has no managed access to is not held for (fail closed:
++ * nobody is charged and the network carries the cost).
++ */
++ async function holdRoute({ card = null, provider = null, model = null, access = null, maxTokens = 4096 } = {}) {
++ if (s.settled || s.released) return null;
++ if (!card || !card.nano_usd_per_m) return null;
++ const base = card;
++ const retail = applyDiscount(base, discount);
++ s.retailBase = base;
++ s.answeredProvider = provider;
++ s.answeredModel = model;
++ if (!MANAGED_ACCESS.has(String(access || ''))) return null; // not a managed paid offer: nobody is charged
++ if (!retail) return null;
++ const plan = affordable(retail, { inputTokens: s.inputTokens, maxTokens });
++ s.retail = retail;
++ if (!s.holdId) {
++ s.heldNano = plan.amount;
++ const fingerprint = sha256({
++ resolved_model: model || null, revision: null,
++ input_tokens: s.inputTokens, output_tokens: plan.maxTokens, max_cost_usd: s.maxCostNano == null ? null : String(s.maxCostNano),
++ rate_version: retail.version, subject: s.subject, project_id: s.projectId,
++ });
++ const created = await manager.hold({
++ idempotency_key: holdKey, subject: s.subject, ...(s.projectId ? { project_id: s.projectId } : {}), ...(s.credentialId ? { credential_id: s.credentialId } : {}),
++ max_nano_usd: String(plan.amount), rate_version: retail.version, request_fingerprint: fingerprint, service: 'ai',
++ }, s.requestId);
++ s.holdId = created.hold_id;
++ s.heldNano = BigInt(String(created.held_nano_usd == null ? plan.amount : created.held_nano_usd));
++ }
++ return { maxTokens: plan.maxTokens };
++ }
++
/** The projected output charge of what has streamed so far. */
function projectedNano() {
const output = Math.ceil(s.streamedChars / 4);
@@@ -421,7 +457,7 @@ function createSession({ manager, requestId, payer, projectId, credentialId = nu
}
return {
-- onRoute, observe, settle, release, projectedNano, usageOf, requestId: s.requestId,
++ onRoute, holdRoute, observe, settle, release, projectedNano, usageOf, requestId: s.requestId,
get holdId() { return s.holdId; }, get heldNano() { return s.heldNano; }, get retail() { return s.retail; },
get streamedChars() { return s.streamedChars; }, get settled() { return s.settled; }, get released() { return s.released; },
get inputTokens() { return s.inputTokens; }, get payer() { return s.payer; },
@@@ -484,6 +520,34 @@ function createCredits({ config, db = null, registry = null, env = process.env,
pendingSettlements: settlements ? () => settlements.pending() : async () => [],
startSettlements: settlements ? () => settlements.start() : () => null,
stopSettlements: settlements ? () => settlements.stop() : () => {},
++ /**
++ * BA2: can this payer cover a run of this size at this card?
++ *
++ * The check is a real hold (Billing answers 402 when the Balance is short) that is released at
++ * once: nothing is reserved for a run that has not answered yet, and the money itself is taken
++ * when the answer arrives. A batch asks this before the provider is given the work, so a customer
++ * who cannot pay is refused before the network owes the provider anything.
++ *
++ * → { ok: true } when the payer can cover it, { ok: false, reason } when it cannot.
++ */
++ async probe({ requestId, ctx = {}, input = {}, card = null, provider = null, model = null, access = null, maxTokens = 4096 } = {}) {
++ if (!card) return { ok: true, skipped: 'no card' };
++ // The probe holds under its own key: the run's own hold (taken when the answer arrives) must
++ // not find the key taken.
++ const session = manager.begin({ requestId: `probe:${String(requestId || ctx.runId || '')}`, ctx, input, attribution: ctx.attribution || null, onBehalfOf: ctx.onBehalfOf || null, personPayer: ctx.personPays !== false });
++ if (!session) return { ok: true, skipped: 'no payer' }; // nobody pays: the network carries the run
++ try {
++ await session.holdRoute({ card, provider, model, access, maxTokens });
++ } catch (err) {
++ if (err && (err.status === 402 || err.code === 'billing.insufficient_funds')) {
++ return { ok: false, reason: 'billing.insufficient_funds', detail: String((err && err.detail) || 'Not enough Balance') };
++ }
++ throw err;
++ }
++ await session.release().catch(() => null);
++ return { ok: true };
++ },
++
/**
* Begin a request's Balance session, or null when there is no payer (the pool then refuses paid offers
* that carry a versioned card), when an app names no project, or for a BYOK run (no managed route).
ddiff --git a/server/index.js b/server/index.js
iindex 795d837..494dd62 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -129,6 +129,11 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
if (!credits) throw new AiError(503, 'billing.unavailable', 'Billing Credits is not configured on this service');
return credits.begin(o);
},
++ // BA2: the balance check a batch asks before the provider is given the work.
++ async probe(o) {
++ if (!credits) return { ok: true, skipped: 'no credits' };
++ return credits.probe(o);
++ },
pendingSettlements() { return credits ? credits.pendingSettlements() : Promise.resolve([]); },
drainSettlements(o) { return credits ? credits.drainSettlements(o) : Promise.resolve({ skipped: 'disabled' }); },
startSettlements() { return credits ? credits.startSettlements() : null; },
@@@ -139,7 +144,13 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
// a GPU key and a monthly cap AND Balance billing is on; the public status says which one is missing.
const ownEndpoints = require('./gpu/endpoints').createOwnEndpoints({ db, gpus, credits: creditsRef, config, clock, fetchImpl, log });
gpus.attachEndpoints(ownEndpoints);
-- const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules, credentials, preferences, tiers, projectSecrets, pool, runBudgets, credits: creditsRef, community, presets, receipts, abuse, webhooks });
++ const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules, credentials, preferences, tiers, projectSecrets, pool, runBudgets, credits: creditsRef, community, presets, receipts, abuse, webhooks,
++ // 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) => {
++ const row = await db.prepare("SELECT 1 AS one FROM batch_queue WHERE run_id = ? AND status IN ('queued', 'submitted', 'settling') LIMIT 1").get(runId).catch(() => null);
++ return Boolean(row);
++ } });
await seed({ registry, quotas, config, env, db });
// The seed creates or moves providers from the environment: publish their origins (the schema and every key's
// bind follow), so a key is bound to the host it is for from the first boot.
@@@ -172,8 +183,11 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
// The measured refusal rates (server/refusal/): the weekly job that asks every offer the same published,
// harmless questions and publishes how often it refused them. Due-based, like the price book.
const refusal = require('./refusal/job').createRefusalJob({ db, registry, pool, config, env, clock, log });
++ // The native batcher (BA2): the providers' own batch APIs, behind one queue.
++ const batcher = require('./batch/batcher').createBatcher({ db, runs, registry, credits: creditsRef, config, clock, log, env, fetchImpl });
// The half-price batches (server/batch/): many prompts, answered later, billed through Billing.
-- const batches = require('./batch/service').createBatchService({ db, runs, registry, community, priceBook: offPeak, config, clock, log });
++ const batches = require('./batch/service').createBatchService({ db, runs, registry, community, priceBook: offPeak, batcher, config, clock, log });
++ batcher.attach(batches);
// The Events consumer side: a signed bench.rating.changed webhook, subscribed at boot (server/bench/events.js).
const accountData = accountDataLib.create({ db, log, memory });
const accountSend = env.OV_OAUTH_CLIENT_SECRET ? createNetworkSender({
@@@ -233,6 +247,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
if (listen && config.refusal && config.refusal.enabled) refusal.start();
// The batch worker picks up the queued batches. The queue is in the database, so a restart keeps it.
if (listen && config.batches && config.batches.enabled && config.nodeEnv !== 'test') batches.start();
++ // The native batcher submits the queued groups and settles the answers. It is in the database, so a
++ // restart keeps its queue.
++ if (listen && config.batches && config.batches.enabled && config.nodeEnv !== 'test') batcher.start();
// Durable Credit settles: the same jobs rule as Memory (never in test mode, never under a provider drill; the
// queue's runOnce checks the drill itself).
if (listen && credits && config.nodeEnv !== 'test') credits.startSettlements();
@@@ -272,6 +289,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
priceBook.stop();
refusal.stop();
batches.stop();
++ batcher.stop();
webhooks.stop();
if (credits) credits.stopSettlements();
if (jwks) jwks.stop();
@@@ -288,7 +306,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
if (!givenDb) await db.close();
}
-- return { config, db, registry, pool, quotas, tiers, cache, fetcher, engine, runs, keys, keyLoaded, auth, app, server, bench, benchEvents, memory, memoryJobs, memoryServices, memorySharing, priceBook, refusal, batches, credits: creditsRef, settings, vault, community, endpoints, webhooks, gpus, abuse, ownEndpoints, close };
++ return { config, db, registry, pool, quotas, tiers, cache, fetcher, engine, runs, keys, keyLoaded, auth, app, server, bench, benchEvents, memory, memoryJobs, memoryServices, memorySharing, priceBook, refusal, batches, batcher, credits: creditsRef, settings, vault, community, endpoints, webhooks, gpus, abuse, ownEndpoints, close };
}
if (require.main === module) {
ddiff --git a/server/runs.js b/server/runs.js
iindex 0e8986e..d349853 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -107,8 +107,10 @@ function routePinRequest(value, principal, wf) {
return { provider, model };
}
--function createRuns({ 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 }) {
++function createRuns({ 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 }) {
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 }
const queue = []; // run ids waiting for the worker
let active = 0;
let shuttingDown = false;
@@@ -331,7 +333,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
// The calling service's own budget (a run token's registered budget): with it, a run for a named person may
// use paid capacity, counted against that budget; without it, only free capacity serves the run.
const tokenBudget = runBudgets && principal.jti ? await runBudgets.spent(principal.jti) : null;
-- const serviceBudgeted = Boolean(tokenBudget && (tokenBudget.budget_usd != null || tokenBudget.budget_tokens != null));
++ // BA2: a batch of the network's own is paid by the network: the batcher says so on the principal.
++ const serviceBudgeted = Boolean(principal && principal.servicePaid) || Boolean(tokenBudget && (tokenBudget.budget_usd != null || tokenBudget.budget_tokens != null));
// Spend safety (server/free-tier.js, server/providers/index.js): a run nobody pays for — the free
// allowance (a service naming a person, without their own token) or the Exchange's anonymous Try it (its
// service token, no person and no run-token budget, on ai.complete) — may use only offers from
@@@ -480,7 +483,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
const hold = credentialSubject !== null || projectProvider || selfKeys || !tiers ? null : await tiers.admit(ctx, id);
let reserved;
try {
-- reserved = credentialSubject !== null || selfKeys ? null : await quotas.reserve(ctx);
++ // BA2: a batch run reserves its quota when the answer is here, not while it waits.
++ reserved = credentialSubject !== null || selfKeys || (principal && principal.deferBatch) ? null : await quotas.reserve(ctx);
} catch (err) { if (hold) await tiers.release(hold); throw err; }
try {
@@@ -503,6 +507,18 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
}
}
++ // BA2: a batch run waits for the provider's own batch API. It is created now (the customer sees
++ // it, and its latency class says 'batch'), but it takes no execution slot and no quota until the
++ // answer is here; then it finishes on the ordinary path, with the ordinary settle and receipt.
++ if (principal && principal.deferBatch) {
++ const controller = new AbortController();
++ let resolveDone;
++ const done = new Promise((r) => { resolveDone = r; });
++ deferred.set(id, { controller, scope, ctx, input, hold, resolveDone });
++ inflight.set(id, { controller, promise: done, release: () => {}, caller, priority });
++ return { run: await get(id), created: true, promise: done };
++ }
++
const controller = new AbortController();
let release;
const gate = new Promise((r) => { release = r; });
@@@ -809,7 +825,20 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
// With Credits off there is no Balance payer at all, so every run is the free tier for a new community machine;
// with it on only a run that could not hold a paid offer is.
const communityFree = !(credits && credits.enabled) || Boolean(ctx.creditsRequired);
-- const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash, preferences,
++ // BA2: a batch run's answer is the provider's own batch answer, not a provider call. The hold
++ // for the customer's money is taken here, on the card the ordinary path would have used.
++ const batchAnswer = ctx.batchAnswer ? await ctx.batchAnswer : null;
++ if (batchAnswer && batchAnswer.ok === false) {
++ throw new AiError(502, String((batchAnswer.error && batchAnswer.error.code) || 'batch.item_failed'), String((batchAnswer.error && batchAnswer.error.detail) || 'the provider did not answer this request'));
++ }
++ if (batchAnswer && batchAnswer.ok && credit && typeof credit.holdRoute === 'function') {
++ await credit.holdRoute({ card: batchAnswer.holdRetail || null, provider: batchAnswer.provider || null, model: batchAnswer.model || null, access: batchAnswer.access || null, maxTokens: batchAnswer.maxTokens || 4096 });
++ }
++ let r = null;
++ if (batchAnswer && batchAnswer.ok) {
++ r = { output: batchAnswer.output || { text: '' }, usage: batchAnswer.usage || { input: 0, output: 0, cached: 0, cache_write: 0 }, cost: Number(batchAnswer.cost) || 0, provider: batchAnswer.provider || null, model: batchAnswer.model || null, citations: [], explain: batchAnswer.explain || null, grounding: null, synthetic: false, fallbackUsed: false };
++ }
++ if (!r) r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash, preferences,
// A developer app's run (ADR-014): the provider pool excludes a person's byok key, and the run is
// limited to free and local capacity when it is a sandbox run or when no tier budget governs its
// project (AI_GOVERN_TIERS off): metered capacity is only ever spent under a project budget.
@@@ -1054,8 +1083,22 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
async function recoverInterrupted() {
return await db.tx(async () => {
const open = await db.prepare("SELECT id FROM runs WHERE status IN ('queued', 'running')").all();
-- const r = await db.prepare("UPDATE runs SET status = 'failed', error_code = 'run.interrupted', error_detail = 'the service restarted while this run was in progress', finished_at = ? WHERE status IN ('queued', 'running')").run(iso(clock.now()));
-- for (const { id } of open) await events.runChanged(await getRow(id));
++ // BA2: a run that waits for a provider's batch answer is not interrupted. It waits in the queue
++ // (the database, not this process), the provider is still working on it, and the batcher finishes
++ // it when the answer is there - after this restart too. Failing it here would lose the answer
++ // the provider is already going to charge us for.
++ const waiting = new Set();
++ if (batchWaiting) {
++ for (const { id } of open) {
++ if (await batchWaiting(id)) waiting.add(id);
++ }
++ }
++ const interrupted = open.filter(({ id }) => !waiting.has(id));
++ if (!interrupted.length) return 0;
++ const placeholders = interrupted.map(() => '?').join(',');
++ const r = await db.prepare(`UPDATE runs SET status = 'failed', error_code = 'run.interrupted', error_detail = 'the service restarted while this run was in progress', finished_at = ? WHERE status IN ('queued', 'running') AND id IN (${placeholders})`)
++ .run(iso(clock.now()), ...interrupted.map(({ id }) => id));
++ for (const { id } of interrupted) await events.runChanged(await getRow(id));
return r.changes;
});
}
@@@ -1076,14 +1119,107 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
/** Shutdown: in-flight runs end as failed/run.interrupted (retryable), never as a silent loss. */
async function drain() {
shuttingDown = true;
++ // BA2: a run that waits for a provider's batch answer is not interrupted by the shutdown: it stays
++ // queued, the provider keeps working on it, and the batcher finishes it after the restart. It is
++ // released here (its promise resolves, so the shutdown is not held by it) and never aborted.
++ for (const [id, d] of deferred) {
++ d.resolveDone();
++ inflight.delete(id);
++ }
++ deferred.clear();
for (const h of inflight.values()) { h.controller.abort(); h.release(); }
await Promise.allSettled([...inflight.values()].map(h => h.promise));
}
++ /**
++ * BA2: finish a batch run. The answer is the provider's own batch answer; the run then finishes on the
++ * ordinary path, so the settle, the receipt, the usage record and the events are the ordinary ones and
++ * a customer watching the run sees nothing new.
++ *
++ * A run that is already finished (or was never a batch run) is left alone: two pollers can never
++ * finish the same run twice.
++ */
++ async function completeBatch(id, answer, { plan = null, costCheck = null } = {}) {
++ const key = String(id);
++ const d = deferred.get(key);
++ if (d) {
++ deferred.delete(key);
++ const ctx = { ...d.ctx, batchAnswer: Promise.resolve(answer), ...(costCheck ? { costCheck } : {}) };
++ try {
++ const reserved = await quotas.reserve(ctx).catch(() => null);
++ await execute(key, { controller: d.controller, reserved, scope: d.scope, ctx, input: d.input, hold: d.hold });
++ } finally {
++ if (d.hold && tiers && tiers.release) await tiers.release(d.hold).catch(() => {});
++ d.resolveDone();
++ }
++ return get(key);
++ }
++ // The service restarted while the run waited: everything execute() needs is rebuilt from the
++ // database - the run row, the run's input and the plan the batcher stored with the item.
++ const row = await getRow(key);
++ if (!row) return null;
++ if (TERMINAL.has(row.status)) return null; // it already finished, or it was cancelled
++ const data = plan || {};
++ // The answer is the provider's own batch answer, exactly as in the fast path.
++ const ctx = { ...(await rebuildCtx(row, data, costCheck)), batchAnswer: Promise.resolve(answer) };
++ const input = data.input && typeof data.input === 'object' ? data.input : parseJson(row.input, {});
++ const controller = new AbortController();
++ let hold = null;
++ if (tiers) {
++ try { hold = await tiers.admit(ctx, key); }
++ catch (err) { log.warn(`[runs] ${key}: the tier budget refused the finished batch run: ${(err && err.message) || err}`); }
++ }
++ try {
++ const reserved = await quotas.reserve(ctx).catch(() => null);
++ await execute(key, { controller, reserved, scope: null, ctx, input, hold });
++ } finally {
++ if (hold && tiers && tiers.release) await tiers.release(hold).catch(() => {});
++ }
++ return get(key);
++ }
++
++ /**
++ * BA2: the context a finished batch run needs, rebuilt from the run row and the plan the batcher
++ * stored with the item. Only the fields the run's completion reads are here; the plan's principal
++ * snapshot holds the fields create() reads, and never a token.
++ */
++ async function rebuildCtx(row, data, costCheck) {
++ const opts = parseJson(row.options, {});
++ const attribution = parseJson(row.attribution, null);
++ const onBehalfOf = parseJson(row.on_behalf_of, null);
++ const principal = data.principal || {};
++ // The person the run was made for: the row names them, and the options say whether they pay.
++ const personSubject = opts.person_pays !== undefined ? (onBehalfOf && onBehalfOf.id ? String(onBehalfOf.id) : null) : null;
++ const tokenBudget = runBudgets && principal.jti ? await runBudgets.spent(principal.jti).catch(() => null) : null;
++ const serviceBudgeted = Boolean(principal.servicePaid)
++ || Boolean(tokenBudget && (tokenBudget.budget_usd != null || tokenBudget.budget_tokens != null));
++ const isApp = principal.kind === 'app';
++ return {
++ requesterType: row.requester_type,
++ requesterId: row.requester_id,
++ ...(principal.jti ? { runJti: principal.jti } : {}),
++ actorKey: subjectKey(onBehalfOf),
++ targetKey: row.target_key,
++ attributionKey: entityKey(attribution),
++ workflowKey: row.workflow_key,
++ ...(isApp ? { appRun: true, projectId: principal.projectId, sandbox: principal.env === 'sandbox', appId: principal.appId } : {}),
++ onBehalfOf: personSubject || principal.onBehalfOf || null,
++ personSubject,
++ personPays: Boolean(opts.person_pays),
++ serviceBudgeted,
++ freeUse: false, // a batch run is never the anonymous Try it
++ ...(data.discount ? { discount: String(data.discount) } : {}),
++ ...(data.maxCostNano != null ? { maxCostNano: String(data.maxCostNano) } : {}),
++ ...(costCheck ? { costCheck } : {}),
++ latencyClass: row.latency_class || 'batch',
++ runId: row.id,
++ };
++ }
++
/** Queue depth for /metrics and /api/ready: runs waiting for a slot, and runs holding one. */
function stats() { return { queued: queue.length, running: active, inflight: inflight.size, max_concurrent: config.runs.maxConcurrent, max_queued: config.runs.maxQueued }; }
-- return { create, wait, get, getReceipt, list, listOnBehalfOf, cancel, retry, citations, addCitations, requestsFor, explained, recoverInterrupted, prune, drain, stats, inflight, TERMINAL, decode };
++ return { create, wait, get, getReceipt, list, listOnBehalfOf, cancel, retry, citations, addCitations, requestsFor, explained, recoverInterrupted, prune, drain, stats, inflight, TERMINAL, decode, completeBatch, deferred };
}
module.exports = { createRuns, entityKey, subjectKey };
ddiff --git a/test/batch-native.test.js b/test/batch-native.test.js
nnew file mode 100644
iindex 0000000..a56442a
---- /dev/null
++++ b/test/batch-native.test.js
@@@ -0,0 +1,427 @@
++'use strict';
++/**
++ * BA2: the native batcher, end to end against a stub provider (no real key, no real provider).
++ *
++ * The stub speaks OpenAI's batch API (POST /v1/files, POST /v1/batches, GET /v1/batches/:id and the
++ * content of an output or error file), so the whole path is exercised: the queue, the group, the
++ * provider's file, the answers, the failure of one request, the re-route of a batch that misses its
++ * deadline, and the money of a customer batch settled at half the price.
++ */
++const assert = require('assert');
++const http = require('http');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const { boot, request, token, suite, ALL } = require('./helpers');
++const { createOpenAiCompatibleAdapter } = require('../server/batch/providers/openai-compatible');
++
++const t = suite('batch-native');
++const MODEL = 'gpt-5-nano';
++const PRINCIPAL = { kind: 'service', sub: 'svc:test', subject: { type: 'service', id: 'test' }, onBehalfOf: null };
++
++/** A stub provider that speaks OpenAI's batch API. */
++function stubProvider() {
++ 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 path = new URL(req.url, 'http://x').pathname;
++ let body = {};
++ try { body = raw ? JSON.parse(raw) : {}; } catch { body = {}; }
++ state.calls.push({ path, 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 (path === '/v1/files' && req.method === 'POST') {
++ const id = `file_${++state.seq}`;
++ state.files.set(id, body.file || '');
++ return send(200, { id, object: 'file', bytes: (body.file || '').length });
++ }
++ if (path === '/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, error_file_id: null });
++ return send(200, { id, object: 'batch', status: 'in_progress' });
++ }
++ if (path.startsWith('/v1/batches/')) {
++ const id = path.split('/')[3];
++ const b = state.batches.get(id);
++ if (!b) return send(404, { error: { message: 'no such batch' } });
++ return send(200, { id, object: 'batch', status: b.status, output_file_id: b.output_file_id, error_file_id: b.error_file_id });
++ }
++ if (path.startsWith('/v1/files/') && path.endsWith('/content')) {
++ const id = path.split('/')[3];
++ return send(200, state.files.get(id) || '', 'application/x-ndjson');
++ }
++ if (path === '/v1/chat/completions') {
++ return send(200, { id: 'cmpl_1', object: 'chat.completion', model: body.model || MODEL, choices: [{ index: 0, message: { role: 'assistant', content: 'the ordinary path answered' }, finish_reason: 'stop' }], usage: { prompt_tokens: 5, completion_tokens: 5, total_tokens: 10 } });
++ }
++ return send(404, { error: { message: 'not found' } });
++ });
++ });
++ return new Promise((resolve) => {
++ server.listen(0, '127.0.0.1', () => {
++ const url = `http://127.0.0.1:${server.address().port}`;
++ resolve({
++ url, state,
++ close: () => new Promise((r) => { server.closeAllConnections?.(); server.close(() => r()); }),
++ /** The provider's answers for a batch: one line per request, plus the errors of the refused ones. */
++ answer(id, { text = null, error = null } = {}) {
++ const b = state.batches.get(id);
++ const file = state.files.get(b.input_file_id) || '';
++ const lines = file.split('\n').filter(Boolean).map((l) => JSON.parse(l));
++ const out = [];
++ const errs = [];
++ for (const line of lines) {
++ if (error && error.custom_id === line.custom_id) {
++ errs.push(JSON.stringify({ id: 'req_x', custom_id: line.custom_id, response: null, error: { code: error.code || 'server_error', message: error.message || 'the provider failed this request' } }));
++ continue;
++ }
++ out.push(JSON.stringify({
++ id: 'req_1', custom_id: line.custom_id,
++ response: { status_code: 200, body: { id: 'cmpl_1', object: 'chat.completion', model: line.body.model, choices: [{ index: 0, message: { role: 'assistant', content: (text && text[line.custom_id]) || `answer for ${line.custom_id}` }, finish_reason: 'stop' }], usage: { prompt_tokens: 100, completion_tokens: 50, total_tokens: 150 } } },
++ error: null,
++ }));
++ }
++ const outId = `file_out_${++state.seq}`;
++ state.files.set(outId, out.join('\n'));
++ b.output_file_id = outId;
++ if (errs.length) {
++ const errId = `file_err_${++state.seq}`;
++ state.files.set(errId, errs.join('\n'));
++ b.error_file_id = errId;
++ }
++ b.status = 'completed';
++ return id;
++ },
++ });
++ });
++ });
++}
++
++/** A booted service whose shared provider is the stub, with a price card for the model. */
++async function bootWith(stub, env = {}, dir = undefined) {
++ const h = await boot({ dir, env: { AI_BATCH_NATIVE_MAX_ITEMS: '2', AI_BATCH_NATIVE_MAX_AGE_MS: '60000', ...env } });
++ await h.registry.upsertProvider({ key: 'shared', base_url: `${stub.url}/v1`, provider_access: 'OWN_DEPLOYMENT' }, { actor: 'test' });
++ await h.registry.upsertModel({ provider_key: 'shared', model_key: MODEL, type: 'chat', cost: { in_per_mtok: 1, out_per_mtok: 2, source: 'urn:openvibe:ai:test', verified_at: '2026-01-01' } }, { actor: 'test' });
++ return h;
++}
++
++t.test('the JSONL file has the provider\'s own shape, and the provider\'s own limits', () => {
++ const adapter = createOpenAiCompatibleAdapter({ baseUrl: 'http://x/v1', apiKey: 'k', limits: { maxItems: 3, maxBytes: 400, oneModelPerFile: true } });
++ const file = adapter.buildFile([{ customId: 'bni_1', body: { model: 'm1', messages: [{ role: 'user', content: 'hi' }] } }]);
++ assert.deepStrictEqual(JSON.parse(file.lines[0]), {
++ custom_id: 'bni_1', method: 'POST', url: '/v1/chat/completions',
++ body: { model: 'm1', messages: [{ role: 'user', content: 'hi' }] },
++ });
++ assert.ok(file.bytes > 0);
++ // The provider's own limits: too many requests, two models in one file, a file that is too big.
++ assert.throws(() => adapter.buildFile([1, 2, 3, 4].map((n) => ({ customId: `bni_${n}`, body: { model: 'm1', messages: [] } }))), /at most 3 requests/);
++ assert.throws(() => adapter.buildFile([{ customId: 'a', body: { model: 'm1', messages: [] } }, { customId: 'b', body: { model: 'm2', messages: [] } }]), /one model per file/);
++ assert.throws(() => adapter.buildFile([{ customId: 'a', body: { model: 'm1', messages: [{ role: 'user', content: 'x'.repeat(500) }] } }]), /file limit/);
++});
++
++t.test('a full group is submitted to the provider as one batch', async () => {
++ const stub = await stubProvider();
++ const h = await bootWith(stub);
++ try {
++ await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'one' });
++ // Not full yet (the limit is two).
++ await h.batcher.submitOnce();
++ assert.strictEqual(stub.state.calls.filter((c) => c.path === '/v1/batches').length, 0);
++ // The second item makes the group full: one file, one batch.
++ await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'two' });
++ const res = await h.batcher.submitOnce();
++ assert.strictEqual(res.sent, 1);
++ const files = stub.state.calls.filter((c) => c.path === '/v1/files');
++ const batches = stub.state.calls.filter((c) => c.path === '/v1/batches');
++ assert.strictEqual(files.length, 1);
++ assert.strictEqual(batches.length, 1);
++ const lines = files[0].body.file.split('\n').filter(Boolean);
++ assert.strictEqual(lines.length, 2); // the group is two items, the third waits
++ assert.strictEqual(batches[0].body.completion_window, '24h');
++ assert.strictEqual(batches[0].body.endpoint, '/v1/chat/completions');
++ } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('an old group is submitted even when it is not full', async () => {
++ const stub = await stubProvider();
++ const h = await bootWith(stub, { AI_BATCH_NATIVE_MAX_AGE_MS: '1000' });
++ try {
++ const item = await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'lonely' });
++ // The group is one minute old: old enough to be sent.
++ await h.db.prepare('UPDATE batch_queue SET created_at = ? WHERE id = ?').run(new Date(Date.now() - 60000).toISOString(), item.id);
++ const res = await h.batcher.submitOnce();
++ assert.strictEqual(res.sent, 1);
++ assert.strictEqual(stub.state.calls.filter((c) => c.path === '/v1/batches').length, 1);
++ } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('the answers are mapped back onto their own runs by custom_id', async () => {
++ const stub = await stubProvider();
++ const h = await bootWith(stub);
++ try {
++ const a = await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'first' });
++ const b = await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'second' });
++ await h.batcher.submitOnce();
++ const batchId = stub.state.batches.keys().next().value;
++ stub.answer(batchId, { text: { [a.id]: 'the answer for a', [b.id]: 'the answer for b' } });
++ await h.batcher.pollOnce();
++ const runA = await h.db.prepare('SELECT * FROM runs WHERE id = ?').get(a.runId);
++ const runB = await h.db.prepare('SELECT * FROM runs WHERE id = ?').get(b.runId);
++ assert.strictEqual(runA.status, 'succeeded');
++ assert.strictEqual(runB.status, 'succeeded');
++ assert.strictEqual(JSON.parse(runA.output).text, 'the answer for a');
++ assert.strictEqual(JSON.parse(runB.output).text, 'the answer for b');
++ // The tokens are the provider's own, and the cost is the provider's batch price (half of its list price).
++ assert.strictEqual(Number(runA.tokens_in), 100);
++ assert.strictEqual(Number(runA.tokens_out), 50);
++ // 100 input tokens at 1 USD per million, halved by the provider's batch price, plus the output.
++ const expected = (100n * 1000000000n / 2n / 1000000n) + (50n * 2000000000n / 2n / 1000000n);
++ assert.strictEqual(String(runA.cost_usd), String(Number(expected) / 1e9));
++ } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('a partial failure fails only its own item', async () => {
++ const stub = await stubProvider();
++ const h = await bootWith(stub);
++ try {
++ const a = await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'first' });
++ const b = await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'second' });
++ await h.batcher.submitOnce();
++ const batchId = stub.state.batches.keys().next().value;
++ stub.answer(batchId, { error: { custom_id: b.id, code: 'server_error', message: 'the provider refused this one' } });
++ await h.batcher.pollOnce();
++ const runA = await h.db.prepare('SELECT * FROM runs WHERE id = ?').get(a.runId);
++ const runB = await h.db.prepare('SELECT * FROM runs WHERE id = ?').get(b.runId);
++ assert.strictEqual(runA.status, 'succeeded');
++ assert.strictEqual(runB.status, 'failed');
++ assert.strictEqual(runB.error_code, 'server_error');
++ const items = await h.db.prepare('SELECT status FROM batch_queue WHERE id IN (?, ?) ORDER BY id').all(a.id, b.id);
++ assert.deepStrictEqual(items.map((i) => i.status).sort(), ['failed', 'succeeded']);
++ } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('a batch the provider cannot finish in time is re-routed to the ordinary path', async () => {
++ const stub = await stubProvider();
++ const h = await bootWith(stub);
++ try {
++ const item = await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'late' });
++ await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'later' });
++ // The deadline is already gone: the provider cannot meet it.
++ await h.db.prepare('UPDATE batch_queue SET deadline_at = ? WHERE status = ?').run(new Date(Date.now() - 1000).toISOString(), 'queued');
++ await h.batcher.submitOnce();
++ await h.batcher.pollOnce();
++ const row = await h.db.prepare('SELECT status, error_code FROM batch_queue WHERE id = ?').get(item.id);
++ assert.strictEqual(row.status, 'rerouted');
++ assert.strictEqual(row.error_code, 'batch.rerouted');
++ // The run went back to the ordinary path: it is not waiting for the provider any more.
++ const run = await h.db.prepare('SELECT status FROM runs WHERE id = ?').get(item.runId);
++ assert.notStrictEqual(run.status, 'queued');
++ } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('two pollers never settle the same item twice', async () => {
++ const stub = await stubProvider();
++ const h = await bootWith(stub);
++ try {
++ const a = await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'first' });
++ await h.batcher.enqueue({ principal: PRINCIPAL, provider: 'shared', model: MODEL, prompt: 'second' });
++ await h.batcher.submitOnce();
++ const batchId = stub.state.batches.keys().next().value;
++ stub.answer(batchId, { text: { [a.id]: 'only once' } });
++ await h.batcher.pollOnce();
++ await h.batcher.pollOnce(); // the second poller finds the batch settled
++ const row = await h.db.prepare('SELECT status, cost_nano_usd FROM batch_queue WHERE id = ?').get(a.id);
++ assert.strictEqual(row.status, 'succeeded');
++ const run = await h.db.prepare('SELECT status, tokens_in, tokens_out FROM runs WHERE id = ?').get(a.runId);
++ assert.strictEqual(run.status, 'succeeded');
++ assert.strictEqual(Number(run.tokens_in), 100);
++ assert.strictEqual(Number(run.tokens_out), 50);
++ } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('a customer batch runs on the native route and is settled at half the price', async () => {
++ const stub = await stubProvider();
++ const billing = await fakeBilling();
++ const h = await bootWith(stub, {
++ AI_CREDITS: '1', USAGE_SAMPLES: 'off',
++ OV_BILLING_INTERNAL_URL: billing.url, OV_NETWORK_INTERNAL_URL: billing.url,
++ OV_OAUTH_CLIENT_SECRET: 'test-secret', OV_OAUTH_CLIENT_ID: 'ai',
++ });
++ try {
++ const ULID = '01JAB2C3D4E5F6G7H8J9K0MNPQ';
++ const PRJ = `prj_${ULID}`;
++ const appTok = token('app', ['ai.app.run'], { sub: `app:app_${ULID}`, actorType: 'app', ns: [PRJ, `app.${PRJ}.*`], extra: { project_id: PRJ, env: 'production' } });
++ const res = await request(h.base, 'POST', '/api/v1/batches', { tok: appTok, body: { items: [{ prompt: 'a customer prompt' }], model: MODEL } });
++ assert.strictEqual(res.status, 201);
++ const batchId = res.body.batch.id;
++ // The capacity: the native batch route of the shared provider is eligible, and it costs at most half.
++ const model = await h.registry.listModels().then((rows) => rows.find((r) => r.model_key === MODEL));
++ const capacity = await h.batcher.cardsFor('shared', MODEL);
++ assert.ok(capacity.batch, 'the provider has a batch card');
++ assert.ok(BigInt(capacity.batch.nano_usd_per_m.input) <= BigInt(String(Math.floor(1000000000 / 2))), 'the batch price is at most half the list price');
++ // The batch worker puts the item on the native route.
++ const run = await h.batches.runOnce();
++ assert.ok(run.route === 'native-batch', JSON.stringify(run));
++ assert.strictEqual(run.queued, true);
++ // The provider answers it. The group is one item and not full, so it goes out when it is old.
++ 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 providerBatch = stub.state.batches.keys().next().value;
++ stub.answer(providerBatch, { text: {} });
++ await h.batcher.pollOnce();
++ // The customer is charged half the retail price, and the network's real cost is the batch price.
++ const settle = billing.state.settlements[0];
++ assert.ok(settle, 'the customer was charged');
++ assert.strictEqual(settle.body.retail.nano_usd_per_m.input, String(Math.floor(1000000000 / 2)));
++ assert.strictEqual(settle.body.retail.nano_usd_per_m.output, String(Math.floor(2000000000 / 2)));
++ const item = await h.db.prepare('SELECT status, answer, charged_nano_usd FROM batch_items WHERE batch_id = ?').get(batchId);
++ assert.strictEqual(item.status, 'succeeded');
++ assert.strictEqual(String(item.charged_nano_usd), String(settle.body.supply_cost_nano_usd));
++ } finally { await h.stop(); await stub.close(); await billing.close(); }
++});
++
++/** A small fake Billing: the hold and the settle, and what they were told. */
++async function fakeBilling({ balance = 1000000000000n } = {}) {
++ const state = { balance: BigInt(balance), held: 0n, holds: new Map(), settlements: [], seq: 0 };
++ const server = http.createServer((req, res) => {
++ let raw = '';
++ req.on('data', (c) => { raw += c; });
++ req.on('end', () => {
++ const path = new URL(req.url, 'http://x').pathname;
++ let body = {};
++ try { body = raw ? JSON.parse(raw) : {}; } catch { body = {}; }
++ const send = (status, out) => { res.writeHead(status, { 'Content-Type': 'application/json' }); res.end(JSON.stringify(out)); };
++ if (path === '/internal/credits/holds') {
++ const want = BigInt(String(body.max_nano_usd || '0'));
++ if (want > state.balance - state.held) {
++ return send(402, { detail: 'not enough balance', title: 'Insufficient funds' });
++ }
++ const prev = state.holds.get(body.idempotency_key);
++ if (prev) return send(201, { hold_id: prev.hold_id, held_nano_usd: prev.held_nano_usd, available_nano_usd: String(state.balance - state.held), lease_expires_at: new Date(Date.now() + 300000).toISOString() });
++ const id = `hld_${++state.seq}`;
++ state.holds.set(body.idempotency_key, { hold_id: id, held_nano_usd: String(body.max_nano_usd) });
++ state.held += BigInt(String(body.max_nano_usd));
++ return send(201, { hold_id: id, held_nano_usd: String(body.max_nano_usd), available_nano_usd: String(state.balance - state.held), lease_expires_at: new Date(Date.now() + 300000).toISOString() });
++ }
++ if (path.startsWith('/internal/credits/holds/') && path.endsWith('/settle')) {
++ state.settlements.push({ path, body });
++ const hold = [...state.holds.values()].find((x) => x.hold_id === decodeURIComponent(path.split('/')[4]));
++ const held = hold ? BigInt(hold.held_nano_usd) : 0n;
++ return send(201, { charge_nano_usd: '0', released_nano_usd: String(held), journal_id: `txn_${state.settlements.length}`, shortfall_nano_usd: '0' });
++ }
++ if (path.startsWith('/internal/credits/holds/') && path.endsWith('/release')) {
++ return send(200, { released: true });
++ }
++ if (path === '/oauth/token') return send(200, { access_token: 'test-token', token_type: 'Bearer', expires_in: 300, scope: '' });
++ if (path === '/api/v1/usage') return send(201, { ok: true });
++ return send(404, { error: 'not found' });
++ });
++ });
++ await new Promise((r) => server.listen(0, '127.0.0.1', r));
++ const url = `http://127.0.0.1:${server.address().port}`;
++ return { url, state, close: () => new Promise((r) => { server.closeAllConnections?.(); server.close(() => r()); }) };
++}
++
++t.test('a first-party service with the ai.batches.create grant may open a batch of the network\'s own', async () => {
++ const stub = await stubProvider();
++ const h = await bootWith(stub);
++ try {
++ // The grant: the batch is the network's own, and it names the service that asked.
++ const granted = token('live', ['ai.run.create', 'ai.batches.create'], { sub: 'svc:live' });
++ const res = await request(h.base, 'POST', '/api/v1/batches', { tok: granted, body: { items: [{ prompt: 'a service prompt' }], model: MODEL } });
++ assert.strictEqual(res.status, 201, res.text);
++ assert.strictEqual(res.body.batch.owner_type, 'service');
++ assert.strictEqual(res.body.batch.owner_id, 'live');
++ // A service without the grant may not open one.
++ const plain = token('live', ['ai.run.create'], { sub: 'svc:live' });
++ const refused = await request(h.base, 'POST', '/api/v1/batches', { tok: plain, body: { items: [{ prompt: 'a service prompt' }] } });
++ assert.strictEqual(refused.status, 403);
++ assert.strictEqual(refused.body.code, 'batch.no_grant');
++ } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('a restart while the batch waits leaves the run queued, and the answer still finishes it', async () => {
++ const stub = await stubProvider();
++ const billing = await fakeBilling();
++ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-batch-restart-'));
++ const env = {
++ AI_CREDITS: '1', USAGE_SAMPLES: 'off',
++ OV_BILLING_INTERNAL_URL: billing.url, OV_NETWORK_INTERNAL_URL: billing.url,
++ OV_OAUTH_CLIENT_SECRET: 'test-secret', OV_OAUTH_CLIENT_ID: 'ai',
++ };
++ const h1 = await bootWith(stub, env, dir);
++ try {
++ const ULID = '01JAB2C3D4E5F6G7H8J9K0MNPQ';
++ const PRJ = `prj_${ULID}`;
++ const appTok = token('app', ['ai.app.run'], { sub: `app:app_${ULID}`, actorType: 'app', ns: [PRJ, `app.${PRJ}.*`], extra: { project_id: PRJ, env: 'production' } });
++ const res = await request(h1.base, 'POST', '/api/v1/batches', { tok: appTok, body: { items: [{ prompt: 'a customer prompt' }], model: MODEL } });
++ assert.strictEqual(res.status, 201, res.text);
++ const run = await h1.batches.runOnce();
++ assert.strictEqual(run.route, 'native-batch');
++ assert.strictEqual(run.queued, true);
++ const item = await h1.db.prepare('SELECT * FROM batch_items WHERE batch_id = ?').get(res.body.batch.id);
++ // The group goes out to the provider.
++ await h1.db.prepare("UPDATE batch_queue SET created_at = ? WHERE status = 'queued'").run(new Date(Date.now() - 60000).toISOString());
++ await h1.batcher.submitOnce();
++ assert.strictEqual(stub.state.calls.filter((c) => c.path === '/v1/batches').length, 1);
++ // The service goes away with the run still waiting for the provider.
++ await h1.stop();
++
++ // A restart on the same database: the run is still waiting, so it is not interrupted.
++ const h2 = await bootWith(stub, env, dir);
++ try {
++ const row = await h2.db.prepare('SELECT status FROM runs WHERE id = ?').get(item.run_id);
++ assert.strictEqual(row.status, 'queued', 'the restart did not fail the waiting run');
++ // The provider answers, and the run finishes on the ordinary path.
++ const providerBatch = stub.state.batches.keys().next().value;
++ stub.answer(providerBatch, { text: {} });
++ await h2.batcher.pollOnce();
++ const done = await h2.db.prepare('SELECT * FROM runs WHERE id = ?').get(item.run_id);
++ assert.strictEqual(done.status, 'succeeded');
++ assert.ok(String(done.output).includes('answer for'), 'the run carries the provider\'s answer');
++ assert.ok(Number(done.cost_usd) > 0, 'the run carries the batch cost');
++ assert.ok(billing.state.settlements.length >= 1, 'the customer was charged the ordinary way');
++ } finally { await h2.stop(); }
++ } finally { await stub.close(); await billing.close(); }
++});
++
++t.test('a customer who cannot pay is refused before the provider is given the work', async () => {
++ const stub = await stubProvider();
++ const billing = await fakeBilling({ balance: 1n });
++ const h = await bootWith(stub, {
++ AI_CREDITS: '1', USAGE_SAMPLES: 'off',
++ OV_BILLING_INTERNAL_URL: billing.url, OV_NETWORK_INTERNAL_URL: billing.url,
++ OV_OAUTH_CLIENT_SECRET: 'test-secret', OV_OAUTH_CLIENT_ID: 'ai',
++ });
++ try {
++ const ULID = '01JAB2C3D4E5F6G7H8J9K0MNPQ';
++ const PRJ = `prj_${ULID}`;
++ const appTok = token('app', ['ai.app.run'], { sub: `app:app_${ULID}`, actorType: 'app', ns: [PRJ, `app.${PRJ}.*`], extra: { project_id: PRJ, env: 'production' } });
++ const res = await request(h.base, 'POST', '/api/v1/batches', { tok: appTok, body: { items: [{ prompt: 'a customer prompt' }], model: MODEL } });
++ assert.strictEqual(res.status, 201, res.text);
++ const run = await h.batches.runOnce();
++ assert.strictEqual(run.queued, false);
++ assert.strictEqual(run.error, 'billing.insufficient_funds');
++ // The provider was never asked to do the work: nobody owes it anything.
++ assert.strictEqual(stub.state.calls.filter((c) => c.path === '/v1/batches').length, 0);
++ const item = await h.db.prepare('SELECT * FROM batch_items WHERE batch_id = ?').get(res.body.batch.id);
++ assert.strictEqual(item.status, 'failed');
++ assert.strictEqual(item.error_code, 'billing.insufficient_funds');
++ } finally { await h.stop(); await stub.close(); await billing.close(); }
++});
++
++t.test('a token without ai.batches.create is refused even when it carries every other AI capability', async () => {
++ const stub = await stubProvider();
++ const h = await bootWith(stub);
++ try {
++ const every = token('live', ALL, { sub: 'svc:live' });
++ const refused = await request(h.base, 'POST', '/api/v1/batches', { tok: every, body: { items: [{ prompt: 'a service prompt' }], model: MODEL } });
++ assert.strictEqual(refused.status, 403);
++ assert.strictEqual(refused.body.code, 'batch.no_grant');
++ } finally { await h.stop(); await stub.close(); }
++});
++
++t.run();