BA3: batch-class runs wait for a provider's own batch API (#363)

OP OpenVibers committed 5 days ago · 91ccf27019 · 9 files, +733 −55

A run whose resolved latency class is 'batch' is now enqueued through the
native batcher when the route offers a provider whose batch API is enabled
and whose batch price fits the half-price card. The run is created at once
with status queued and latency_class 'batch'; the provider's answer finishes
it on the ordinary path (events, webhooks, receipt and the batch price).
deadline_hours (1-72) bounds the wait: an item past it re-routes to the
background class once. With no eligible batch provider the run runs in the
background class at once and says it was clamped. A ?wait= caller gets 202
with the queued run at once.

The engine prepares the provider request of a single-step workflow and
resumes that step with the stored answer, so ai.complete, ai.generate and
the background product workflows all keep their own output shaping. The
batcher's plan also runs BA2's Balance probe before the run is created.

Changes

ddiff --git a/server/batch/batcher.js b/server/batch/batcher.js
iindex 88d796b..3f13dd9 100644
---- a/server/batch/batcher.js
++++ b/server/batch/batcher.js
@@@ -84,7 +84,7 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
      // 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,
++        enabled: !(config.batches && config.batches.enabled === false),
          // 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,
@@@ -196,6 +196,90 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
          };
      }
  
++    /** What a restarted service needs of the run's context: scalars only, never a function or a token. */
++    function ctxSnapshot(ctx) {
++        const out = {};
++        for (const k of ['appRun', 'projectId', 'sandbox', 'appId', 'onBehalfOf', 'personSubject', 'personPays', 'serviceBudgeted',
++            'freeUse', 'byok', 'maxCostNano', 'communityOptIn', 'contentPolicy', 'presetPolicy', 'presetName']) {
++            if (ctx && ctx[k] !== undefined) out[k] = ctx[k];
++        }
++        return out;
++    }
++
++    /**
++     * BA3: where a batch-class run waits. The run's own workflow has one provider step (the engine built its
++     * request); the route's offers are tried in order, and the first provider whose own batch API is enabled
++     * and whose real batch price fits inside the half of the retail card the customer is charged takes it.
++     * A payer whose Balance cannot cover the run is refused here, before the provider is given the work.
++     *
++     * → { provider, model, body, prompt, stepKind, operation, maxTokens, holdCard, data } or null (no
++     *   eligible provider: the run runs in the background class and says it was clamped).
++     */
++    async function planRun({ prepared = null, offers = [], principal = {}, ctx = {}, input = null, requestId = null, maxCostNano = null } = {}) {
++        if (!settings.enabled || !prepared || typeof prepared.bodyFor !== 'function' || !prepared.route) return null;
++        const capacity = require('./capacity');                     // lazy: capacity requires this module for its cards
++        for (const offer of offers || []) {
++            if (!offer || !offer.provider || !offer.model) continue;
++            const cfg = providerConfig(offer.provider);
++            // Only a provider whose batch API is enabled and whose adapter speaks the OpenAI batch shape.
++            if (!cfg || !cfg.enabled || cfg.adapter !== 'openai-compatible') continue;
++            const provider = await registry.getProvider(offer.provider);
++            if (!provider || provider.status !== 'active') continue;
++            const cards = await cardsFor(offer.provider, offer.model);
++            if (!cards || !cards.batch) continue;
++            // The same money rule as capacity route (d): the provider's real batch price fits inside the
++            // retail card the run is charged at.
++            const charge = capacity.chargeOf(cards.row);
++            if (!charge || !capacity.within(cards.batch.nano_usd_per_m, charge)) continue;
++            let body;
++            try { body = prepared.bodyFor(offer.model); } catch { continue; }
++            if (!body || !Array.isArray(body.messages) || !body.messages.length) continue;
++            body.model = offer.model;
++            // BA2's probe: a real hold, released at once. A payer who cannot cover the run is refused now,
++            // so the network never owes a provider for an answer nobody can pay for.
++            if (credits && typeof credits.probe === 'function') {
++                const verdict = await credits.probe({
++                    requestId: requestId || String(ctx.runId || ''),
++                    ctx: payerCtx(principal, { discount: 'half', maxCostNano, requestId }),
++                    input, card: cards.retail, provider: offer.provider, model: offer.model, access: cards.access,
++                    maxTokens: prepared.maxTokens,
++                });
++                if (!verdict.ok) throw new AiError(402, String(verdict.reason || 'billing.insufficient_funds'), String(verdict.detail || 'Not enough Balance'));
++            }
++            return {
++                provider: offer.provider, model: offer.model, body, prompt: prepared.prompt || '',
++                stepKind: prepared.stepKind, operation: prepared.operation, maxTokens: prepared.maxTokens,
++                holdCard: cards.retail,
++                data: {
++                    principal: planOf({ principal, input, card: cards.retail, maxTokens: prepared.maxTokens, discount: 'half', maxCostNano, requestId, servicePaid: Boolean(ctx.serviceBudgeted) }).principal,
++                    input: input || null, card: cards.retail, maxTokens: Number(prepared.maxTokens) || 4096,
++                    discount: 'half', maxCostNano: maxCostNano == null ? null : String(maxCostNano),
++                    requestId: requestId || null, route: null, charge: null,
++                    runLevel: true, stepKind: prepared.stepKind, ctx: ctxSnapshot(ctx),
++                },
++            };
++        }
++        return null;
++    }
++
++    /**
++     * BA3: put a run that already exists (created by runs.create) into the queue, with the deadline its
++     * caller asked for. The run itself is never created here: the API saw it at once, and the answer is
++     * written onto that same run when the provider answers.
++     */
++    async function enqueueRun({ runId, plan, deadlineHours = null } = {}) {
++        if (!runId || !plan || !plan.body) throw new AiError(500, 'batch.no_run', 'a batch-class run needs its own run');
++        const hours = Number(deadlineHours) > 0 ? Math.max(1, Math.min(72, Math.trunc(Number(deadlineHours)))) : Number(settings.deadlineHours) || 24;
++        const deadline = new Date(clock.now() + hours * 3600 * 1000).toISOString();
++        const id = `bni_${crypto.randomBytes(10).toString('hex')}`;
++        const data = { ...(plan.data || {}), deadlineHours: hours };
++        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 (?, ?, 'service', NULL, NULL, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?)`)
++            .run(id, runId, plan.provider, plan.model, plan.prompt || '', JSON.stringify(plan.body),
++                plan.holdCard ? JSON.stringify(plan.holdCard) : null, JSON.stringify(data), deadline, iso(), iso());
++        return { id, runId };
++    }
++
      /**
       * 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.
@@@ -244,8 +328,11 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
          if (running) return { skipped: 'running' };
          running = true;
          try {
++            // BA3: an item whose deadline passed before the provider took it is never sent: it re-routes
++            // to the ordinary (background) path now, exactly once.
++            const rerouted = await reroutePending('the deadline passed before the provider took the batch');
              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();
++                FROM batch_queue WHERE status = 'queued' AND deadline_at > ?::timestamptz GROUP BY provider, model`).all(iso());
              const sent = [];
              for (const group of groups || []) {
                  const age = clock.now() - new Date(group.oldest).getTime();
@@@ -253,7 +340,7 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
                  const job = await submitGroup(group.provider, group.model);
                  if (job) sent.push(job);
              }
--            return { sent: sent.length, jobs: sent };
++            return { sent: sent.length, jobs: sent, ...(rerouted ? { rerouted } : {}) };
          } catch (err) {
              log.warn('[batch] the submitter stopped early: ' + String((err && err.message) || err));
              return { error: String((err && err.code) || 'error') };
@@@ -265,7 +352,7 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
      /** 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));
++            AND deadline_at > ?::timestamptz ORDER BY created_at LIMIT ?`).all(providerKey, modelKey, iso(), Number(settings.maxItemsPerGroup));
          if (!items || !items.length) return null;
          const adapter = await adapterFor(providerKey);
          const cfg = providerConfig(providerKey);
@@@ -320,8 +407,10 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
          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 <= ?")
++            // provider's answer is still there, and the run must not wait for ever. An item with no
++            // provider batch (a re-route that was interrupted) goes back to 'queued', so the deadline
++            // is looked at again instead of waiting for a job that does not exist.
++            await db.prepare("UPDATE batch_queue SET status = CASE WHEN job_id IS NULL THEN 'queued' ELSE 'submitted' END, 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' };
@@@ -425,6 +514,9 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
                  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,
++                // BA3: a run-level item resumes its own workflow step with this answer; a customer batch
++                // item's run is the ai.complete run the batcher created, which the shortcut finishes.
++                ...(plan.runLevel ? { resume: true, stepKind: plan.stepKind || null } : {}),
              }, { 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);
@@@ -440,19 +532,40 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
  
      /**
       * 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.
++     * run answers through the pool as any other run does, and the item says it was re-routed. A run-level
++     * run (BA3) re-routes into the background class, exactly once: the item is then no longer queued, and
++     * the run itself never asks for a batch again.
       */
++    async function rerouteItem(item, reason, from = 'submitted') {
++        const claimed = await db.prepare(`UPDATE batch_queue SET status = 'settling', updated_at = ? WHERE id = ? AND status = ? RETURNING id`)
++            .get(iso(), item.id, from).catch(() => null);
++        if (!claimed) return false;
++        const plan = item.plan ? JSON.parse(item.plan) : {};
++        if (plan.runLevel && item.run_id) {
++            const row = await db.prepare('SELECT options FROM runs WHERE id = ?').get(item.run_id).catch(() => null);
++            const opts = row && row.options ? (JSON.parse(row.options) || {}) : {};
++            opts.latency_clamped = { asked: 'batch', used: 'background' };
++            await db.prepare("UPDATE runs SET latency_class = 'background', options = ? WHERE id = ?")
++                .run(JSON.stringify(opts), item.run_id).catch(() => null);
++        }
++        await runs.completeBatch(item.run_id, { reroute: true }, { 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);
++        return true;
++    }
++
++    /** The items whose deadline passed before the provider ever took them. */
++    async function reroutePending(reason) {
++        const rows = await db.prepare(`SELECT * FROM batch_queue WHERE status = 'queued' AND deadline_at <= ?::timestamptz`).all(iso()).catch(() => []);
++        let n = 0;
++        for (const item of rows || []) if (await rerouteItem(item, reason, 'queued')) n++;
++        return n;
++    }
++
      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);
--        }
++        for (const item of items || []) await rerouteItem(item, reason, 'submitted');
          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);
      }
