openvibers/openvibe.ai Commits

Route batch work by its deadline and record why, and add Together's batch API (#492)

OP OpenVibers committed 1 days ago · 9d608673da · 6 files, +132 −19

Changes

ddiff --git a/migrations/0080_batch_route_receipt.sql b/migrations/0080_batch_route_receipt.sql
nnew file mode 100644
iindex 0000000..4f36c6b
---- /dev/null
++++ b/migrations/0080_batch_route_receipt.sql
@@@ -0,0 +1,4 @@
++-- phase: expand
++-- The batch run receipt says which route answered and why, and what an item is waiting for.
++ALTER TABLE batch_items ADD COLUMN IF NOT EXISTS route_reason text COLLATE "C";
++ALTER TABLE batch_items ADD COLUMN IF NOT EXISTS waiting_reason text COLLATE "C";
ddiff --git a/server/batch/capacity.js b/server/batch/capacity.js
iindex 6f8d6c0..72de6e7 100644
---- a/server/batch/capacity.js
++++ b/server/batch/capacity.js
@@@ -191,6 +191,7 @@ async function batchCapacity(model, at, { registry = null, community = null, pri
cost: cost.nano_usd_per_m,
batch_discount: cfg.discount,
window: cfg.window,
++ window_hours: Number(String(cfg.window).replace(/[^0-9.]/g, '')) || 24,
});
}
}
@@@ -198,4 +199,40 @@ async function batchCapacity(model, at, { registry = null, community = null, pri
return { charge, routes };
}
--module.exports = { batchCapacity, chargeOf, within, insideWindow: insideWindowLegacy, halfOf };
++/**
++ * May this route meet a deadline? A provider's own batch API promises a completion window (24h for
++ * every adapter here); a deadline sooner than that cannot use it. Every other route answers now.
++ */
++function meetsDeadline(route, deadlineAt, now = Date.now()) {
++ const hours = Number(route && route.window_hours);
++ if (!Number.isFinite(hours) || hours <= 0) return true;
++ const deadline = new Date(deadlineAt).getTime();
++ // A minute of grace: a 24-hour window against a deadline set 24 hours from the same moment does
++ // meet it; only a deadline genuinely sooner than the window is refused.
++ return Number.isFinite(deadline) && now + hours * 3600 * 1000 <= deadline + 60_000;
++}
++
++/** The total per-million cost of a route, for the cheapest-route comparison (input + output). */
++function totalCost(route) {
++ const cost = (route && route.cost) || {};
++ const add = (a, b) => BigInt(a == null ? 0 : String(a)) + BigInt(b == null ? 0 : String(b));
++ return add(cost.input, cost.output);
++}
++
++/** The plain sentence the run receipt carries: which route answered and why it was the one. */
++function routeReason(route) {
++ if (!route) return 'the ordinary route (the deadline passed and the loss budget covered it)';
++ if (route.source === 'native-batch') {
++ return `${route.provider} batch API at ${Math.round((Number(route.batch_discount) || 0.5) * 100)}% of the list price, ${route.window || '24h'} window`;
++ }
++ if (route.source === 'discount-window') {
++ const w = route.window || {};
++ return `${route.provider} off-peak price inside ${w.start_hour}:00-${w.end_hour}:00 UTC`;
++ }
++ if (route.source === 'community') return 'a community machine offer inside half the retail card';
++ if (route.source === 'own') return 'OpenVibe\'s own GPU capacity, already paid for';
++ if (route.source === 'free') return 'a free provider route';
++ return String(route.source || 'low-cost capacity');
++}
++
++module.exports = { batchCapacity, chargeOf, within, insideWindow: insideWindowLegacy, halfOf, meetsDeadline, totalCost, routeReason };
ddiff --git a/server/batch/providers/index.js b/server/batch/providers/index.js
iindex d32aa59..36d0d1f 100644
---- a/server/batch/providers/index.js
++++ b/server/batch/providers/index.js
@@@ -61,6 +61,18 @@ const PROVIDERS = Object.freeze({
limits: { maxItems: 10000, maxBytes: 100 * 1024 * 1024, oneModelPerFile: true },
source: 'https://platform.kimi.ai/docs/guide/use-batch-api.md',
},
++ // Together's Batch API: the same file-and-batch shape as OpenAI's, half the list price, answered
++ // within 24 hours (Together's own pricing page calls it an introductory 50% discount).
++ together: {
++ adapter: 'openai-compatible',
++ label: 'Together Batch API',
++ discount: 0.5,
++ window: '24h',
++ enabled: false,
++ when_configured: true,
++ limits: { maxItems: 50000, maxBytes: 200 * 1024 * 1024, oneModelPerFile: true },
++ source: 'https://docs.together.ai/docs/batch',
++ },
// DeepInfra's OpenAI-compatible Batch API: 20% cheaper than real time, answered within 24 hours.
deepinfra: {
adapter: 'openai-compatible',
ddiff --git a/server/batch/service.js b/server/batch/service.js
iindex f065a55..e217072 100644
---- a/server/batch/service.js
++++ b/server/batch/service.js
@@@ -35,7 +35,7 @@ const { AiError } = require('../util');
const { hasCap } = require('../auth');
const { retailRate } = require('../providers/model-catalog');
const { toCard } = require('./batcher');
--const { batchCapacity, chargeOf } = require('./capacity');
++const { batchCapacity, chargeOf, meetsDeadline, totalCost, routeReason } = require('./capacity');
const { retailChargeNano } = require('../credits');
const DEFAULT_MAX_ITEMS = 1000;
@@@ -282,9 +282,9 @@ function createBatchService({ db, runs, registry = null, community = null, price
}
/** Give the claim back, so another worker (or this one, later) may take the item. */
-- async function release(item, status = 'waiting') {
-- await db.prepare('UPDATE batch_items SET status = ?, claimed_until = NULL, claimed_by = NULL, updated_at = ? WHERE batch_id = ? AND idx = ?')
-- .run(status, iso(), item.batch_id, item.idx);
++ async function release(item, status = 'waiting', waitingReason = null) {
++ await db.prepare('UPDATE batch_items SET status = ?, waiting_reason = ?, claimed_until = NULL, claimed_by = NULL, updated_at = ? WHERE batch_id = ? AND idx = ?')
++ .run(status, waitingReason == null ? null : String(waitingReason).slice(0, 300), iso(), item.batch_id, item.idx);
}
/** What the network absorbed today (the loss budget is per UTC day). */
@@@ -362,7 +362,7 @@ function createBatchService({ db, runs, registry = null, community = null, price
* Put one prompt of the batch on the provider's own batch API. The run is created at once (the customer
* sees it) and waits for the answer; the batcher settles it when the provider answers.
*/
-- async function enqueueNative(batch, item, route, charge, model) {
++ async function enqueueNative(batch, item, route, charge, model, reason = null) {
try {
const principal = principalFor(batch);
const check = settleCheck({ item, route, charge, batch });
@@@ -386,9 +386,9 @@ function createBatchService({ db, runs, registry = null, community = null, price
charge,
servicePaid: principal.kind === 'service',
});
-- await db.prepare(`UPDATE batch_items SET status = 'running', run_id = ?, route_provider = ?, route_model = ?, claimed_until = NULL, claimed_by = NULL, updated_at = ?
++ await db.prepare(`UPDATE batch_items SET status = 'running', run_id = ?, route_provider = ?, route_model = ?, route_reason = ?, claimed_until = NULL, claimed_by = NULL, updated_at = ?
WHERE batch_id = ? AND idx = ?`)
-- .run(queued.runId || null, route.provider, route.model, iso(), batch.id, item.idx);
++ .run(queued.runId || null, route.provider, route.model, reason, iso(), batch.id, item.idx);
return { ok: true };
} catch (err) {
log.warn('[batch] an item could not be queued for the provider\'s batch API: ' + String((err && err.message) || err));
@@@ -404,7 +404,7 @@ function createBatchService({ db, runs, registry = null, community = null, price
}
/** Run one prompt of the batch: an ordinary run, billed with the half-price card. */
-- async function runItem(batch, item, route, charge, regular = false) {
++ async function runItem(batch, item, route, charge, regular = false, reason = null) {
const principal = principalFor(batch);
const body = { workflow: 'ai.complete', input: { messages: [{ role: 'user', content: item.prompt }] } };
if (batch.owner_type === 'user') body.on_behalf_of = { type: 'user', id: batch.owner_id };
@@@ -444,10 +444,11 @@ function createBatchService({ db, runs, registry = null, community = null, price
return { ok: false, code: 'batch.cost_exceeded' };
}
const answer = textOf(run);
-- await db.prepare(`UPDATE batch_items SET status = 'succeeded', run_id = ?, answer = ?, capacity = ?, route_provider = ?, route_model = ?,
-- claimed_until = NULL, claimed_by = NULL, updated_at = ? WHERE batch_id = ? AND idx = ?`)
++ await db.prepare(`UPDATE batch_items SET status = 'succeeded', run_id = ?, answer = ?, capacity = ?, route_provider = ?, route_model = ?, route_reason = ?,
++ waiting_reason = NULL, claimed_until = NULL, claimed_by = NULL, updated_at = ? WHERE batch_id = ? AND idx = ?`)
.run(run.id || null, answer, regular ? 'regular' : (route ? route.source : 'low-cost'),
-- route ? route.provider : null, route ? route.model : null, iso(), batch.id, item.idx);
++ route ? route.provider : null, route ? route.model : null,
++ regular ? 'the deadline passed and the owner\'s loss budget covered ordinary capacity' : reason, iso(), batch.id, item.idx);
return { ok: true, run };
}
@@@ -486,27 +487,33 @@ function createBatchService({ db, runs, registry = null, community = null, price
const capacity = model && charge
? await batchCapacity(model, clock.now(), { registry, community, priceBook })
: { charge, routes: [] };
-- const routes = (capacity.routes || []).slice().sort((a, b) => {
-- const ai = BigInt(a.cost && a.cost.input != null ? a.cost.input : '0');
-- const bi = BigInt(b.cost && b.cost.input != null ? b.cost.input : '0');
++ // A route must meet the customer's latest finish time: a provider batch that promises 24
++ // hours is not a route for a one-hour deadline. The cheapest of the rest wins.
++ const routes = (capacity.routes || []).filter((r) => meetsDeadline(r, item.deadline_at, clock.now())).sort((a, b) => {
++ const ai = totalCost(a);
++ const bi = totalCost(b);
return ai < bi ? -1 : ai > bi ? 1 : 0;
});
if (routes.length) {
const route = routes[0];
++ const reason = routeReason(route);
// (d) A provider's own batch API: the item waits for the provider's answer instead of being
// answered now. The batcher writes the answer onto the run and the run finishes as usual.
if (route.source === 'native-batch' && batcher) {
-- const res = await enqueueNative(batch, item, route, capacity.charge, model);
++ const res = await enqueueNative(batch, item, route, capacity.charge, model, reason);
return { batch: batch.id, item: item.idx, route: route.source, queued: res.ok, error: res.code };
}
-- const res = await runItem(batch, item, route, capacity.charge);
++ const res = await runItem(batch, item, route, capacity.charge, false, reason);
if (res.ok) await finishBatch(batch);
else await release(item, 'waiting');
return { batch: batch.id, item: item.idx, route: route.source, ok: res.ok };
}
if (!deadlinePassed) {
// No low-cost capacity yet: the item waits, and the customer is charged nothing.
-- await release(item, 'waiting');
++ const tooLong = (capacity.routes || []).find((r) => !meetsDeadline(r, item.deadline_at, clock.now()));
++ await release(item, 'waiting', tooLong
++ ? `${tooLong.provider} batch API promises ${tooLong.window || '24h'}, after this batch's deadline`
++ : 'waiting for a route inside the customer\'s deadline; nothing has been charged');
return { batch: batch.id, item: item.idx, waiting: true };
}
// The deadline passed with no eligible capacity: the loss budget may pay for ordinary capacity.
ddiff --git a/server/pricebook/offpeak.js b/server/pricebook/offpeak.js
iindex a8fc553..f4856ad 100644
---- a/server/pricebook/offpeak.js
++++ b/server/pricebook/offpeak.js
@@@ -52,6 +52,23 @@ function insideWindow(window, at) {
return start <= end ? (hour >= start && hour < end) : (hour >= start || hour < end);
}
++/**
++ * The next moment at or after `from` that is inside the window, or null when the window has no usable
++ * hours. The weekday blocks are honoured; a bounded search (eight days) never loops.
++ */
++function nextWindowAt(window, from = Date.now()) {
++ if (!window) return null;
++ const start = Number(window.start_hour);
++ const end = Number(window.end_hour);
++ if (!Number.isFinite(start) || !Number.isFinite(end) || start === end) return null;
++ const days = Array.isArray(window.weekdays) ? window.weekdays : [];
++ for (let step = 0; step <= 8 * 24; step += 1) {
++ const at = Number(from) + step * 60 * 60 * 1000;
++ if (insideWindow(window, at)) return new Date(at).toISOString();
++ }
++ return null;
++}
++
function createOffPeak({ db, clock = { now: () => Date.now() } }) {
const iso = () => new Date(clock.now()).toISOString();
@@@ -192,4 +209,4 @@ function createOffPeak({ db, clock = { now: () => Date.now() } }) {
return { record, clear, offPeakRows, recordBlock, clearBlocks, offPeakBlocks, windows, priceAt };
}
--module.exports = { createOffPeak, FIELDS, insideWindow, isoWeekday, parseWeekdays, formatWeekdays };
++module.exports = { nextWindowAt, createOffPeak, FIELDS, insideWindow, isoWeekday, parseWeekdays, formatWeekdays };
ddiff --git a/test/batch-route.test.js b/test/batch-route.test.js
nnew file mode 100644
iindex 0000000..0c90e35
---- /dev/null
++++ b/test/batch-route.test.js
@@@ -0,0 +1,36 @@
++'use strict';
++/** The batch lane's route receipt: deadline, cheapest route, off-peak waiting. Pure functions. */
++const assert = require('node:assert/strict');
++const { PROVIDERS } = require('../server/batch/providers');
++const { meetsDeadline, totalCost, routeReason } = require('../server/batch/capacity');
++const { nextWindowAt } = require('../server/pricebook/offpeak');
++
++const now = Date.parse('2026-11-11T20:00:00Z');
++const iso = (ms) => new Date(ms).toISOString();
++
++// Together's own batch API is a half-price, 24-hour route, off until a provider key is configured.
++assert.equal(PROVIDERS.together.discount, 0.5);
++assert.equal(PROVIDERS.together.window, '24h');
++assert.equal(PROVIDERS.together.enabled, false);
++assert.equal(PROVIDERS.together.when_configured, true);
++
++// A 24-hour provider window cannot meet a one-hour deadline; it can meet tomorrow's.
++assert.equal(meetsDeadline({ window_hours: 24 }, iso(now + 3600_000), now), false);
++assert.equal(meetsDeadline({ window_hours: 24 }, iso(now + 25 * 3600_000), now), true);
++assert.equal(meetsDeadline({ source: 'own' }, iso(now + 1000), now), true);
++
++// Cheapest is the whole route: input plus output, not input alone.
++assert.equal(totalCost({ cost: { input: '3', output: '5' } }), 8n);
++assert.ok(totalCost({ cost: { input: '1', output: '20' } }) > totalCost({ cost: { input: '5', output: '5' } }));
++assert.match(routeReason({ source: 'native-batch', provider: 'together', batch_discount: 0.5, window: '24h' }), /together.*50%.*24h/i);
++assert.match(routeReason({ source: 'discount-window', provider: 'deepseek', window: { start_hour: 16, end_hour: 0 } }), /off-peak.*16:00-0:00 UTC/);
++
++// The next off-peak moment is read from the window, never from a hardcoded DeepSeek hour.
++const at = Date.parse('2026-11-11T20:00:00Z'); // a Wednesday, 20:00 UTC
++const next = nextWindowAt({ start_hour: 16, end_hour: 0, weekdays: [1, 2, 3, 4, 5] }, at);
++assert.equal(next, '2026-11-11T20:00:00.000Z', 'inside the 16:00-00:00 weekday window now');
++const weekend = nextWindowAt({ start_hour: 16, end_hour: 0, weekdays: [1, 2, 3, 4, 5] }, Date.parse('2026-11-14T12:00:00Z'));
++assert.equal(weekend, '2026-11-16T16:00:00.000Z', 'a Saturday waits for Monday 16:00 UTC');
++assert.equal(nextWindowAt({ start_hour: 4, end_hour: 4 }), null, 'an empty window has no next moment');
++
++console.log('Batch route receipt: Together half-price, deadline windows, cheapest total, next off-peak window: passed');