openvibers/openvibe.codes Commits
Codes: the budget of a run token, and the attempt that stops at it (#181)
Changes
nnew file mode 100644@@@ -0,0 +1,15 @@++-- phase: expand++-- The budget a run token carries (Codes C3), and where the money an attempt spent went.++--++-- codes_run_tokens: the budget the minted token carries (null: no cap). OpenVibe.AI holds the counter for it, keyed by++-- the token's jti; these columns say what the token was given, so a run can be traced without asking AI.++-- codes_attempts: the budget the attempt was given, and the limit that stopped it ('usd', 'tokens', 'steps', 'minutes').++--++-- The budget is never raised here: it is what the task's limits (a benchmark's per-task limits) or the person's default++-- says, and it is written once, when the attempt's token is minted.++ALTER TABLE codes_run_tokens ADD COLUMN IF NOT EXISTS budget_usd double precision;++ALTER TABLE codes_run_tokens ADD COLUMN IF NOT EXISTS budget_tokens bigint;++++ALTER TABLE codes_attempts ADD COLUMN IF NOT EXISTS budget_usd double precision;++ALTER TABLE codes_attempts ADD COLUMN IF NOT EXISTS budget_tokens bigint;++ALTER TABLE codes_attempts ADD COLUMN IF NOT EXISTS budget_hit text COLLATE "C"; -- usd | tokens | steps | minutes@@@ -120,6 +120,8 @@ function load(env = process.env) {steps: int(env.CODES_TASK_STEPS, 0),minutes: int(env.CODES_TASK_MINUTES, 0),usd: Number(env.CODES_TASK_USD || 0) || 0,++ // The tokens one attempt may spend, when the task does not say (a run token's budget, Codes C3).++ tokens: int(env.CODES_TASK_TOKENS, 0),},// The task's timeline (stage C2): how many automatic continues one task may have before it needs a person.task: { maxContinues: Math.max(0, int(env.CODES_TASK_MAX_CONTINUES, 2)) },@@@ -86,6 +86,22 @@ const clip = (s, n) => String(s == null ? '' : s).slice(0, n);const runnerOf = (agent) => RUNNER_OF[agent] || (agent || 'auto');const wordsOf = (reason) => REASON_TEXT[reason] || REASON_TEXT.error;++/**++ * What an attempt that stopped at its budget says, in plain words: which budget it reached and what it spent.++ * The dollars and the tokens are what OpenVibe.AI counted for the run token.++ */++function budgetNote(a) {++ if (!a) return 'stopped at its budget';++ const usd = a.cost_usd == null ? null : Number(a.cost_usd);++ const tokens = a.tokens == null ? null : Number(a.tokens);++ const limitUsd = a.budget_usd == null ? null : Number(a.budget_usd);++ const limitTokens = a.budget_tokens == null ? null : Number(a.budget_tokens);++ const parts = [];++ if (limitUsd != null) parts.push(`$${usd == null ? 0 : usd.toFixed(4)} of $${limitUsd.toFixed(2)}`);++ if (limitTokens != null) parts.push(`${tokens == null ? 0 : tokens} of ${limitTokens} tokens`);++ return parts.length ? `stopped at its budget: ${parts.join(', ')} spent` : 'stopped at its budget';++}++function createAttempts({ store, config = {}, log = console, harnesses = null }) {const db = store.db;const catalog = harnesses || require('./harnesses').createHarnesses();@@@ -131,16 +147,18 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })* One attempt. The fence: follow_up_of is unique, so an ending that starts a follow-up twice answers the same* attempt the second time and starts no second job. created tells the caller which of the two it was.*/-- async function create({ task, agent = null, model = null, machine = null, session = null, why = null, kind = 'first', followUpOf = null, state = 'queued' }) {++ async function create({ task, agent = null, model = null, machine = null, session = null, why = null, kind = 'first', followUpOf = null, state = 'queued', budget = null }) {if (!task || !task.id) throw new Error('an attempt belongs to a task');for (let tries = 0; tries < 5; tries++) {const n = await db.one('SELECT COALESCE(MAX(n), 0) + 1 AS n FROM codes_attempts WHERE task_id = $1', [task.id]);const id = store.newId('att');const inserted = await db.maybe(-- `INSERT INTO codes_attempts (id, task_id, n, agent, model, machine, session, state, why, kind, follow_up_of, created_at)-- VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::timestamptz)++ `INSERT INTO codes_attempts (id, task_id, n, agent, model, machine, session, state, why, kind, follow_up_of, budget_usd, budget_tokens, created_at)++ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14::timestamptz)ON CONFLICT DO NOTHING RETURNING *`,-- [id, task.id, n.n, agent, model, machine, session, state, why, kind, followUpOf, store.iso()],++ [id, task.id, n.n, agent, model, machine, session, state, why, kind, followUpOf,++ budget && Number(budget.usd) > 0 ? Number(budget.usd) : null,++ budget && Number(budget.tokens) > 0 ? Math.round(Number(budget.tokens)) : null, store.iso()],);if (inserted) {await db.exec('UPDATE run_tasks SET attempts_count = $2 WHERE id = $1', [task.id, n.n]);@@@ -167,7 +185,7 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })}/** An attempt ended: its state, why, what it cost. */-- async function end(id, { state = 'failed', reason = null, why = null, cost_usd = null, tokens = null, steps = null, session = null, model = null, machine = null, finished_at = null } = {}) {++ async function end(id, { state = 'failed', reason = null, why = null, cost_usd = null, tokens = null, steps = null, session = null, model = null, machine = null, budget_hit = null, finished_at = null } = {}) {return await db.one(`UPDATE codes_attempts SET state = $2,reason = COALESCE($3, reason),@@@ -178,9 +196,18 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })session = COALESCE($8, session),model = COALESCE($9, model),machine = COALESCE($10, machine),-- finished_at = COALESCE($11::timestamptz, finished_at)++ budget_hit = COALESCE($11, budget_hit),++ finished_at = COALESCE($12::timestamptz, finished_at)WHERE id = $1 RETURNING *`,-- [id, state, reason, why, cost_usd == null ? null : String(cost_usd), tokens, steps, session, model, machine, finished_at || store.iso()],++ [id, state, reason, why, cost_usd == null ? null : String(cost_usd), tokens, steps, session, model, machine, budget_hit, finished_at || store.iso()],++ );++ }++++ /** What an attempt really spent, counted by OpenVibe.AI (a run refused for its budget stops mid-turn). */++ async function spend(id, { cost_usd = null, tokens = null } = {}) {++ return await db.one(++ `UPDATE codes_attempts SET cost_usd = COALESCE($2::numeric, cost_usd), tokens = COALESCE($3, tokens) WHERE id = $1 RETURNING *`,++ [id, cost_usd == null ? null : String(cost_usd), tokens],);}@@@ -267,7 +294,7 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })prev: prev ? prev.id : null,parent: a.follow_up_of || null,stopped: a.state === 'stopped',-- budget: reason === 'budget' ? { hit: 'steps' } : undefined,++ budget: reason === 'budget' ? { hit: a.budget_hit || 'steps' } : undefined,wrapUp: a.kind === 'wrap-up',hung: reason === 'hung',facts: {@@@ -338,6 +365,11 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })const words = wordsOf(reason);const continues = upto.filter((a) => a.kind === 'continue').length;++ // An attempt that ran out of money (OpenVibe.AI refused it) ends the task there: no wrap-up turn buys++ // another request the budget will not pay for, and no other agent takes over with the same cap.++ if (reason === 'budget' && ['usd', 'tokens'].includes(ending.budget_hit)) {++ return { outcome: 'unfinished', note: budgetNote(ending) };++ }if (step.action === 'done') return { outcome: 'done', note: 'the task finished' };if (step.action === 'stop') return { outcome: 'stopped', note: 'the task was stopped on purpose' };if (step.action === 'wait') return { outcome: null, note: null };@@@ -411,8 +443,8 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })return {STATES, REASONS, KINDS, REASON_TEXT, RUNNER_OF, HARNESS_OF,-- list, get, byJob, create, bind, end, decide, present, routing, nextAgent, nameOf, wordsOf,++ list, get, byJob, create, bind, end, spend, decide, present, routing, nextAgent, nameOf, wordsOf, budgetNote,};}--module.exports = { createAttempts, RUNNER_OF, HARNESS_OF, STATES, REASONS, KINDS, REASON_TEXT, ERROR_TEXT };++module.exports = { createAttempts, RUNNER_OF, HARNESS_OF, STATES, REASONS, KINDS, REASON_TEXT, ERROR_TEXT, budgetNote };@@@ -179,12 +179,12 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou* The environment the job's harness gets (Codes C3): a run token that names the person and says where to* send its requests. A failure to mint one is not fatal: the job runs as it did before, without the token.*/-- async function harnessEnv({ subject, taskId, attemptId = null, agent = null }) {++ async function harnessEnv({ subject, taskId, attemptId = null, agent = null, limits = null }) {if (!runTokens) return null;try {const provider = providerFor(agent);const own = provider ? await hasOwnKey(subject, provider) : false;-- const minted = await runTokens.mint({ subject, task: taskId, attempt: attemptId, harness: agent, provider, ownKey: own });++ const minted = await runTokens.mint({ subject, task: taskId, attempt: attemptId, harness: agent, provider, ownKey: own, budget: budgetFor(limits) });return minted.env;} catch (e) {log.warn(`[Codes] no run token for task ${taskId}: ${String(e && e.message).slice(0, 200)}`);@@@ -192,6 +192,53 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou}}++ /**++ * The budget of one attempt: the task's limits (a benchmark's per-task limits) first, then the person's default++ * in Codes' settings, and nothing at all when neither says (an uncapped run is an uncapped run).++ */++ function budgetFor(limits) {++ const pick = (from, fallback) => {++ const n = Number(from);++ if (Number.isFinite(n) && n > 0) return n;++ const f = Number(fallback);++ return Number.isFinite(f) && f > 0 ? f : null;++ };++ const l = limits && typeof limits === 'object' ? limits : {};++ return {++ usd: pick(l.usd, config.budget && config.budget.usd),++ tokens: pick(l.tokens, config.budget && config.budget.tokens),++ };++ }++++ /**++ * The attempt's run token and what it spent, per OpenVibe.AI: the budget the token carried and the counter of++ * what the run really spent (the numbers AI holds, not what the harness reported).++ */++ async function tokenSpend(attemptId) {++ if (!runTokens || !store || !store.db || !attemptId) return null;++ const row = await store.db.maybe('SELECT jti FROM codes_run_tokens WHERE attempt_id = $1 ORDER BY created_at DESC LIMIT 1', [attemptId]);++ if (!row || !row.jti) return null;++ return await runTokens.spend(row.jti);++ }++++ /** Whether the run was refused for its budget (OpenVibe.AI said 402 ai.run_budget_reached). */++ function budgetRefused(result, error) {++ return /ai\.run_budget_reached/.test(budgetText(result, error));++ }++++ /** Which budget ran out, as the refusal says it: the tokens or the dollars. */++ function budgetKind(result, error) {++ const text = budgetText(result, error);++ return /tokens? budget|tokens? of/i.test(text) ? 'tokens' : 'usd';++ }++++ /** Everything the run said: the error and the session lines, as one text. */++ function budgetText(result, error) {++ const doc = asObject(result) || {};++ const lines = Array.isArray(doc.log) ? doc.log : [];++ return [String(error || ''), ...lines.map((l) => (l && l.text ? String(l.text) : ''))].join('\n');++ }++/** The provider a harness runs on: the catalog's own word for it (anthropic, openai, deepseek, …). */function providerFor(agent) {if (!agent) return null;@@@ -218,6 +265,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ouasync function recordState(row, next) {const state = STATES.has(next && next.state) ? next.state : null;if (!state || !row || !row.id) return row;++ let budgetAttempt = null; // the attempt OpenVibe.AI refused for its budget (its spend is read after the write)const saved = await store.tx(async () => {const current = await db.maybe('SELECT * FROM run_tasks WHERE id = $1 FOR UPDATE', [row.id]);if (!current) return row;@@@ -247,13 +295,20 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ouif (attempt) {if (TERMINAL.has(state)) {const usage = usageOf(updated.result);-- const why = reasonOf(state, updated.error);-- const hit = budgetHit(attempt, usage, why);++ let why = reasonOf(state, updated.error);++ const localHit = budgetHit(attempt, usage, why);++ // OpenVibe.AI refused the run for its budget (402 ai.run_budget_reached): the attempt ends at its++ // budget, not as a crash. Which budget ran out is in the refusal itself.++ const refused = !localHit && budgetRefused(updated.result, updated.error);++ const hit = localHit || (refused ? budgetKind(updated.result, updated.error) : null);++ if (refused) why = 'budget';++ if (refused) budgetAttempt = { id: attempt.id };await attemptStore.end(attempt.id, {state: hit ? 'budget-stopped' : (ATTEMPT_STATE[state] || 'failed'),reason: hit ? 'budget' : why,cost_usd: usage.cost_usd, tokens: usage.tokens, steps: usage.steps, session: usage.session,model: usage.model, machine: machineOf(next.job),++ budget_hit: hit,finished_at: updated.finished_at || store.iso(),});} else {@@@ -264,6 +319,11 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou}return updated;});++ // What the refused run really spent: OpenVibe.AI counted it (the harness may have stopped mid-turn).++ if (budgetAttempt) {++ const spend = await tokenSpend(budgetAttempt.id);++ if (spend) await attemptStore.spend(budgetAttempt.id, { cost_usd: spend.spent_usd, tokens: spend.spent_tokens });++ }if (TERMINAL.has(saved.state) && benchAttemptEnded && saved.actor_ref) {const meta = asObject(saved.actor_ref);if (meta && meta.bench) await benchAttemptEnded(saved).catch((e) => log.warn(`[Codes] benchmark publish failed for task ${saved.id}: ${String(e && e.message).slice(0, 200)}`));@@@ -298,16 +358,18 @@ function createRunTasks({ store, run, config, log = console, sessions = null, oureturn inserted;});// The task's first attempt: every Run job is one attempt, and this one is the task as it was asked.++ // The budget of the attempt: the task's limits (a benchmark's per-task limits) or the person's default.++ const budget = budgetFor(args.limits);const first = await attemptStore.create({task: inserted, agent: args.agent || null, model: args.model || null,why: args.agent ? `the first try, with ${attemptStore.nameOf(args.agent)}` : 'the first try, with the agent that fits the task',-- kind: 'first', state: 'queued',++ kind: 'first', state: 'queued', budget,});let job;try {// The job's harness runs with a short-lived run token that names the person (Codes C3): the environment// travels with the job's input and the machine's worker hands it to the harness.-- const env = await harnessEnv({ subject, taskId: id, agent: args.agent || null });++ const env = await harnessEnv({ subject, taskId: id, attemptId: first.attempt.id, agent: args.agent || null, limits: args.limits });job = await run.submit({ subject, project, args: env ? { ...args, env } : args, key: id });} catch (e) {const why = e instanceof RunError ? `${e.code}: ${e.message}` : e.message;@@@ -399,8 +461,13 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ouconst actorRef = asObject(row.actor_ref);if (actorRef && actorRef.bench) {const done = row.state === 'succeeded';++ // A benchmark attempt that ran out of money says so, with what it spent: it is not a crash.++ const ended = (await attemptStore.list(row.id)).slice(-1)[0] || null;++ const note = ended && ended.reason === 'budget' && ['usd', 'tokens'].includes(ended.budget_hit)++ ? attemptStore.budgetNote(ended)++ : (done ? 'benchmark attempt finished' : (row.error || 'benchmark attempt failed'));return await db.one('UPDATE run_tasks SET outcome = $2, outcome_note = $3, updated_at = $4 WHERE id = $1 RETURNING *',-- [row.id, done ? 'done' : 'unfinished', done ? 'benchmark attempt finished' : (row.error || 'benchmark attempt failed'), store.now()]);++ [row.id, done ? 'done' : 'unfinished', note, store.now()]);}const attempts = await attemptStore.list(row.id);const ending = attempts[attempts.length - 1];@@@ -16,10 +16,17 @@* The credential header (x-openvibe-credential) names the provider whose key the run uses, when the person has one.** createRunTokens({ config, store, fetchImpl, log })-- * mint({ subject, task, attempt, harness, provider, ownKey }) → { token, expiresAt, jti, env }-- * record({ subject, task, attempt, provider, jti, expiresAt }) → the row (no token in it)++ * mint({ subject, task, attempt, harness, provider, ownKey, budget }) → { token, expiresAt, jti, env }++ * record({ subject, task, attempt, provider, jti, expiresAt, budget }) → the row (no token in it)++ * spend(jti) → what the run spent, per OpenVibe.AI (null when it cannot be read)++ *++ * The budget of a run token (the hard spend cap): the token's jti is registered with OpenVibe.AI, which counts++ * every request of that token and refuses the next one when the budget is spent. The budget is a row in AI, keyed by++ * the jti, not a claim in the token: the holder of a token can neither raise the budget nor name another run's.*/++const { createServiceTokenClient } = require('openvibe-sdk/auth');++const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;/** The harnesses that speak Anthropic's Messages API (the rest speak OpenAI's Chat Completions). */@@@ -39,6 +46,11 @@ function aiBase(config) {return String((config.ai && config.ai.url) || 'https://ai.openvibe.services').replace(/\/+$/, '');}++/** The address Codes itself reaches OpenVibe.AI at (the vault is host-local: the internal address). */++function aiInternal(config) {++ return String((config.ai && (config.ai.internalUrl || config.ai.url)) || 'https://ai.openvibe.services').replace(/\/+$/, '');++}++/*** The environment a harness of this kind gets: where to send its requests, the token to send, and (when the person* has a key for that provider) the header that says which key.@@@ -68,12 +80,18 @@ function envFor({ config, harness, token, provider, ownKey }) {function createRunTokens({ config, store = null, fetchImpl = globalThis.fetch, log = console } = {}) {const ttl = Math.min(7200, Math.max(60, Number((config.runTokens && config.runTokens.ttlSeconds) || 7200)));++ // Codes' own token for OpenVibe.AI (ai.run.create): the budget of a run token is registered there, under the++ // token's jti, and the counter of what the run spent is read back from it.++ const scope = (config.ai && config.ai.scope) || 'ai.credential.manage ai.run.create';++ const tokens = config.oauth && config.oauth.clientSecret++ ? createServiceTokenClient({ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.ai', scope, fetch: fetchImpl })++ : null;/*** Mint a run token for one attempt of one task of one person.* → { token, expiresAt, jti, env }; the token is in the answer and nowhere else.*/-- async function mint({ subject, task = null, attempt = null, harness = null, provider = null, ownKey = false } = {}) {++ async function mint({ subject, task = null, attempt = null, harness = null, provider = null, ownKey = false, budget = null } = {}) {if (!USER.test(String(subject || ''))) throw Object.assign(new Error('subject must be a usr_ subject'), { status: 400, code: 'subject.required' });const clientId = config.oauth && config.oauth.clientId;const clientSecret = config.oauth && config.oauth.clientSecret;@@@ -109,19 +127,72 @@ function createRunTokens({ config, store = null, fetchImpl = globalThis.fetch, lenv: envFor({ config, harness, token, provider, ownKey }),};require('./redact').remember(token);-- await record({ subject, task, attempt, provider, jti: out.jti, expiresAt: out.expiresAt });++ // The budget of the token: OpenVibe.AI holds the counter for it. A token with a budget that cannot be++ // registered is not handed out: a run that spends without a cap is a hole in the owner's money.++ if (budget && (Number(budget.usd) > 0 || Number(budget.tokens) > 0)) await registerBudget(out.jti, budget, out.expiresAt);++ await record({ subject, task, attempt, provider, jti: out.jti, expiresAt: out.expiresAt, budget });return out;}++ /**++ * Register the budget of a run token with OpenVibe.AI, under the token's jti. The budget is a row there: the++ * holder of the token cannot change it, and only Codes (with its own token) writes it.++ */++ async function registerBudget(jti, budget, expiresAt) {++ if (!jti) throw Object.assign(new Error('the run token has no id, so its budget cannot be registered'), { status: 503, code: 'run_token.budget_unavailable' });++ if (!tokens) throw Object.assign(new Error('Codes has no Network client secret, so it cannot register a run budget'), { status: 503, code: 'run_token.budget_unavailable' });++ const expires = expiresAt ? Math.floor(Date.parse(expiresAt) / 1000) : Math.floor(Date.now() / 1000) + ttl;++ let res;++ try {++ res = await fetchImpl(`${aiInternal(config)}/api/v1/run-budgets/${encodeURIComponent(jti)}`, {++ method: 'PUT',++ headers: { ...(await tokens.authHeaders()), Accept: 'application/json', 'Content-Type': 'application/json' },++ body: JSON.stringify({ budget_usd: Number(budget.usd) > 0 ? Number(budget.usd) : null, budget_tokens: Number(budget.tokens) > 0 ? Math.round(Number(budget.tokens)) : null, expires_at: expires }),++ signal: AbortSignal.timeout(20_000),++ });++ } catch (e) {++ throw Object.assign(new Error(`OpenVibe.AI did not answer: ${e.message}`), { status: 503, code: 'run_token.budget_unavailable' });++ }++ if (!res.ok) {++ const doc = await res.json().catch(() => null);++ throw Object.assign(new Error((doc && (doc.detail || doc.title)) || `OpenVibe.AI answered ${res.status}`), { status: 503, code: 'run_token.budget_unavailable' });++ }++ return true;++ }++++ /** What a run token has spent, per OpenVibe.AI (the counter of its jti): null when it cannot be read. */++ async function spend(jti) {++ if (!jti || !tokens) return null;++ try {++ const res = await fetchImpl(`${aiInternal(config)}/api/v1/run-budgets/${encodeURIComponent(jti)}`, {++ method: 'GET', headers: { ...(await tokens.authHeaders()), Accept: 'application/json' }, signal: AbortSignal.timeout(15_000),++ });++ if (!res.ok) return null;++ const doc = await res.json().catch(() => null);++ if (!doc) return null;++ return {++ budget_usd: doc.budget_usd == null ? null : Number(doc.budget_usd),++ budget_tokens: doc.budget_tokens == null ? null : Number(doc.budget_tokens),++ spent_usd: Number(doc.spent_usd) || 0,++ spent_tokens: Number(doc.spent_tokens) || 0,++ reached: Boolean(doc.reached),++ };++ } catch (e) {++ log.warn(`[Codes] the spend of a run token could not be read: ${String(e && e.message).slice(0, 200)}`);++ return null;++ }++ }++/** The row that says a run token was minted for a task: its jti and its expiry, never the token. */-- async function record({ subject, task = null, attempt = null, provider = null, jti = null, expiresAt = null } = {}) {++ async function record({ subject, task = null, attempt = null, provider = null, jti = null, expiresAt = null, budget = null } = {}) {if (!store || !store.db) return null;try {return await store.db.one(-- `INSERT INTO codes_run_tokens (id, task_id, attempt_id, subject, provider, jti, expires_at, created_at)-- VALUES ($1, $2, $3, $4, $5, $6, $7::timestamptz, $8::timestamptz) RETURNING *`,++ `INSERT INTO codes_run_tokens (id, task_id, attempt_id, subject, provider, jti, expires_at, budget_usd, budget_tokens, created_at)++ VALUES ($1, $2, $3, $4, $5, $6, $7::timestamptz, $8, $9, $10::timestamptz) RETURNING *`,[store.newId('rtk'), task || null, attempt || null, subject || null, provider || null, jti || null,-- expiresAt || null, store.iso()],++ expiresAt || null, budget && Number(budget.usd) > 0 ? Number(budget.usd) : null,++ budget && Number(budget.tokens) > 0 ? Math.round(Number(budget.tokens)) : null, store.iso()],);} catch (e) {// The row is a trace, not the point: a database that cannot take it must not stop the run.@@@ -137,7 +208,7 @@ function createRunTokens({ config, store = null, fetchImpl = globalThis.fetch, lreturn rows.map((r) => ({ id: r.id, task_id: r.task_id, attempt_id: r.attempt_id, subject: r.subject, provider: r.provider, jti: r.jti, expires_at: r.expires_at, created_at: r.created_at }));}-- return { mint, record, forTask, envFor, ttl };++ return { mint, record, forTask, envFor, ttl, spend, registerBudget };}module.exports = { createRunTokens, envFor, ANTHROPIC_HARNESSES };nnew file mode 100644@@@ -0,0 +1,110 @@++'use strict';++/**++ * The hard spend cap of a Codes run (Codes C3): the attempt's run token carries the budget, OpenVibe.AI counts what++ * the run spends, and the request after the budget is spent is refused (402 ai.run_budget_reached). The attempt then++ * ends as "stopped at its budget" with what it spent: the outcome line, not a crash.++ */++const assert = require('assert');++const http = require('http');++const { boot, check, done } = require('./helpers/boot');++++(async () => {++ // A stand-in for OpenVibe.AI: it keeps the budgets Codes registers and what each run spent.++ const budgets = new Map();++ const registered = [];++ const ai = http.createServer((req, res) => {++ const chunks = [];++ req.on('data', (c) => chunks.push(c));++ req.on('end', () => {++ const send = (status, doc) => { res.writeHead(status, { 'Content-Type': 'application/json' }); res.end(JSON.stringify(doc)); };++ if (req.url.startsWith('/api/v1/credentials')) return send(200, { keys: [] });++ const jti = decodeURIComponent(req.url.split('/').pop());++ if (req.method === 'PUT') {++ const body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}');++ budgets.set(jti, { ...body, spent_usd: 0.42, spent_tokens: 1000 });++ registered.push({ jti, body });++ return send(200, { jti, ...body });++ }++ const row = budgets.get(jti) || { budget_usd: null, budget_tokens: null, spent_usd: 0, spent_tokens: 0 };++ return send(200, { jti, ...row, reached: true });++ });++ });++ await new Promise((r) => ai.listen(0, '127.0.0.1', r));++ const aiUrl = `http://127.0.0.1:${ai.address().port}`;++++ let state = 'queued';++ let jobError = null;++ const run = {++ enabled: true,++ async submit() { return { id: 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J1', state: 'queued' }; },++ async read(jobId) { return { id: jobId, state, result: null, error: jobError }; },++ async cancel() { state = 'cancelled'; return { id: 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J1', state }; },++ async streamUrl() { throw new Error('not used'); },++ };++ const t = await boot({++ run,++ env: { CODES_AI_URL: aiUrl, CODES_AI_INTERNAL_URL: aiUrl, CODES_TASK_TOKENS: '1000', CODES_TASK_USD: '2' },++ });++ const db = t.ctx.store.db;++ const ada = t.network.addUser('ada');++ const actor = t.network.mintService('svc:actor', 'openvibe.codes', ['codes.task.start', 'codes.task.read']);++ const project = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';++ await db.exec(`INSERT INTO run_tasks (id, subject, project_id, job_id, repo, prompt, state, created_at, updated_at)++ VALUES ('tsk_01J8Z3Q4R5S6T7V8W9X0Y1Z2T1', $1, $2, 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2Z200', 'https://github.com/a/b', 'by hand', 'succeeded', 1, 2)`, [ada.subject, project]);++++ const start = async () => {++ const res = await fetch(`${t.base}/api/v1/tasks`, {++ method: 'POST',++ headers: { Accept: 'application/json', 'Content-Type': 'application/json', Authorization: `Bearer ${actor}`, 'X-OV-Subject': ada.subject },++ body: JSON.stringify({ repo: 'https://github.com/octocat/Hello-World', prompt: 'Add a line to the README.', agent: 'deepseek', idempotency_key: `act_budget_${Date.now()}` }),++ });++ return { status: res.status, body: await res.json().catch(() => null) };++ };++++ try {++ let taskId;++ await check('the attempt\'s run token carries the budget from Codes\' settings', async () => {++ const r = await start();++ assert.strictEqual(r.status, 201, JSON.stringify(r.body));++ taskId = r.body.task.id;++ assert.strictEqual(registered.length, 1, 'the budget was registered with OpenVibe.AI');++ assert.strictEqual(registered[0].body.budget_tokens, 1000);++ assert.strictEqual(registered[0].body.budget_usd, 2);++ assert.ok(registered[0].jti, 'the budget is registered under the token\'s own id');++ });++++ await check('the token row keeps the budget, and the attempt knows it', async () => {++ const token = await db.maybe('SELECT jti, budget_usd, budget_tokens FROM codes_run_tokens WHERE task_id = $1', [taskId]);++ assert.ok(token, 'the run token was recorded');++ assert.strictEqual(Number(token.budget_tokens), 1000);++ const attempt = await db.maybe('SELECT state, budget_usd, budget_tokens FROM codes_attempts WHERE task_id = $1', [taskId]);++ assert.strictEqual(Number(attempt.budget_tokens), 1000);++ });++++ await check('a run refused for its budget ends as "stopped at its budget", with what it spent', async () => {++ state = 'failed';++ jobError = 'the run was refused: ai.run_budget_reached: this run reached its tokens budget: 1000 of 1000 tokens spent.';++ const row = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [taskId]);++ await t.ctx.tasks.refresh(row, { force: true });++ const attempt = await db.maybe('SELECT state, reason, budget_hit, cost_usd, tokens FROM codes_attempts WHERE task_id = $1', [taskId]);++ assert.strictEqual(attempt.state, 'budget-stopped');++ assert.strictEqual(attempt.reason, 'budget');++ assert.strictEqual(attempt.budget_hit, 'tokens');++ assert.strictEqual(Number(attempt.tokens), 1000);++ await t.ctx.tasks.afterEnd(taskId);++ const task = await db.maybe('SELECT outcome, outcome_note FROM run_tasks WHERE id = $1', [taskId]);++ assert.strictEqual(task.outcome, 'unfinished');++ assert.ok(/stopped at its budget/.test(task.outcome_note), `the outcome line says it: ${task.outcome_note}`);++ assert.ok(/1000 of 1000 tokens/.test(task.outcome_note), `the outcome line says what it spent: ${task.outcome_note}`);++ });++++ await check('the task carries on: nothing was spent twice', async () => {++ const attempts = await db.many('SELECT n, state FROM codes_attempts WHERE task_id = $1 ORDER BY n', [taskId]);++ assert.strictEqual(attempts.length, 1, 'a budget stop buys no wrap-up turn');++ });++ } finally {++ await t.close();++ await new Promise((r) => ai.close(r));++ }++ done();++})().catch((e) => { console.error(e); process.exit(1); });
new file mode 100644
new file mode 100644
@@ -0,0 +1,15 @@
@@ -0,0 +1,15 @@
+-- phase: expand
+-- The budget a run token carries (Codes C3), and where the money an attempt spent went.
+--
+-- codes_run_tokens: the budget the minted token carries (null: no cap). OpenVibe.AI holds the counter for it, keyed by
+-- the token's jti; these columns say what the token was given, so a run can be traced without asking AI.
+-- codes_attempts: the budget the attempt was given, and the limit that stopped it ('usd', 'tokens', 'steps', 'minutes').
+--
+-- The budget is never raised here: it is what the task's limits (a benchmark's per-task limits) or the person's default
+-- says, and it is written once, when the attempt's token is minted.
+ALTER TABLE codes_run_tokens ADD COLUMN IF NOT EXISTS budget_usd double precision;
+ALTER TABLE codes_run_tokens ADD COLUMN IF NOT EXISTS budget_tokens bigint;
+
+ALTER TABLE codes_attempts ADD COLUMN IF NOT EXISTS budget_usd double precision;
+ALTER TABLE codes_attempts ADD COLUMN IF NOT EXISTS budget_tokens bigint;
+ALTER TABLE codes_attempts ADD COLUMN IF NOT EXISTS budget_hit text COLLATE "C"; -- usd | tokens | steps | minutes
@@ -120,6 +120,8 @@ function load(env = process.env) {
@@ -120,6 +120,8 @@ function load(env = process.env) {
steps: int(env.CODES_TASK_STEPS, 0),
steps: int(env.CODES_TASK_STEPS, 0),
minutes: int(env.CODES_TASK_MINUTES, 0),
minutes: int(env.CODES_TASK_MINUTES, 0),
usd: Number(env.CODES_TASK_USD || 0) || 0,
usd: Number(env.CODES_TASK_USD || 0) || 0,
+ // The tokens one attempt may spend, when the task does not say (a run token's budget, Codes C3).
+ tokens: int(env.CODES_TASK_TOKENS, 0),
},
},
// The task's timeline (stage C2): how many automatic continues one task may have before it needs a person.
// The task's timeline (stage C2): how many automatic continues one task may have before it needs a person.
task: { maxContinues: Math.max(0, int(env.CODES_TASK_MAX_CONTINUES, 2)) },
task: { maxContinues: Math.max(0, int(env.CODES_TASK_MAX_CONTINUES, 2)) },
@@ -86,6 +86,22 @@ const clip = (s, n) => String(s == null ? '' : s).slice(0, n);
@@ -86,6 +86,22 @@ const clip = (s, n) => String(s == null ? '' : s).slice(0, n);
const runnerOf = (agent) => RUNNER_OF[agent] || (agent || 'auto');
const runnerOf = (agent) => RUNNER_OF[agent] || (agent || 'auto');
const wordsOf = (reason) => REASON_TEXT[reason] || REASON_TEXT.error;
const wordsOf = (reason) => REASON_TEXT[reason] || REASON_TEXT.error;
+/**
+ * What an attempt that stopped at its budget says, in plain words: which budget it reached and what it spent.
+ * The dollars and the tokens are what OpenVibe.AI counted for the run token.
+ */
+function budgetNote(a) {
+ if (!a) return 'stopped at its budget';
+ const usd = a.cost_usd == null ? null : Number(a.cost_usd);
+ const tokens = a.tokens == null ? null : Number(a.tokens);
+ const limitUsd = a.budget_usd == null ? null : Number(a.budget_usd);
+ const limitTokens = a.budget_tokens == null ? null : Number(a.budget_tokens);
+ const parts = [];
+ if (limitUsd != null) parts.push(`$${usd == null ? 0 : usd.toFixed(4)} of $${limitUsd.toFixed(2)}`);
+ if (limitTokens != null) parts.push(`${tokens == null ? 0 : tokens} of ${limitTokens} tokens`);
+ return parts.length ? `stopped at its budget: ${parts.join(', ')} spent` : 'stopped at its budget';
+}
+
function createAttempts({ store, config = {}, log = console, harnesses = null }) {
function createAttempts({ store, config = {}, log = console, harnesses = null }) {
const db = store.db;
const db = store.db;
const catalog = harnesses || require('./harnesses').createHarnesses();
const catalog = harnesses || require('./harnesses').createHarnesses();
@@ -131,16 +147,18 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
@@ -131,16 +147,18 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
* One attempt. The fence: follow_up_of is unique, so an ending that starts a follow-up twice answers the same
* One attempt. The fence: follow_up_of is unique, so an ending that starts a follow-up twice answers the same
* attempt the second time and starts no second job. created tells the caller which of the two it was.
* attempt the second time and starts no second job. created tells the caller which of the two it was.
*/
*/
- async function create({ task, agent = null, model = null, machine = null, session = null, why = null, kind = 'first', followUpOf = null, state = 'queued' }) {
+ async function create({ task, agent = null, model = null, machine = null, session = null, why = null, kind = 'first', followUpOf = null, state = 'queued', budget = null }) {
if (!task || !task.id) throw new Error('an attempt belongs to a task');
if (!task || !task.id) throw new Error('an attempt belongs to a task');
for (let tries = 0; tries < 5; tries++) {
for (let tries = 0; tries < 5; tries++) {
const n = await db.one('SELECT COALESCE(MAX(n), 0) + 1 AS n FROM codes_attempts WHERE task_id = $1', [task.id]);
const n = await db.one('SELECT COALESCE(MAX(n), 0) + 1 AS n FROM codes_attempts WHERE task_id = $1', [task.id]);
const id = store.newId('att');
const id = store.newId('att');
const inserted = await db.maybe(
const inserted = await db.maybe(
- `INSERT INTO codes_attempts (id, task_id, n, agent, model, machine, session, state, why, kind, follow_up_of, created_at)
- VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::timestamptz)
+ `INSERT INTO codes_attempts (id, task_id, n, agent, model, machine, session, state, why, kind, follow_up_of, budget_usd, budget_tokens, created_at)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14::timestamptz)
ON CONFLICT DO NOTHING RETURNING *`,
ON CONFLICT DO NOTHING RETURNING *`,
- [id, task.id, n.n, agent, model, machine, session, state, why, kind, followUpOf, store.iso()],
+ [id, task.id, n.n, agent, model, machine, session, state, why, kind, followUpOf,
+ budget && Number(budget.usd) > 0 ? Number(budget.usd) : null,
+ budget && Number(budget.tokens) > 0 ? Math.round(Number(budget.tokens)) : null, store.iso()],
);
);
if (inserted) {
if (inserted) {
await db.exec('UPDATE run_tasks SET attempts_count = $2 WHERE id = $1', [task.id, n.n]);
await db.exec('UPDATE run_tasks SET attempts_count = $2 WHERE id = $1', [task.id, n.n]);
@@ -167,7 +185,7 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
@@ -167,7 +185,7 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
}
}
/** An attempt ended: its state, why, what it cost. */
/** An attempt ended: its state, why, what it cost. */
- async function end(id, { state = 'failed', reason = null, why = null, cost_usd = null, tokens = null, steps = null, session = null, model = null, machine = null, finished_at = null } = {}) {
+ async function end(id, { state = 'failed', reason = null, why = null, cost_usd = null, tokens = null, steps = null, session = null, model = null, machine = null, budget_hit = null, finished_at = null } = {}) {
return await db.one(
return await db.one(
`UPDATE codes_attempts SET state = $2,
`UPDATE codes_attempts SET state = $2,
reason = COALESCE($3, reason),
reason = COALESCE($3, reason),
@@ -178,9 +196,18 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
@@ -178,9 +196,18 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
session = COALESCE($8, session),
session = COALESCE($8, session),
model = COALESCE($9, model),
model = COALESCE($9, model),
machine = COALESCE($10, machine),
machine = COALESCE($10, machine),
- finished_at = COALESCE($11::timestamptz, finished_at)
+ budget_hit = COALESCE($11, budget_hit),
+ finished_at = COALESCE($12::timestamptz, finished_at)
WHERE id = $1 RETURNING *`,
WHERE id = $1 RETURNING *`,
- [id, state, reason, why, cost_usd == null ? null : String(cost_usd), tokens, steps, session, model, machine, finished_at || store.iso()],
+ [id, state, reason, why, cost_usd == null ? null : String(cost_usd), tokens, steps, session, model, machine, budget_hit, finished_at || store.iso()],
+ );
+ }
+
+ /** What an attempt really spent, counted by OpenVibe.AI (a run refused for its budget stops mid-turn). */
+ async function spend(id, { cost_usd = null, tokens = null } = {}) {
+ return await db.one(
+ `UPDATE codes_attempts SET cost_usd = COALESCE($2::numeric, cost_usd), tokens = COALESCE($3, tokens) WHERE id = $1 RETURNING *`,
+ [id, cost_usd == null ? null : String(cost_usd), tokens],
);
);
}
}
@@ -267,7 +294,7 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
@@ -267,7 +294,7 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
prev: prev ? prev.id : null,
prev: prev ? prev.id : null,
parent: a.follow_up_of || null,
parent: a.follow_up_of || null,
stopped: a.state === 'stopped',
stopped: a.state === 'stopped',
- budget: reason === 'budget' ? { hit: 'steps' } : undefined,
+ budget: reason === 'budget' ? { hit: a.budget_hit || 'steps' } : undefined,
wrapUp: a.kind === 'wrap-up',
wrapUp: a.kind === 'wrap-up',
hung: reason === 'hung',
hung: reason === 'hung',
facts: {
facts: {
@@ -338,6 +365,11 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
@@ -338,6 +365,11 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
const words = wordsOf(reason);
const words = wordsOf(reason);
const continues = upto.filter((a) => a.kind === 'continue').length;
const continues = upto.filter((a) => a.kind === 'continue').length;
+ // An attempt that ran out of money (OpenVibe.AI refused it) ends the task there: no wrap-up turn buys
+ // another request the budget will not pay for, and no other agent takes over with the same cap.
+ if (reason === 'budget' && ['usd', 'tokens'].includes(ending.budget_hit)) {
+ return { outcome: 'unfinished', note: budgetNote(ending) };
+ }
if (step.action === 'done') return { outcome: 'done', note: 'the task finished' };
if (step.action === 'done') return { outcome: 'done', note: 'the task finished' };
if (step.action === 'stop') return { outcome: 'stopped', note: 'the task was stopped on purpose' };
if (step.action === 'stop') return { outcome: 'stopped', note: 'the task was stopped on purpose' };
if (step.action === 'wait') return { outcome: null, note: null };
if (step.action === 'wait') return { outcome: null, note: null };
@@ -411,8 +443,8 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
@@ -411,8 +443,8 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
return {
return {
STATES, REASONS, KINDS, REASON_TEXT, RUNNER_OF, HARNESS_OF,
STATES, REASONS, KINDS, REASON_TEXT, RUNNER_OF, HARNESS_OF,
- list, get, byJob, create, bind, end, decide, present, routing, nextAgent, nameOf, wordsOf,
+ list, get, byJob, create, bind, end, spend, decide, present, routing, nextAgent, nameOf, wordsOf, budgetNote,
};
};
}
}
-module.exports = { createAttempts, RUNNER_OF, HARNESS_OF, STATES, REASONS, KINDS, REASON_TEXT, ERROR_TEXT };
+module.exports = { createAttempts, RUNNER_OF, HARNESS_OF, STATES, REASONS, KINDS, REASON_TEXT, ERROR_TEXT, budgetNote };
@@ -179,12 +179,12 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -179,12 +179,12 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
* The environment the job's harness gets (Codes C3): a run token that names the person and says where to
* The environment the job's harness gets (Codes C3): a run token that names the person and says where to
* send its requests. A failure to mint one is not fatal: the job runs as it did before, without the token.
* send its requests. A failure to mint one is not fatal: the job runs as it did before, without the token.
*/
*/
- async function harnessEnv({ subject, taskId, attemptId = null, agent = null }) {
+ async function harnessEnv({ subject, taskId, attemptId = null, agent = null, limits = null }) {
if (!runTokens) return null;
if (!runTokens) return null;
try {
try {
const provider = providerFor(agent);
const provider = providerFor(agent);
const own = provider ? await hasOwnKey(subject, provider) : false;
const own = provider ? await hasOwnKey(subject, provider) : false;
- const minted = await runTokens.mint({ subject, task: taskId, attempt: attemptId, harness: agent, provider, ownKey: own });
+ const minted = await runTokens.mint({ subject, task: taskId, attempt: attemptId, harness: agent, provider, ownKey: own, budget: budgetFor(limits) });
return minted.env;
return minted.env;
} catch (e) {
} catch (e) {
log.warn(`[Codes] no run token for task ${taskId}: ${String(e && e.message).slice(0, 200)}`);
log.warn(`[Codes] no run token for task ${taskId}: ${String(e && e.message).slice(0, 200)}`);
@@ -192,6 +192,53 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -192,6 +192,53 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
}
}
}
}
+ /**
+ * The budget of one attempt: the task's limits (a benchmark's per-task limits) first, then the person's default
+ * in Codes' settings, and nothing at all when neither says (an uncapped run is an uncapped run).
+ */
+ function budgetFor(limits) {
+ const pick = (from, fallback) => {
+ const n = Number(from);
+ if (Number.isFinite(n) && n > 0) return n;
+ const f = Number(fallback);
+ return Number.isFinite(f) && f > 0 ? f : null;
+ };
+ const l = limits && typeof limits === 'object' ? limits : {};
+ return {
+ usd: pick(l.usd, config.budget && config.budget.usd),
+ tokens: pick(l.tokens, config.budget && config.budget.tokens),
+ };
+ }
+
+ /**
+ * The attempt's run token and what it spent, per OpenVibe.AI: the budget the token carried and the counter of
+ * what the run really spent (the numbers AI holds, not what the harness reported).
+ */
+ async function tokenSpend(attemptId) {
+ if (!runTokens || !store || !store.db || !attemptId) return null;
+ const row = await store.db.maybe('SELECT jti FROM codes_run_tokens WHERE attempt_id = $1 ORDER BY created_at DESC LIMIT 1', [attemptId]);
+ if (!row || !row.jti) return null;
+ return await runTokens.spend(row.jti);
+ }
+
+ /** Whether the run was refused for its budget (OpenVibe.AI said 402 ai.run_budget_reached). */
+ function budgetRefused(result, error) {
+ return /ai\.run_budget_reached/.test(budgetText(result, error));
+ }
+
+ /** Which budget ran out, as the refusal says it: the tokens or the dollars. */
+ function budgetKind(result, error) {
+ const text = budgetText(result, error);
+ return /tokens? budget|tokens? of/i.test(text) ? 'tokens' : 'usd';
+ }
+
+ /** Everything the run said: the error and the session lines, as one text. */
+ function budgetText(result, error) {
+ const doc = asObject(result) || {};
+ const lines = Array.isArray(doc.log) ? doc.log : [];
+ return [String(error || ''), ...lines.map((l) => (l && l.text ? String(l.text) : ''))].join('\n');
+ }
+
/** The provider a harness runs on: the catalog's own word for it (anthropic, openai, deepseek, …). */
/** The provider a harness runs on: the catalog's own word for it (anthropic, openai, deepseek, …). */
function providerFor(agent) {
function providerFor(agent) {
if (!agent) return null;
if (!agent) return null;
@@ -218,6 +265,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -218,6 +265,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
async function recordState(row, next) {
async function recordState(row, next) {
const state = STATES.has(next && next.state) ? next.state : null;
const state = STATES.has(next && next.state) ? next.state : null;
if (!state || !row || !row.id) return row;
if (!state || !row || !row.id) return row;
+ let budgetAttempt = null; // the attempt OpenVibe.AI refused for its budget (its spend is read after the write)
const saved = await store.tx(async () => {
const saved = await store.tx(async () => {
const current = await db.maybe('SELECT * FROM run_tasks WHERE id = $1 FOR UPDATE', [row.id]);
const current = await db.maybe('SELECT * FROM run_tasks WHERE id = $1 FOR UPDATE', [row.id]);
if (!current) return row;
if (!current) return row;
@@ -247,13 +295,20 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -247,13 +295,20 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
if (attempt) {
if (attempt) {
if (TERMINAL.has(state)) {
if (TERMINAL.has(state)) {
const usage = usageOf(updated.result);
const usage = usageOf(updated.result);
- const why = reasonOf(state, updated.error);
- const hit = budgetHit(attempt, usage, why);
+ let why = reasonOf(state, updated.error);
+ const localHit = budgetHit(attempt, usage, why);
+ // OpenVibe.AI refused the run for its budget (402 ai.run_budget_reached): the attempt ends at its
+ // budget, not as a crash. Which budget ran out is in the refusal itself.
+ const refused = !localHit && budgetRefused(updated.result, updated.error);
+ const hit = localHit || (refused ? budgetKind(updated.result, updated.error) : null);
+ if (refused) why = 'budget';
+ if (refused) budgetAttempt = { id: attempt.id };
await attemptStore.end(attempt.id, {
await attemptStore.end(attempt.id, {
state: hit ? 'budget-stopped' : (ATTEMPT_STATE[state] || 'failed'),
state: hit ? 'budget-stopped' : (ATTEMPT_STATE[state] || 'failed'),
reason: hit ? 'budget' : why,
reason: hit ? 'budget' : why,
cost_usd: usage.cost_usd, tokens: usage.tokens, steps: usage.steps, session: usage.session,
cost_usd: usage.cost_usd, tokens: usage.tokens, steps: usage.steps, session: usage.session,
model: usage.model, machine: machineOf(next.job),
model: usage.model, machine: machineOf(next.job),
+ budget_hit: hit,
finished_at: updated.finished_at || store.iso(),
finished_at: updated.finished_at || store.iso(),
});
});
} else {
} else {
@@ -264,6 +319,11 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -264,6 +319,11 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
}
}
return updated;
return updated;
});
});
+ // What the refused run really spent: OpenVibe.AI counted it (the harness may have stopped mid-turn).
+ if (budgetAttempt) {
+ const spend = await tokenSpend(budgetAttempt.id);
+ if (spend) await attemptStore.spend(budgetAttempt.id, { cost_usd: spend.spent_usd, tokens: spend.spent_tokens });
+ }
if (TERMINAL.has(saved.state) && benchAttemptEnded && saved.actor_ref) {
if (TERMINAL.has(saved.state) && benchAttemptEnded && saved.actor_ref) {
const meta = asObject(saved.actor_ref);
const meta = asObject(saved.actor_ref);
if (meta && meta.bench) await benchAttemptEnded(saved).catch((e) => log.warn(`[Codes] benchmark publish failed for task ${saved.id}: ${String(e && e.message).slice(0, 200)}`));
if (meta && meta.bench) await benchAttemptEnded(saved).catch((e) => log.warn(`[Codes] benchmark publish failed for task ${saved.id}: ${String(e && e.message).slice(0, 200)}`));
@@ -298,16 +358,18 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -298,16 +358,18 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
return inserted;
return inserted;
});
});
// The task's first attempt: every Run job is one attempt, and this one is the task as it was asked.
// The task's first attempt: every Run job is one attempt, and this one is the task as it was asked.
+ // The budget of the attempt: the task's limits (a benchmark's per-task limits) or the person's default.
+ const budget = budgetFor(args.limits);
const first = await attemptStore.create({
const first = await attemptStore.create({
task: inserted, agent: args.agent || null, model: args.model || null,
task: inserted, agent: args.agent || null, model: args.model || null,
why: args.agent ? `the first try, with ${attemptStore.nameOf(args.agent)}` : 'the first try, with the agent that fits the task',
why: args.agent ? `the first try, with ${attemptStore.nameOf(args.agent)}` : 'the first try, with the agent that fits the task',
- kind: 'first', state: 'queued',
+ kind: 'first', state: 'queued', budget,
});
});
let job;
let job;
try {
try {
// The job's harness runs with a short-lived run token that names the person (Codes C3): the environment
// The job's harness runs with a short-lived run token that names the person (Codes C3): the environment
// travels with the job's input and the machine's worker hands it to the harness.
// travels with the job's input and the machine's worker hands it to the harness.
- const env = await harnessEnv({ subject, taskId: id, agent: args.agent || null });
+ const env = await harnessEnv({ subject, taskId: id, attemptId: first.attempt.id, agent: args.agent || null, limits: args.limits });
job = await run.submit({ subject, project, args: env ? { ...args, env } : args, key: id });
job = await run.submit({ subject, project, args: env ? { ...args, env } : args, key: id });
} catch (e) {
} catch (e) {
const why = e instanceof RunError ? `${e.code}: ${e.message}` : e.message;
const why = e instanceof RunError ? `${e.code}: ${e.message}` : e.message;
@@ -399,8 +461,13 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -399,8 +461,13 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
const actorRef = asObject(row.actor_ref);
const actorRef = asObject(row.actor_ref);
if (actorRef && actorRef.bench) {
if (actorRef && actorRef.bench) {
const done = row.state === 'succeeded';
const done = row.state === 'succeeded';
+ // A benchmark attempt that ran out of money says so, with what it spent: it is not a crash.
+ const ended = (await attemptStore.list(row.id)).slice(-1)[0] || null;
+ const note = ended && ended.reason === 'budget' && ['usd', 'tokens'].includes(ended.budget_hit)
+ ? attemptStore.budgetNote(ended)
+ : (done ? 'benchmark attempt finished' : (row.error || 'benchmark attempt failed'));
return await db.one('UPDATE run_tasks SET outcome = $2, outcome_note = $3, updated_at = $4 WHERE id = $1 RETURNING *',
return await db.one('UPDATE run_tasks SET outcome = $2, outcome_note = $3, updated_at = $4 WHERE id = $1 RETURNING *',
- [row.id, done ? 'done' : 'unfinished', done ? 'benchmark attempt finished' : (row.error || 'benchmark attempt failed'), store.now()]);
+ [row.id, done ? 'done' : 'unfinished', note, store.now()]);
}
}
const attempts = await attemptStore.list(row.id);
const attempts = await attemptStore.list(row.id);
const ending = attempts[attempts.length - 1];
const ending = attempts[attempts.length - 1];
@@ -16,10 +16,17 @@
@@ -16,10 +16,17 @@
* The credential header (x-openvibe-credential) names the provider whose key the run uses, when the person has one.
* The credential header (x-openvibe-credential) names the provider whose key the run uses, when the person has one.
*
*
* createRunTokens({ config, store, fetchImpl, log })
* createRunTokens({ config, store, fetchImpl, log })
- * mint({ subject, task, attempt, harness, provider, ownKey }) → { token, expiresAt, jti, env }
- * record({ subject, task, attempt, provider, jti, expiresAt }) → the row (no token in it)
+ * mint({ subject, task, attempt, harness, provider, ownKey, budget }) → { token, expiresAt, jti, env }
+ * record({ subject, task, attempt, provider, jti, expiresAt, budget }) → the row (no token in it)
+ * spend(jti) → what the run spent, per OpenVibe.AI (null when it cannot be read)
+ *
+ * The budget of a run token (the hard spend cap): the token's jti is registered with OpenVibe.AI, which counts
+ * every request of that token and refuses the next one when the budget is spent. The budget is a row in AI, keyed by
+ * the jti, not a claim in the token: the holder of a token can neither raise the budget nor name another run's.
*/
*/
+const { createServiceTokenClient } = require('openvibe-sdk/auth');
+
const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
/** The harnesses that speak Anthropic's Messages API (the rest speak OpenAI's Chat Completions). */
/** The harnesses that speak Anthropic's Messages API (the rest speak OpenAI's Chat Completions). */
@@ -39,6 +46,11 @@ function aiBase(config) {
@@ -39,6 +46,11 @@ function aiBase(config) {
return String((config.ai && config.ai.url) || 'https://ai.openvibe.services').replace(/\/+$/, '');
return String((config.ai && config.ai.url) || 'https://ai.openvibe.services').replace(/\/+$/, '');
}
}
+/** The address Codes itself reaches OpenVibe.AI at (the vault is host-local: the internal address). */
+function aiInternal(config) {
+ return String((config.ai && (config.ai.internalUrl || config.ai.url)) || 'https://ai.openvibe.services').replace(/\/+$/, '');
+}
+
/**
/**
* The environment a harness of this kind gets: where to send its requests, the token to send, and (when the person
* The environment a harness of this kind gets: where to send its requests, the token to send, and (when the person
* has a key for that provider) the header that says which key.
* has a key for that provider) the header that says which key.
@@ -68,12 +80,18 @@ function envFor({ config, harness, token, provider, ownKey }) {
@@ -68,12 +80,18 @@ function envFor({ config, harness, token, provider, ownKey }) {
function createRunTokens({ config, store = null, fetchImpl = globalThis.fetch, log = console } = {}) {
function createRunTokens({ config, store = null, fetchImpl = globalThis.fetch, log = console } = {}) {
const ttl = Math.min(7200, Math.max(60, Number((config.runTokens && config.runTokens.ttlSeconds) || 7200)));
const ttl = Math.min(7200, Math.max(60, Number((config.runTokens && config.runTokens.ttlSeconds) || 7200)));
+ // Codes' own token for OpenVibe.AI (ai.run.create): the budget of a run token is registered there, under the
+ // token's jti, and the counter of what the run spent is read back from it.
+ const scope = (config.ai && config.ai.scope) || 'ai.credential.manage ai.run.create';
+ const tokens = config.oauth && config.oauth.clientSecret
+ ? createServiceTokenClient({ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.ai', scope, fetch: fetchImpl })
+ : null;
/**
/**
* Mint a run token for one attempt of one task of one person.
* Mint a run token for one attempt of one task of one person.
* → { token, expiresAt, jti, env }; the token is in the answer and nowhere else.
* → { token, expiresAt, jti, env }; the token is in the answer and nowhere else.
*/
*/
- async function mint({ subject, task = null, attempt = null, harness = null, provider = null, ownKey = false } = {}) {
+ async function mint({ subject, task = null, attempt = null, harness = null, provider = null, ownKey = false, budget = null } = {}) {
if (!USER.test(String(subject || ''))) throw Object.assign(new Error('subject must be a usr_ subject'), { status: 400, code: 'subject.required' });
if (!USER.test(String(subject || ''))) throw Object.assign(new Error('subject must be a usr_ subject'), { status: 400, code: 'subject.required' });
const clientId = config.oauth && config.oauth.clientId;
const clientId = config.oauth && config.oauth.clientId;
const clientSecret = config.oauth && config.oauth.clientSecret;
const clientSecret = config.oauth && config.oauth.clientSecret;
@@ -109,19 +127,72 @@ function createRunTokens({ config, store = null, fetchImpl = globalThis.fetch, l
@@ -109,19 +127,72 @@ function createRunTokens({ config, store = null, fetchImpl = globalThis.fetch, l
env: envFor({ config, harness, token, provider, ownKey }),
env: envFor({ config, harness, token, provider, ownKey }),
};
};
require('./redact').remember(token);
require('./redact').remember(token);
- await record({ subject, task, attempt, provider, jti: out.jti, expiresAt: out.expiresAt });
+ // The budget of the token: OpenVibe.AI holds the counter for it. A token with a budget that cannot be
+ // registered is not handed out: a run that spends without a cap is a hole in the owner's money.
+ if (budget && (Number(budget.usd) > 0 || Number(budget.tokens) > 0)) await registerBudget(out.jti, budget, out.expiresAt);
+ await record({ subject, task, attempt, provider, jti: out.jti, expiresAt: out.expiresAt, budget });
return out;
return out;
}
}
+ /**
+ * Register the budget of a run token with OpenVibe.AI, under the token's jti. The budget is a row there: the
+ * holder of the token cannot change it, and only Codes (with its own token) writes it.
+ */
+ async function registerBudget(jti, budget, expiresAt) {
+ if (!jti) throw Object.assign(new Error('the run token has no id, so its budget cannot be registered'), { status: 503, code: 'run_token.budget_unavailable' });
+ if (!tokens) throw Object.assign(new Error('Codes has no Network client secret, so it cannot register a run budget'), { status: 503, code: 'run_token.budget_unavailable' });
+ const expires = expiresAt ? Math.floor(Date.parse(expiresAt) / 1000) : Math.floor(Date.now() / 1000) + ttl;
+ let res;
+ try {
+ res = await fetchImpl(`${aiInternal(config)}/api/v1/run-budgets/${encodeURIComponent(jti)}`, {
+ method: 'PUT',
+ headers: { ...(await tokens.authHeaders()), Accept: 'application/json', 'Content-Type': 'application/json' },
+ body: JSON.stringify({ budget_usd: Number(budget.usd) > 0 ? Number(budget.usd) : null, budget_tokens: Number(budget.tokens) > 0 ? Math.round(Number(budget.tokens)) : null, expires_at: expires }),
+ signal: AbortSignal.timeout(20_000),
+ });
+ } catch (e) {
+ throw Object.assign(new Error(`OpenVibe.AI did not answer: ${e.message}`), { status: 503, code: 'run_token.budget_unavailable' });
+ }
+ if (!res.ok) {
+ const doc = await res.json().catch(() => null);
+ throw Object.assign(new Error((doc && (doc.detail || doc.title)) || `OpenVibe.AI answered ${res.status}`), { status: 503, code: 'run_token.budget_unavailable' });
+ }
+ return true;
+ }
+
+ /** What a run token has spent, per OpenVibe.AI (the counter of its jti): null when it cannot be read. */
+ async function spend(jti) {
+ if (!jti || !tokens) return null;
+ try {
+ const res = await fetchImpl(`${aiInternal(config)}/api/v1/run-budgets/${encodeURIComponent(jti)}`, {
+ method: 'GET', headers: { ...(await tokens.authHeaders()), Accept: 'application/json' }, signal: AbortSignal.timeout(15_000),
+ });
+ if (!res.ok) return null;
+ const doc = await res.json().catch(() => null);
+ if (!doc) return null;
+ return {
+ budget_usd: doc.budget_usd == null ? null : Number(doc.budget_usd),
+ budget_tokens: doc.budget_tokens == null ? null : Number(doc.budget_tokens),
+ spent_usd: Number(doc.spent_usd) || 0,
+ spent_tokens: Number(doc.spent_tokens) || 0,
+ reached: Boolean(doc.reached),
+ };
+ } catch (e) {
+ log.warn(`[Codes] the spend of a run token could not be read: ${String(e && e.message).slice(0, 200)}`);
+ return null;
+ }
+ }
+
/** The row that says a run token was minted for a task: its jti and its expiry, never the token. */
/** The row that says a run token was minted for a task: its jti and its expiry, never the token. */
- async function record({ subject, task = null, attempt = null, provider = null, jti = null, expiresAt = null } = {}) {
+ async function record({ subject, task = null, attempt = null, provider = null, jti = null, expiresAt = null, budget = null } = {}) {
if (!store || !store.db) return null;
if (!store || !store.db) return null;
try {
try {
return await store.db.one(
return await store.db.one(
- `INSERT INTO codes_run_tokens (id, task_id, attempt_id, subject, provider, jti, expires_at, created_at)
- VALUES ($1, $2, $3, $4, $5, $6, $7::timestamptz, $8::timestamptz) RETURNING *`,
+ `INSERT INTO codes_run_tokens (id, task_id, attempt_id, subject, provider, jti, expires_at, budget_usd, budget_tokens, created_at)
+ VALUES ($1, $2, $3, $4, $5, $6, $7::timestamptz, $8, $9, $10::timestamptz) RETURNING *`,
[store.newId('rtk'), task || null, attempt || null, subject || null, provider || null, jti || null,
[store.newId('rtk'), task || null, attempt || null, subject || null, provider || null, jti || null,
- expiresAt || null, store.iso()],
+ expiresAt || null, budget && Number(budget.usd) > 0 ? Number(budget.usd) : null,
+ budget && Number(budget.tokens) > 0 ? Math.round(Number(budget.tokens)) : null, store.iso()],
);
);
} catch (e) {
} catch (e) {
// The row is a trace, not the point: a database that cannot take it must not stop the run.
// The row is a trace, not the point: a database that cannot take it must not stop the run.
@@ -137,7 +208,7 @@ function createRunTokens({ config, store = null, fetchImpl = globalThis.fetch, l
@@ -137,7 +208,7 @@ function createRunTokens({ config, store = null, fetchImpl = globalThis.fetch, l
return rows.map((r) => ({ id: r.id, task_id: r.task_id, attempt_id: r.attempt_id, subject: r.subject, provider: r.provider, jti: r.jti, expires_at: r.expires_at, created_at: r.created_at }));
return rows.map((r) => ({ id: r.id, task_id: r.task_id, attempt_id: r.attempt_id, subject: r.subject, provider: r.provider, jti: r.jti, expires_at: r.expires_at, created_at: r.created_at }));
}
}
- return { mint, record, forTask, envFor, ttl };
+ return { mint, record, forTask, envFor, ttl, spend, registerBudget };
}
}
module.exports = { createRunTokens, envFor, ANTHROPIC_HARNESSES };
module.exports = { createRunTokens, envFor, ANTHROPIC_HARNESSES };
new file mode 100644
new file mode 100644
@@ -0,0 +1,110 @@
@@ -0,0 +1,110 @@
+'use strict';
+/**
+ * The hard spend cap of a Codes run (Codes C3): the attempt's run token carries the budget, OpenVibe.AI counts what
+ * the run spends, and the request after the budget is spent is refused (402 ai.run_budget_reached). The attempt then
+ * ends as "stopped at its budget" with what it spent: the outcome line, not a crash.
+ */
+const assert = require('assert');
+const http = require('http');
+const { boot, check, done } = require('./helpers/boot');
+
+(async () => {
+ // A stand-in for OpenVibe.AI: it keeps the budgets Codes registers and what each run spent.
+ const budgets = new Map();
+ const registered = [];
+ const ai = http.createServer((req, res) => {
+ const chunks = [];
+ req.on('data', (c) => chunks.push(c));
+ req.on('end', () => {
+ const send = (status, doc) => { res.writeHead(status, { 'Content-Type': 'application/json' }); res.end(JSON.stringify(doc)); };
+ if (req.url.startsWith('/api/v1/credentials')) return send(200, { keys: [] });
+ const jti = decodeURIComponent(req.url.split('/').pop());
+ if (req.method === 'PUT') {
+ const body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}');
+ budgets.set(jti, { ...body, spent_usd: 0.42, spent_tokens: 1000 });
+ registered.push({ jti, body });
+ return send(200, { jti, ...body });
+ }
+ const row = budgets.get(jti) || { budget_usd: null, budget_tokens: null, spent_usd: 0, spent_tokens: 0 };
+ return send(200, { jti, ...row, reached: true });
+ });
+ });
+ await new Promise((r) => ai.listen(0, '127.0.0.1', r));
+ const aiUrl = `http://127.0.0.1:${ai.address().port}`;
+
+ let state = 'queued';
+ let jobError = null;
+ const run = {
+ enabled: true,
+ async submit() { return { id: 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J1', state: 'queued' }; },
+ async read(jobId) { return { id: jobId, state, result: null, error: jobError }; },
+ async cancel() { state = 'cancelled'; return { id: 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J1', state }; },
+ async streamUrl() { throw new Error('not used'); },
+ };
+ const t = await boot({
+ run,
+ env: { CODES_AI_URL: aiUrl, CODES_AI_INTERNAL_URL: aiUrl, CODES_TASK_TOKENS: '1000', CODES_TASK_USD: '2' },
+ });
+ const db = t.ctx.store.db;
+ const ada = t.network.addUser('ada');
+ const actor = t.network.mintService('svc:actor', 'openvibe.codes', ['codes.task.start', 'codes.task.read']);
+ const project = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';
+ await db.exec(`INSERT INTO run_tasks (id, subject, project_id, job_id, repo, prompt, state, created_at, updated_at)
+ VALUES ('tsk_01J8Z3Q4R5S6T7V8W9X0Y1Z2T1', $1, $2, 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2Z200', 'https://github.com/a/b', 'by hand', 'succeeded', 1, 2)`, [ada.subject, project]);
+
+ const start = async () => {
+ const res = await fetch(`${t.base}/api/v1/tasks`, {
+ method: 'POST',
+ headers: { Accept: 'application/json', 'Content-Type': 'application/json', Authorization: `Bearer ${actor}`, 'X-OV-Subject': ada.subject },
+ body: JSON.stringify({ repo: 'https://github.com/octocat/Hello-World', prompt: 'Add a line to the README.', agent: 'deepseek', idempotency_key: `act_budget_${Date.now()}` }),
+ });
+ return { status: res.status, body: await res.json().catch(() => null) };
+ };
+
+ try {
+ let taskId;
+ await check('the attempt\'s run token carries the budget from Codes\' settings', async () => {
+ const r = await start();
+ assert.strictEqual(r.status, 201, JSON.stringify(r.body));
+ taskId = r.body.task.id;
+ assert.strictEqual(registered.length, 1, 'the budget was registered with OpenVibe.AI');
+ assert.strictEqual(registered[0].body.budget_tokens, 1000);
+ assert.strictEqual(registered[0].body.budget_usd, 2);
+ assert.ok(registered[0].jti, 'the budget is registered under the token\'s own id');
+ });
+
+ await check('the token row keeps the budget, and the attempt knows it', async () => {
+ const token = await db.maybe('SELECT jti, budget_usd, budget_tokens FROM codes_run_tokens WHERE task_id = $1', [taskId]);
+ assert.ok(token, 'the run token was recorded');
+ assert.strictEqual(Number(token.budget_tokens), 1000);
+ const attempt = await db.maybe('SELECT state, budget_usd, budget_tokens FROM codes_attempts WHERE task_id = $1', [taskId]);
+ assert.strictEqual(Number(attempt.budget_tokens), 1000);
+ });
+
+ await check('a run refused for its budget ends as "stopped at its budget", with what it spent', async () => {
+ state = 'failed';
+ jobError = 'the run was refused: ai.run_budget_reached: this run reached its tokens budget: 1000 of 1000 tokens spent.';
+ const row = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [taskId]);
+ await t.ctx.tasks.refresh(row, { force: true });
+ const attempt = await db.maybe('SELECT state, reason, budget_hit, cost_usd, tokens FROM codes_attempts WHERE task_id = $1', [taskId]);
+ assert.strictEqual(attempt.state, 'budget-stopped');
+ assert.strictEqual(attempt.reason, 'budget');
+ assert.strictEqual(attempt.budget_hit, 'tokens');
+ assert.strictEqual(Number(attempt.tokens), 1000);
+ await t.ctx.tasks.afterEnd(taskId);
+ const task = await db.maybe('SELECT outcome, outcome_note FROM run_tasks WHERE id = $1', [taskId]);
+ assert.strictEqual(task.outcome, 'unfinished');
+ assert.ok(/stopped at its budget/.test(task.outcome_note), `the outcome line says it: ${task.outcome_note}`);
+ assert.ok(/1000 of 1000 tokens/.test(task.outcome_note), `the outcome line says what it spent: ${task.outcome_note}`);
+ });
+
+ await check('the task carries on: nothing was spent twice', async () => {
+ const attempts = await db.many('SELECT n, state FROM codes_attempts WHERE task_id = $1 ORDER BY n', [taskId]);
+ assert.strictEqual(attempts.length, 1, 'a budget stop buys no wrap-up turn');
+ });
+ } finally {
+ await t.close();
+ await new Promise((r) => ai.close(r));
+ }
+ done();
+})().catch((e) => { console.error(e); process.exit(1); });