openvibers/openvibe.ai Commits
BA3: the run says it was clamped, and a batch can be cancelled (#361)
Changes
nnew file mode 100644@@@ -0,0 +1,12 @@++-- phase: expand++-- A batch can be cancelled: the prompts that had not run are cancelled, and nothing is charged for them.++--++-- The check on the status of a batch and of one of its prompts learns the new value. Nothing else changes: a++-- batch that is cancelled keeps the answers it already has, and the prompts that already answered keep their++-- status and their charge.++ALTER TABLE batches DROP CONSTRAINT IF EXISTS batches_status_check;++ALTER TABLE batches ADD CONSTRAINT batches_status_check++ CHECK (status IN ('queued', 'running', 'succeeded', 'failed', 'expired', 'cancelled'));++ALTER TABLE batch_items DROP CONSTRAINT IF EXISTS batch_items_status_check;++ALTER TABLE batch_items ADD CONSTRAINT batch_items_status_check++ CHECK (status IN ('queued', 'waiting', 'running', 'succeeded', 'failed', 'expired', 'cancelled'));@@@ -5,6 +5,7 @@* POST /api/v1/batches submit a batch (ai.run.create, or an app's ai.app.run)* GET /api/v1/batches the caller's batches* GET /api/v1/batches/:id one batch and its items++ * POST /api/v1/batches/:id/cancel cancel a batch (the items that had not run)** A batch is billed at half the retail price, through Billing like every other request: the batch* table is the queue and the status, never a second ledger. The owner of the batch is the account@@@ -50,6 +51,15 @@ function batchesRouter({ batches, auth, config, clock = { now: () => Date.now()return { batch };}));++ // BA3: a batch can be cancelled. Only the prompts that never had a run are cancelled, and nothing is++ // charged for them. A prompt that already has a run (BA2 puts one on the provider's own batch API) goes on++ // and settles on its own path: it keeps its status, and the answer says so.++ r.post('/api/v1/batches/:id/cancel', create, wrap(async (req) => {++ const batch = await batches.cancel(req.params.id, req.principal);++ if (!batch) throw new AiError(404, 'batch.not_found', 'no such batch');++ return { batch };++ }));++return r;}@@@ -40,6 +40,8 @@ const { retailChargeNano } = require('../credits');const DEFAULT_MAX_ITEMS = 1000;const EXPIRED_REASON = 'No low-cost capacity came free before your deadline. You were not charged.';++const CANCELLED_REASON = 'The batch was cancelled before this prompt ran. You were not charged.';++const SENT_REASON = 'Already sent to the provider: it finishes and is charged.';/*** The grant a first-party service needs to open a batch of the network's own (BA2). It is checked here,@@@ -219,11 +221,46 @@ function createBatchService({ db, runs, registry = null, community = null, pricereturn { ...rest, item_count: count, items: items || [] };}-- /** The caller's own batches, newest first. */++ /** The caller's own batches, newest first: the same shape one batch has (the items are not in the list). */async function list(principal) {const owner = ownerOf(principal);const rows = await db.prepare('SELECT * FROM batches WHERE owner_type = ? AND owner_id = ? ORDER BY created_at DESC LIMIT 100').all(owner.type, owner.id);-- return rows || [];++ return (rows || []).map((row) => { const { items: count, ...rest } = row; return { ...rest, item_count: count, items: [] }; });++ }++++ /**++ * Cancel a batch: only the prompts that never had a run are cancelled, and nothing is charged for them.++ * A prompt that already has a run goes on and settles on its own path: it keeps its status and the answer++ * says so. The prompts that already answered keep their answers.++ * → the batch, as GET /api/v1/batches/:id shows it (or null when it is not the caller's).++ */++ async function cancel(id, principal) {++ const row = await db.prepare('SELECT * FROM batches WHERE id = ?').get(String(id)).catch(() => null);++ if (!row) return null;++ const owner = ownerOf(principal);++ if (!(row.owner_type === owner.type && String(row.owner_id) === owner.id)) return null;++ const at = iso();++ await db.tx(async () => {++ // Only the prompts that never had a run are cancelled: they are not charged for.++ await db.prepare(`UPDATE batch_items SET status = 'cancelled', reason = ?, charged_nano_usd = 0, claimed_until = NULL, claimed_by = NULL, updated_at = ?++ WHERE batch_id = ? AND run_id IS NULL AND status IN ('queued', 'waiting')`).run(CANCELLED_REASON, at, row.id);++ // A prompt that already has a run (BA2 puts one on the provider's own batch API; the ordinary++ // path keeps it) goes on and settles on its own path. It keeps its status and says so: nobody is++ // told it was not charged.++ await db.prepare(`UPDATE batch_items SET reason = ?, updated_at = ?++ WHERE batch_id = ? AND run_id IS NOT NULL AND status IN ('waiting', 'running')`).run(SENT_REASON, at, row.id);++ });++ // The batch is cancelled only when nothing is left in flight: a prompt that has a run finishes on its own.++ const left = await db.prepare(`SELECT COUNT(*) AS n FROM batch_items WHERE batch_id = ? AND run_id IS NOT NULL AND status IN ('waiting', 'running')`).get(row.id);++ if (Number(left && left.n || 0) > 0) {++ await db.prepare(`UPDATE batches SET status = CASE WHEN status IN ('queued', 'running') THEN 'running' ELSE status END,++ claimed_until = NULL, claimed_by = NULL, updated_at = ? WHERE id = ?`).run(at, row.id);++ } else {++ await db.prepare(`UPDATE batches SET status = CASE WHEN status IN ('queued', 'running') THEN 'cancelled' ELSE status END,++ claimed_until = NULL, claimed_by = NULL, finished_at = CASE WHEN status IN ('queued', 'running') THEN ? ELSE finished_at END, updated_at = ?++ WHERE id = ?`).run(at, at, row.id);++ }++ return get(id, principal);}/**@@@ -517,7 +554,7 @@ function createBatchService({ db, runs, registry = null, community = null, priceif (timer) { clearInterval(timer); timer = null; }}-- return { submit, get, list, runOnce, start, stop, settings, maxItems, EXPIRED_REASON, resolveModel, finishBatch, settleCheckFor };++ return { submit, get, list, cancel, runOnce, start, stop, settings, maxItems, EXPIRED_REASON, CANCELLED_REASON, SENT_REASON, resolveModel, finishBatch, settleCheckFor };}--module.exports = { createBatchService, DEFAULT_MAX_ITEMS, EXPIRED_REASON };++module.exports = { createBatchService, DEFAULT_MAX_ITEMS, EXPIRED_REASON, CANCELLED_REASON, SENT_REASON };@@@ -356,19 +356,20 @@ const batchItem = {type: 'object', required: ['idx', 'status'],properties: {idx: integer({ minimum: 0 }),-- status: str({ enum: ['queued', 'running', 'succeeded', 'failed'] }),++ status: str({ enum: ['queued', 'waiting', 'running', 'succeeded', 'failed', 'expired', 'cancelled'] }),model: nullable('string'),run_id: nullable('string', { description: 'The run that answered this prompt' }),answer: nullable('string'),error_code: nullable('string'),error_detail: nullable('string'),++ reason: nullable('string', { description: 'The plain note about the prompt: it expired (it was not charged), it was cancelled before it ran (it was not charged), or it was already sent to the provider (it finishes and is charged).' }),},};const batch = {type: 'object', required: ['id', 'status', 'items'],properties: {id: str(), owner_type: str({ enum: ['user', 'app'] }), owner_id: str(), project_id: nullable('string'),-- status: str({ enum: ['queued', 'running', 'succeeded', 'failed'] }),++ status: str({ enum: ['queued', 'running', 'succeeded', 'failed', 'expired', 'cancelled'] }),item_count: integer({ minimum: 1, description: 'How many prompts the batch carries' }),items: arrayOf(batchItem),discount: str({ description: 'The price the batch is charged at: half the retail rate, through Billing like every other request.' }),@@@ -396,6 +397,12 @@ route('get', '/batches/{id}', {parameters: [path('id', str())],responses: { 200: json('The batch', { type: 'object', required: ['batch'], properties: { batch: batch } }), 404: problem, ...errors([401, 403]) },});++route('post', '/batches/{id}/cancel', {++ operationId: 'cancelBatch', summary: 'Cancel a batch: the prompts that never had a run are cancelled, and nothing is charged for them', tags: ['batches'], security: SERVICE_OR_APP,++ description: 'ai.run.create (or a developer app\'s ai.app.run). Only the prompts of the batch that never had a run are cancelled, and they are not charged for. A prompt that already has a run (BA2 puts one on the provider\'s own batch API) goes on and settles on its own path: it keeps its status, the answer says it was already sent to the provider, and nobody is told it was not charged. The prompts that already answered keep their answers. The batch itself is cancelled only when nothing is left in flight. Somebody else\'s batch is 404 batch.not_found.',++ parameters: [path('id', str())],++ responses: { 200: json('The cancelled batch', { type: 'object', required: ['batch'], properties: { batch: batch } }), 404: problem, ...errors([401, 403]) },++});// ── Exchange read API (OpenVibe.Exchange) ───────────────────────────────────────────────────const exchangeOffer = {@@@ -138,6 +138,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now// BA1: how long the caller may wait, and the tier the run really ran on.latency_class: r.latency_class || latency.DEFAULT,service_tier: r.service_tier || null,++ // BA3: a caller that asked for a faster class than the workflow allows is told so (the class asked++ // for and the class the run really ran in); nothing else about the run changes.++ latency_clamped: (parseJson(r.options, {}) || {}).latency_clamped || null,usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts, usage_sample_ids: r.usage_sample_ids || (r.usage_sample_id ? [r.usage_sample_id] : null) },explain: parseJson(r.explain, null),citations_count: r.citations_count,@@@ -415,6 +418,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowinput: null, input_hash: inputHash, idempotency_key: idem, trace_id: trace, request_id: requestId, retry_of: retryOf,latency_class: latencyClass,options: JSON.stringify({ cache: options.cache !== false && credentialSubject === null && !projectProvider && !pinned && !selfKeys && !communityOptIn, debug: Boolean(options.debug), input_retained: kept.full,++ ...(options.latency_clamped ? { latency_clamped: options.latency_clamped } : {}),...(credentialSubject !== null ? { credential_subject: credentialSubject } : {}), ...(credentialProvider !== null ? { credential_provider: credentialProvider } : {}),...(selfKeys ? { self_subject: runPerson } : {}),...(runPerson ? { person_pays: personPays } : {}),nnew file mode 100644@@@ -0,0 +1,79 @@++'use strict';++/**++ * BA3: a batch can be cancelled, and the answer never tells a customer that a prompt which is already++ * with the provider was not charged.++ *++ * - a prompt that never had a run is cancelled, and it is not charged for;++ * - a prompt that already has a run (BA2 puts one on the provider's own batch API, or the ordinary path++ * keeps it) goes on and settles on its own path: it keeps its status, and the answer says so;++ * - the batch is cancelled only when nothing is left in flight;++ * - the list of batches answers like one batch: the same shape, without the items.++ */++const assert = require('assert');++const { boot, request, suite, silent, token, ALL } = require('./helpers');++++const t = suite('batch-cancel');++++const PERSON = 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2B2';++++t.test('a batch can be cancelled, and the prompts that had not run are not charged for', async () => {++ const h = await boot({ env: {} });++ try {++ const all = token('dana', ALL, { sub: PERSON, actorType: 'user' });++ const made = await request(h.base, 'POST', '/api/v1/batches', { tok: all, body: { items: ['one prompt', 'another prompt'], deadline_hours: 6 } });++ assert.strictEqual(made.status, 201, made.text);++ const id = made.body.batch.id;++ assert.strictEqual(made.body.batch.item_count, 2, 'the batch says how many prompts it carries');++ assert.strictEqual(made.body.batch.items.length, 2, 'and carries them');++++ const cancelled = await request(h.base, 'POST', `/api/v1/batches/${id}/cancel`, { tok: all });++ assert.strictEqual(cancelled.status, 200, cancelled.text);++ assert.strictEqual(cancelled.body.batch.status, 'cancelled', 'nothing was in flight: the batch is cancelled');++ for (const item of cancelled.body.batch.items) {++ assert.strictEqual(item.status, 'cancelled', 'a prompt that had not run is cancelled');++ assert.ok(item.charged_nano_usd === '0' || item.charged_nano_usd === 0 || item.charged_nano_usd === null,++ 'and nothing was charged for it');++ assert.match(String(item.reason || ''), /not charged/i, 'and the answer says it was not charged');++ }++++ // The list answers like one batch: the same shape, without the items.++ const list = await request(h.base, 'GET', '/api/v1/batches', { tok: all });++ assert.strictEqual(list.status, 200, list.text);++ const row = list.body.batches.find((b) => b.id === id);++ assert.ok(row, 'the batch is in the list');++ assert.strictEqual(row.item_count, 2, 'the list says how many prompts it carries');++ assert.deepStrictEqual(row.items, [], 'and carries no items');++++ // A batch that is not there is not found.++ const missing = await request(h.base, 'POST', '/api/v1/batches/bat_nope/cancel', { tok: all });++ assert.strictEqual(missing.status, 404, 'a batch that does not exist is 404');++ } finally { await h.stop(); }++});++++t.test('a prompt that is already with the provider is not cancelled, and nobody is told it was not charged', async () => {++ const h = await boot({ env: {} });++ try {++ const all = token('dana', ALL, { sub: PERSON, actorType: 'user' });++ const made = await request(h.base, 'POST', '/api/v1/batches', { tok: all, body: { items: ['one prompt', 'another prompt'], deadline_hours: 6 } });++ assert.strictEqual(made.status, 201, made.text);++ const id = made.body.batch.id;++ // BA2: an item whose run exists waits with that run: the run goes on and settles on the ordinary path.++ await h.testdb.db.prepare("UPDATE batch_items SET status = 'waiting', run_id = ? WHERE batch_id = ? AND idx = 0")++ .run('run_01J8Z3Q4R5S6T7V8W9X0Y1Z2B2', id);++++ const cancelled = await request(h.base, 'POST', `/api/v1/batches/${id}/cancel`, { tok: all });++ assert.strictEqual(cancelled.status, 200, cancelled.text);++ const items = cancelled.body.batch.items;++ const sent = items.find((i) => i.idx === 0);++ const other = items.find((i) => i.idx === 1);++ assert.strictEqual(sent.status, 'waiting', 'the prompt that has a run keeps its status');++ assert.strictEqual(sent.run_id, 'run_01J8Z3Q4R5S6T7V8W9X0Y1Z2B2', 'and keeps its run');++ assert.doesNotMatch(String(sent.reason || ''), /not charged/i, 'nobody is told it was not charged');++ assert.match(String(sent.reason || ''), /already sent to the provider/i, 'the answer says it was sent to the provider');++ assert.strictEqual(other.status, 'cancelled', 'the prompt that had not run is cancelled');++ assert.match(String(other.reason || ''), /not charged/i, 'and that one was not charged');++ assert.strictEqual(cancelled.body.batch.status, 'running', 'the batch is still running: a prompt is in flight');++ } finally { await h.stop(); }++});++++t.run();@@@ -242,10 +242,13 @@ t.test('a caller cannot make a background workflow interactive (it still runs, iconst fast = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: all, body: { workflow: 'live.easter_egg', input: {}, options: { latency: 'interactive', cache: false } } });assert.strictEqual(fast.status, 201, `the run happens (${fast.text})`);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.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');// 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);@@@ -254,6 +257,7 @@ t.test('a caller cannot make a background workflow interactive (it still runs, iconst read = await request(h.base, 'GET', `/api/v1/runs/${run.body.run.id}`, { tok: all });assert.strictEqual(read.body.run.service_tier, 'flex', 'the wire shape carries the tier');assert.strictEqual(read.body.run.latency_class, 'background');++ assert.strictEqual(read.body.run.latency_clamped, null, 'the wire shape carries the clamp');// The gauges say the same: the tier the run ran on, and what waiting earned.const metrics = await (await fetch(`${h.base}/metrics`)).text();assert.ok(/ai_runs_by_tier\{tier="flex",provider="shared"\} \d+\n/.test(metrics), `the tier is on the metrics:\n${metrics}`);
new file mode 100644
new file mode 100644
@@ -0,0 +1,12 @@
@@ -0,0 +1,12 @@
+-- phase: expand
+-- A batch can be cancelled: the prompts that had not run are cancelled, and nothing is charged for them.
+--
+-- The check on the status of a batch and of one of its prompts learns the new value. Nothing else changes: a
+-- batch that is cancelled keeps the answers it already has, and the prompts that already answered keep their
+-- status and their charge.
+ALTER TABLE batches DROP CONSTRAINT IF EXISTS batches_status_check;
+ALTER TABLE batches ADD CONSTRAINT batches_status_check
+ CHECK (status IN ('queued', 'running', 'succeeded', 'failed', 'expired', 'cancelled'));
+ALTER TABLE batch_items DROP CONSTRAINT IF EXISTS batch_items_status_check;
+ALTER TABLE batch_items ADD CONSTRAINT batch_items_status_check
+ CHECK (status IN ('queued', 'waiting', 'running', 'succeeded', 'failed', 'expired', 'cancelled'));
@@ -5,6 +5,7 @@
@@ -5,6 +5,7 @@
* POST /api/v1/batches submit a batch (ai.run.create, or an app's ai.app.run)
* POST /api/v1/batches submit a batch (ai.run.create, or an app's ai.app.run)
* GET /api/v1/batches the caller's batches
* GET /api/v1/batches the caller's batches
* GET /api/v1/batches/:id one batch and its items
* GET /api/v1/batches/:id one batch and its items
+ * POST /api/v1/batches/:id/cancel cancel a batch (the items that had not run)
*
*
* A batch is billed at half the retail price, through Billing like every other request: the batch
* A batch is billed at half the retail price, through Billing like every other request: the batch
* table is the queue and the status, never a second ledger. The owner of the batch is the account
* table is the queue and the status, never a second ledger. The owner of the batch is the account
@@ -50,6 +51,15 @@ function batchesRouter({ batches, auth, config, clock = { now: () => Date.now()
@@ -50,6 +51,15 @@ function batchesRouter({ batches, auth, config, clock = { now: () => Date.now()
return { batch };
return { batch };
}));
}));
+ // BA3: a batch can be cancelled. Only the prompts that never had a run are cancelled, and nothing is
+ // charged for them. A prompt that already has a run (BA2 puts one on the provider's own batch API) goes on
+ // and settles on its own path: it keeps its status, and the answer says so.
+ r.post('/api/v1/batches/:id/cancel', create, wrap(async (req) => {
+ const batch = await batches.cancel(req.params.id, req.principal);
+ if (!batch) throw new AiError(404, 'batch.not_found', 'no such batch');
+ return { batch };
+ }));
+
return r;
return r;
}
}
@@ -40,6 +40,8 @@ const { retailChargeNano } = require('../credits');
@@ -40,6 +40,8 @@ const { retailChargeNano } = require('../credits');
const DEFAULT_MAX_ITEMS = 1000;
const DEFAULT_MAX_ITEMS = 1000;
const EXPIRED_REASON = 'No low-cost capacity came free before your deadline. You were not charged.';
const EXPIRED_REASON = 'No low-cost capacity came free before your deadline. You were not charged.';
+const CANCELLED_REASON = 'The batch was cancelled before this prompt ran. You were not charged.';
+const SENT_REASON = 'Already sent to the provider: it finishes and is charged.';
/**
/**
* The grant a first-party service needs to open a batch of the network's own (BA2). It is checked here,
* The grant a first-party service needs to open a batch of the network's own (BA2). It is checked here,
@@ -219,11 +221,46 @@ function createBatchService({ db, runs, registry = null, community = null, price
@@ -219,11 +221,46 @@ function createBatchService({ db, runs, registry = null, community = null, price
return { ...rest, item_count: count, items: items || [] };
return { ...rest, item_count: count, items: items || [] };
}
}
- /** The caller's own batches, newest first. */
+ /** The caller's own batches, newest first: the same shape one batch has (the items are not in the list). */
async function list(principal) {
async function list(principal) {
const owner = ownerOf(principal);
const owner = ownerOf(principal);
const rows = await db.prepare('SELECT * FROM batches WHERE owner_type = ? AND owner_id = ? ORDER BY created_at DESC LIMIT 100').all(owner.type, owner.id);
const rows = await db.prepare('SELECT * FROM batches WHERE owner_type = ? AND owner_id = ? ORDER BY created_at DESC LIMIT 100').all(owner.type, owner.id);
- return rows || [];
+ return (rows || []).map((row) => { const { items: count, ...rest } = row; return { ...rest, item_count: count, items: [] }; });
+ }
+
+ /**
+ * Cancel a batch: only the prompts that never had a run are cancelled, and nothing is charged for them.
+ * A prompt that already has a run goes on and settles on its own path: it keeps its status and the answer
+ * says so. The prompts that already answered keep their answers.
+ * → the batch, as GET /api/v1/batches/:id shows it (or null when it is not the caller's).
+ */
+ async function cancel(id, principal) {
+ const row = await db.prepare('SELECT * FROM batches WHERE id = ?').get(String(id)).catch(() => null);
+ if (!row) return null;
+ const owner = ownerOf(principal);
+ if (!(row.owner_type === owner.type && String(row.owner_id) === owner.id)) return null;
+ const at = iso();
+ await db.tx(async () => {
+ // Only the prompts that never had a run are cancelled: they are not charged for.
+ await db.prepare(`UPDATE batch_items SET status = 'cancelled', reason = ?, charged_nano_usd = 0, claimed_until = NULL, claimed_by = NULL, updated_at = ?
+ WHERE batch_id = ? AND run_id IS NULL AND status IN ('queued', 'waiting')`).run(CANCELLED_REASON, at, row.id);
+ // A prompt that already has a run (BA2 puts one on the provider's own batch API; the ordinary
+ // path keeps it) goes on and settles on its own path. It keeps its status and says so: nobody is
+ // told it was not charged.
+ await db.prepare(`UPDATE batch_items SET reason = ?, updated_at = ?
+ WHERE batch_id = ? AND run_id IS NOT NULL AND status IN ('waiting', 'running')`).run(SENT_REASON, at, row.id);
+ });
+ // The batch is cancelled only when nothing is left in flight: a prompt that has a run finishes on its own.
+ const left = await db.prepare(`SELECT COUNT(*) AS n FROM batch_items WHERE batch_id = ? AND run_id IS NOT NULL AND status IN ('waiting', 'running')`).get(row.id);
+ if (Number(left && left.n || 0) > 0) {
+ await db.prepare(`UPDATE batches SET status = CASE WHEN status IN ('queued', 'running') THEN 'running' ELSE status END,
+ claimed_until = NULL, claimed_by = NULL, updated_at = ? WHERE id = ?`).run(at, row.id);
+ } else {
+ await db.prepare(`UPDATE batches SET status = CASE WHEN status IN ('queued', 'running') THEN 'cancelled' ELSE status END,
+ claimed_until = NULL, claimed_by = NULL, finished_at = CASE WHEN status IN ('queued', 'running') THEN ? ELSE finished_at END, updated_at = ?
+ WHERE id = ?`).run(at, at, row.id);
+ }
+ return get(id, principal);
}
}
/**
/**
@@ -517,7 +554,7 @@ function createBatchService({ db, runs, registry = null, community = null, price
@@ -517,7 +554,7 @@ function createBatchService({ db, runs, registry = null, community = null, price
if (timer) { clearInterval(timer); timer = null; }
if (timer) { clearInterval(timer); timer = null; }
}
}
- return { submit, get, list, runOnce, start, stop, settings, maxItems, EXPIRED_REASON, resolveModel, finishBatch, settleCheckFor };
+ return { submit, get, list, cancel, runOnce, start, stop, settings, maxItems, EXPIRED_REASON, CANCELLED_REASON, SENT_REASON, resolveModel, finishBatch, settleCheckFor };
}
}
-module.exports = { createBatchService, DEFAULT_MAX_ITEMS, EXPIRED_REASON };
+module.exports = { createBatchService, DEFAULT_MAX_ITEMS, EXPIRED_REASON, CANCELLED_REASON, SENT_REASON };
@@ -356,19 +356,20 @@ const batchItem = {
@@ -356,19 +356,20 @@ const batchItem = {
type: 'object', required: ['idx', 'status'],
type: 'object', required: ['idx', 'status'],
properties: {
properties: {
idx: integer({ minimum: 0 }),
idx: integer({ minimum: 0 }),
- status: str({ enum: ['queued', 'running', 'succeeded', 'failed'] }),
+ status: str({ enum: ['queued', 'waiting', 'running', 'succeeded', 'failed', 'expired', 'cancelled'] }),
model: nullable('string'),
model: nullable('string'),
run_id: nullable('string', { description: 'The run that answered this prompt' }),
run_id: nullable('string', { description: 'The run that answered this prompt' }),
answer: nullable('string'),
answer: nullable('string'),
error_code: nullable('string'),
error_code: nullable('string'),
error_detail: nullable('string'),
error_detail: nullable('string'),
+ reason: nullable('string', { description: 'The plain note about the prompt: it expired (it was not charged), it was cancelled before it ran (it was not charged), or it was already sent to the provider (it finishes and is charged).' }),
},
},
};
};
const batch = {
const batch = {
type: 'object', required: ['id', 'status', 'items'],
type: 'object', required: ['id', 'status', 'items'],
properties: {
properties: {
id: str(), owner_type: str({ enum: ['user', 'app'] }), owner_id: str(), project_id: nullable('string'),
id: str(), owner_type: str({ enum: ['user', 'app'] }), owner_id: str(), project_id: nullable('string'),
- status: str({ enum: ['queued', 'running', 'succeeded', 'failed'] }),
+ status: str({ enum: ['queued', 'running', 'succeeded', 'failed', 'expired', 'cancelled'] }),
item_count: integer({ minimum: 1, description: 'How many prompts the batch carries' }),
item_count: integer({ minimum: 1, description: 'How many prompts the batch carries' }),
items: arrayOf(batchItem),
items: arrayOf(batchItem),
discount: str({ description: 'The price the batch is charged at: half the retail rate, through Billing like every other request.' }),
discount: str({ description: 'The price the batch is charged at: half the retail rate, through Billing like every other request.' }),
@@ -396,6 +397,12 @@ route('get', '/batches/{id}', {
@@ -396,6 +397,12 @@ route('get', '/batches/{id}', {
parameters: [path('id', str())],
parameters: [path('id', str())],
responses: { 200: json('The batch', { type: 'object', required: ['batch'], properties: { batch: batch } }), 404: problem, ...errors([401, 403]) },
responses: { 200: json('The batch', { type: 'object', required: ['batch'], properties: { batch: batch } }), 404: problem, ...errors([401, 403]) },
});
});
+route('post', '/batches/{id}/cancel', {
+ operationId: 'cancelBatch', summary: 'Cancel a batch: the prompts that never had a run are cancelled, and nothing is charged for them', tags: ['batches'], security: SERVICE_OR_APP,
+ description: 'ai.run.create (or a developer app\'s ai.app.run). Only the prompts of the batch that never had a run are cancelled, and they are not charged for. A prompt that already has a run (BA2 puts one on the provider\'s own batch API) goes on and settles on its own path: it keeps its status, the answer says it was already sent to the provider, and nobody is told it was not charged. The prompts that already answered keep their answers. The batch itself is cancelled only when nothing is left in flight. Somebody else\'s batch is 404 batch.not_found.',
+ parameters: [path('id', str())],
+ responses: { 200: json('The cancelled batch', { type: 'object', required: ['batch'], properties: { batch: batch } }), 404: problem, ...errors([401, 403]) },
+});
// ── Exchange read API (OpenVibe.Exchange) ───────────────────────────────────────────────────
// ── Exchange read API (OpenVibe.Exchange) ───────────────────────────────────────────────────
const exchangeOffer = {
const exchangeOffer = {
@@ -138,6 +138,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -138,6 +138,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
// BA1: how long the caller may wait, and the tier the run really ran on.
// BA1: how long the caller may wait, and the tier the run really ran on.
latency_class: r.latency_class || latency.DEFAULT,
latency_class: r.latency_class || latency.DEFAULT,
service_tier: r.service_tier || null,
service_tier: r.service_tier || null,
+ // BA3: a caller that asked for a faster class than the workflow allows is told so (the class asked
+ // for and the class the run really ran in); nothing else about the run changes.
+ latency_clamped: (parseJson(r.options, {}) || {}).latency_clamped || null,
usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts, usage_sample_ids: r.usage_sample_ids || (r.usage_sample_id ? [r.usage_sample_id] : null) },
usage: { tokens_in: r.tokens_in, tokens_out: r.tokens_out, cost_usd: r.cost_usd, attempts: r.attempts, usage_sample_ids: r.usage_sample_ids || (r.usage_sample_id ? [r.usage_sample_id] : null) },
explain: parseJson(r.explain, null),
explain: parseJson(r.explain, null),
citations_count: r.citations_count,
citations_count: r.citations_count,
@@ -415,6 +418,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -415,6 +418,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
input: null, input_hash: inputHash, idempotency_key: idem, trace_id: trace, request_id: requestId, retry_of: retryOf,
input: null, input_hash: inputHash, idempotency_key: idem, trace_id: trace, request_id: requestId, retry_of: retryOf,
latency_class: latencyClass,
latency_class: latencyClass,
options: JSON.stringify({ cache: options.cache !== false && credentialSubject === null && !projectProvider && !pinned && !selfKeys && !communityOptIn, debug: Boolean(options.debug), input_retained: kept.full,
options: JSON.stringify({ cache: options.cache !== false && credentialSubject === null && !projectProvider && !pinned && !selfKeys && !communityOptIn, debug: Boolean(options.debug), input_retained: kept.full,
+ ...(options.latency_clamped ? { latency_clamped: options.latency_clamped } : {}),
...(credentialSubject !== null ? { credential_subject: credentialSubject } : {}), ...(credentialProvider !== null ? { credential_provider: credentialProvider } : {}),
...(credentialSubject !== null ? { credential_subject: credentialSubject } : {}), ...(credentialProvider !== null ? { credential_provider: credentialProvider } : {}),
...(selfKeys ? { self_subject: runPerson } : {}),
...(selfKeys ? { self_subject: runPerson } : {}),
...(runPerson ? { person_pays: personPays } : {}),
...(runPerson ? { person_pays: personPays } : {}),
new file mode 100644
new file mode 100644
@@ -0,0 +1,79 @@
@@ -0,0 +1,79 @@
+'use strict';
+/**
+ * BA3: a batch can be cancelled, and the answer never tells a customer that a prompt which is already
+ * with the provider was not charged.
+ *
+ * - a prompt that never had a run is cancelled, and it is not charged for;
+ * - a prompt that already has a run (BA2 puts one on the provider's own batch API, or the ordinary path
+ * keeps it) goes on and settles on its own path: it keeps its status, and the answer says so;
+ * - the batch is cancelled only when nothing is left in flight;
+ * - the list of batches answers like one batch: the same shape, without the items.
+ */
+const assert = require('assert');
+const { boot, request, suite, silent, token, ALL } = require('./helpers');
+
+const t = suite('batch-cancel');
+
+const PERSON = 'usr_01J8Z3Q4R5S6T7V8W9X0Y1Z2B2';
+
+t.test('a batch can be cancelled, and the prompts that had not run are not charged for', async () => {
+ const h = await boot({ env: {} });
+ try {
+ const all = token('dana', ALL, { sub: PERSON, actorType: 'user' });
+ const made = await request(h.base, 'POST', '/api/v1/batches', { tok: all, body: { items: ['one prompt', 'another prompt'], deadline_hours: 6 } });
+ assert.strictEqual(made.status, 201, made.text);
+ const id = made.body.batch.id;
+ assert.strictEqual(made.body.batch.item_count, 2, 'the batch says how many prompts it carries');
+ assert.strictEqual(made.body.batch.items.length, 2, 'and carries them');
+
+ const cancelled = await request(h.base, 'POST', `/api/v1/batches/${id}/cancel`, { tok: all });
+ assert.strictEqual(cancelled.status, 200, cancelled.text);
+ assert.strictEqual(cancelled.body.batch.status, 'cancelled', 'nothing was in flight: the batch is cancelled');
+ for (const item of cancelled.body.batch.items) {
+ assert.strictEqual(item.status, 'cancelled', 'a prompt that had not run is cancelled');
+ assert.ok(item.charged_nano_usd === '0' || item.charged_nano_usd === 0 || item.charged_nano_usd === null,
+ 'and nothing was charged for it');
+ assert.match(String(item.reason || ''), /not charged/i, 'and the answer says it was not charged');
+ }
+
+ // The list answers like one batch: the same shape, without the items.
+ const list = await request(h.base, 'GET', '/api/v1/batches', { tok: all });
+ assert.strictEqual(list.status, 200, list.text);
+ const row = list.body.batches.find((b) => b.id === id);
+ assert.ok(row, 'the batch is in the list');
+ assert.strictEqual(row.item_count, 2, 'the list says how many prompts it carries');
+ assert.deepStrictEqual(row.items, [], 'and carries no items');
+
+ // A batch that is not there is not found.
+ const missing = await request(h.base, 'POST', '/api/v1/batches/bat_nope/cancel', { tok: all });
+ assert.strictEqual(missing.status, 404, 'a batch that does not exist is 404');
+ } finally { await h.stop(); }
+});
+
+t.test('a prompt that is already with the provider is not cancelled, and nobody is told it was not charged', async () => {
+ const h = await boot({ env: {} });
+ try {
+ const all = token('dana', ALL, { sub: PERSON, actorType: 'user' });
+ const made = await request(h.base, 'POST', '/api/v1/batches', { tok: all, body: { items: ['one prompt', 'another prompt'], deadline_hours: 6 } });
+ assert.strictEqual(made.status, 201, made.text);
+ const id = made.body.batch.id;
+ // BA2: an item whose run exists waits with that run: the run goes on and settles on the ordinary path.
+ await h.testdb.db.prepare("UPDATE batch_items SET status = 'waiting', run_id = ? WHERE batch_id = ? AND idx = 0")
+ .run('run_01J8Z3Q4R5S6T7V8W9X0Y1Z2B2', id);
+
+ const cancelled = await request(h.base, 'POST', `/api/v1/batches/${id}/cancel`, { tok: all });
+ assert.strictEqual(cancelled.status, 200, cancelled.text);
+ const items = cancelled.body.batch.items;
+ const sent = items.find((i) => i.idx === 0);
+ const other = items.find((i) => i.idx === 1);
+ assert.strictEqual(sent.status, 'waiting', 'the prompt that has a run keeps its status');
+ assert.strictEqual(sent.run_id, 'run_01J8Z3Q4R5S6T7V8W9X0Y1Z2B2', 'and keeps its run');
+ assert.doesNotMatch(String(sent.reason || ''), /not charged/i, 'nobody is told it was not charged');
+ assert.match(String(sent.reason || ''), /already sent to the provider/i, 'the answer says it was sent to the provider');
+ assert.strictEqual(other.status, 'cancelled', 'the prompt that had not run is cancelled');
+ assert.match(String(other.reason || ''), /not charged/i, 'and that one was not charged');
+ assert.strictEqual(cancelled.body.batch.status, 'running', 'the batch is still running: a prompt is in flight');
+ } finally { await h.stop(); }
+});
+
+t.run();
@@ -242,10 +242,13 @@ t.test('a caller cannot make a background workflow interactive (it still runs, i
@@ -242,10 +242,13 @@ t.test('a caller cannot make a background workflow interactive (it still runs, i
const fast = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: all, body: { workflow: 'live.easter_egg', input: {}, options: { latency: 'interactive', cache: false } } });
const fast = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: all, body: { workflow: 'live.easter_egg', input: {}, options: { latency: 'interactive', cache: false } } });
assert.strictEqual(fast.status, 201, `the run happens (${fast.text})`);
assert.strictEqual(fast.status, 201, `the run happens (${fast.text})`);
assert.strictEqual(fast.body.run.latency_class, 'background', 'a background workflow stays in the background');
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.
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 } } });
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.status, 201, `a slower class is allowed (${slower.text})`);
assert.strictEqual(slower.body.run.latency_class, 'batch');
assert.strictEqual(slower.body.run.latency_class, 'batch');
+ assert.strictEqual(slower.body.run.latency_clamped, null, 'a slower ask is not clamped');
// The plain run is background, on the Flex tier, and 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 } } });
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);
assert.strictEqual(run.status, 201, run.text);
@@ -254,6 +257,7 @@ t.test('a caller cannot make a background workflow interactive (it still runs, i
@@ -254,6 +257,7 @@ t.test('a caller cannot make a background workflow interactive (it still runs, i
const read = await request(h.base, 'GET', `/api/v1/runs/${run.body.run.id}`, { tok: all });
const read = await request(h.base, 'GET', `/api/v1/runs/${run.body.run.id}`, { tok: all });
assert.strictEqual(read.body.run.service_tier, 'flex', 'the wire shape carries the tier');
assert.strictEqual(read.body.run.service_tier, 'flex', 'the wire shape carries the tier');
assert.strictEqual(read.body.run.latency_class, 'background');
assert.strictEqual(read.body.run.latency_class, 'background');
+ assert.strictEqual(read.body.run.latency_clamped, null, 'the wire shape carries the clamp');
// The gauges say the same: the tier the run ran on, and what waiting earned.
// The gauges say the same: the tier the run ran on, and what waiting earned.
const metrics = await (await fetch(`${h.base}/metrics`)).text();
const metrics = await (await fetch(`${h.base}/metrics`)).text();
assert.ok(/ai_runs_by_tier\{tier="flex",provider="shared"\} \d+\n/.test(metrics), `the tier is on the metrics:\n${metrics}`);
assert.ok(/ai_runs_by_tier\{tier="flex",provider="shared"\} \d+\n/.test(metrics), `the tier is on the metrics:\n${metrics}`);