@@@ -510,7 +623,7 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock
      /** 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 };
++    const api = { enqueue, enqueueRun, planRun, submitOnce, pollOnce, runOnce, start, stop, settings, cardsFor, scaleCard, attach };
      return api;
  }
  
ddiff --git a/server/batch/providers/openai-compatible.js b/server/batch/providers/openai-compatible.js
iindex 0ab2464..2921338 100644
---- a/server/batch/providers/openai-compatible.js
++++ b/server/batch/providers/openai-compatible.js
@@@ -195,7 +195,9 @@ function createOpenAiCompatibleAdapter({ baseUrl, apiKey, fetchImpl = globalThis
                  out.push({
                      custom_id: row.custom_id,
                      ok: true,
--                    output: { text },
++                    // The raw answer's choices, so a complete run (ai.complete) keeps its tool calls and finish
++                    // reason too; the text is what a chat/generate step shapes.
++                    output: { text, completion: { choices: body.choices || [], system_fingerprint: body.system_fingerprint || null } },
                      usage: {
                          input: Number(usage.prompt_tokens || 0),
                          output: Number(usage.completion_tokens || 0),
ddiff --git a/server/index.js b/server/index.js
iindex 494dd62..f9dd805 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -144,7 +144,10 @@ 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,
++    // BA3: the runs and the batcher need each other (a batch-class run is created first, then queued), so
++    // the runs hold this facade and the batcher fills it in below, before any request can arrive.
++    const batchRef = {};
++    const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules, credentials, preferences, tiers, projectSecrets, pool, runBudgets, credits: creditsRef, community, presets, receipts, abuse, webhooks, batch: batchRef,
          // 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) => {
@@@ -185,6 +188,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
      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 });
++    // BA3: a batch-class run asks the batcher where it may wait, and hands its own run to the queue.
++    batchRef.planRun = batcher.planRun;
++    batchRef.enqueueRun = batcher.enqueueRun;
      // 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, batcher, config, clock, log });
      batcher.attach(batches);
ddiff --git a/server/openapi.js b/server/openapi.js
iindex e1241d6..e13e18b 100644
---- a/server/openapi.js
++++ b/server/openapi.js
@@@ -63,7 +63,7 @@ const runBody = (required = []) => ({
          workflow: str({ description: 'The workflow key, e.g. ai.generate' }), version: integer({ minimum: 1 }),
          input: { type: 'object' }, idempotency_key: str({ minLength: 8, maxLength: 128 }),
          target: { type: 'object' }, attribution: { type: 'object' }, on_behalf_of: ref('SubjectRef'),
--        options: { type: 'object' }, route: { type: 'object', description: '{ provider, model } — one offer from GET /api/v1/routes?operation=…' },
++        options: { type: 'object', description: 'cache, debug, and the latency class: latency (interactive | background | batch); a batch-class run also takes deadline_hours (1-72, the default is 24)' }, route: { type: 'object', description: '{ provider, model } — one offer from GET /api/v1/routes?operation=…' },
      },
  });
  const directBody = {
@@@ -708,6 +708,9 @@ function openapi() {
                      source_service: nullable('string'), target: nullable('object'), input: nullable('object'), output: nullable('object'),
                      error: nullable('object', { properties: { code: str(), detail: str() } }), synthetic: { type: 'boolean' },
                      provenance: { type: 'object' },
++                    latency_class: { type: 'string', enum: ['interactive', 'background', 'batch'] },
++                    service_tier: nullable('string', { description: 'the tier the answer really ran on: standard, flex, batch, offpeak or local' }),
++                    latency_clamped: nullable('object', { description: '{ asked, used }: the class the caller asked for and the class the run really took' }),
                      usage: { type: 'object', properties: { tokens_in: nullable('integer'), tokens_out: nullable('integer'), cost_usd: nullable('number'), attempts: nullable('integer'), usage_sample_ids: nullable(arrayOf(str())) } },
                      explain: nullable('object'), citations_count: nullable('integer'), grounding: nullable('object'),
                      retry_of: nullable('string'), idempotency_key: nullable('string'), trace_id: nullable('string'),
ddiff --git a/server/providers/openai.js b/server/providers/openai.js
iindex 5787054..321a35d 100644
---- a/server/providers/openai.js
++++ b/server/providers/openai.js
@@@ -57,6 +57,29 @@ function buildBody({ model, system, messages, image, json, maxTokens, temperatur
      return body;
  }
  
++/**
++ * The Chat Completions body of a complete operation (ai.complete), for a provider batch request too:
++ * the caller's own request with the routed model, a reasoning model's allowance, and the Flex tier.
++ */
++function buildCompleteBody({ model, openai = {}, maxTokens, temperature, serviceTier } = {}) {
++    const o = openai || {};
++    const body = { model, messages: o.messages };
++    for (const k of ['tools', 'tool_choice', 'parallel_tool_calls', 'response_format', 'stop', 'seed', 'top_p', 'presence_penalty', 'frequency_penalty']) {
++        if (o[k] !== undefined && o[k] !== null) body[k] = o[k];
++    }
++    if (isReasoningModel(model)) {
++        // max_tokens is the answer the caller wants; a reasoning model thinks first, out of the same allowance.
++        if (maxTokens) body.max_completion_tokens = Math.max(maxTokens, 256) + 1024;
++        body.reasoning_effort = 'low';
++        for (const k of ['top_p', 'presence_penalty', 'frequency_penalty']) delete body[k];
++    } else {
++        if (maxTokens) body.max_tokens = maxTokens;
++        if (temperature != null) body.temperature = temperature;
++    }
++    if (serviceTier) body.service_tier = String(serviceTier);
++    return body;
++}
++
  function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetch } = {}) {
      const baseUrl = String(record.base_url || DEFAULT_BASE_URL).replace(/\/+$/, '');
      // A local llama.cpp process serves one model, so the embedding endpoint may live on another local
@@@ -146,23 +169,7 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc
       */
      async function complete(req) {
          const o = req.openai || {};
--        const body = { model: req.model, messages: o.messages };
--        for (const k of ['tools', 'tool_choice', 'parallel_tool_calls', 'response_format', 'stop', 'seed', 'top_p', 'presence_penalty', 'frequency_penalty']) {
--            if (o[k] !== undefined && o[k] !== null) body[k] = o[k];
--        }
--        if (isReasoningModel(req.model)) {
--            // max_tokens is the answer the caller wants; a reasoning model thinks first, out of the same allowance. Without
--            // room for that, and with its default (medium) effort, a long tool result could use it all up thinking and
--            // answer nothing: Chat's first production inbox action (2026-10-25) ended "finish_reason: length", empty.
--            if (req.maxTokens) body.max_completion_tokens = Math.max(req.maxTokens, 256) + 1024;
--            body.reasoning_effort = 'low';
--            for (const k of ['top_p', 'presence_penalty', 'frequency_penalty']) delete body[k];
--        } else {
--            if (req.maxTokens) body.max_tokens = req.maxTokens;
--            if (req.temperature != null) body.temperature = req.temperature;
--        }
--        // The service tier (BA1): a Flex tier travels with the request and the answer says which tier it ran on.
--        if (req.serviceTier) body.service_tier = String(req.serviceTier);
++        const body = buildCompleteBody({ model: req.model, openai: o, maxTokens: req.maxTokens, temperature: req.temperature, serviceTier: req.serviceTier });
          if (typeof req.onDelta === 'function') return await completeStreaming(body, req);
          const j = await postJson(`${baseUrl}/chat/completions`, headers(), body, { signal: req.signal, timeoutMs: req.timeoutMs, fetchImpl });
          if (!Array.isArray(j.choices) || !j.choices.length) throw new ProviderError('completion has no choices', { code: 'provider.output' });
@@@ -250,4 +257,4 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc
      return adapter;
  }
  
--module.exports = { createOpenAiProvider, buildBody, DEFAULT_BASE_URL };
++module.exports = { createOpenAiProvider, buildBody, buildCompleteBody, DEFAULT_BASE_URL };
ddiff --git a/server/runs.js b/server/runs.js
iindex 3b4ae77..3e734fd 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -35,6 +35,18 @@ const TERMINAL = new Set(['succeeded', 'failed', 'cancelled', 'cached']);
  const QUEUE_RETRY_AFTER_S = 5;
  const newRunId = () => `run_${ids.ulid()}`;
  
++/**
++ * BA3: how long a batch-class run may wait for a provider's own batch API, in hours (1 to 72). The
++ * caller names it in options.deadline_hours; the default is the batcher's own 24 hours.
++ */
++function batchDeadlineHours(value) {
++    if (value == null || value === '') return 24;
++    const n = Number(value);
++    if (!Number.isFinite(n)) throw new AiError(422, 'input.invalid', 'deadline_hours must be a number of hours');
++    if (n < 1 || n > 72) throw new AiError(422, 'input.invalid', 'deadline_hours is between 1 and 72');
++    return Math.trunc(n);
++}
++
  function entityKey(ref) { return ref ? `${ref.service}:${ref.type}:${ref.id}` : null; }
  function subjectKey(ref) { return ref ? `${ref.type}:${ref.id}` : null; }
  
@@@ -107,7 +119,7 @@ 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, batchWaiting = 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, batch = 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 }
@@@ -268,7 +280,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
          const askedLatency = options.latency != null ? options.latency : null;
          const resolvedLatency = latency.resolve(workflowClass, askedLatency);
          if (resolvedLatency.error) throw new AiError(422, 'input.invalid', resolvedLatency.error, { latency_class: workflowClass });
--        const latencyClass = resolvedLatency.class;
++        let latencyClass = resolvedLatency.class;         // BA3 may move a batch-class run to the background class
          // A faster class than the workflow allows was asked for: it runs in its own class, and the run's options say so.
          if (resolvedLatency.clamped) options.latency_clamped = resolvedLatency.clamped;
          // Community capacity is opt-in per request: a person's machine sees the prompt, so only a caller that says so
@@@ -475,6 +487,51 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
              }
          }
  
++        // BA3: a run whose resolved class is 'batch' waits for a provider's own batch API whenever the
++        // route has one whose real batch price fits the half-price card (the same money rule as capacity
++        // route (d)). The run is created at once; the provider's answer finishes it on the ordinary path.
++        // When no provider on the route can batch, the run runs in the background class at once and says so.
++        let autoBatch = null;
++        let batchDeadline = null;
++        if (latencyClass === 'batch' && !(principal && principal.deferBatch)) {
++            batchDeadline = batchDeadlineHours(options.deadline_hours);
++            // A run paid by somebody else's key or by a free allowance is not put on the network's batch
++            // provider; a free-only app run cannot be placed on paid capacity either.
++            const appFreeOnly = Boolean(ctx.appRun && (ctx.sandbox || !(tiers && tiers.enabled)));
++            let prepared = null;
++            if (batch && typeof batch.planRun === 'function' && engine.prepareBatch
++                && credentialSubject === null && !projectProvider && !selfKeys && !freeUse && !communityOptIn && !appFreeOnly) {
++                try {
++                    prepared = await engine.prepareBatch(wf, { template_key: tpl ? tpl.key : null, template_version: tpl ? tpl.version : null }, input, { ...ctx, runId: id });
++                } catch (err) {
++                    log.warn(`[runs] ${id}: the batch request could not be prepared: ${(err && err.message) || err}`);
++                }
++            }
++            if (prepared && pool && typeof pool.offers === 'function') {
++                let offers = await pool.offers(prepared.route, prepared.operation, ctx).catch(() => []);
++                // A pinned run waits on the provider it pinned, never on another eligible one.
++                if (pinned) offers = offers.filter((o) => o.provider === pinned.provider && (pinned.model == null || o.model === pinned.model));
++                if (offers.length) autoBatch = await batch.planRun({ prepared, offers, principal, ctx, input, requestId: requestId || id, maxCostNano });
++            }
++            if (autoBatch) {
++                ctx.discount = 'half';
++                ctx.latencyClass = 'batch';
++                options.deadline_hours = batchDeadline;
++                options.batch = { provider: autoBatch.provider, model: autoBatch.model, deadline_hours: batchDeadline };
++            } else {
++                latencyClass = 'background';
++                ctx.latencyClass = 'background';
++                delete ctx.discount;
++                options.latency_clamped = { asked: 'batch', used: 'background' };
++            }
++            base.latency_class = latencyClass;
++            base.options = JSON.stringify({
++                ...parseJson(base.options, {}),
++                ...(options.latency_clamped ? { latency_clamped: options.latency_clamped } : {}),
++                ...(autoBatch ? { deadline_hours: batchDeadline, batch: { provider: autoBatch.provider, model: autoBatch.model } } : {}),
++            });
++        }
++
          // Queue caps, then quota: both before any provider call (a refused run is not counted).
          const caller = `${requester.type}:${requester.id}`;
          // Kudos bought with volunteer service start this run earlier in the queue (never cash, never a bypass of limits).
@@@ -487,8 +544,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 {
--            // 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);
++            // BA2/BA3: a batch run reserves its quota when the answer is here, not while it waits.
++            reserved = credentialSubject !== null || selfKeys || (principal && principal.deferBatch) || autoBatch ? null : await quotas.reserve(ctx);
          } catch (err) { if (hold) await tiers.release(hold); throw err; }
  
          try {
@@@ -511,6 +568,46 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
              }
          }
  
++        // BA3: the run waits for the provider's own batch API. It is created now (its class says 'batch')
++        // and takes no execution slot and no quota until the answer is here; the batcher then finishes it
++        // on the ordinary path, with the ordinary settle, receipt and events. A caller that waits (?wait=)
++        // is given the queued run at once (202 from the API), never held for hours.
++        if (autoBatch) {
++            const controller = new AbortController();
++            let resolveDone;
++            const done = new Promise((r) => { resolveDone = r; });
++            try {
++                deferred.set(id, { controller, scope, ctx, input, hold, resolveDone, runLevel: true });
++                inflight.set(id, { controller, promise: done, release: () => {}, caller, priority });
++                await batch.enqueueRun({ runId: id, plan: autoBatch, deadlineHours: batchDeadline });
++                return { run: await get(id), created: true, promise: null };
++            } catch (err) {
++                // The queue could not take it (the provider's batch table is unreachable, say): run it in
++                // the background class now rather than leave it queued for ever.
++                deferred.delete(id);
++                inflight.delete(id);
++                log.warn(`[runs] ${id}: the provider's batch queue did not take the run (${(err && err.message) || err}); it runs in the background class`);
++                latencyClass = 'background';
++                ctx.latencyClass = 'background';
++                delete ctx.discount;
++                const opts = { ...parseJson(base.options, {}), latency_clamped: { asked: 'batch', used: 'background' } };
++                delete opts.deadline_hours;
++                delete opts.batch;
++                base.latency_class = 'background';
++                base.options = JSON.stringify(opts);
++                await db.prepare("UPDATE runs SET latency_class = 'background', options = ? WHERE id = ?").run(base.options, id);
++                try {
++                    reserved = credentialSubject !== null || selfKeys ? null : await quotas.reserve(ctx);
++                } catch (qerr) {
++                    await db.prepare("UPDATE runs SET status = 'failed', error_code = ?, error_detail = ?, finished_at = ? WHERE id = ?")
++                        .run(qerr.code || 'quota.refused', String(qerr.detail || qerr.message || 'quota refused').slice(0, 500), iso(clock.now()), id);
++                    await events.runChanged(await getRow(id));
++                    if (hold) await tiers.release(hold);
++                    throw qerr;
++                }
++            }
++        }
++
          // 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.
@@@ -839,7 +936,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
                  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) {
++            // BA3: a run-level answer resumes the workflow's own step (engine.execute reads it); a customer
++            // batch item's run is the ai.complete run the batcher created, finished by this shortcut.
++            if (batchAnswer && batchAnswer.ok && !batchAnswer.resume) {
                  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,
@@@ -872,6 +971,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
                  ...(credential ? { credential, role: input && typeof input.role === 'string' ? input.role : null } : {}),
                  ...(billingDelta ? { onDelta: billingDelta } : {}),
                  ...(onRoute ? { onRoute } : {}),
++                // BA3: a run-level batch answer resumes the workflow's own step.
++                ...(batchAnswer ? { batchAnswer } : {}),
                  // BA1: the class this run may wait in: the router picks the tier from it.
                  latencyClass: ctx.latencyClass || latency.DEFAULT,
                  ...(ctx.flexBudgetMs ? { flexBudgetMs: ctx.flexBudgetMs } : {}) });
@@@ -1010,6 +1111,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
          if (TERMINAL.has(row.status)) throw new AiError(409, 'run.terminal', `run is already ${row.status}`);
          const h = inflight.get(id);
          const r = await db.prepare("UPDATE runs SET status = 'cancelled', error_code = 'run.cancelled', error_detail = 'cancelled by caller', finished_at = ? WHERE id = ? AND status IN ('queued', 'running')").run(iso(clock.now()), id);
++        // BA3: a run that has not been handed to a provider yet leaves its batch queue, so the provider is
++        // never given work for a run nobody will read; a submitted one finishes and is charged as it must.
++        await db.prepare("DELETE FROM batch_queue WHERE run_id = ? AND status = 'queued'").run(id).catch(() => null);
++        deferred.delete(id);
          if (h) {
              h.controller.abort();
              const i = queue.indexOf(id);
@@@ -1148,7 +1253,15 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
          const d = deferred.get(key);
          if (d) {
              deferred.delete(key);
--            const ctx = { ...d.ctx, batchAnswer: Promise.resolve(answer), ...(costCheck ? { costCheck } : {}) };
++            let baseCtx = d.ctx;
++            if (answer && answer.reroute) {
++                // The deadline passed: the batcher moved the run to the background class before this call.
++                const row = await getRow(key);
++                baseCtx = { ...baseCtx, latencyClass: (row && row.latency_class) || 'background' };
++                // A run-level run re-routed to the background path is charged at the ordinary price, not half.
++                if (d.runLevel) { const rest = { ...baseCtx }; delete rest.discount; baseCtx = rest; }
++            }
++            const ctx = { ...baseCtx, 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 });
@@@ -1163,7 +1276,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
          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 || {};
++        // A run-level run re-routed to the background path is charged at the ordinary price, not half.
++        const data = answer && answer.reroute && plan && plan.runLevel ? { ...plan, discount: null } : (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, {});
@@@ -1192,6 +1306,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
          const attribution = parseJson(row.attribution, null);
          const onBehalfOf = parseJson(row.on_behalf_of, null);
          const principal = data.principal || {};
++        // BA3: a run-level batch stores the scalars its own completion needs (the app, person and policy
++        // flags); a customer batch item has none and is rebuilt from its principal as before.
++        const saved = data.ctx && typeof data.ctx === 'object' ? data.ctx : {};
          // 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;
@@@ -1199,6 +1316,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
              || Boolean(tokenBudget && (tokenBudget.budget_usd != null || tokenBudget.budget_tokens != null));
          const isApp = principal.kind === 'app';
          return {
++            ...saved,
              requesterType: row.requester_type,
              requesterId: row.requester_id,
              ...(principal.jti ? { runJti: principal.jti } : {}),
@@@ -1211,7 +1329,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
              personSubject,
              personPays: Boolean(opts.person_pays),
              serviceBudgeted,
--            freeUse: false,                                        // a batch run is never the anonymous Try it
++            freeUse: saved.freeUse === true,                        // a customer batch run is never the anonymous Try it
              ...(data.discount ? { discount: String(data.discount) } : {}),
              ...(data.maxCostNano != null ? { maxCostNano: String(data.maxCostNano) } : {}),
              ...(costCheck ? { costCheck } : {}),
ddiff --git a/server/workflows/engine.js b/server/workflows/engine.js
iindex ae558b2..dd0b38d 100644
---- a/server/workflows/engine.js
++++ b/server/workflows/engine.js
@@@ -34,6 +34,16 @@ const { preferenceLines } = require('../user-modules');
  const schemas = require('../schemas');
  const { PREPARE, POSTPROCESS } = require('./hooks');
  const modelCatalog = require('../providers/model-catalog');
++const { buildBody, buildCompleteBody } = require('../providers/openai');
++
++/** The last user message's text, for the queue's own record of what one item asks (never sent as-is). */
++function lastUserText(messages) {
++    for (let i = (messages || []).length - 1; i >= 0; i--) {
++        const m = messages[i];
++        if (m && m.role === 'user' && typeof m.content === 'string') return m.content.slice(0, 20000);
++    }
++    return '';
++}
  
  const ROLE_TIMEOUT = { chat: 20000, vision: 30000, director: 25000, summary: 30000, legacy: 30000 };
  const INLINE_AUDIO_MAX = 1024 * 1024;
@@@ -120,6 +130,28 @@ function normMessages(messages, user) {
      return out;
  }
  
++/**
++ * BA3: the answer of a provider's own batch API, when it is resuming the step that asked for it. The
++ * batcher says `resume: true` and which kind of step the request was; the step then shapes the stored
++ * answer exactly as it shapes a live one.
++ */
++function batchAnswerFor(ctx, stepKind) {
++    const b = ctx && ctx.batchAnswer;
++    return b && b.ok === true && b.resume === true && b.stepKind === stepKind ? b : null;
++}
++
++/** The exec shape the steps use, built from a batch answer (no provider call, the batch cost). */
++function batchExec(b, result) {
++    const usage = b.usage || {};
++    return {
++        result,
++        provider: b.provider || null, model: b.model || null,
++        usage: { input: Number(usage.input || 0), output: Number(usage.output || 0), cached: Number(usage.cached || 0) },
++        cost: Number(b.cost) || 0, serviceTier: 'batch', savedUsd: 0,
++        explain: b.explain || null, synthetic: false, fallbackUsed: false,
++    };
++}
++
  function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spawnImpl, log = console, research = null }) {
      // One loopback reader for everything that streams media (server/media-proxy.js): transcripts and media.analyze.
      let proxy = null;
@@@ -134,7 +166,13 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
          return r;
      }
  
--    async function llmStep(step, input, wf, run, ctx) {
++    /**
++     * Everything an llm step needs before its provider call: the template it pinned, the rendered system and
++     * user prompts, the route, the optional image and the request the pool would send. BA3 splits this out of
++     * llmStep so the same request can be sent to a provider's own batch API (prepareBatch) and so a batch
++     * answer can finish the step on the ordinary path (batchAnswer) without calling a provider again.
++     */
++    async function prepareLlm(step, input, wf, run, ctx) {
          // The run pinned its template version when it was created; later steps use the active one.
          const template = run.template_key === step.template && run.template_version
              ? await registry.getTemplate(step.template, run.template_version)
@@@ -174,7 +212,19 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
              timeoutMs: p.timeout_ms || route.timeout_ms || 30000,
              cacheKey: `${wf.key}:${wf.version}`,
          };
--        const exec = await pool.execute(route, step.operation || 'chat', req, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: sha256({ s: system, u: user, i: image ? image.base64.length : 0 }) });
++        return { template, prep, vars, system, user, route, jsonSchema, image, req };
++    }
++
++    async function llmStep(step, input, wf, run, ctx) {
++        const p = await prepareLlm(step, input, wf, run, ctx);
++        if (p.output !== undefined) return { output: p.output, noProvider: true, template: p.template };
++        const { template, system, user, route, jsonSchema, image, req } = p;
++        // BA3: the answer came from the provider's own batch API. The step shapes it as it shapes a live
++        // answer (JSON parse, postprocess, citations); no provider is called again.
++        const batch = batchAnswerFor(ctx, 'llm');
++        const exec = batch
++            ? batchExec(batch, { text: (batch.output && batch.output.text) || '' })
++            : await pool.execute(route, step.operation || 'chat', req, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: sha256({ s: system, u: user, i: image ? image.base64.length : 0 }) });
          // synthetic: the stub answered (no provider configured, or a test); a postprocess that checks an answer's
          // substance (a space needs a real index.html) lets a synthetic answer through in the right shape.
          const r = { text: exec.result.text || '', json: exec.result.json || null, synthetic: Boolean(exec.synthetic) };
@@@ -210,7 +260,10 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
              cacheKey: input.cache_key || null,
              prefer: ctx.prefer || null,
          };
--        const exec = await pool.execute(route, 'chat', req, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: sha256({ s: sys, m: messages }) });
++        const batch = batchAnswerFor(ctx, 'passthrough');
++        const exec = batch
++            ? batchExec(batch, { text: (batch.output && batch.output.text) || '' })
++            : await pool.execute(route, 'chat', req, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: sha256({ s: sys, m: messages }) });
          let json = exec.result.json || null;
          if (input.json && !json) json = parseJsonLoose(exec.result.text);
          return { output: { text: exec.result.text || '', json }, exec, route };
@@@ -234,6 +287,59 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
              : { policy: selection.policy, aliasPolicy: selection.policy };
      }
  
++    /**
++     * BA3: the provider request of a run that may wait for a provider's own batch. Only a workflow with a
++     * single provider step can be batched: a batch API answers one request, not a workflow's later steps.
++     * The body is built for the model the run is placed on (the batcher knows it after the money check).
++     *
++     * → { stepKind, operation, route, maxTokens, prompt, bodyFor(model) } or null.
++     */
++    async function prepareBatch(wf, run, input, ctx = {}) {
++        if (!wf || !Array.isArray(wf.steps) || wf.steps.length !== 1) return null;
++        const step = wf.steps[0];
++        if (step.kind === 'llm') {
++            // An image cannot travel in a provider batch request here: the answer would need the media file.
++            if (step.image && input && input[step.image]) return null;
++            const p = await prepareLlm(step, input, wf, run, ctx);
++            if (!p || p.output !== undefined || p.image) return null;
++            const bodyFor = (model) => buildBody({
++                model, system: p.req.system, messages: p.req.messages, image: null,
++                json: p.req.json, maxTokens: p.req.maxTokens, temperature: p.req.temperature,
++                cacheKey: null, baseUrl: '', serviceTier: null,
++            }, 'schema');
++            return { stepKind: 'llm', operation: step.operation || 'chat', route: p.route, maxTokens: p.req.maxTokens, bodyFor, prompt: lastUserText(p.req.messages) };
++        }
++        if (step.kind === 'complete') {
++            const { route } = await selectedRoute(await routeFor(step.route || 'default.chat'), input);
++            const openai = {};
++            for (const k of ['messages', 'tools', 'tool_choice', 'parallel_tool_calls', 'response_format', 'stop', 'seed', 'top_p', 'presence_penalty', 'frequency_penalty']) {
++                if (input[k] !== undefined) openai[k] = input[k];
++            }
++            if (!Array.isArray(openai.messages) || !openai.messages.length) return null;
++            const maxTokens = Math.max(1, Math.round(input.max_tokens || route.max_output_tokens || 1024));
++            const temperature = input.temperature == null ? null : input.temperature;
++            return {
++                stepKind: 'complete', operation: 'complete', route, maxTokens, prompt: lastUserText(openai.messages),
++                bodyFor: (model) => buildCompleteBody({ model, openai, maxTokens, temperature }),
++            };
++        }
++        if (step.kind === 'passthrough') {
++            if (input.image) return null;
++            const route = await routeFor(step.route || `${step.route_prefix || 'live'}.${input.role || 'legacy'}`);
++            const sys = typeof input.system === 'string' ? (input.system.trim() ? [{ text: input.system, cache: false }] : [])
++                : Array.isArray(input.system) ? input.system.filter(x => x && String(x.text || '').trim()).map(x => ({ text: String(x.text), cache: Boolean(x.cache) })) : [];
++            const messages = normMessages(input.messages, input.user);
++            const json = input.json ? { name: input.json.name || 'result', schema: input.json.schema, strict: input.json.strict !== false, description: input.json.description } : null;
++            const maxTokens = Math.max(1, Math.round(input.max_tokens || 400));
++            const temperature = input.temperature == null ? null : input.temperature;
++            return {
++                stepKind: 'passthrough', operation: 'chat', route, maxTokens, prompt: lastUserText(messages),
++                bodyFor: (model) => buildBody({ model, system: sys, messages, image: null, json, maxTokens, temperature, cacheKey: null, baseUrl: '', serviceTier: null }, 'schema'),
++            };
++        }
++        return null;
++    }
++
      /** complete: the OpenAI Chat Completions request as the caller wrote it (ai.complete), to an upstream that speaks it. */
      async function completeStep(step, input, wf, run, ctx) {
          const { route, selection } = await selectedRoute(await routeFor(step.route || 'default.chat'), input);
@@@ -249,7 +355,10 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
              prefer: ctx.prefer || null,
              ...(typeof ctx.onDelta === 'function' ? { onDelta: ctx.onDelta } : {}),
          };
--        const exec = await pool.execute(route, 'complete', req, { ...ctx, ...selectionCtx(selection), routeKey: route.key, routeVersion: route.version, promptHash: sha256({ m: openai.messages, t: openai.tools || null }) });
++        const batch = batchAnswerFor(ctx, 'complete');
++        const exec = batch
++            ? batchExec(batch, { completion: (batch.output && batch.output.completion) || null })
++            : await pool.execute(route, 'complete', req, { ...ctx, ...selectionCtx(selection), routeKey: route.key, routeVersion: route.version, promptHash: sha256({ m: openai.messages, t: openai.tools || null }) });
          const c = exec.result.completion;
          if (!c || !Array.isArray(c.choices) || !c.choices.length) throw new AiError(502, 'output.empty', 'the provider answered, but with no choices');
          return { output: { choices: c.choices, system_fingerprint: c.system_fingerprint || null, model: exec.model }, exec, route };
@@@ -396,7 +505,7 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa
          };
      }
  
--    return { execute, narrowCitations, research: () => megasource() };
++    return { execute, prepareBatch, narrowCitations, research: () => megasource() };
  }
  
  module.exports = { createEngine, narrowCitations, collectCited, collectGaps, groundingOf };
ddiff --git a/test/batch-run.test.js b/test/batch-run.test.js
nnew file mode 100644
iindex 0000000..2ff3dbd
---- /dev/null
++++ b/test/batch-run.test.js
@@@ -0,0 +1,319 @@
++'use strict';
++/**
++ * BA3: a run whose latency class is 'batch' waits for a provider's own batch API.
++ *
++ *   - the run is created at once (202, status queued, latency_class 'batch') and takes no provider call;
++ *   - the provider's batch answer finishes the same run, at the provider's batch price;
++ *   - a restart while it waits leaves it queued, and the answer still finishes it;
++ *   - ?wait= is answered at once with the queued run, never held for hours;
++ *   - past deadline_hours the item re-routes to the background class once, and the run says it was clamped;
++ *   - a route with no batch provider runs in the background class at once and says it was clamped.
++ */
++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 t = suite('batch-run');
++const MODEL = 'gpt-4o-mini';
++
++/** A stub provider that speaks the OpenAI batch API and answers an ordinary chat call too. */
++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 p = new URL(req.url, 'http://x').pathname;
++            let body = {};
++            try { body = raw ? JSON.parse(raw) : {}; } catch { body = {}; }
++            state.calls.push({ path: p, method: req.method, body });
++            const send = (status, out, type) => {
++                res.writeHead(status, { 'Content-Type': type || 'application/json' });
++                res.end(typeof out === 'string' ? out : JSON.stringify(out));
++            };
++            if (p === '/v1/files' && req.method === 'POST') {
++                const id = `file_${++state.seq}`;
++                state.files.set(id, body.file || '');
++                return send(200, { id, object: 'file', bytes: (body.file || '').length });
++            }
++            if (p === '/v1/batches' && req.method === 'POST') {
++                const id = `batch_${++state.seq}`;
++                state.batches.set(id, { input_file_id: body.input_file_id, status: 'in_progress', output_file_id: null });
++                return send(200, { id, object: 'batch', status: 'in_progress' });
++            }
++            if (p.startsWith('/v1/batches/')) {
++                const b = state.batches.get(p.split('/')[3]);
++                if (!b) return send(404, { error: { message: 'no such batch' } });
++                return send(200, { id: p.split('/')[3], object: 'batch', status: b.status, output_file_id: b.output_file_id, error_file_id: null });
++            }
++            if (p.startsWith('/v1/files/') && p.endsWith('/content')) {
++                return send(200, state.files.get(p.split('/')[3]) || '', 'application/x-ndjson');
++            }
++            if (p === '/v1/chat/completions') {
++                return send(200, { id: '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 answers every request in its file. */
++                answer(id, { text = 'the batch answer' } = {}) {
++                    const b = state.batches.get(id);
++                    const lines = (state.files.get(b.input_file_id) || '').split('\n').filter(Boolean).map((l) => JSON.parse(l));
++                    const out = lines.map((line) => JSON.stringify({
++                        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 }, 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;
++                    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, priced = true } = {}) {
++    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`, auth_mode: 'none', provider_access: 'OWN_DEPLOYMENT' }, { actor: 'test' });
++    if (priced) 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;
++}
++
++/** A small fake Billing: the hold and the settle, and what they were told. */
++async function fakeBilling() {
++    const state = { balance: 1000000000000n, 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 p = 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 (p === '/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 id = `hld_${++state.seq}`;
++                state.holds.set(body.idempotency_key, { hold_id: id, held_nano_usd: String(body.max_nano_usd) });
++                state.held += want;
++                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 (p.startsWith('/internal/credits/holds/') && p.endsWith('/settle')) {
++                state.settlements.push({ path: p, body });
++                const hold = [...state.holds.values()].find((x) => x.hold_id === decodeURIComponent(p.split('/')[4]));
++                return send(201, { charge_nano_usd: '0', released_nano_usd: String(hold ? hold.held_nano_usd : '0'), journal_id: `txn_${state.settlements.length}`, shortfall_nano_usd: '0' });
++            }
++            if (p.startsWith('/internal/credits/holds/') && p.endsWith('/release')) return send(200, { released: true });
++            if (p === '/oauth/token') return send(200, { access_token: 'test-token', token_type: 'Bearer', expires_in: 300, scope: '' });
++            if (p === '/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()); }) };
++}
++
++async function ageGroup(h) {
++    await h.db.prepare("UPDATE batch_queue SET created_at = ? WHERE status = 'queued'").run(new Date(Date.now() - 60000).toISOString());
++}
++
++const tok = () => token('all', ALL, { sub: 'svc:test' });
++const body = (extra = {}) => ({ workflow: 'ai.generate', input: { prompt: 'take your time' }, options: { latency: 'batch', cache: false }, ...extra });
++
++t.test('a batch-class run is queued, answered by the provider\'s batch, and charged the batch price', async () => {
++    const stub = await stubProvider();
++    const h = await bootWith(stub);
++    try {
++        const started = Date.now();
++        const res = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: tok(), body: body() });
++        const took = Date.now() - started;
++        assert.strictEqual(res.status, 202, `the queued run is 202 (${res.text})`);
++        assert.strictEqual(res.body.run.status, 'queued');
++        assert.strictEqual(res.body.run.latency_class, 'batch');
++        assert.strictEqual(res.body.run.latency_clamped, null);
++        assert.ok(took < 3000, `?wait= did not hold the caller (${took} ms)`);
++        // The run has no provider call and waits in the database.
++        const requests = await h.db.prepare('SELECT COUNT(*) AS n FROM requests WHERE run_id = ?').get(res.body.run.id);
++        assert.strictEqual(Number(requests.n), 0);
++        const item = await h.db.prepare('SELECT * FROM batch_queue WHERE run_id = ?').get(res.body.run.id);
++        assert.ok(item, 'the run waits in the batch queue');
++        assert.strictEqual(item.kind, 'service');
++        assert.strictEqual(JSON.parse(item.plan).runLevel, true, `the plan is a run-level plan: ${item.plan}`);
++        assert.strictEqual(JSON.parse(item.plan).stepKind, 'llm');
++        assert.strictEqual(stub.state.calls.filter((c) => c.path === '/v1/batches').length, 0);
++
++        await ageGroup(h);
++        const submitted = await h.batcher.submitOnce();
++        assert.strictEqual(submitted.sent, 1);
++        const providerBatch = stub.state.batches.keys().next().value;
++        stub.answer(providerBatch, { text: 'the batch answer' });
++        await h.batcher.pollOnce();
++
++        const run = await h.db.prepare('SELECT * FROM runs WHERE id = ?').get(res.body.run.id);
++        assert.strictEqual(run.status, 'succeeded');
++        assert.strictEqual(run.latency_class, 'batch', 'the run it was created as');
++        assert.strictEqual(run.service_tier, 'batch', 'the tier the provider really ran');
++        assert.strictEqual(JSON.parse(run.output).text, 'the batch answer');
++        // 100 input tokens at 1 USD/M and 50 output at 2 USD/M, both at the provider's half-price batch.
++        const expected = Number((100n * 500000000n + 50n * 1000000000n) / 1000000n) / 1e9;
++        assert.strictEqual(Number(run.cost_usd), expected, 'the run cost is the batch price');
++        assert.ok(Number(run.cost_usd) > 0);
++    } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('an ai.complete run is answered by the provider\'s batch with its choices intact', async () => {
++    const stub = await stubProvider();
++    const h = await bootWith(stub);
++    try {
++        const res = await request(h.base, 'POST', '/api/v1/runs?wait=5000', {
++            tok: tok(),
++            body: { workflow: 'ai.complete', input: { messages: [{ role: 'user', content: 'hello' }] }, options: { latency: 'batch', cache: false } },
++        });
++        assert.strictEqual(res.status, 202, res.text);
++        await ageGroup(h);
++        await h.batcher.submitOnce();
++        stub.answer(stub.state.batches.keys().next().value, { text: 'the complete batch answer' });
++        await h.batcher.pollOnce();
++        const run = await h.db.prepare('SELECT * FROM runs WHERE id = ?').get(res.body.run.id);
++        assert.strictEqual(run.status, 'succeeded');
++        const out = JSON.parse(run.output);
++        assert.strictEqual(out.choices[0].message.content, 'the complete batch answer');
++        assert.strictEqual(Number(run.tokens_in), 100);
++        assert.strictEqual(Number(run.tokens_out), 50);
++    } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('a restart while the run waits leaves it queued, and the batch answer still finishes it', async () => {
++    const stub = await stubProvider();
++    const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-batch-run-restart-'));
++    const h1 = await bootWith(stub, { dir });
++    let runId;
++    try {
++        const res = await request(h1.base, 'POST', '/api/v1/runs?wait=5000', { tok: tok(), body: body() });
++        assert.strictEqual(res.status, 202, res.text);
++        runId = res.body.run.id;
++        await h1.stop();
++
++        const h2 = await bootWith(stub, { dir });
++        try {
++            const left = await h2.db.prepare('SELECT status, latency_class FROM runs WHERE id = ?').get(runId);
++            assert.strictEqual(left.status, 'queued', 'the restart did not fail the waiting run');
++            assert.strictEqual(left.latency_class, 'batch');
++            await ageGroup(h2);
++            await h2.batcher.submitOnce();
++            stub.answer(stub.state.batches.keys().next().value, { text: 'the answer after the restart' });
++            await h2.batcher.pollOnce();
++            const done = await h2.db.prepare('SELECT status, output, cost_usd FROM runs WHERE id = ?').get(runId);
++            assert.strictEqual(done.status, 'succeeded');
++            assert.strictEqual(JSON.parse(done.output).text, 'the answer after the restart');
++            assert.ok(Number(done.cost_usd) > 0, 'the batch cost is on the run');
++        } finally { await h2.stop(); }
++    } finally { await stub.close(); }
++});
++
++t.test('past deadline_hours the run re-routes to the background class once, and says it was clamped', async () => {
++    const stub = await stubProvider();
++    const h = await bootWith(stub);
++    try {
++        const res = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: tok(), body: body({ options: { latency: 'batch', deadline_hours: 1, cache: false } }) });
++        assert.strictEqual(res.status, 202, res.text);
++        // The deadline passes while the item still waits for a group.
++        await h.db.prepare('UPDATE batch_queue SET deadline_at = ? WHERE run_id = ?').run(new Date(Date.now() - 1000).toISOString(), res.body.run.id);
++        await h.batcher.submitOnce();
++        // It was never handed to the provider: it re-routed instead.
++        assert.strictEqual(stub.state.calls.filter((c) => c.path === '/v1/batches').length, 0);
++        const item = await h.db.prepare('SELECT status FROM batch_queue WHERE run_id = ?').get(res.body.run.id);
++        assert.strictEqual(item.status, 'rerouted');
++        const run = await h.db.prepare('SELECT * FROM runs WHERE id = ?').get(res.body.run.id);
++        assert.strictEqual(run.status, 'succeeded', 'the background path answered it');
++        assert.strictEqual(run.latency_class, 'background');
++        assert.deepStrictEqual(JSON.parse(run.options).latency_clamped, { asked: 'batch', used: 'background' });
++        assert.strictEqual(JSON.parse(run.output).text, 'the ordinary path answered');
++        // Once: another pass does not send it or run it again.
++        await h.batcher.submitOnce();
++        await h.batcher.pollOnce();
++        assert.strictEqual(stub.state.calls.filter((c) => c.path === '/v1/batches').length, 0);
++        const after = await h.db.prepare('SELECT status FROM batch_queue WHERE run_id = ?').get(res.body.run.id);
++        assert.strictEqual(after.status, 'rerouted');
++    } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('a route with no batch provider runs in the background class at once and says it was clamped', async () => {
++    const stub = await stubProvider();
++    const h = await bootWith(stub, { priced: false });
++    try {
++        const res = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: tok(), body: body() });
++        assert.strictEqual(res.status, 201, `no provider can batch, so the run answers (${res.text})`);
++        assert.strictEqual(res.body.run.latency_class, 'background');
++        assert.deepStrictEqual(res.body.run.latency_clamped, { asked: 'batch', used: 'background' });
++        assert.strictEqual((await h.db.prepare('SELECT COUNT(*) AS n FROM batch_queue').get()).n, 0, 'nothing waits for a batch');
++        assert.strictEqual(JSON.parse(JSON.stringify(res.body.run.output)).text, 'the ordinary path answered');
++    } finally { await h.stop(); await stub.close(); }
++});
++
++t.test('the ordinary payer is charged the batch price, and the Balance is checked when the run is queued', async () => {
++    const stub = await stubProvider();
++    const billing = await fakeBilling();
++    const h = await bootWith(stub, {
++        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',
++        },
++    });
++    try {
++        const person = token('dana', ALL, { sub: 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2B3', actorType: 'user' });
++        const res = await request(h.base, 'POST', '/api/v1/runs?wait=5000', {
++            tok: person,
++            body: { workflow: 'ai.complete', input: { messages: [{ role: 'user', content: 'hello' }] }, options: { latency: 'batch', cache: false } },
++        });
++        assert.strictEqual(res.status, 202, res.text);
++        // BA2's probe checked the person's Balance when the run was queued (under its own key).
++        assert.ok([...billing.state.holds.keys()].some((k) => String(k).includes(':probe:')), 'the queued run probed the Balance');
++        await ageGroup(h);
++        await h.batcher.submitOnce();
++        stub.answer(stub.state.batches.keys().next().value, { text: 'the paid batch answer' });
++        await h.batcher.pollOnce();
++
++        const settle = billing.state.settlements[0];
++        assert.ok(settle, 'the ordinary payer was charged');
++        assert.strictEqual(settle.body.retail.nano_usd_per_m.input, '500000000', 'half the retail input price');
++        assert.strictEqual(settle.body.retail.nano_usd_per_m.output, '1000000000', 'half the retail output price');
++        const run = await h.db.prepare('SELECT status, cost_usd, service_tier FROM runs WHERE id = ?').get(res.body.run.id);
++        assert.strictEqual(run.status, 'succeeded');
++        assert.strictEqual(run.service_tier, 'batch');
++        assert.strictEqual(String(settle.body.supply_cost_nano_usd), String(Math.round(Number(run.cost_usd) * 1e9)));
++    } finally { await h.stop(); await stub.close(); await billing.close(); }
++});
++
++t.test('deadline_hours is 1 to 72', async () => {
++    const stub = await stubProvider();
++    const h = await bootWith(stub);
++    try {
++        for (const hours of [0, 73, 'soon']) {
++            const res = await request(h.base, 'POST', '/api/v1/runs', { tok: tok(), body: body({ options: { latency: 'batch', deadline_hours: hours } }) });
++            assert.strictEqual(res.status, 422, `${hours}: ${res.text}`);
++            assert.strictEqual(res.body.code, 'input.invalid');
++        }
++    } finally { await h.stop(); await stub.close(); }
++});
++
++t.run();
ddiff --git a/test/latency-classes.test.js b/test/latency-classes.test.js
iindex 67fed05..613057f 100644
---- a/test/latency-classes.test.js
++++ b/test/latency-classes.test.js
@@@ -244,11 +244,12 @@ t.test('a caller cannot make a background workflow interactive (it still runs, i
          assert.strictEqual(fast.body.run.latency_class, 'background', 'a background workflow stays in the background');
          // BA3: the run says that it was asked for more than it can give: the class asked for and the class used.
          assert.deepStrictEqual(fast.body.run.latency_clamped, { asked: 'interactive', used: 'background' }, 'the run says it was clamped');
--        // A slower class is allowed.
++        // A slower class is allowed. This route has no priced model, so no provider's batch price can be
++        // checked: the run stays on the ordinary path, in the background class, and says it was clamped.
          const slower = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: all, body: { workflow: 'live.easter_egg', input: {}, options: { latency: 'batch', cache: false } } });
--        assert.strictEqual(slower.status, 201, `a slower class is allowed (${slower.text})`);
--        assert.strictEqual(slower.body.run.latency_class, 'batch');
--        assert.strictEqual(slower.body.run.latency_clamped, null, 'a slower ask is not clamped');
++        assert.strictEqual(slower.status, 201, `a slower class is answered (${slower.text})`);
++        assert.strictEqual(slower.body.run.latency_class, 'background', 'no batch provider on the route: the background class');
++        assert.deepStrictEqual(slower.body.run.latency_clamped, { asked: 'batch', used: 'background' }, 'and the run says so');
          // The plain run is background, on the Flex tier, and says so.
          const run = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: all, body: { workflow: 'live.easter_egg', input: {}, options: { cache: false } } });
          assert.strictEqual(run.status, 201, run.text);

diff --git a/server/batch/batcher.js b/server/batch/batcher.js

index 88d796b..3f13dd9 100644

--- a/server/batch/batcher.js

+++ b/server/batch/batcher.js

@@ -84,7 +84,7 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock

// 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,

+ enabled: !(config.batches && config.batches.enabled === false),

// 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,

@@ -196,6 +196,90 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock

};

}

+ /** What a restarted service needs of the run's context: scalars only, never a function or a token. */

+ function ctxSnapshot(ctx) {

+ const out = {};

+ for (const k of ['appRun', 'projectId', 'sandbox', 'appId', 'onBehalfOf', 'personSubject', 'personPays', 'serviceBudgeted',

+ 'freeUse', 'byok', 'maxCostNano', 'communityOptIn', 'contentPolicy', 'presetPolicy', 'presetName']) {

+ if (ctx && ctx[k] !== undefined) out[k] = ctx[k];

+ }

+ return out;

+ }

+ /**

+ * BA3: where a batch-class run waits. The run's own workflow has one provider step (the engine built its

+ * request); the route's offers are tried in order, and the first provider whose own batch API is enabled

+ * and whose real batch price fits inside the half of the retail card the customer is charged takes it.

+ * A payer whose Balance cannot cover the run is refused here, before the provider is given the work.

+ *

+ * → { provider, model, body, prompt, stepKind, operation, maxTokens, holdCard, data } or null (no

+ * eligible provider: the run runs in the background class and says it was clamped).

+ */

+ async function planRun({ prepared = null, offers = [], principal = {}, ctx = {}, input = null, requestId = null, maxCostNano = null } = {}) {

+ if (!settings.enabled || !prepared || typeof prepared.bodyFor !== 'function' || !prepared.route) return null;

+ const capacity = require('./capacity'); // lazy: capacity requires this module for its cards

+ for (const offer of offers || []) {

+ if (!offer || !offer.provider || !offer.model) continue;

+ const cfg = providerConfig(offer.provider);

+ // Only a provider whose batch API is enabled and whose adapter speaks the OpenAI batch shape.

+ if (!cfg || !cfg.enabled || cfg.adapter !== 'openai-compatible') continue;

+ const provider = await registry.getProvider(offer.provider);

+ if (!provider || provider.status !== 'active') continue;

+ const cards = await cardsFor(offer.provider, offer.model);

+ if (!cards || !cards.batch) continue;

+ // The same money rule as capacity route (d): the provider's real batch price fits inside the

+ // retail card the run is charged at.

+ const charge = capacity.chargeOf(cards.row);

+ if (!charge || !capacity.within(cards.batch.nano_usd_per_m, charge)) continue;

+ let body;

+ try { body = prepared.bodyFor(offer.model); } catch { continue; }

+ if (!body || !Array.isArray(body.messages) || !body.messages.length) continue;

+ body.model = offer.model;

+ // BA2's probe: a real hold, released at once. A payer who cannot cover the run is refused now,

+ // so the network never owes a provider for an answer nobody can pay for.

+ if (credits && typeof credits.probe === 'function') {

+ const verdict = await credits.probe({

+ requestId: requestId || String(ctx.runId || ''),

+ ctx: payerCtx(principal, { discount: 'half', maxCostNano, requestId }),

+ input, card: cards.retail, provider: offer.provider, model: offer.model, access: cards.access,

+ maxTokens: prepared.maxTokens,

+ });

+ if (!verdict.ok) throw new AiError(402, String(verdict.reason || 'billing.insufficient_funds'), String(verdict.detail || 'Not enough Balance'));

+ }

+ return {

+ provider: offer.provider, model: offer.model, body, prompt: prepared.prompt || '',

+ stepKind: prepared.stepKind, operation: prepared.operation, maxTokens: prepared.maxTokens,

+ holdCard: cards.retail,

+ data: {

+ principal: planOf({ principal, input, card: cards.retail, maxTokens: prepared.maxTokens, discount: 'half', maxCostNano, requestId, servicePaid: Boolean(ctx.serviceBudgeted) }).principal,

+ input: input || null, card: cards.retail, maxTokens: Number(prepared.maxTokens) || 4096,

+ discount: 'half', maxCostNano: maxCostNano == null ? null : String(maxCostNano),

+ requestId: requestId || null, route: null, charge: null,

+ runLevel: true, stepKind: prepared.stepKind, ctx: ctxSnapshot(ctx),

+ },

+ };

+ }

+ return null;

+ }

+ /**

+ * BA3: put a run that already exists (created by runs.create) into the queue, with the deadline its

+ * caller asked for. The run itself is never created here: the API saw it at once, and the answer is

+ * written onto that same run when the provider answers.

+ */

+ async function enqueueRun({ runId, plan, deadlineHours = null } = {}) {

+ if (!runId || !plan || !plan.body) throw new AiError(500, 'batch.no_run', 'a batch-class run needs its own run');

+ const hours = Number(deadlineHours) > 0 ? Math.max(1, Math.min(72, Math.trunc(Number(deadlineHours)))) : Number(settings.deadlineHours) || 24;

+ const deadline = new Date(clock.now() + hours * 3600 * 1000).toISOString();

+ const id = `bni_${crypto.randomBytes(10).toString('hex')}`;

+ const data = { ...(plan.data || {}), deadlineHours: hours };

+ 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 (?, ?, 'service', NULL, NULL, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?)`)

+ .run(id, runId, plan.provider, plan.model, plan.prompt || '', JSON.stringify(plan.body),

+ plan.holdCard ? JSON.stringify(plan.holdCard) : null, JSON.stringify(data), deadline, iso(), iso());

+ return { id, runId };

+ }

/**

* 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.

@@ -244,8 +328,11 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock

if (running) return { skipped: 'running' };

running = true;

try {

+ // BA3: an item whose deadline passed before the provider took it is never sent: it re-routes

+ // to the ordinary (background) path now, exactly once.

+ const rerouted = await reroutePending('the deadline passed before the provider took the batch');

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();

+ FROM batch_queue WHERE status = 'queued' AND deadline_at > ?::timestamptz GROUP BY provider, model`).all(iso());

const sent = [];

for (const group of groups || []) {

const age = clock.now() - new Date(group.oldest).getTime();

@@ -253,7 +340,7 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock

const job = await submitGroup(group.provider, group.model);

if (job) sent.push(job);

}

- return { sent: sent.length, jobs: sent };

+ return { sent: sent.length, jobs: sent, ...(rerouted ? { rerouted } : {}) };

} catch (err) {

log.warn('[batch] the submitter stopped early: ' + String((err && err.message) || err));

return { error: String((err && err.code) || 'error') };

@@ -265,7 +352,7 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock

/** 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));

+ AND deadline_at > ?::timestamptz ORDER BY created_at LIMIT ?`).all(providerKey, modelKey, iso(), Number(settings.maxItemsPerGroup));

if (!items || !items.length) return null;

const adapter = await adapterFor(providerKey);

const cfg = providerConfig(providerKey);

@@ -320,8 +407,10 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock

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 <= ?")

+ // provider's answer is still there, and the run must not wait for ever. An item with no

+ // provider batch (a re-route that was interrupted) goes back to 'queued', so the deadline

+ // is looked at again instead of waiting for a job that does not exist.

+ await db.prepare("UPDATE batch_queue SET status = CASE WHEN job_id IS NULL THEN 'queued' ELSE 'submitted' END, 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' };

@@ -425,6 +514,9 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock

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,

+ // BA3: a run-level item resumes its own workflow step with this answer; a customer batch

+ // item's run is the ai.complete run the batcher created, which the shortcut finishes.

+ ...(plan.runLevel ? { resume: true, stepKind: plan.stepKind || null } : {}),

}, { 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);

@@ -440,19 +532,40 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock

/**

* 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.

+ * run answers through the pool as any other run does, and the item says it was re-routed. A run-level

+ * run (BA3) re-routes into the background class, exactly once: the item is then no longer queued, and

+ * the run itself never asks for a batch again.

+ async function rerouteItem(item, reason, from = 'submitted') {

+ const claimed = await db.prepare(`UPDATE batch_queue SET status = 'settling', updated_at = ? WHERE id = ? AND status = ? RETURNING id`)

+ .get(iso(), item.id, from).catch(() => null);

+ if (!claimed) return false;

+ const plan = item.plan ? JSON.parse(item.plan) : {};

+ if (plan.runLevel && item.run_id) {

+ const row = await db.prepare('SELECT options FROM runs WHERE id = ?').get(item.run_id).catch(() => null);

+ const opts = row && row.options ? (JSON.parse(row.options) || {}) : {};

+ opts.latency_clamped = { asked: 'batch', used: 'background' };

+ await db.prepare("UPDATE runs SET latency_class = 'background', options = ? WHERE id = ?")

+ .run(JSON.stringify(opts), item.run_id).catch(() => null);

+ }

+ await runs.completeBatch(item.run_id, { reroute: true }, { 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);

+ return true;

+ }

+ /** The items whose deadline passed before the provider ever took them. */

+ async function reroutePending(reason) {

+ const rows = await db.prepare(`SELECT * FROM batch_queue WHERE status = 'queued' AND deadline_at <= ?::timestamptz`).all(iso()).catch(() => []);

+ let n = 0;

+ for (const item of rows || []) if (await rerouteItem(item, reason, 'queued')) n++;

+ return n;

+ }

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);

- }

+ for (const item of items || []) await rerouteItem(item, reason, 'submitted');

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);

}

@@ -510,7 +623,7 @@ function createBatcher({ db, runs, registry, credits = null, config = {}, clock

/** 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 };

+ const api = { enqueue, enqueueRun, planRun, submitOnce, pollOnce, runOnce, start, stop, settings, cardsFor, scaleCard, attach };

return api;

}

diff --git a/server/batch/providers/openai-compatible.js b/server/batch/providers/openai-compatible.js

index 0ab2464..2921338 100644

--- a/server/batch/providers/openai-compatible.js

+++ b/server/batch/providers/openai-compatible.js

@@ -195,7 +195,9 @@ function createOpenAiCompatibleAdapter({ baseUrl, apiKey, fetchImpl = globalThis

out.push({

custom_id: row.custom_id,

ok: true,

- output: { text },

+ // The raw answer's choices, so a complete run (ai.complete) keeps its tool calls and finish

+ // reason too; the text is what a chat/generate step shapes.

+ output: { text, completion: { choices: body.choices || [], system_fingerprint: body.system_fingerprint || null } },

usage: {

input: Number(usage.prompt_tokens || 0),

output: Number(usage.completion_tokens || 0),

diff --git a/server/index.js b/server/index.js

index 494dd62..f9dd805 100644

--- a/server/index.js

+++ b/server/index.js

@@ -144,7 +144,10 @@ 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,

+ // BA3: the runs and the batcher need each other (a batch-class run is created first, then queued), so

+ // the runs hold this facade and the batcher fills it in below, before any request can arrive.

+ const batchRef = {};

+ const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules, credentials, preferences, tiers, projectSecrets, pool, runBudgets, credits: creditsRef, community, presets, receipts, abuse, webhooks, batch: batchRef,

// 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) => {

@@ -185,6 +188,9 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now

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 });

+ // BA3: a batch-class run asks the batcher where it may wait, and hands its own run to the queue.

+ batchRef.planRun = batcher.planRun;

+ batchRef.enqueueRun = batcher.enqueueRun;

// 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, batcher, config, clock, log });

batcher.attach(batches);

diff --git a/server/openapi.js b/server/openapi.js

index e1241d6..e13e18b 100644

--- a/server/openapi.js

+++ b/server/openapi.js

@@ -63,7 +63,7 @@ const runBody = (required = []) => ({

workflow: str({ description: 'The workflow key, e.g. ai.generate' }), version: integer({ minimum: 1 }),

input: { type: 'object' }, idempotency_key: str({ minLength: 8, maxLength: 128 }),

target: { type: 'object' }, attribution: { type: 'object' }, on_behalf_of: ref('SubjectRef'),

- options: { type: 'object' }, route: { type: 'object', description: '{ provider, model } — one offer from GET /api/v1/routes?operation=…' },

+ options: { type: 'object', description: 'cache, debug, and the latency class: latency (interactive | background | batch); a batch-class run also takes deadline_hours (1-72, the default is 24)' }, route: { type: 'object', description: '{ provider, model } — one offer from GET /api/v1/routes?operation=…' },

});

const directBody = {

@@ -708,6 +708,9 @@ function openapi() {

source_service: nullable('string'), target: nullable('object'), input: nullable('object'), output: nullable('object'),

error: nullable('object', { properties: { code: str(), detail: str() } }), synthetic: { type: 'boolean' },

provenance: { type: 'object' },

+ latency_class: { type: 'string', enum: ['interactive', 'background', 'batch'] },

+ service_tier: nullable('string', { description: 'the tier the answer really ran on: standard, flex, batch, offpeak or local' }),

+ latency_clamped: nullable('object', { description: '{ asked, used }: the class the caller asked for and the class the run really took' }),

usage: { type: 'object', properties: { tokens_in: nullable('integer'), tokens_out: nullable('integer'), cost_usd: nullable('number'), attempts: nullable('integer'), usage_sample_ids: nullable(arrayOf(str())) } },

explain: nullable('object'), citations_count: nullable('integer'), grounding: nullable('object'),

retry_of: nullable('string'), idempotency_key: nullable('string'), trace_id: nullable('string'),

diff --git a/server/providers/openai.js b/server/providers/openai.js

index 5787054..321a35d 100644

--- a/server/providers/openai.js

+++ b/server/providers/openai.js

@@ -57,6 +57,29 @@ function buildBody({ model, system, messages, image, json, maxTokens, temperatur

return body;

}

+/**

+ * The Chat Completions body of a complete operation (ai.complete), for a provider batch request too:

+ * the caller's own request with the routed model, a reasoning model's allowance, and the Flex tier.

+ */

+function buildCompleteBody({ model, openai = {}, maxTokens, temperature, serviceTier } = {}) {

+ const o = openai || {};

+ const body = { model, messages: o.messages };

+ for (const k of ['tools', 'tool_choice', 'parallel_tool_calls', 'response_format', 'stop', 'seed', 'top_p', 'presence_penalty', 'frequency_penalty']) {

+ if (o[k] !== undefined && o[k] !== null) body[k] = o[k];

+ }

+ if (isReasoningModel(model)) {

+ // max_tokens is the answer the caller wants; a reasoning model thinks first, out of the same allowance.

+ if (maxTokens) body.max_completion_tokens = Math.max(maxTokens, 256) + 1024;

+ body.reasoning_effort = 'low';

+ for (const k of ['top_p', 'presence_penalty', 'frequency_penalty']) delete body[k];

+ } else {

+ if (maxTokens) body.max_tokens = maxTokens;

+ if (temperature != null) body.temperature = temperature;

+ }

+ if (serviceTier) body.service_tier = String(serviceTier);

+ return body;

function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetch } = {}) {

const baseUrl = String(record.base_url || DEFAULT_BASE_URL).replace(/\/+$/, '');

// A local llama.cpp process serves one model, so the embedding endpoint may live on another local

@@ -146,23 +169,7 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc

async function complete(req) {

const o = req.openai || {};

- const body = { model: req.model, messages: o.messages };

- for (const k of ['tools', 'tool_choice', 'parallel_tool_calls', 'response_format', 'stop', 'seed', 'top_p', 'presence_penalty', 'frequency_penalty']) {

- if (o[k] !== undefined && o[k] !== null) body[k] = o[k];

- }

- if (isReasoningModel(req.model)) {

- // max_tokens is the answer the caller wants; a reasoning model thinks first, out of the same allowance. Without

- // room for that, and with its default (medium) effort, a long tool result could use it all up thinking and

- // answer nothing: Chat's first production inbox action (2026-10-25) ended "finish_reason: length", empty.

- if (req.maxTokens) body.max_completion_tokens = Math.max(req.maxTokens, 256) + 1024;

- body.reasoning_effort = 'low';

- for (const k of ['top_p', 'presence_penalty', 'frequency_penalty']) delete body[k];

- } else {

- if (req.maxTokens) body.max_tokens = req.maxTokens;

- if (req.temperature != null) body.temperature = req.temperature;

- }

- // The service tier (BA1): a Flex tier travels with the request and the answer says which tier it ran on.

- if (req.serviceTier) body.service_tier = String(req.serviceTier);

+ const body = buildCompleteBody({ model: req.model, openai: o, maxTokens: req.maxTokens, temperature: req.temperature, serviceTier: req.serviceTier });

if (typeof req.onDelta === 'function') return await completeStreaming(body, req);

const j = await postJson(`${baseUrl}/chat/completions`, headers(), body, { signal: req.signal, timeoutMs: req.timeoutMs, fetchImpl });

if (!Array.isArray(j.choices) || !j.choices.length) throw new ProviderError('completion has no choices', { code: 'provider.output' });

@@ -250,4 +257,4 @@ function createOpenAiProvider(record, { apiKey = '', fetchImpl = globalThis.fetc

return adapter;

}

-module.exports = { createOpenAiProvider, buildBody, DEFAULT_BASE_URL };

+module.exports = { createOpenAiProvider, buildBody, buildCompleteBody, DEFAULT_BASE_URL };

diff --git a/server/runs.js b/server/runs.js

index 3b4ae77..3e734fd 100644

--- a/server/runs.js

+++ b/server/runs.js

@@ -35,6 +35,18 @@ const TERMINAL = new Set(['succeeded', 'failed', 'cancelled', 'cached']);

const QUEUE_RETRY_AFTER_S = 5;

const newRunId = () => `run_${ids.ulid()}`;

+/**

+ * BA3: how long a batch-class run may wait for a provider's own batch API, in hours (1 to 72). The

+ * caller names it in options.deadline_hours; the default is the batcher's own 24 hours.

+ */

+function batchDeadlineHours(value) {

+ if (value == null || value === '') return 24;

+ const n = Number(value);

+ if (!Number.isFinite(n)) throw new AiError(422, 'input.invalid', 'deadline_hours must be a number of hours');

+ if (n < 1 || n > 72) throw new AiError(422, 'input.invalid', 'deadline_hours is between 1 and 72');

+ return Math.trunc(n);

function entityKey(ref) { return ref ? `${ref.service}:${ref.type}:${ref.id}` : null; }

function subjectKey(ref) { return ref ? `${ref.type}:${ref.id}` : null; }

@@ -107,7 +119,7 @@ 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, batchWaiting = 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, batch = 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 }

@@ -268,7 +280,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

const askedLatency = options.latency != null ? options.latency : null;

const resolvedLatency = latency.resolve(workflowClass, askedLatency);

if (resolvedLatency.error) throw new AiError(422, 'input.invalid', resolvedLatency.error, { latency_class: workflowClass });

- const latencyClass = resolvedLatency.class;

+ let latencyClass = resolvedLatency.class; // BA3 may move a batch-class run to the background class

// A faster class than the workflow allows was asked for: it runs in its own class, and the run's options say so.

if (resolvedLatency.clamped) options.latency_clamped = resolvedLatency.clamped;

// Community capacity is opt-in per request: a person's machine sees the prompt, so only a caller that says so

@@ -475,6 +487,51 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

}

+ // BA3: a run whose resolved class is 'batch' waits for a provider's own batch API whenever the

+ // route has one whose real batch price fits the half-price card (the same money rule as capacity

+ // route (d)). The run is created at once; the provider's answer finishes it on the ordinary path.

+ // When no provider on the route can batch, the run runs in the background class at once and says so.

+ let autoBatch = null;

+ let batchDeadline = null;

+ if (latencyClass === 'batch' && !(principal && principal.deferBatch)) {

+ batchDeadline = batchDeadlineHours(options.deadline_hours);

+ // A run paid by somebody else's key or by a free allowance is not put on the network's batch

+ // provider; a free-only app run cannot be placed on paid capacity either.

+ const appFreeOnly = Boolean(ctx.appRun && (ctx.sandbox || !(tiers && tiers.enabled)));

+ let prepared = null;

+ if (batch && typeof batch.planRun === 'function' && engine.prepareBatch

+ && credentialSubject === null && !projectProvider && !selfKeys && !freeUse && !communityOptIn && !appFreeOnly) {

+ try {

+ prepared = await engine.prepareBatch(wf, { template_key: tpl ? tpl.key : null, template_version: tpl ? tpl.version : null }, input, { ...ctx, runId: id });

+ } catch (err) {

+ log.warn(`[runs] ${id}: the batch request could not be prepared: ${(err && err.message) || err}`);

+ }

+ if (prepared && pool && typeof pool.offers === 'function') {

+ let offers = await pool.offers(prepared.route, prepared.operation, ctx).catch(() => []);

+ // A pinned run waits on the provider it pinned, never on another eligible one.

+ if (pinned) offers = offers.filter((o) => o.provider === pinned.provider && (pinned.model == null || o.model === pinned.model));

+ if (offers.length) autoBatch = await batch.planRun({ prepared, offers, principal, ctx, input, requestId: requestId || id, maxCostNano });

+ }

+ if (autoBatch) {

+ ctx.discount = 'half';

+ ctx.latencyClass = 'batch';

+ options.deadline_hours = batchDeadline;

+ options.batch = { provider: autoBatch.provider, model: autoBatch.model, deadline_hours: batchDeadline };

+ } else {

+ latencyClass = 'background';

+ ctx.latencyClass = 'background';

+ delete ctx.discount;

+ options.latency_clamped = { asked: 'batch', used: 'background' };

+ }

+ base.latency_class = latencyClass;

+ base.options = JSON.stringify({

+ ...parseJson(base.options, {}),

+ ...(options.latency_clamped ? { latency_clamped: options.latency_clamped } : {}),

+ ...(autoBatch ? { deadline_hours: batchDeadline, batch: { provider: autoBatch.provider, model: autoBatch.model } } : {}),

+ });

+ }

// Queue caps, then quota: both before any provider call (a refused run is not counted).

const caller = `${requester.type}:${requester.id}`;

// Kudos bought with volunteer service start this run earlier in the queue (never cash, never a bypass of limits).

@@ -487,8 +544,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 {

- // 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);

+ // BA2/BA3: a batch run reserves its quota when the answer is here, not while it waits.

+ reserved = credentialSubject !== null || selfKeys || (principal && principal.deferBatch) || autoBatch ? null : await quotas.reserve(ctx);

} catch (err) { if (hold) await tiers.release(hold); throw err; }

try {

@@ -511,6 +568,46 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

}

+ // BA3: the run waits for the provider's own batch API. It is created now (its class says 'batch')

+ // and takes no execution slot and no quota until the answer is here; the batcher then finishes it

+ // on the ordinary path, with the ordinary settle, receipt and events. A caller that waits (?wait=)

+ // is given the queued run at once (202 from the API), never held for hours.

+ if (autoBatch) {

+ const controller = new AbortController();

+ let resolveDone;

+ const done = new Promise((r) => { resolveDone = r; });

+ try {

+ deferred.set(id, { controller, scope, ctx, input, hold, resolveDone, runLevel: true });

+ inflight.set(id, { controller, promise: done, release: () => {}, caller, priority });

+ await batch.enqueueRun({ runId: id, plan: autoBatch, deadlineHours: batchDeadline });

+ return { run: await get(id), created: true, promise: null };

+ } catch (err) {

+ // The queue could not take it (the provider's batch table is unreachable, say): run it in

+ // the background class now rather than leave it queued for ever.

+ deferred.delete(id);

+ inflight.delete(id);

+ log.warn(`[runs] ${id}: the provider's batch queue did not take the run (${(err && err.message) || err}); it runs in the background class`);

+ latencyClass = 'background';

+ ctx.latencyClass = 'background';

+ delete ctx.discount;

+ const opts = { ...parseJson(base.options, {}), latency_clamped: { asked: 'batch', used: 'background' } };

+ delete opts.deadline_hours;

+ delete opts.batch;

+ base.latency_class = 'background';

+ base.options = JSON.stringify(opts);

+ await db.prepare("UPDATE runs SET latency_class = 'background', options = ? WHERE id = ?").run(base.options, id);

+ try {

+ reserved = credentialSubject !== null || selfKeys ? null : await quotas.reserve(ctx);

+ } catch (qerr) {

+ await db.prepare("UPDATE runs SET status = 'failed', error_code = ?, error_detail = ?, finished_at = ? WHERE id = ?")

+ .run(qerr.code || 'quota.refused', String(qerr.detail || qerr.message || 'quota refused').slice(0, 500), iso(clock.now()), id);

+ await events.runChanged(await getRow(id));

+ if (hold) await tiers.release(hold);

+ throw qerr;

+ }

// 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.

@@ -839,7 +936,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

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) {

+ // BA3: a run-level answer resumes the workflow's own step (engine.execute reads it); a customer

+ // batch item's run is the ai.complete run the batcher created, finished by this shortcut.

+ if (batchAnswer && batchAnswer.ok && !batchAnswer.resume) {

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,

@@ -872,6 +971,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

...(credential ? { credential, role: input && typeof input.role === 'string' ? input.role : null } : {}),

...(billingDelta ? { onDelta: billingDelta } : {}),

...(onRoute ? { onRoute } : {}),

+ // BA3: a run-level batch answer resumes the workflow's own step.

+ ...(batchAnswer ? { batchAnswer } : {}),

// BA1: the class this run may wait in: the router picks the tier from it.

latencyClass: ctx.latencyClass || latency.DEFAULT,

...(ctx.flexBudgetMs ? { flexBudgetMs: ctx.flexBudgetMs } : {}) });

@@ -1010,6 +1111,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

if (TERMINAL.has(row.status)) throw new AiError(409, 'run.terminal', `run is already ${row.status}`);

const h = inflight.get(id);

const r = await db.prepare("UPDATE runs SET status = 'cancelled', error_code = 'run.cancelled', error_detail = 'cancelled by caller', finished_at = ? WHERE id = ? AND status IN ('queued', 'running')").run(iso(clock.now()), id);

+ // BA3: a run that has not been handed to a provider yet leaves its batch queue, so the provider is

+ // never given work for a run nobody will read; a submitted one finishes and is charged as it must.

+ await db.prepare("DELETE FROM batch_queue WHERE run_id = ? AND status = 'queued'").run(id).catch(() => null);

+ deferred.delete(id);

if (h) {

h.controller.abort();

const i = queue.indexOf(id);

@@ -1148,7 +1253,15 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

const d = deferred.get(key);

if (d) {

deferred.delete(key);

- const ctx = { ...d.ctx, batchAnswer: Promise.resolve(answer), ...(costCheck ? { costCheck } : {}) };

+ let baseCtx = d.ctx;

+ if (answer && answer.reroute) {

+ // The deadline passed: the batcher moved the run to the background class before this call.

+ const row = await getRow(key);

+ baseCtx = { ...baseCtx, latencyClass: (row && row.latency_class) || 'background' };

+ // A run-level run re-routed to the background path is charged at the ordinary price, not half.

+ if (d.runLevel) { const rest = { ...baseCtx }; delete rest.discount; baseCtx = rest; }

+ }

+ const ctx = { ...baseCtx, 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 });

@@ -1163,7 +1276,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

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 || {};

+ // A run-level run re-routed to the background path is charged at the ordinary price, not half.

+ const data = answer && answer.reroute && plan && plan.runLevel ? { ...plan, discount: null } : (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, {});

@@ -1192,6 +1306,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

const attribution = parseJson(row.attribution, null);

const onBehalfOf = parseJson(row.on_behalf_of, null);

const principal = data.principal || {};

+ // BA3: a run-level batch stores the scalars its own completion needs (the app, person and policy

+ // flags); a customer batch item has none and is rebuilt from its principal as before.

+ const saved = data.ctx && typeof data.ctx === 'object' ? data.ctx : {};

// 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;

@@ -1199,6 +1316,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

|| Boolean(tokenBudget && (tokenBudget.budget_usd != null || tokenBudget.budget_tokens != null));

const isApp = principal.kind === 'app';

return {

+ ...saved,

requesterType: row.requester_type,

requesterId: row.requester_id,

...(principal.jti ? { runJti: principal.jti } : {}),

@@ -1211,7 +1329,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now

personSubject,

personPays: Boolean(opts.person_pays),

serviceBudgeted,

- freeUse: false, // a batch run is never the anonymous Try it

+ freeUse: saved.freeUse === true, // a customer batch run is never the anonymous Try it

...(data.discount ? { discount: String(data.discount) } : {}),

...(data.maxCostNano != null ? { maxCostNano: String(data.maxCostNano) } : {}),

...(costCheck ? { costCheck } : {}),

diff --git a/server/workflows/engine.js b/server/workflows/engine.js

index ae558b2..dd0b38d 100644

--- a/server/workflows/engine.js

+++ b/server/workflows/engine.js

@@ -34,6 +34,16 @@ const { preferenceLines } = require('../user-modules');

const schemas = require('../schemas');

const { PREPARE, POSTPROCESS } = require('./hooks');

const modelCatalog = require('../providers/model-catalog');

+const { buildBody, buildCompleteBody } = require('../providers/openai');

+/** The last user message's text, for the queue's own record of what one item asks (never sent as-is). */

+function lastUserText(messages) {

+ for (let i = (messages || []).length - 1; i >= 0; i--) {

+ const m = messages[i];

+ if (m && m.role === 'user' && typeof m.content === 'string') return m.content.slice(0, 20000);

+ }

+ return '';

const ROLE_TIMEOUT = { chat: 20000, vision: 30000, director: 25000, summary: 30000, legacy: 30000 };

const INLINE_AUDIO_MAX = 1024 * 1024;

@@ -120,6 +130,28 @@ function normMessages(messages, user) {

return out;

}

+/**

+ * BA3: the answer of a provider's own batch API, when it is resuming the step that asked for it. The

+ * batcher says `resume: true` and which kind of step the request was; the step then shapes the stored

+ * answer exactly as it shapes a live one.

+ */

+function batchAnswerFor(ctx, stepKind) {

+ const b = ctx && ctx.batchAnswer;

+ return b && b.ok === true && b.resume === true && b.stepKind === stepKind ? b : null;

+/** The exec shape the steps use, built from a batch answer (no provider call, the batch cost). */

+function batchExec(b, result) {

+ const usage = b.usage || {};

+ return {

+ result,

+ provider: b.provider || null, model: b.model || null,

+ usage: { input: Number(usage.input || 0), output: Number(usage.output || 0), cached: Number(usage.cached || 0) },

+ cost: Number(b.cost) || 0, serviceTier: 'batch', savedUsd: 0,

+ explain: b.explain || null, synthetic: false, fallbackUsed: false,

+ };

function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spawnImpl, log = console, research = null }) {

// One loopback reader for everything that streams media (server/media-proxy.js): transcripts and media.analyze.

let proxy = null;

@@ -134,7 +166,13 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa

return r;

}

- async function llmStep(step, input, wf, run, ctx) {

+ /**

+ * Everything an llm step needs before its provider call: the template it pinned, the rendered system and

+ * user prompts, the route, the optional image and the request the pool would send. BA3 splits this out of

+ * llmStep so the same request can be sent to a provider's own batch API (prepareBatch) and so a batch

+ * answer can finish the step on the ordinary path (batchAnswer) without calling a provider again.

+ */

+ async function prepareLlm(step, input, wf, run, ctx) {

// The run pinned its template version when it was created; later steps use the active one.

const template = run.template_key === step.template && run.template_version

? await registry.getTemplate(step.template, run.template_version)

@@ -174,7 +212,19 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa

timeoutMs: p.timeout_ms || route.timeout_ms || 30000,

cacheKey: `${wf.key}:${wf.version}`,

};

- const exec = await pool.execute(route, step.operation || 'chat', req, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: sha256({ s: system, u: user, i: image ? image.base64.length : 0 }) });

+ return { template, prep, vars, system, user, route, jsonSchema, image, req };

+ }

+ async function llmStep(step, input, wf, run, ctx) {

+ const p = await prepareLlm(step, input, wf, run, ctx);

+ if (p.output !== undefined) return { output: p.output, noProvider: true, template: p.template };

+ const { template, system, user, route, jsonSchema, image, req } = p;

+ // BA3: the answer came from the provider's own batch API. The step shapes it as it shapes a live

+ // answer (JSON parse, postprocess, citations); no provider is called again.

+ const batch = batchAnswerFor(ctx, 'llm');

+ const exec = batch

+ ? batchExec(batch, { text: (batch.output && batch.output.text) || '' })

+ : await pool.execute(route, step.operation || 'chat', req, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: sha256({ s: system, u: user, i: image ? image.base64.length : 0 }) });

// synthetic: the stub answered (no provider configured, or a test); a postprocess that checks an answer's

// substance (a space needs a real index.html) lets a synthetic answer through in the right shape.

const r = { text: exec.result.text || '', json: exec.result.json || null, synthetic: Boolean(exec.synthetic) };

@@ -210,7 +260,10 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa

cacheKey: input.cache_key || null,

prefer: ctx.prefer || null,

};

- const exec = await pool.execute(route, 'chat', req, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: sha256({ s: sys, m: messages }) });

+ const batch = batchAnswerFor(ctx, 'passthrough');

+ const exec = batch

+ ? batchExec(batch, { text: (batch.output && batch.output.text) || '' })

+ : await pool.execute(route, 'chat', req, { ...ctx, routeKey: route.key, routeVersion: route.version, promptHash: sha256({ s: sys, m: messages }) });

let json = exec.result.json || null;

if (input.json && !json) json = parseJsonLoose(exec.result.text);

return { output: { text: exec.result.text || '', json }, exec, route };

@@ -234,6 +287,59 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa

: { policy: selection.policy, aliasPolicy: selection.policy };

}

+ /**

+ * BA3: the provider request of a run that may wait for a provider's own batch. Only a workflow with a

+ * single provider step can be batched: a batch API answers one request, not a workflow's later steps.

+ * The body is built for the model the run is placed on (the batcher knows it after the money check).

+ *

+ * → { stepKind, operation, route, maxTokens, prompt, bodyFor(model) } or null.

+ */

+ async function prepareBatch(wf, run, input, ctx = {}) {

+ if (!wf || !Array.isArray(wf.steps) || wf.steps.length !== 1) return null;

+ const step = wf.steps[0];

+ if (step.kind === 'llm') {

+ // An image cannot travel in a provider batch request here: the answer would need the media file.

+ if (step.image && input && input[step.image]) return null;

+ const p = await prepareLlm(step, input, wf, run, ctx);

+ if (!p || p.output !== undefined || p.image) return null;

+ const bodyFor = (model) => buildBody({

+ model, system: p.req.system, messages: p.req.messages, image: null,

+ json: p.req.json, maxTokens: p.req.maxTokens, temperature: p.req.temperature,

+ cacheKey: null, baseUrl: '', serviceTier: null,

+ }, 'schema');

+ return { stepKind: 'llm', operation: step.operation || 'chat', route: p.route, maxTokens: p.req.maxTokens, bodyFor, prompt: lastUserText(p.req.messages) };

+ }

+ if (step.kind === 'complete') {

+ const { route } = await selectedRoute(await routeFor(step.route || 'default.chat'), input);

+ const openai = {};

+ for (const k of ['messages', 'tools', 'tool_choice', 'parallel_tool_calls', 'response_format', 'stop', 'seed', 'top_p', 'presence_penalty', 'frequency_penalty']) {

+ if (input[k] !== undefined) openai[k] = input[k];

+ }

+ if (!Array.isArray(openai.messages) || !openai.messages.length) return null;

+ const maxTokens = Math.max(1, Math.round(input.max_tokens || route.max_output_tokens || 1024));

+ const temperature = input.temperature == null ? null : input.temperature;

+ return {

+ stepKind: 'complete', operation: 'complete', route, maxTokens, prompt: lastUserText(openai.messages),

+ bodyFor: (model) => buildCompleteBody({ model, openai, maxTokens, temperature }),

+ };

+ }

+ if (step.kind === 'passthrough') {

+ if (input.image) return null;

+ const route = await routeFor(step.route || `${step.route_prefix || 'live'}.${input.role || 'legacy'}`);

+ const sys = typeof input.system === 'string' ? (input.system.trim() ? [{ text: input.system, cache: false }] : [])

+ : Array.isArray(input.system) ? input.system.filter(x => x && String(x.text || '').trim()).map(x => ({ text: String(x.text), cache: Boolean(x.cache) })) : [];

+ const messages = normMessages(input.messages, input.user);

+ const json = input.json ? { name: input.json.name || 'result', schema: input.json.schema, strict: input.json.strict !== false, description: input.json.description } : null;

+ const maxTokens = Math.max(1, Math.round(input.max_tokens || 400));

+ const temperature = input.temperature == null ? null : input.temperature;

+ return {

+ stepKind: 'passthrough', operation: 'chat', route, maxTokens, prompt: lastUserText(messages),

+ bodyFor: (model) => buildBody({ model, system: sys, messages, image: null, json, maxTokens, temperature, cacheKey: null, baseUrl: '', serviceTier: null }, 'schema'),

+ };

+ }

+ return null;

+ }

/** complete: the OpenAI Chat Completions request as the caller wrote it (ai.complete), to an upstream that speaks it. */

async function completeStep(step, input, wf, run, ctx) {

const { route, selection } = await selectedRoute(await routeFor(step.route || 'default.chat'), input);

@@ -249,7 +355,10 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa

prefer: ctx.prefer || null,

...(typeof ctx.onDelta === 'function' ? { onDelta: ctx.onDelta } : {}),

};

- const exec = await pool.execute(route, 'complete', req, { ...ctx, ...selectionCtx(selection), routeKey: route.key, routeVersion: route.version, promptHash: sha256({ m: openai.messages, t: openai.tools || null }) });

+ const batch = batchAnswerFor(ctx, 'complete');

+ const exec = batch

+ ? batchExec(batch, { completion: (batch.output && batch.output.completion) || null })

+ : await pool.execute(route, 'complete', req, { ...ctx, ...selectionCtx(selection), routeKey: route.key, routeVersion: route.version, promptHash: sha256({ m: openai.messages, t: openai.tools || null }) });

const c = exec.result.completion;

if (!c || !Array.isArray(c.choices) || !c.choices.length) throw new AiError(502, 'output.empty', 'the provider answered, but with no choices');

return { output: { choices: c.choices, system_fingerprint: c.system_fingerprint || null, model: exec.model }, exec, route };

@@ -396,7 +505,7 @@ function createEngine({ registry, pool, fetcher, quotas = null, config = {}, spa

};

}

- return { execute, narrowCitations, research: () => megasource() };

+ return { execute, prepareBatch, narrowCitations, research: () => megasource() };

}

module.exports = { createEngine, narrowCitations, collectCited, collectGaps, groundingOf };

diff --git a/test/batch-run.test.js b/test/batch-run.test.js

new file mode 100644

index 0000000..2ff3dbd

--- /dev/null

+++ b/test/batch-run.test.js

@@ -0,0 +1,319 @@

+'use strict';

+/**

+ * BA3: a run whose latency class is 'batch' waits for a provider's own batch API.

+ *

+ * - the run is created at once (202, status queued, latency_class 'batch') and takes no provider call;

+ * - the provider's batch answer finishes the same run, at the provider's batch price;

+ * - a restart while it waits leaves it queued, and the answer still finishes it;

+ * - ?wait= is answered at once with the queued run, never held for hours;

+ * - past deadline_hours the item re-routes to the background class once, and the run says it was clamped;

+ * - a route with no batch provider runs in the background class at once and says it was clamped.

+ */

+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 t = suite('batch-run');

+const MODEL = 'gpt-4o-mini';

+/** A stub provider that speaks the OpenAI batch API and answers an ordinary chat call too. */

+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 p = new URL(req.url, 'http://x').pathname;

+ let body = {};

+ try { body = raw ? JSON.parse(raw) : {}; } catch { body = {}; }

+ state.calls.push({ path: p, method: req.method, body });

+ const send = (status, out, type) => {

+ res.writeHead(status, { 'Content-Type': type || 'application/json' });

+ res.end(typeof out === 'string' ? out : JSON.stringify(out));

+ };

+ if (p === '/v1/files' && req.method === 'POST') {

+ const id = `file_${++state.seq}`;

+ state.files.set(id, body.file || '');

+ return send(200, { id, object: 'file', bytes: (body.file || '').length });

+ }

+ if (p === '/v1/batches' && req.method === 'POST') {

+ const id = `batch_${++state.seq}`;

+ state.batches.set(id, { input_file_id: body.input_file_id, status: 'in_progress', output_file_id: null });

+ return send(200, { id, object: 'batch', status: 'in_progress' });

+ }

+ if (p.startsWith('/v1/batches/')) {

+ const b = state.batches.get(p.split('/')[3]);

+ if (!b) return send(404, { error: { message: 'no such batch' } });

+ return send(200, { id: p.split('/')[3], object: 'batch', status: b.status, output_file_id: b.output_file_id, error_file_id: null });

+ }

+ if (p.startsWith('/v1/files/') && p.endsWith('/content')) {

+ return send(200, state.files.get(p.split('/')[3]) || '', 'application/x-ndjson');

+ }

+ if (p === '/v1/chat/completions') {

+ return send(200, { id: '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 answers every request in its file. */

+ answer(id, { text = 'the batch answer' } = {}) {

+ const b = state.batches.get(id);

+ const lines = (state.files.get(b.input_file_id) || '').split('\n').filter(Boolean).map((l) => JSON.parse(l));

+ const out = lines.map((line) => JSON.stringify({

+ 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 }, 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;

+ 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, priced = true } = {}) {

+ 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`, auth_mode: 'none', provider_access: 'OWN_DEPLOYMENT' }, { actor: 'test' });

+ if (priced) 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;

+/** A small fake Billing: the hold and the settle, and what they were told. */

+async function fakeBilling() {

+ const state = { balance: 1000000000000n, 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 p = 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 (p === '/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 id = `hld_${++state.seq}`;

+ state.holds.set(body.idempotency_key, { hold_id: id, held_nano_usd: String(body.max_nano_usd) });

+ state.held += want;

+ 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 (p.startsWith('/internal/credits/holds/') && p.endsWith('/settle')) {

+ state.settlements.push({ path: p, body });

+ const hold = [...state.holds.values()].find((x) => x.hold_id === decodeURIComponent(p.split('/')[4]));

+ return send(201, { charge_nano_usd: '0', released_nano_usd: String(hold ? hold.held_nano_usd : '0'), journal_id: `txn_${state.settlements.length}`, shortfall_nano_usd: '0' });

+ }

+ if (p.startsWith('/internal/credits/holds/') && p.endsWith('/release')) return send(200, { released: true });

+ if (p === '/oauth/token') return send(200, { access_token: 'test-token', token_type: 'Bearer', expires_in: 300, scope: '' });

+ if (p === '/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()); }) };

+async function ageGroup(h) {

+ await h.db.prepare("UPDATE batch_queue SET created_at = ? WHERE status = 'queued'").run(new Date(Date.now() - 60000).toISOString());

+const tok = () => token('all', ALL, { sub: 'svc:test' });

+const body = (extra = {}) => ({ workflow: 'ai.generate', input: { prompt: 'take your time' }, options: { latency: 'batch', cache: false }, ...extra });

+t.test('a batch-class run is queued, answered by the provider\'s batch, and charged the batch price', async () => {

+ const stub = await stubProvider();

+ const h = await bootWith(stub);

+ try {

+ const started = Date.now();

+ const res = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: tok(), body: body() });

+ const took = Date.now() - started;

+ assert.strictEqual(res.status, 202, `the queued run is 202 (${res.text})`);

+ assert.strictEqual(res.body.run.status, 'queued');

+ assert.strictEqual(res.body.run.latency_class, 'batch');

+ assert.strictEqual(res.body.run.latency_clamped, null);

+ assert.ok(took < 3000, `?wait= did not hold the caller (${took} ms)`);

+ // The run has no provider call and waits in the database.

+ const requests = await h.db.prepare('SELECT COUNT(*) AS n FROM requests WHERE run_id = ?').get(res.body.run.id);

+ assert.strictEqual(Number(requests.n), 0);

+ const item = await h.db.prepare('SELECT * FROM batch_queue WHERE run_id = ?').get(res.body.run.id);

+ assert.ok(item, 'the run waits in the batch queue');

+ assert.strictEqual(item.kind, 'service');

+ assert.strictEqual(JSON.parse(item.plan).runLevel, true, `the plan is a run-level plan: ${item.plan}`);

+ assert.strictEqual(JSON.parse(item.plan).stepKind, 'llm');

+ assert.strictEqual(stub.state.calls.filter((c) => c.path === '/v1/batches').length, 0);

+ await ageGroup(h);

+ const submitted = await h.batcher.submitOnce();

+ assert.strictEqual(submitted.sent, 1);

+ const providerBatch = stub.state.batches.keys().next().value;

+ stub.answer(providerBatch, { text: 'the batch answer' });

+ await h.batcher.pollOnce();

+ const run = await h.db.prepare('SELECT * FROM runs WHERE id = ?').get(res.body.run.id);

+ assert.strictEqual(run.status, 'succeeded');

+ assert.strictEqual(run.latency_class, 'batch', 'the run it was created as');

+ assert.strictEqual(run.service_tier, 'batch', 'the tier the provider really ran');

+ assert.strictEqual(JSON.parse(run.output).text, 'the batch answer');

+ // 100 input tokens at 1 USD/M and 50 output at 2 USD/M, both at the provider's half-price batch.

+ const expected = Number((100n * 500000000n + 50n * 1000000000n) / 1000000n) / 1e9;

+ assert.strictEqual(Number(run.cost_usd), expected, 'the run cost is the batch price');

+ assert.ok(Number(run.cost_usd) > 0);

+ } finally { await h.stop(); await stub.close(); }

+});

+t.test('an ai.complete run is answered by the provider\'s batch with its choices intact', async () => {

+ const stub = await stubProvider();

+ const h = await bootWith(stub);

+ try {

+ const res = await request(h.base, 'POST', '/api/v1/runs?wait=5000', {

+ tok: tok(),

+ body: { workflow: 'ai.complete', input: { messages: [{ role: 'user', content: 'hello' }] }, options: { latency: 'batch', cache: false } },

+ });

+ assert.strictEqual(res.status, 202, res.text);

+ await ageGroup(h);

+ await h.batcher.submitOnce();

+ stub.answer(stub.state.batches.keys().next().value, { text: 'the complete batch answer' });

+ await h.batcher.pollOnce();

+ const run = await h.db.prepare('SELECT * FROM runs WHERE id = ?').get(res.body.run.id);

+ assert.strictEqual(run.status, 'succeeded');

+ const out = JSON.parse(run.output);

+ assert.strictEqual(out.choices[0].message.content, 'the complete batch answer');

+ assert.strictEqual(Number(run.tokens_in), 100);

+ assert.strictEqual(Number(run.tokens_out), 50);

+ } finally { await h.stop(); await stub.close(); }

+});

+t.test('a restart while the run waits leaves it queued, and the batch answer still finishes it', async () => {

+ const stub = await stubProvider();

+ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-batch-run-restart-'));

+ const h1 = await bootWith(stub, { dir });

+ let runId;

+ try {

+ const res = await request(h1.base, 'POST', '/api/v1/runs?wait=5000', { tok: tok(), body: body() });

+ assert.strictEqual(res.status, 202, res.text);

+ runId = res.body.run.id;

+ await h1.stop();

+ const h2 = await bootWith(stub, { dir });

+ try {

+ const left = await h2.db.prepare('SELECT status, latency_class FROM runs WHERE id = ?').get(runId);

+ assert.strictEqual(left.status, 'queued', 'the restart did not fail the waiting run');

+ assert.strictEqual(left.latency_class, 'batch');

+ await ageGroup(h2);

+ await h2.batcher.submitOnce();

+ stub.answer(stub.state.batches.keys().next().value, { text: 'the answer after the restart' });

+ await h2.batcher.pollOnce();

+ const done = await h2.db.prepare('SELECT status, output, cost_usd FROM runs WHERE id = ?').get(runId);

+ assert.strictEqual(done.status, 'succeeded');

+ assert.strictEqual(JSON.parse(done.output).text, 'the answer after the restart');

+ assert.ok(Number(done.cost_usd) > 0, 'the batch cost is on the run');

+ } finally { await h2.stop(); }

+ } finally { await stub.close(); }

+});

+t.test('past deadline_hours the run re-routes to the background class once, and says it was clamped', async () => {

+ const stub = await stubProvider();

+ const h = await bootWith(stub);

+ try {

+ const res = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: tok(), body: body({ options: { latency: 'batch', deadline_hours: 1, cache: false } }) });

+ assert.strictEqual(res.status, 202, res.text);

+ // The deadline passes while the item still waits for a group.

+ await h.db.prepare('UPDATE batch_queue SET deadline_at = ? WHERE run_id = ?').run(new Date(Date.now() - 1000).toISOString(), res.body.run.id);

+ await h.batcher.submitOnce();

+ // It was never handed to the provider: it re-routed instead.

+ assert.strictEqual(stub.state.calls.filter((c) => c.path === '/v1/batches').length, 0);

+ const item = await h.db.prepare('SELECT status FROM batch_queue WHERE run_id = ?').get(res.body.run.id);

+ assert.strictEqual(item.status, 'rerouted');

+ const run = await h.db.prepare('SELECT * FROM runs WHERE id = ?').get(res.body.run.id);

+ assert.strictEqual(run.status, 'succeeded', 'the background path answered it');

+ assert.strictEqual(run.latency_class, 'background');

+ assert.deepStrictEqual(JSON.parse(run.options).latency_clamped, { asked: 'batch', used: 'background' });

+ assert.strictEqual(JSON.parse(run.output).text, 'the ordinary path answered');

+ // Once: another pass does not send it or run it again.

+ await h.batcher.submitOnce();

+ await h.batcher.pollOnce();

+ assert.strictEqual(stub.state.calls.filter((c) => c.path === '/v1/batches').length, 0);

+ const after = await h.db.prepare('SELECT status FROM batch_queue WHERE run_id = ?').get(res.body.run.id);

+ assert.strictEqual(after.status, 'rerouted');

+ } finally { await h.stop(); await stub.close(); }

+});

+t.test('a route with no batch provider runs in the background class at once and says it was clamped', async () => {

+ const stub = await stubProvider();

+ const h = await bootWith(stub, { priced: false });

+ try {

+ const res = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: tok(), body: body() });

+ assert.strictEqual(res.status, 201, `no provider can batch, so the run answers (${res.text})`);

+ assert.strictEqual(res.body.run.latency_class, 'background');

+ assert.deepStrictEqual(res.body.run.latency_clamped, { asked: 'batch', used: 'background' });

+ assert.strictEqual((await h.db.prepare('SELECT COUNT(*) AS n FROM batch_queue').get()).n, 0, 'nothing waits for a batch');

+ assert.strictEqual(JSON.parse(JSON.stringify(res.body.run.output)).text, 'the ordinary path answered');

+ } finally { await h.stop(); await stub.close(); }

+});

+t.test('the ordinary payer is charged the batch price, and the Balance is checked when the run is queued', async () => {

+ const stub = await stubProvider();

+ const billing = await fakeBilling();

+ const h = await bootWith(stub, {

+ 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',

+ },

+ });

+ try {

+ const person = token('dana', ALL, { sub: 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2B3', actorType: 'user' });

+ const res = await request(h.base, 'POST', '/api/v1/runs?wait=5000', {

+ tok: person,

+ body: { workflow: 'ai.complete', input: { messages: [{ role: 'user', content: 'hello' }] }, options: { latency: 'batch', cache: false } },

+ });

+ assert.strictEqual(res.status, 202, res.text);

+ // BA2's probe checked the person's Balance when the run was queued (under its own key).

+ assert.ok([...billing.state.holds.keys()].some((k) => String(k).includes(':probe:')), 'the queued run probed the Balance');

+ await ageGroup(h);

+ await h.batcher.submitOnce();

+ stub.answer(stub.state.batches.keys().next().value, { text: 'the paid batch answer' });

+ await h.batcher.pollOnce();

+ const settle = billing.state.settlements[0];

+ assert.ok(settle, 'the ordinary payer was charged');

+ assert.strictEqual(settle.body.retail.nano_usd_per_m.input, '500000000', 'half the retail input price');

+ assert.strictEqual(settle.body.retail.nano_usd_per_m.output, '1000000000', 'half the retail output price');

+ const run = await h.db.prepare('SELECT status, cost_usd, service_tier FROM runs WHERE id = ?').get(res.body.run.id);

+ assert.strictEqual(run.status, 'succeeded');

+ assert.strictEqual(run.service_tier, 'batch');

+ assert.strictEqual(String(settle.body.supply_cost_nano_usd), String(Math.round(Number(run.cost_usd) * 1e9)));

+ } finally { await h.stop(); await stub.close(); await billing.close(); }

+});

+t.test('deadline_hours is 1 to 72', async () => {

+ const stub = await stubProvider();

+ const h = await bootWith(stub);

+ try {

+ for (const hours of [0, 73, 'soon']) {

+ const res = await request(h.base, 'POST', '/api/v1/runs', { tok: tok(), body: body({ options: { latency: 'batch', deadline_hours: hours } }) });

+ assert.strictEqual(res.status, 422, `${hours}: ${res.text}`);

+ assert.strictEqual(res.body.code, 'input.invalid');

+ }

+ } finally { await h.stop(); await stub.close(); }

+});

+t.run();

diff --git a/test/latency-classes.test.js b/test/latency-classes.test.js

index 67fed05..613057f 100644

--- a/test/latency-classes.test.js

+++ b/test/latency-classes.test.js

@@ -244,11 +244,12 @@ t.test('a caller cannot make a background workflow interactive (it still runs, i

assert.strictEqual(fast.body.run.latency_class, 'background', 'a background workflow stays in the background');

// BA3: the run says that it was asked for more than it can give: the class asked for and the class used.

assert.deepStrictEqual(fast.body.run.latency_clamped, { asked: 'interactive', used: 'background' }, 'the run says it was clamped');

- // A slower class is allowed.

+ // A slower class is allowed. This route has no priced model, so no provider's batch price can be

+ // checked: the run stays on the ordinary path, in the background class, and says it was clamped.

const slower = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: all, body: { workflow: 'live.easter_egg', input: {}, options: { latency: 'batch', cache: false } } });

- assert.strictEqual(slower.status, 201, `a slower class is allowed (${slower.text})`);

- assert.strictEqual(slower.body.run.latency_class, 'batch');

- assert.strictEqual(slower.body.run.latency_clamped, null, 'a slower ask is not clamped');

+ assert.strictEqual(slower.status, 201, `a slower class is answered (${slower.text})`);

+ assert.strictEqual(slower.body.run.latency_class, 'background', 'no batch provider on the route: the background class');

+ assert.deepStrictEqual(slower.body.run.latency_clamped, { asked: 'batch', used: 'background' }, 'and the run says so');

// The plain run is background, on the Flex tier, and says so.

const run = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: all, body: { workflow: 'live.easter_egg', input: {}, options: { cache: false } } });

assert.strictEqual(run.status, 201, run.text);