openvibers/openvibe.codes Commits
Codes: run and publish benchmark matrices (#175)
Changes
@@@ -95,11 +95,14 @@ function parseArgs(raw, env = process.env) {let a;try { a = JSON.parse(String(raw || '').trim() || '{}'); } catch { throw new Error('the job input is not JSON'); }if (!a || typeof a !== 'object' || Array.isArray(a)) throw new Error('the job input must be an object');-- let repo;-- try { repo = new URL(String(a.repo || '')); } catch { throw new Error('repo must be a repository URL'); }-- const fileOk = env.OPENVIBE_CODES_ALLOW_FILE_REPO === '1' && repo.protocol === 'file:';-- if (!fileOk && repo.protocol !== 'https:') throw new Error('repo must be an https URL');-- if (repo.username || repo.password) throw new Error('repo must not carry credentials');++ const kind = a.kind === 'bench' ? 'bench' : 'code';++ let repo = null;++ if (kind === 'code') {++ try { repo = new URL(String(a.repo || '')); } catch { throw new Error('repo must be a repository URL'); }++ const fileOk = env.OPENVIBE_CODES_ALLOW_FILE_REPO === '1' && repo.protocol === 'file:';++ if (!fileOk && repo.protocol !== 'https:') throw new Error('repo must be an https URL');++ if (repo.username || repo.password) throw new Error('repo must not carry credentials');++ }const prompt = typeof a.prompt === 'string' ? a.prompt.trim() : '';if (!prompt || prompt.length > MAX_PROMPT) throw new Error(`prompt must be 1-${MAX_PROMPT} characters`);const ref = a.ref == null || a.ref === '' ? null : String(a.ref);@@@ -118,8 +121,14 @@ function parseArgs(raw, env = process.env) {const resumeIn = a.resume == null || a.resume === '' ? env.RESUME : a.resume;const resume = typeof resumeIn === 'string' && /^[A-Za-z0-9._:-]{1,200}$/.test(resumeIn) ? resumeIn : null;const wanted = jobEnv(a);++ const limits = kind === 'bench' && a.limits && typeof a.limits === 'object' ? {++ ms: Number.isFinite(Number(a.limits.ms)) && Number(a.limits.ms) > 0 ? Math.min(300000, Number(a.limits.ms)) : null,++ tokens: Number.isFinite(Number(a.limits.tokens)) && Number(a.limits.tokens) > 0 ? Math.min(1000000, Number(a.limits.tokens)) : null,++ usd: Number.isFinite(Number(a.limits.usd)) && Number(a.limits.usd) > 0 ? Math.min(100, Number(a.limits.usd)) : null,++ } : null;return {-- repo: repo.toString(), ref, prompt, permission, push,++ kind, repo: repo ? repo.toString() : null, ref, prompt, permission, push,++ ...(limits ? { limits } : {}),...(resume ? { resume } : {}),...(Object.keys(wanted).length ? { env: wanted } : {}),agent: typeof a.agent === 'string' && /^[a-z0-9-]{1,40}$/.test(a.agent) ? a.agent : null,@@@ -292,13 +301,19 @@ async function runJob({ input, harness, emit, writeResult, env = process.env, tm}const dir = fs.mkdtempSync(path.join(tmp, 'codes-'));const work = path.join(dir, 'repo');-- const result = { protocol: JOB_PROTOCOL, ok: false, repo: scrub(job.repo), ref: job.ref, agents: [], summary: '' };++ const result = { protocol: JOB_PROTOCOL, ok: false, kind: job.kind, repo: job.repo ? scrub(job.repo) : null, ref: job.ref, agents: [], summary: '' };++ let base = null;try {-- say({ t: 'step', text: `Cloning ${scrub(job.repo)}${job.ref ? ` (${job.ref})` : ''}` });-- await git(['clone', '--depth', '50', ...(job.ref ? ['--branch', job.ref] : []), '--', job.repo, work], { timeout: 300_000 });-- const base = (await git(['rev-parse', '--abbrev-ref', 'HEAD'], { cwd: work })).trim();-- result.base = base;-- say({ t: 'step', text: `Working on ${base}` });++ if (job.kind === 'code') {++ say({ t: 'step', text: `Cloning ${scrub(job.repo)}${job.ref ? ` (${job.ref})` : ''}` });++ await git(['clone', '--depth', '50', ...(job.ref ? ['--branch', job.ref] : []), '--', job.repo, work], { timeout: 300_000 });++ base = (await git(['rev-parse', '--abbrev-ref', 'HEAD'], { cwd: work })).trim();++ result.base = base;++ say({ t: 'step', text: `Working on ${base}` });++ } else {++ fs.mkdirSync(work, { recursive: true });++ say({ t: 'step', text: 'Preparing an empty benchmark workspace' });++ }const run = harness.run({prompt: job.prompt, task: 'edit', agent: job.agent, model: job.model, cwd: work,permission: job.permission, handoff: !job.agent, maxAttempts: 3,@@@ -314,9 +329,13 @@ async function runJob({ input, harness, emit, writeResult, env = process.env, tmif (e.type === 'system' && e.subtype === 'init' && e.agent && !result.agents.includes(e.agent)) result.agents.push(e.agent);}result.ok = !!(last && last.type === 'result' && !last.is_error);-- result.summary = last && typeof last.result === 'string' ? last.result.slice(0, 4000) : '';++ const answer = last && typeof last.result === 'string' ? last.result : '';++ result.summary = answer.slice(0, 4000);++ if (job.kind === 'bench') result.output = answer.slice(0, 200000);result.cost_usd = last && last.total_cost_usd ? last.total_cost_usd : 0;-- Object.assign(result, await changes(work));++ const usage = last && last.usage && typeof last.usage === 'object' ? last.usage : {};++ result.tokens = Number(last && (last.total_tokens || last.tokens)) || (Number(usage.input_tokens) || 0) + (Number(usage.output_tokens) || 0);++ Object.assign(result, job.kind === 'bench' ? { files: [], patch: '', patch_truncated: false } : await changes(work));say({ t: 'step', text: result.files.length ? `${result.files.length} file${result.files.length === 1 ? '' : 's'} changed` : 'No files changed' });// The proof of the change: the tests it reaches, the pages it touched, the diff, and a second model's opinion.// It is collected here, in the clone the agent worked in, so the evidence is about this machine's work.@@@ -36,6 +36,7 @@ const { createRunTasks } = require('./domain/run-tasks');const { createConnections } = require('./domain/connections');const { createRunTokens } = require('./domain/run-tokens');const { createAttempts } = require('./domain/attempts');++const { createBenchPublisher } = require('./domain/bench-publisher');const { createSessionLines } = require('./domain/session-lines');const { createOutbox, createPublisher } = require('./events/outbox');const { createEventsConsumer } = require('./events-consumer');@@@ -76,8 +77,9 @@ async function createApp(opts = {}) {const runTokens = createRunTokens({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });// The task's attempts (stage C2): every Run job is one attempt, and the engine decides what happens next.const attempts = createAttempts({ store, config, log, harnesses });-- const tasks = createRunTasks({ store, run, config, log, sessions, outbox, runTokens, connections, attempts });-- const ctx = { config, store, keys, sso, harnesses, log, sessions, outbox, run, tasks, connections, runTokens, attempts };++ const benchPublisher = createBenchPublisher({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });++ const tasks = createRunTasks({ store, run, config, log, sessions, outbox, runTokens, connections, attempts, benchAttemptEnded: benchPublisher.onTaskTerminal });++ const ctx = { config, store, keys, sso, harnesses, log, sessions, outbox, run, tasks, connections, runTokens, attempts, benchPublisher };ctx.forgejo = createForgejoClient({ config, log });const app = express();@@@ -86,6 +86,12 @@ function load(env = process.env) {'OPENVIBE_AI_TOKEN', 'OPENVIBE_CODES_BASE_URL', 'OPENVIBE_CODES_API_KEY', 'OPENVIBE_CODES_MODEL','OPENVIBE_CODES_HEADERS'],},++ bench: {++ url: trim(env.CODES_BENCH_URL || 'https://bench.openvibe.services'),++ renderOrigin: trim(env.CODES_BENCH_RENDER_ORIGIN || 'https://bench-render.openvibe.services'),++ dailyTasks: Math.max(1, int(env.CODES_BENCH_DAILY_TASKS, 50)),++ maxMatrixSize: Math.max(1, int(env.CODES_BENCH_MAX_MATRIX, 100)),++ },// OpenVibe.Events (ADR-033): account export and deletion arrive at POST /internal/events, signed with one of// these; run.job.* subscriptions are created at boot. The outbox relay publishes codes.task.* with the// service's events.event.publish token.nnew file mode 100644@@@ -0,0 +1,82 @@++'use strict';++++const { createServiceTokenClient } = require('openvibe-sdk/auth');++++const object = (v) => {++ if (v && typeof v === 'object' && !Array.isArray(v)) return v;++ if (typeof v === 'string') { try { return JSON.parse(v); } catch { return null; } }++ return null;++};++const normalize = (v) => String(v == null ? '' : v).replace(/\s+/g, ' ').trim();++++function grade(expected, output) {++ if (expected == null) return null;++ if (typeof expected === 'string') return normalize(expected) === normalize(output) ? 100 : 0;++ if (!expected || typeof expected !== 'object' || Array.isArray(expected)) return null;++ if (typeof expected.exactly === 'string') return normalize(expected.exactly) === normalize(output) ? 100 : 0;++ if (typeof expected.contains === 'string') return String(output || '').includes(expected.contains) ? 100 : 0;++ if (typeof expected.not_contains === 'string') return String(output || '').includes(expected.not_contains) ? 0 : 100;++ return null;++}++++function createBenchPublisher({ config, store, fetchImpl = globalThis.fetch, log = console }) {++ const tokens = config.oauth.clientSecret ? createServiceTokenClient({++ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId,++ clientSecret: config.oauth.clientSecret, audience: 'openvibe.bench', scope: 'bench.run.publish', fetch: fetchImpl,++ }) : null;++ const publishing = new Set();++++ async function onTaskTerminal(row) {++ const actorRef = object(row && row.actor_ref);++ const meta = actorRef && actorRef.bench;++ if (!meta || !meta.run_id || !meta.task_count || !tokens) return;++ const key = String(meta.run_id);++ if (publishing.has(key)) return;++ const siblings = await store.db.many(++ "SELECT * FROM run_tasks WHERE subject = $1 AND actor_ref->'bench'->>'run_id' = $2 ORDER BY created_at, id",++ [row.subject, key]);++ if (siblings.length < Number(meta.task_count) || siblings.some((x) => !['succeeded', 'failed', 'cancelled', 'expired'].includes(x.state))) return;++ publishing.add(key);++ try {++ const details = siblings.map((taskRow) => {++ const taskMeta = (object(taskRow.actor_ref) || {}).bench || {};++ const result = object(taskRow.result) || {};++ const automatic = (taskMeta.judges || []).some((j) => ['test_suite', 'automatic_grader'].includes(j.type));++ const score = automatic ? (grade(taskMeta.expected, result.output) ?? 0) : null;++ const taskId = String(taskMeta.task_id || taskRow.id);++ const renderKey = `${meta.run_id}~${Buffer.from(taskId).toString('base64url')}`;++ const base = String(config.bench.renderOrigin).replace(/\/+$/, '');++ return {++ task_id: taskId, score,++ cost_usd: Number(result.cost_usd) || 0,++ ms: Number(result.duration_ms) || 0,++ tokens: Number(result.tokens) || 0,++ artifacts: [{ kind: String(taskMeta.kind || 'text'), url: `${base}/render/w/${renderKey}`, payload: { text: String(result.output || result.summary || taskRow.error || '').slice(0, 200000) } }],++ transcript_url: `${String(config.baseUrl).replace(/\/+$/, '')}/run/tasks/${taskRow.id}`,++ };++ });++ const scoreRows = details.filter((t) => t.score != null);++ const score = scoreRows.length === details.length ? scoreRows.reduce((n, t) => n + t.score, 0) / scoreRows.length : null;++ const body = {++ id: meta.run_id, benchmark: meta.benchmark, model: meta.model, provider: meta.provider || 'unknown',++ route: { provider: meta.provider || 'unknown', model: meta.model },++ harness: { name: meta.harness, version: meta.harness_version || 'unknown' }, settings: {},++ executor: { kind: 'codes', id: String(row.id) }, tasks: details, score,++ cost_usd: details.reduce((n, t) => n + t.cost_usd, 0), ms: details.reduce((n, t) => n + t.ms, 0),++ tokens: details.reduce((n, t) => n + t.tokens, 0), provenance: 'verified', session: meta.session,++ };++ const response = await fetchImpl(`${String(config.bench.url).replace(/\/+$/, '')}/api/v1/runs`, {++ method: 'POST', headers: { ...(await tokens.authHeaders()), 'Content-Type': 'application/json' },++ body: JSON.stringify(body), signal: AbortSignal.timeout(20000),++ });++ if (!response.ok) throw new Error(`Bench answered ${response.status}`);++ } catch (e) {++ publishing.delete(key);++ throw e;++ }++ }++++ return { onTaskTerminal };++}++++module.exports = { createBenchPublisher, grade };nnew file mode 100644@@@ -0,0 +1,18 @@++'use strict';++++/** Expand a frozen benchmark into one output attempt for every task × model × harness × repetition. */++function expandMatrix(definition, models, harnesses, repeat) {++ const tasks = Array.isArray(definition && definition.tasks) ? definition.tasks : [];++ const cells = [];++ for (const model of models) for (const harness of harnesses) for (let repetition = 1; repetition <= repeat; repetition++) {++ for (const task of tasks) {++ if (!task || typeof task.id !== 'string' || typeof task.prompt !== 'string' || !task.prompt.trim()) {++ throw Object.assign(new Error('the executor definition must reveal every benchmark task prompt'), { code: 'codes.bench_holdout_hidden' });++ }++ cells.push({ task, model, harness, repetition });++ }++ }++ return cells;++}++++module.exports = { expandMatrix };@@@ -52,7 +52,11 @@ function createRunClient({ config, fetchImpl = globalThis.fetch, now = () => Datasync function submit({ subject, project, args, key }) {const body = {class: 'code', artifact: { name: 'openvibe-codes', version: JOB_PROTOCOL }, args,-- limits: { wall_ms: 30 * 60_000, cpu_ms: 30 * 60_000, mem_bytes: 4 * 1024 ** 3 },++ limits: {++ wall_ms: Math.max(1000, Math.min(30 * 60_000, Number(args.limits && args.limits.ms) || 30 * 60_000)),++ cpu_ms: Math.max(1000, Math.min(30 * 60_000, Number(args.limits && args.limits.ms) || 30 * 60_000)),++ mem_bytes: 4 * 1024 ** 3,++ },ttl_ms: 45 * 60_000, egress: 'public',requirements: { kind: 'run.code', mobility: 'job', latency_class: 'background', objective: 'balanced', trust: ['user-owned'] },};@@@ -140,7 +140,7 @@ function usageOf(result) {};}--function createRunTasks({ store, run, config, log = console, sessions = null, outbox = null, runTokens = null, connections = null, attempts = null, proof = null }) {++function createRunTasks({ store, run, config, log = console, sessions = null, outbox = null, runTokens = null, connections = null, attempts = null, proof = null, benchAttemptEnded = null }) {const db = store.db;const who = (row) => ({ subject: row.subject, project: row.project_id });// Lazily, so a caller that only reads tasks does not need the session/outbox modules (and there is no cycle).@@@ -218,7 +218,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;-- return 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]);if (!current) return row;if (RANK[state] < RANK[current.state]) return current;@@@ -264,6 +264,11 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou}return updated;});++ 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)}`));++ }++ return saved;}/**@@@ -288,7 +293,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ouawait store.tx(async () => {inserted = await db.one(`INSERT INTO run_tasks (id, subject, project_id, repo, ref, prompt, agent, push, pull_request, state, via, idempotency_key, actor_ref, created_at, updated_at)VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'queued', $10, $11, $12::jsonb, $13, $13) RETURNING *`,-- [id, subject, project, args.repo, args.ref, args.prompt, args.agent, !!args.push, !!(args.push && args.push.pull_request), via, key, actorRefValue, at]);++ [id, subject, project, args.repo || 'bench://', args.ref, args.prompt, args.agent, !!args.push, !!(args.push && args.push.pull_request), via, key, actorRefValue, at]);await taskOutbox.emitTask(inserted, null);return inserted;});@@@ -388,6 +393,12 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ouif (!row) return null;if (!TERMINAL.has(row.state)) return row;if (row.outcome === 'stopped') return row; // a person stopped it: nothing carries it on++ const actorRef = asObject(row.actor_ref);++ if (actorRef && actorRef.bench) {++ const done = row.state === 'succeeded';++ 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()]);++ }const attempts = await attemptStore.list(row.id);const ending = attempts[attempts.length - 1];if (!ending || !ENDED.has(ending.state)) return row;@@@ -33,6 +33,8 @@ const { asyncRouter } = require('./router');const { parseArgs } = require('../../harness/job');const { createRunTasks, TERMINAL } = require('../domain/run-tasks');const { createTaskEvents } = require('./task-events');++const { createServiceTokenClient } = require('openvibe-sdk/auth');++const { expandMatrix } = require('../domain/bench');const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;const TASK = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;@@@ -44,6 +46,10 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, taskconst taskEvents = createTaskEvents({ store, config, log, attempts: taskStore.attempts });const problem = (req, res, status, code, detail) => http.sendProblem(res, status, code, { detail, ctx: req.ov });const json = express.json({ limit: '32kb' });++ const benchTokens = config.oauth.clientSecret ? createServiceTokenClient({++ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId,++ clientSecret: config.oauth.clientSecret, audience: 'openvibe.bench', scope: 'bench.run.publish',++ }) : null;/** The route's guard: the service token, its capability and the person it acts for → req.principal. */const guard = (cap) => async (req, res, next) => {@@@ -174,6 +180,84 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, taskreturn res.status(201).json({ task: taskStore.present(row), created: true });});++ // Bench is the only caller: it supplies a service token for Codes plus the signed-in user's short-lived++ // Network token in a private server-to-server header. The latter is used only to provision the user's own project.++ r.post('/benchmarks/:id/run', guardAny('codes.task.start'), json, async (req, res) => {++ const p = req.principal;++ if (p.owner) return problem(req, res, 403, 'codes.bench_service_required', 'Bench starts verified benchmark tasks.');++ const userToken = String(req.get('x-ov-user-token') || '').trim();++ const user = await keys.verifyUser(userToken, { issuer: config.networkIssuer, audience: config.oauth.sessionAudience });++ if (!user.ok || user.claims.subject_id !== p.subject) return problem(req, res, 403, 'codes.bench_subject_mismatch', 'The signed-in person must match X-OV-Subject.');++ if (!benchTokens) return problem(req, res, 503, 'codes.bench_unconfigured', 'Codes has no Network client credentials for Bench.');++ const b = req.body || {};++ const models = Array.isArray(b.models) ? b.models.map(String) : [];++ const requestedHarnesses = Array.isArray(b.harnesses) ? b.harnesses.map(String) : [];++ const repeat = Number(b.repeat);++ if (!models.length || !requestedHarnesses.length || !Number.isInteger(repeat) || repeat < 1 || repeat > 10++ || models.some((m) => !m || m.length > 120) || new Set(requestedHarnesses).size !== requestedHarnesses.length) {++ return problem(req, res, 422, 'codes.bench_invalid', 'models and harnesses are required; repeat must be 1–10.');++ }++ const available = new Set(harnesses.list().map((h) => h.id));++ if (requestedHarnesses.some((h) => !available.has(h))) return problem(req, res, 422, 'codes.bench_harness_unknown', 'A harness is not in GET /api/v1/harnesses.');++ let definition;++ try {++ const response = await fetch(`${config.bench.url}/api/v1/benchmarks/${encodeURIComponent(String(req.params.id))}`, {++ headers: { ...(await benchTokens.authHeaders()), Accept: 'application/json' }, signal: AbortSignal.timeout(15000),++ });++ const body = await response.json().catch(() => ({}));++ if (!response.ok) return problem(req, res, response.status, body.code || 'codes.bench_definition_failed', body.detail || 'Bench did not return the benchmark definition.');++ definition = body.definition;++ } catch (e) { return problem(req, res, 503, 'codes.bench_unavailable', `Bench did not answer: ${e.message}`); }++ let matrix;++ try { matrix = expandMatrix(definition, models, requestedHarnesses, repeat); }++ catch (e) { return problem(req, res, 422, e.code || 'codes.bench_empty', e.message); }++ if (!matrix.length) return problem(req, res, 422, 'codes.bench_empty', 'The benchmark has no tasks.');++ const total = matrix.length;++ if (total > config.bench.maxMatrixSize) return problem(req, res, 422, 'codes.bench_matrix_too_large', `This matrix has ${total} tasks; the limit is ${config.bench.maxMatrixSize}.`);++ const running = Number(await db.value("SELECT COUNT(*)::int AS n FROM (SELECT actor_ref->'bench'->>'session' AS session FROM run_tasks WHERE subject = $1 AND actor_ref ? 'bench' GROUP BY actor_ref->'bench'->>'session' HAVING bool_or(state NOT IN ('succeeded','failed','cancelled','expired'))) active", [p.subject]));++ if (running >= 3) return problem(req, res, 429, 'codes.bench_concurrency_limit', 'A person may have at most three benchmark runs active.');++ const today = store.now() - (store.now() % 86400000);++ const daily = Number(await db.value("SELECT COUNT(*)::int AS n FROM run_tasks WHERE subject = $1 AND actor_ref ? 'bench' AND created_at >= $2", [p.subject, today]));++ if (daily + total > config.bench.dailyTasks) return problem(req, res, 429, 'codes.bench_daily_limit', `The daily benchmark task cap is ${config.bench.dailyTasks}.`);++++ let project;++ try {++ const headers = { Authorization: `Bearer ${userToken}`, Accept: 'application/json' };++ const projectsResponse = await fetch(`${config.networkInternalUrl}/api/v1/projects`, { headers, signal: AbortSignal.timeout(8000) });++ if (!projectsResponse.ok) throw new Error(`Network answered ${projectsResponse.status}`);++ const projects = (await projectsResponse.json()).projects || [];++ project = projects.find((x) => x.name === 'bench' && !x.archived_at)?.id;++ if (!project) {++ const created = await fetch(`${config.networkInternalUrl}/api/v1/projects`, {++ method: 'POST', headers: { ...headers, 'Content-Type': 'application/json' }, body: JSON.stringify({ name: 'bench' }), signal: AbortSignal.timeout(8000),++ });++ if (!created.ok) throw new Error(`Network answered ${created.status}`);++ project = (await created.json()).id;++ }++ } catch (e) { return problem(req, res, 502, 'codes.bench_project_failed', `Could not get or create the person's bench project: ${e.message}`); }++++ const session = `${req.params.id}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`;++ const started = [];++ const runIds = new Map();++ try {++ for (const cell of matrix) {++ const { task, model, harness, repetition } = cell;++ const key = `${model}\u0000${harness}\u0000${repetition}`;++ if (!runIds.has(key)) runIds.set(key, store.newId('brn'));++ const runId = runIds.get(key);++ const harnessOffer = harnesses.list().find((h) => h.id === harness);++ const inputs = Array.isArray(task.files) ? JSON.stringify(task.files).slice(0, 3000) : '';++ const prompt = `Benchmark ${b.benchmark || req.params.id}, task ${task.id}.\n\n${String(task.prompt).slice(0, 10000)}${inputs ? `\n\nInput files:\n${inputs}` : ''}\n\nReturn only the requested artifact content. Respect the per-task limits: ${Number(definition.limits && definition.limits.ms) || 0} ms, ${Number(definition.limits && definition.limits.tokens) || 0} tokens, $${Number(definition.limits && definition.limits.usd) || 0}.`;++ const limits = definition.limits || {};++ const args = parseArgs(JSON.stringify({ kind: 'bench', prompt, agent: harness, model, limits }));++ const taskRow = await taskStore.start({ subject: p.subject, project, args, via: p.via,++ actorRef: { bench: { run_id: runId, task_count: matrix.filter((x) => x.model === model && x.harness === harness && x.repetition === repetition).length, benchmark: `${req.params.id}@${Number(definition.version) || 1}`, session, model, provider: (harnessOffer && harnessOffer.provider) || String(model).split('/')[0] || 'unknown', harness, harness_version: (harnessOffer && harnessOffer.version) || 'unknown', repetition, task_id: String(task.id), kind: definition.kind, expected: task.expected, judges: definition.judges || [], scoring: definition.scoring || {}, limits: definition.limits || {} } } });++ started.push(taskStore.present(taskRow));++ }++ } catch (e) { return problem(req, res, 502, 'codes.bench_start_failed', `Codes started ${started.length} of ${total} tasks: ${e.message}`); }++ return res.status(202).json({ session, tasks: started, task: started[0] || null, total: started.length });++ });++/*** The caller's tasks, newest first: the person's own (whoever started them), or the ones this service started for* the person. ?state=narrows it to the words `openvibe-codes task list` prints: running (queued, placed, running),nnew file mode 100644@@@ -0,0 +1,27 @@++'use strict';++const assert = require('assert');++const { expandMatrix } = require('../server/domain/bench');++const { grade } = require('../server/domain/bench-publisher');++const { check, done } = require('./helpers/boot');++++(async () => {++ const def = { tasks: [{ id: 'public', prompt: 'Make a page' }, { id: 'holdout', prompt: 'Make a game', holdout: true }] };++ await check('matrix expands each revealed task across models, harnesses and repetitions', () => {++ const matrix = expandMatrix(def, ['m1', 'm2'], ['h1', 'h2'], 3);++ assert.strictEqual(matrix.length, 24);++ assert.deepStrictEqual(matrix[0], { task: def.tasks[0], model: 'm1', harness: 'h1', repetition: 1 });++ assert.strictEqual(matrix.at(-1).repetition, 3);++ });++ await check('a withheld task cannot enter the executor matrix', () => {++ assert.throws(() => expandMatrix({ tasks: [{ id: 'secret', prompt: null, holdout: true }] }, ['m'], ['h'], 1),++ (e) => e.code === 'codes.bench_holdout_hidden');++ });++ await check('automatic expected-output judges grade exact and contains checks', () => {++ assert.strictEqual(grade(' a\npage ', 'a page'), 100);++ assert.strictEqual(grade({ contains: '<main>' }, '<main>ok</main>'), 100);++ assert.strictEqual(grade({ not_contains: 'script' }, '<p>safe</p>'), 100);++ assert.strictEqual(grade('expected', 'actual'), 0);++ assert.strictEqual(grade(null, 'anything'), null);++ });++ done();++})().catch((e) => { process.stderr.write(`${e.stack || e}\n`); process.exit(1); });@@@ -63,6 +63,23 @@ function fakeHarness({ fail = false } = {}) {assert.strictEqual(parseArgs(JSON.stringify({ repo: 'file:///tmp/x', prompt: 'x' }), env).repo, 'file:///tmp/x');});++ await check('a bench job accepts no repository and returns the model output without a repository diff', async () => {++ const parsed = parseArgs(JSON.stringify({ kind: 'bench', prompt: 'Return a small HTML artifact', agent: 'claude-code', model: 'cheap-model', limits: { ms: 5000, tokens: 10000, usd: 0.05 } }));++ assert.strictEqual(parsed.repo, null);++ assert.strictEqual(parsed.kind, 'bench');++ assert.deepStrictEqual(parsed.limits, { ms: 5000, tokens: 10000, usd: 0.05 });++ const docs = [];++ const harness = fakeHarness();++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'codes-bench-tmp-'));++ const doc = await runJob({ input: JSON.stringify({ kind: 'bench', prompt: 'Return a small HTML artifact' }), harness,++ emit: () => {}, writeResult: (d) => docs.push(d), env, tmp });++ assert.strictEqual(doc.kind, 'bench');++ assert.strictEqual(doc.output, 'Done: README updated.');++ assert.deepStrictEqual(doc.files, []);++ assert.strictEqual(docs.length, 1);++ fs.rmSync(tmp, { recursive: true, force: true });++ });++await check('a task clones, runs the agent there, streams short lines and hands over the patch once', async () => {const src = repository();const lines = [];@@@ -43,9 +43,11 @@ const INTERNAL = [`http://127.0.0.1:3000${PROBE}`, `http://2130706433${PROBE}`,'server/auth/sso.js': 'Network OAuth token and revoke (configured)','server/domain/forgejo.js': 'OpenVibe Git API (CODES_GIT_INTERNAL_URL, configured loopback Forgejo) with a service token; paths built from validated owner/name','server/domain/run-client.js': 'OpenVibe.Run (CODES_RUN_URL, configured) and the Network token endpoint; job ids only, never a URL from a person',++ 'server/domain/bench-publisher.js': 'OpenVibe.Bench (CODES_BENCH_URL, configured) run publishing and the Network token endpoint; artifact URLs use the configured render origin','server/http/run.js': 'Network projects API (configured) with the viewer\'s token, and the Run job stream URL run-client.js builds on its own base','server/http/home.js': 'OpenVibe Git\'s public repository search (CODES_GIT_INTERNAL_URL, configured loopback Forgejo) for the home page; no token','server/http/me-api.js': 'Network projects API (configured) with the viewer\'s token, for the user task API; no URL from a person',++ 'server/http/tasks-api.js': 'OpenVibe.Bench (CODES_BENCH_URL, configured), Network projects API (configured) with the verified user token; benchmark ids and task ids are URL-encoded, no URL from a person','server/http/pages.js': 'Network projects API (configured) for Improve OpenVibe project creation; Forgejo repo read (CODES_GIT_INTERNAL_URL) for repo details; no URL from a person','server/domain/connections.js': 'OpenVibe.AI (CODES_AI_URL, configured) for a person\'s own keys: the credential routes and the two model endpoints, with the provider and the subject as path segments; no URL from a person','server/domain/run-tokens.js': 'OpenVibe.Network\'s token endpoint (OV_NETWORK_INTERNAL_URL, configured) for a run token; no URL from a person',
@@ -95,11 +95,14 @@ function parseArgs(raw, env = process.env) {
@@ -95,11 +95,14 @@ function parseArgs(raw, env = process.env) {
let a;
let a;
try { a = JSON.parse(String(raw || '').trim() || '{}'); } catch { throw new Error('the job input is not JSON'); }
try { a = JSON.parse(String(raw || '').trim() || '{}'); } catch { throw new Error('the job input is not JSON'); }
if (!a || typeof a !== 'object' || Array.isArray(a)) throw new Error('the job input must be an object');
if (!a || typeof a !== 'object' || Array.isArray(a)) throw new Error('the job input must be an object');
- let repo;
- try { repo = new URL(String(a.repo || '')); } catch { throw new Error('repo must be a repository URL'); }
- const fileOk = env.OPENVIBE_CODES_ALLOW_FILE_REPO === '1' && repo.protocol === 'file:';
- if (!fileOk && repo.protocol !== 'https:') throw new Error('repo must be an https URL');
- if (repo.username || repo.password) throw new Error('repo must not carry credentials');
+ const kind = a.kind === 'bench' ? 'bench' : 'code';
+ let repo = null;
+ if (kind === 'code') {
+ try { repo = new URL(String(a.repo || '')); } catch { throw new Error('repo must be a repository URL'); }
+ const fileOk = env.OPENVIBE_CODES_ALLOW_FILE_REPO === '1' && repo.protocol === 'file:';
+ if (!fileOk && repo.protocol !== 'https:') throw new Error('repo must be an https URL');
+ if (repo.username || repo.password) throw new Error('repo must not carry credentials');
+ }
const prompt = typeof a.prompt === 'string' ? a.prompt.trim() : '';
const prompt = typeof a.prompt === 'string' ? a.prompt.trim() : '';
if (!prompt || prompt.length > MAX_PROMPT) throw new Error(`prompt must be 1-${MAX_PROMPT} characters`);
if (!prompt || prompt.length > MAX_PROMPT) throw new Error(`prompt must be 1-${MAX_PROMPT} characters`);
const ref = a.ref == null || a.ref === '' ? null : String(a.ref);
const ref = a.ref == null || a.ref === '' ? null : String(a.ref);
@@ -118,8 +121,14 @@ function parseArgs(raw, env = process.env) {
@@ -118,8 +121,14 @@ function parseArgs(raw, env = process.env) {
const resumeIn = a.resume == null || a.resume === '' ? env.RESUME : a.resume;
const resumeIn = a.resume == null || a.resume === '' ? env.RESUME : a.resume;
const resume = typeof resumeIn === 'string' && /^[A-Za-z0-9._:-]{1,200}$/.test(resumeIn) ? resumeIn : null;
const resume = typeof resumeIn === 'string' && /^[A-Za-z0-9._:-]{1,200}$/.test(resumeIn) ? resumeIn : null;
const wanted = jobEnv(a);
const wanted = jobEnv(a);
+ const limits = kind === 'bench' && a.limits && typeof a.limits === 'object' ? {
+ ms: Number.isFinite(Number(a.limits.ms)) && Number(a.limits.ms) > 0 ? Math.min(300000, Number(a.limits.ms)) : null,
+ tokens: Number.isFinite(Number(a.limits.tokens)) && Number(a.limits.tokens) > 0 ? Math.min(1000000, Number(a.limits.tokens)) : null,
+ usd: Number.isFinite(Number(a.limits.usd)) && Number(a.limits.usd) > 0 ? Math.min(100, Number(a.limits.usd)) : null,
+ } : null;
return {
return {
- repo: repo.toString(), ref, prompt, permission, push,
+ kind, repo: repo ? repo.toString() : null, ref, prompt, permission, push,
+ ...(limits ? { limits } : {}),
...(resume ? { resume } : {}),
...(resume ? { resume } : {}),
...(Object.keys(wanted).length ? { env: wanted } : {}),
...(Object.keys(wanted).length ? { env: wanted } : {}),
agent: typeof a.agent === 'string' && /^[a-z0-9-]{1,40}$/.test(a.agent) ? a.agent : null,
agent: typeof a.agent === 'string' && /^[a-z0-9-]{1,40}$/.test(a.agent) ? a.agent : null,
@@ -292,13 +301,19 @@ async function runJob({ input, harness, emit, writeResult, env = process.env, tm
@@ -292,13 +301,19 @@ async function runJob({ input, harness, emit, writeResult, env = process.env, tm
}
}
const dir = fs.mkdtempSync(path.join(tmp, 'codes-'));
const dir = fs.mkdtempSync(path.join(tmp, 'codes-'));
const work = path.join(dir, 'repo');
const work = path.join(dir, 'repo');
- const result = { protocol: JOB_PROTOCOL, ok: false, repo: scrub(job.repo), ref: job.ref, agents: [], summary: '' };
+ const result = { protocol: JOB_PROTOCOL, ok: false, kind: job.kind, repo: job.repo ? scrub(job.repo) : null, ref: job.ref, agents: [], summary: '' };
+ let base = null;
try {
try {
- say({ t: 'step', text: `Cloning ${scrub(job.repo)}${job.ref ? ` (${job.ref})` : ''}` });
- await git(['clone', '--depth', '50', ...(job.ref ? ['--branch', job.ref] : []), '--', job.repo, work], { timeout: 300_000 });
- const base = (await git(['rev-parse', '--abbrev-ref', 'HEAD'], { cwd: work })).trim();
- result.base = base;
- say({ t: 'step', text: `Working on ${base}` });
+ if (job.kind === 'code') {
+ say({ t: 'step', text: `Cloning ${scrub(job.repo)}${job.ref ? ` (${job.ref})` : ''}` });
+ await git(['clone', '--depth', '50', ...(job.ref ? ['--branch', job.ref] : []), '--', job.repo, work], { timeout: 300_000 });
+ base = (await git(['rev-parse', '--abbrev-ref', 'HEAD'], { cwd: work })).trim();
+ result.base = base;
+ say({ t: 'step', text: `Working on ${base}` });
+ } else {
+ fs.mkdirSync(work, { recursive: true });
+ say({ t: 'step', text: 'Preparing an empty benchmark workspace' });
+ }
const run = harness.run({
const run = harness.run({
prompt: job.prompt, task: 'edit', agent: job.agent, model: job.model, cwd: work,
prompt: job.prompt, task: 'edit', agent: job.agent, model: job.model, cwd: work,
permission: job.permission, handoff: !job.agent, maxAttempts: 3,
permission: job.permission, handoff: !job.agent, maxAttempts: 3,
@@ -314,9 +329,13 @@ async function runJob({ input, harness, emit, writeResult, env = process.env, tm
@@ -314,9 +329,13 @@ async function runJob({ input, harness, emit, writeResult, env = process.env, tm
if (e.type === 'system' && e.subtype === 'init' && e.agent && !result.agents.includes(e.agent)) result.agents.push(e.agent);
if (e.type === 'system' && e.subtype === 'init' && e.agent && !result.agents.includes(e.agent)) result.agents.push(e.agent);
}
}
result.ok = !!(last && last.type === 'result' && !last.is_error);
result.ok = !!(last && last.type === 'result' && !last.is_error);
- result.summary = last && typeof last.result === 'string' ? last.result.slice(0, 4000) : '';
+ const answer = last && typeof last.result === 'string' ? last.result : '';
+ result.summary = answer.slice(0, 4000);
+ if (job.kind === 'bench') result.output = answer.slice(0, 200000);
result.cost_usd = last && last.total_cost_usd ? last.total_cost_usd : 0;
result.cost_usd = last && last.total_cost_usd ? last.total_cost_usd : 0;
- Object.assign(result, await changes(work));
+ const usage = last && last.usage && typeof last.usage === 'object' ? last.usage : {};
+ result.tokens = Number(last && (last.total_tokens || last.tokens)) || (Number(usage.input_tokens) || 0) + (Number(usage.output_tokens) || 0);
+ Object.assign(result, job.kind === 'bench' ? { files: [], patch: '', patch_truncated: false } : await changes(work));
say({ t: 'step', text: result.files.length ? `${result.files.length} file${result.files.length === 1 ? '' : 's'} changed` : 'No files changed' });
say({ t: 'step', text: result.files.length ? `${result.files.length} file${result.files.length === 1 ? '' : 's'} changed` : 'No files changed' });
// The proof of the change: the tests it reaches, the pages it touched, the diff, and a second model's opinion.
// The proof of the change: the tests it reaches, the pages it touched, the diff, and a second model's opinion.
// It is collected here, in the clone the agent worked in, so the evidence is about this machine's work.
// It is collected here, in the clone the agent worked in, so the evidence is about this machine's work.
@@ -36,6 +36,7 @@ const { createRunTasks } = require('./domain/run-tasks');
@@ -36,6 +36,7 @@ const { createRunTasks } = require('./domain/run-tasks');
const { createConnections } = require('./domain/connections');
const { createConnections } = require('./domain/connections');
const { createRunTokens } = require('./domain/run-tokens');
const { createRunTokens } = require('./domain/run-tokens');
const { createAttempts } = require('./domain/attempts');
const { createAttempts } = require('./domain/attempts');
+const { createBenchPublisher } = require('./domain/bench-publisher');
const { createSessionLines } = require('./domain/session-lines');
const { createSessionLines } = require('./domain/session-lines');
const { createOutbox, createPublisher } = require('./events/outbox');
const { createOutbox, createPublisher } = require('./events/outbox');
const { createEventsConsumer } = require('./events-consumer');
const { createEventsConsumer } = require('./events-consumer');
@@ -76,8 +77,9 @@ async function createApp(opts = {}) {
@@ -76,8 +77,9 @@ async function createApp(opts = {}) {
const runTokens = createRunTokens({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });
const runTokens = createRunTokens({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });
// The task's attempts (stage C2): every Run job is one attempt, and the engine decides what happens next.
// The task's attempts (stage C2): every Run job is one attempt, and the engine decides what happens next.
const attempts = createAttempts({ store, config, log, harnesses });
const attempts = createAttempts({ store, config, log, harnesses });
- const tasks = createRunTasks({ store, run, config, log, sessions, outbox, runTokens, connections, attempts });
- const ctx = { config, store, keys, sso, harnesses, log, sessions, outbox, run, tasks, connections, runTokens, attempts };
+ const benchPublisher = createBenchPublisher({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });
+ const tasks = createRunTasks({ store, run, config, log, sessions, outbox, runTokens, connections, attempts, benchAttemptEnded: benchPublisher.onTaskTerminal });
+ const ctx = { config, store, keys, sso, harnesses, log, sessions, outbox, run, tasks, connections, runTokens, attempts, benchPublisher };
ctx.forgejo = createForgejoClient({ config, log });
ctx.forgejo = createForgejoClient({ config, log });
const app = express();
const app = express();
@@ -86,6 +86,12 @@ function load(env = process.env) {
@@ -86,6 +86,12 @@ function load(env = process.env) {
'OPENVIBE_AI_TOKEN', 'OPENVIBE_CODES_BASE_URL', 'OPENVIBE_CODES_API_KEY', 'OPENVIBE_CODES_MODEL',
'OPENVIBE_AI_TOKEN', 'OPENVIBE_CODES_BASE_URL', 'OPENVIBE_CODES_API_KEY', 'OPENVIBE_CODES_MODEL',
'OPENVIBE_CODES_HEADERS'],
'OPENVIBE_CODES_HEADERS'],
},
},
+ bench: {
+ url: trim(env.CODES_BENCH_URL || 'https://bench.openvibe.services'),
+ renderOrigin: trim(env.CODES_BENCH_RENDER_ORIGIN || 'https://bench-render.openvibe.services'),
+ dailyTasks: Math.max(1, int(env.CODES_BENCH_DAILY_TASKS, 50)),
+ maxMatrixSize: Math.max(1, int(env.CODES_BENCH_MAX_MATRIX, 100)),
+ },
// OpenVibe.Events (ADR-033): account export and deletion arrive at POST /internal/events, signed with one of
// OpenVibe.Events (ADR-033): account export and deletion arrive at POST /internal/events, signed with one of
// these; run.job.* subscriptions are created at boot. The outbox relay publishes codes.task.* with the
// these; run.job.* subscriptions are created at boot. The outbox relay publishes codes.task.* with the
// service's events.event.publish token.
// service's events.event.publish token.
new file mode 100644
new file mode 100644
@@ -0,0 +1,82 @@
@@ -0,0 +1,82 @@
+'use strict';
+
+const { createServiceTokenClient } = require('openvibe-sdk/auth');
+
+const object = (v) => {
+ if (v && typeof v === 'object' && !Array.isArray(v)) return v;
+ if (typeof v === 'string') { try { return JSON.parse(v); } catch { return null; } }
+ return null;
+};
+const normalize = (v) => String(v == null ? '' : v).replace(/\s+/g, ' ').trim();
+
+function grade(expected, output) {
+ if (expected == null) return null;
+ if (typeof expected === 'string') return normalize(expected) === normalize(output) ? 100 : 0;
+ if (!expected || typeof expected !== 'object' || Array.isArray(expected)) return null;
+ if (typeof expected.exactly === 'string') return normalize(expected.exactly) === normalize(output) ? 100 : 0;
+ if (typeof expected.contains === 'string') return String(output || '').includes(expected.contains) ? 100 : 0;
+ if (typeof expected.not_contains === 'string') return String(output || '').includes(expected.not_contains) ? 0 : 100;
+ return null;
+}
+
+function createBenchPublisher({ config, store, fetchImpl = globalThis.fetch, log = console }) {
+ const tokens = config.oauth.clientSecret ? createServiceTokenClient({
+ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId,
+ clientSecret: config.oauth.clientSecret, audience: 'openvibe.bench', scope: 'bench.run.publish', fetch: fetchImpl,
+ }) : null;
+ const publishing = new Set();
+
+ async function onTaskTerminal(row) {
+ const actorRef = object(row && row.actor_ref);
+ const meta = actorRef && actorRef.bench;
+ if (!meta || !meta.run_id || !meta.task_count || !tokens) return;
+ const key = String(meta.run_id);
+ if (publishing.has(key)) return;
+ const siblings = await store.db.many(
+ "SELECT * FROM run_tasks WHERE subject = $1 AND actor_ref->'bench'->>'run_id' = $2 ORDER BY created_at, id",
+ [row.subject, key]);
+ if (siblings.length < Number(meta.task_count) || siblings.some((x) => !['succeeded', 'failed', 'cancelled', 'expired'].includes(x.state))) return;
+ publishing.add(key);
+ try {
+ const details = siblings.map((taskRow) => {
+ const taskMeta = (object(taskRow.actor_ref) || {}).bench || {};
+ const result = object(taskRow.result) || {};
+ const automatic = (taskMeta.judges || []).some((j) => ['test_suite', 'automatic_grader'].includes(j.type));
+ const score = automatic ? (grade(taskMeta.expected, result.output) ?? 0) : null;
+ const taskId = String(taskMeta.task_id || taskRow.id);
+ const renderKey = `${meta.run_id}~${Buffer.from(taskId).toString('base64url')}`;
+ const base = String(config.bench.renderOrigin).replace(/\/+$/, '');
+ return {
+ task_id: taskId, score,
+ cost_usd: Number(result.cost_usd) || 0,
+ ms: Number(result.duration_ms) || 0,
+ tokens: Number(result.tokens) || 0,
+ artifacts: [{ kind: String(taskMeta.kind || 'text'), url: `${base}/render/w/${renderKey}`, payload: { text: String(result.output || result.summary || taskRow.error || '').slice(0, 200000) } }],
+ transcript_url: `${String(config.baseUrl).replace(/\/+$/, '')}/run/tasks/${taskRow.id}`,
+ };
+ });
+ const scoreRows = details.filter((t) => t.score != null);
+ const score = scoreRows.length === details.length ? scoreRows.reduce((n, t) => n + t.score, 0) / scoreRows.length : null;
+ const body = {
+ id: meta.run_id, benchmark: meta.benchmark, model: meta.model, provider: meta.provider || 'unknown',
+ route: { provider: meta.provider || 'unknown', model: meta.model },
+ harness: { name: meta.harness, version: meta.harness_version || 'unknown' }, settings: {},
+ executor: { kind: 'codes', id: String(row.id) }, tasks: details, score,
+ cost_usd: details.reduce((n, t) => n + t.cost_usd, 0), ms: details.reduce((n, t) => n + t.ms, 0),
+ tokens: details.reduce((n, t) => n + t.tokens, 0), provenance: 'verified', session: meta.session,
+ };
+ const response = await fetchImpl(`${String(config.bench.url).replace(/\/+$/, '')}/api/v1/runs`, {
+ method: 'POST', headers: { ...(await tokens.authHeaders()), 'Content-Type': 'application/json' },
+ body: JSON.stringify(body), signal: AbortSignal.timeout(20000),
+ });
+ if (!response.ok) throw new Error(`Bench answered ${response.status}`);
+ } catch (e) {
+ publishing.delete(key);
+ throw e;
+ }
+ }
+
+ return { onTaskTerminal };
+}
+
+module.exports = { createBenchPublisher, grade };
new file mode 100644
new file mode 100644
@@ -0,0 +1,18 @@
@@ -0,0 +1,18 @@
+'use strict';
+
+/** Expand a frozen benchmark into one output attempt for every task × model × harness × repetition. */
+function expandMatrix(definition, models, harnesses, repeat) {
+ const tasks = Array.isArray(definition && definition.tasks) ? definition.tasks : [];
+ const cells = [];
+ for (const model of models) for (const harness of harnesses) for (let repetition = 1; repetition <= repeat; repetition++) {
+ for (const task of tasks) {
+ if (!task || typeof task.id !== 'string' || typeof task.prompt !== 'string' || !task.prompt.trim()) {
+ throw Object.assign(new Error('the executor definition must reveal every benchmark task prompt'), { code: 'codes.bench_holdout_hidden' });
+ }
+ cells.push({ task, model, harness, repetition });
+ }
+ }
+ return cells;
+}
+
+module.exports = { expandMatrix };
@@ -52,7 +52,11 @@ function createRunClient({ config, fetchImpl = globalThis.fetch, now = () => Dat
@@ -52,7 +52,11 @@ function createRunClient({ config, fetchImpl = globalThis.fetch, now = () => Dat
async function submit({ subject, project, args, key }) {
async function submit({ subject, project, args, key }) {
const body = {
const body = {
class: 'code', artifact: { name: 'openvibe-codes', version: JOB_PROTOCOL }, args,
class: 'code', artifact: { name: 'openvibe-codes', version: JOB_PROTOCOL }, args,
- limits: { wall_ms: 30 * 60_000, cpu_ms: 30 * 60_000, mem_bytes: 4 * 1024 ** 3 },
+ limits: {
+ wall_ms: Math.max(1000, Math.min(30 * 60_000, Number(args.limits && args.limits.ms) || 30 * 60_000)),
+ cpu_ms: Math.max(1000, Math.min(30 * 60_000, Number(args.limits && args.limits.ms) || 30 * 60_000)),
+ mem_bytes: 4 * 1024 ** 3,
+ },
ttl_ms: 45 * 60_000, egress: 'public',
ttl_ms: 45 * 60_000, egress: 'public',
requirements: { kind: 'run.code', mobility: 'job', latency_class: 'background', objective: 'balanced', trust: ['user-owned'] },
requirements: { kind: 'run.code', mobility: 'job', latency_class: 'background', objective: 'balanced', trust: ['user-owned'] },
};
};
@@ -140,7 +140,7 @@ function usageOf(result) {
@@ -140,7 +140,7 @@ function usageOf(result) {
};
};
}
}
-function createRunTasks({ store, run, config, log = console, sessions = null, outbox = null, runTokens = null, connections = null, attempts = null, proof = null }) {
+function createRunTasks({ store, run, config, log = console, sessions = null, outbox = null, runTokens = null, connections = null, attempts = null, proof = null, benchAttemptEnded = null }) {
const db = store.db;
const db = store.db;
const who = (row) => ({ subject: row.subject, project: row.project_id });
const who = (row) => ({ subject: row.subject, project: row.project_id });
// Lazily, so a caller that only reads tasks does not need the session/outbox modules (and there is no cycle).
// Lazily, so a caller that only reads tasks does not need the session/outbox modules (and there is no cycle).
@@ -218,7 +218,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -218,7 +218,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;
- return 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;
if (RANK[state] < RANK[current.state]) return current;
if (RANK[state] < RANK[current.state]) return current;
@@ -264,6 +264,11 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -264,6 +264,11 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
}
}
return updated;
return updated;
});
});
+ 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)}`));
+ }
+ return saved;
}
}
/**
/**
@@ -288,7 +293,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -288,7 +293,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
await store.tx(async () => {
await store.tx(async () => {
inserted = await db.one(`INSERT INTO run_tasks (id, subject, project_id, repo, ref, prompt, agent, push, pull_request, state, via, idempotency_key, actor_ref, created_at, updated_at)
inserted = await db.one(`INSERT INTO run_tasks (id, subject, project_id, repo, ref, prompt, agent, push, pull_request, state, via, idempotency_key, actor_ref, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'queued', $10, $11, $12::jsonb, $13, $13) RETURNING *`,
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'queued', $10, $11, $12::jsonb, $13, $13) RETURNING *`,
- [id, subject, project, args.repo, args.ref, args.prompt, args.agent, !!args.push, !!(args.push && args.push.pull_request), via, key, actorRefValue, at]);
+ [id, subject, project, args.repo || 'bench://', args.ref, args.prompt, args.agent, !!args.push, !!(args.push && args.push.pull_request), via, key, actorRefValue, at]);
await taskOutbox.emitTask(inserted, null);
await taskOutbox.emitTask(inserted, null);
return inserted;
return inserted;
});
});
@@ -388,6 +393,12 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -388,6 +393,12 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
if (!row) return null;
if (!row) return null;
if (!TERMINAL.has(row.state)) return row;
if (!TERMINAL.has(row.state)) return row;
if (row.outcome === 'stopped') return row; // a person stopped it: nothing carries it on
if (row.outcome === 'stopped') return row; // a person stopped it: nothing carries it on
+ const actorRef = asObject(row.actor_ref);
+ if (actorRef && actorRef.bench) {
+ const done = row.state === 'succeeded';
+ 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()]);
+ }
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];
if (!ending || !ENDED.has(ending.state)) return row;
if (!ending || !ENDED.has(ending.state)) return row;
@@ -33,6 +33,8 @@ const { asyncRouter } = require('./router');
@@ -33,6 +33,8 @@ const { asyncRouter } = require('./router');
const { parseArgs } = require('../../harness/job');
const { parseArgs } = require('../../harness/job');
const { createRunTasks, TERMINAL } = require('../domain/run-tasks');
const { createRunTasks, TERMINAL } = require('../domain/run-tasks');
const { createTaskEvents } = require('./task-events');
const { createTaskEvents } = require('./task-events');
+const { createServiceTokenClient } = require('openvibe-sdk/auth');
+const { expandMatrix } = require('../domain/bench');
const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const TASK = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;
const TASK = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;
@@ -44,6 +46,10 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, task
@@ -44,6 +46,10 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, task
const taskEvents = createTaskEvents({ store, config, log, attempts: taskStore.attempts });
const taskEvents = createTaskEvents({ store, config, log, attempts: taskStore.attempts });
const problem = (req, res, status, code, detail) => http.sendProblem(res, status, code, { detail, ctx: req.ov });
const problem = (req, res, status, code, detail) => http.sendProblem(res, status, code, { detail, ctx: req.ov });
const json = express.json({ limit: '32kb' });
const json = express.json({ limit: '32kb' });
+ const benchTokens = config.oauth.clientSecret ? createServiceTokenClient({
+ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId,
+ clientSecret: config.oauth.clientSecret, audience: 'openvibe.bench', scope: 'bench.run.publish',
+ }) : null;
/** The route's guard: the service token, its capability and the person it acts for → req.principal. */
/** The route's guard: the service token, its capability and the person it acts for → req.principal. */
const guard = (cap) => async (req, res, next) => {
const guard = (cap) => async (req, res, next) => {
@@ -174,6 +180,84 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, task
@@ -174,6 +180,84 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, task
return res.status(201).json({ task: taskStore.present(row), created: true });
return res.status(201).json({ task: taskStore.present(row), created: true });
});
});
+ // Bench is the only caller: it supplies a service token for Codes plus the signed-in user's short-lived
+ // Network token in a private server-to-server header. The latter is used only to provision the user's own project.
+ r.post('/benchmarks/:id/run', guardAny('codes.task.start'), json, async (req, res) => {
+ const p = req.principal;
+ if (p.owner) return problem(req, res, 403, 'codes.bench_service_required', 'Bench starts verified benchmark tasks.');
+ const userToken = String(req.get('x-ov-user-token') || '').trim();
+ const user = await keys.verifyUser(userToken, { issuer: config.networkIssuer, audience: config.oauth.sessionAudience });
+ if (!user.ok || user.claims.subject_id !== p.subject) return problem(req, res, 403, 'codes.bench_subject_mismatch', 'The signed-in person must match X-OV-Subject.');
+ if (!benchTokens) return problem(req, res, 503, 'codes.bench_unconfigured', 'Codes has no Network client credentials for Bench.');
+ const b = req.body || {};
+ const models = Array.isArray(b.models) ? b.models.map(String) : [];
+ const requestedHarnesses = Array.isArray(b.harnesses) ? b.harnesses.map(String) : [];
+ const repeat = Number(b.repeat);
+ if (!models.length || !requestedHarnesses.length || !Number.isInteger(repeat) || repeat < 1 || repeat > 10
+ || models.some((m) => !m || m.length > 120) || new Set(requestedHarnesses).size !== requestedHarnesses.length) {
+ return problem(req, res, 422, 'codes.bench_invalid', 'models and harnesses are required; repeat must be 1–10.');
+ }
+ const available = new Set(harnesses.list().map((h) => h.id));
+ if (requestedHarnesses.some((h) => !available.has(h))) return problem(req, res, 422, 'codes.bench_harness_unknown', 'A harness is not in GET /api/v1/harnesses.');
+ let definition;
+ try {
+ const response = await fetch(`${config.bench.url}/api/v1/benchmarks/${encodeURIComponent(String(req.params.id))}`, {
+ headers: { ...(await benchTokens.authHeaders()), Accept: 'application/json' }, signal: AbortSignal.timeout(15000),
+ });
+ const body = await response.json().catch(() => ({}));
+ if (!response.ok) return problem(req, res, response.status, body.code || 'codes.bench_definition_failed', body.detail || 'Bench did not return the benchmark definition.');
+ definition = body.definition;
+ } catch (e) { return problem(req, res, 503, 'codes.bench_unavailable', `Bench did not answer: ${e.message}`); }
+ let matrix;
+ try { matrix = expandMatrix(definition, models, requestedHarnesses, repeat); }
+ catch (e) { return problem(req, res, 422, e.code || 'codes.bench_empty', e.message); }
+ if (!matrix.length) return problem(req, res, 422, 'codes.bench_empty', 'The benchmark has no tasks.');
+ const total = matrix.length;
+ if (total > config.bench.maxMatrixSize) return problem(req, res, 422, 'codes.bench_matrix_too_large', `This matrix has ${total} tasks; the limit is ${config.bench.maxMatrixSize}.`);
+ const running = Number(await db.value("SELECT COUNT(*)::int AS n FROM (SELECT actor_ref->'bench'->>'session' AS session FROM run_tasks WHERE subject = $1 AND actor_ref ? 'bench' GROUP BY actor_ref->'bench'->>'session' HAVING bool_or(state NOT IN ('succeeded','failed','cancelled','expired'))) active", [p.subject]));
+ if (running >= 3) return problem(req, res, 429, 'codes.bench_concurrency_limit', 'A person may have at most three benchmark runs active.');
+ const today = store.now() - (store.now() % 86400000);
+ const daily = Number(await db.value("SELECT COUNT(*)::int AS n FROM run_tasks WHERE subject = $1 AND actor_ref ? 'bench' AND created_at >= $2", [p.subject, today]));
+ if (daily + total > config.bench.dailyTasks) return problem(req, res, 429, 'codes.bench_daily_limit', `The daily benchmark task cap is ${config.bench.dailyTasks}.`);
+
+ let project;
+ try {
+ const headers = { Authorization: `Bearer ${userToken}`, Accept: 'application/json' };
+ const projectsResponse = await fetch(`${config.networkInternalUrl}/api/v1/projects`, { headers, signal: AbortSignal.timeout(8000) });
+ if (!projectsResponse.ok) throw new Error(`Network answered ${projectsResponse.status}`);
+ const projects = (await projectsResponse.json()).projects || [];
+ project = projects.find((x) => x.name === 'bench' && !x.archived_at)?.id;
+ if (!project) {
+ const created = await fetch(`${config.networkInternalUrl}/api/v1/projects`, {
+ method: 'POST', headers: { ...headers, 'Content-Type': 'application/json' }, body: JSON.stringify({ name: 'bench' }), signal: AbortSignal.timeout(8000),
+ });
+ if (!created.ok) throw new Error(`Network answered ${created.status}`);
+ project = (await created.json()).id;
+ }
+ } catch (e) { return problem(req, res, 502, 'codes.bench_project_failed', `Could not get or create the person's bench project: ${e.message}`); }
+
+ const session = `${req.params.id}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`;
+ const started = [];
+ const runIds = new Map();
+ try {
+ for (const cell of matrix) {
+ const { task, model, harness, repetition } = cell;
+ const key = `${model}\u0000${harness}\u0000${repetition}`;
+ if (!runIds.has(key)) runIds.set(key, store.newId('brn'));
+ const runId = runIds.get(key);
+ const harnessOffer = harnesses.list().find((h) => h.id === harness);
+ const inputs = Array.isArray(task.files) ? JSON.stringify(task.files).slice(0, 3000) : '';
+ const prompt = `Benchmark ${b.benchmark || req.params.id}, task ${task.id}.\n\n${String(task.prompt).slice(0, 10000)}${inputs ? `\n\nInput files:\n${inputs}` : ''}\n\nReturn only the requested artifact content. Respect the per-task limits: ${Number(definition.limits && definition.limits.ms) || 0} ms, ${Number(definition.limits && definition.limits.tokens) || 0} tokens, $${Number(definition.limits && definition.limits.usd) || 0}.`;
+ const limits = definition.limits || {};
+ const args = parseArgs(JSON.stringify({ kind: 'bench', prompt, agent: harness, model, limits }));
+ const taskRow = await taskStore.start({ subject: p.subject, project, args, via: p.via,
+ actorRef: { bench: { run_id: runId, task_count: matrix.filter((x) => x.model === model && x.harness === harness && x.repetition === repetition).length, benchmark: `${req.params.id}@${Number(definition.version) || 1}`, session, model, provider: (harnessOffer && harnessOffer.provider) || String(model).split('/')[0] || 'unknown', harness, harness_version: (harnessOffer && harnessOffer.version) || 'unknown', repetition, task_id: String(task.id), kind: definition.kind, expected: task.expected, judges: definition.judges || [], scoring: definition.scoring || {}, limits: definition.limits || {} } } });
+ started.push(taskStore.present(taskRow));
+ }
+ } catch (e) { return problem(req, res, 502, 'codes.bench_start_failed', `Codes started ${started.length} of ${total} tasks: ${e.message}`); }
+ return res.status(202).json({ session, tasks: started, task: started[0] || null, total: started.length });
+ });
+
/**
/**
* The caller's tasks, newest first: the person's own (whoever started them), or the ones this service started for
* The caller's tasks, newest first: the person's own (whoever started them), or the ones this service started for
* the person. ?state=narrows it to the words `openvibe-codes task list` prints: running (queued, placed, running),
* the person. ?state=narrows it to the words `openvibe-codes task list` prints: running (queued, placed, running),
new file mode 100644
new file mode 100644
@@ -0,0 +1,27 @@
@@ -0,0 +1,27 @@
+'use strict';
+const assert = require('assert');
+const { expandMatrix } = require('../server/domain/bench');
+const { grade } = require('../server/domain/bench-publisher');
+const { check, done } = require('./helpers/boot');
+
+(async () => {
+ const def = { tasks: [{ id: 'public', prompt: 'Make a page' }, { id: 'holdout', prompt: 'Make a game', holdout: true }] };
+ await check('matrix expands each revealed task across models, harnesses and repetitions', () => {
+ const matrix = expandMatrix(def, ['m1', 'm2'], ['h1', 'h2'], 3);
+ assert.strictEqual(matrix.length, 24);
+ assert.deepStrictEqual(matrix[0], { task: def.tasks[0], model: 'm1', harness: 'h1', repetition: 1 });
+ assert.strictEqual(matrix.at(-1).repetition, 3);
+ });
+ await check('a withheld task cannot enter the executor matrix', () => {
+ assert.throws(() => expandMatrix({ tasks: [{ id: 'secret', prompt: null, holdout: true }] }, ['m'], ['h'], 1),
+ (e) => e.code === 'codes.bench_holdout_hidden');
+ });
+ await check('automatic expected-output judges grade exact and contains checks', () => {
+ assert.strictEqual(grade(' a\npage ', 'a page'), 100);
+ assert.strictEqual(grade({ contains: '<main>' }, '<main>ok</main>'), 100);
+ assert.strictEqual(grade({ not_contains: 'script' }, '<p>safe</p>'), 100);
+ assert.strictEqual(grade('expected', 'actual'), 0);
+ assert.strictEqual(grade(null, 'anything'), null);
+ });
+ done();
+})().catch((e) => { process.stderr.write(`${e.stack || e}\n`); process.exit(1); });
@@ -63,6 +63,23 @@ function fakeHarness({ fail = false } = {}) {
@@ -63,6 +63,23 @@ function fakeHarness({ fail = false } = {}) {
assert.strictEqual(parseArgs(JSON.stringify({ repo: 'file:///tmp/x', prompt: 'x' }), env).repo, 'file:///tmp/x');
assert.strictEqual(parseArgs(JSON.stringify({ repo: 'file:///tmp/x', prompt: 'x' }), env).repo, 'file:///tmp/x');
});
});
+ await check('a bench job accepts no repository and returns the model output without a repository diff', async () => {
+ const parsed = parseArgs(JSON.stringify({ kind: 'bench', prompt: 'Return a small HTML artifact', agent: 'claude-code', model: 'cheap-model', limits: { ms: 5000, tokens: 10000, usd: 0.05 } }));
+ assert.strictEqual(parsed.repo, null);
+ assert.strictEqual(parsed.kind, 'bench');
+ assert.deepStrictEqual(parsed.limits, { ms: 5000, tokens: 10000, usd: 0.05 });
+ const docs = [];
+ const harness = fakeHarness();
+ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'codes-bench-tmp-'));
+ const doc = await runJob({ input: JSON.stringify({ kind: 'bench', prompt: 'Return a small HTML artifact' }), harness,
+ emit: () => {}, writeResult: (d) => docs.push(d), env, tmp });
+ assert.strictEqual(doc.kind, 'bench');
+ assert.strictEqual(doc.output, 'Done: README updated.');
+ assert.deepStrictEqual(doc.files, []);
+ assert.strictEqual(docs.length, 1);
+ fs.rmSync(tmp, { recursive: true, force: true });
+ });
+
await check('a task clones, runs the agent there, streams short lines and hands over the patch once', async () => {
await check('a task clones, runs the agent there, streams short lines and hands over the patch once', async () => {
const src = repository();
const src = repository();
const lines = [];
const lines = [];
@@ -43,9 +43,11 @@ const INTERNAL = [`http://127.0.0.1:3000${PROBE}`, `http://2130706433${PROBE}`,
@@ -43,9 +43,11 @@ const INTERNAL = [`http://127.0.0.1:3000${PROBE}`, `http://2130706433${PROBE}`,
'server/auth/sso.js': 'Network OAuth token and revoke (configured)',
'server/auth/sso.js': 'Network OAuth token and revoke (configured)',
'server/domain/forgejo.js': 'OpenVibe Git API (CODES_GIT_INTERNAL_URL, configured loopback Forgejo) with a service token; paths built from validated owner/name',
'server/domain/forgejo.js': 'OpenVibe Git API (CODES_GIT_INTERNAL_URL, configured loopback Forgejo) with a service token; paths built from validated owner/name',
'server/domain/run-client.js': 'OpenVibe.Run (CODES_RUN_URL, configured) and the Network token endpoint; job ids only, never a URL from a person',
'server/domain/run-client.js': 'OpenVibe.Run (CODES_RUN_URL, configured) and the Network token endpoint; job ids only, never a URL from a person',
+ 'server/domain/bench-publisher.js': 'OpenVibe.Bench (CODES_BENCH_URL, configured) run publishing and the Network token endpoint; artifact URLs use the configured render origin',
'server/http/run.js': 'Network projects API (configured) with the viewer\'s token, and the Run job stream URL run-client.js builds on its own base',
'server/http/run.js': 'Network projects API (configured) with the viewer\'s token, and the Run job stream URL run-client.js builds on its own base',
'server/http/home.js': 'OpenVibe Git\'s public repository search (CODES_GIT_INTERNAL_URL, configured loopback Forgejo) for the home page; no token',
'server/http/home.js': 'OpenVibe Git\'s public repository search (CODES_GIT_INTERNAL_URL, configured loopback Forgejo) for the home page; no token',
'server/http/me-api.js': 'Network projects API (configured) with the viewer\'s token, for the user task API; no URL from a person',
'server/http/me-api.js': 'Network projects API (configured) with the viewer\'s token, for the user task API; no URL from a person',
+ 'server/http/tasks-api.js': 'OpenVibe.Bench (CODES_BENCH_URL, configured), Network projects API (configured) with the verified user token; benchmark ids and task ids are URL-encoded, no URL from a person',
'server/http/pages.js': 'Network projects API (configured) for Improve OpenVibe project creation; Forgejo repo read (CODES_GIT_INTERNAL_URL) for repo details; no URL from a person',
'server/http/pages.js': 'Network projects API (configured) for Improve OpenVibe project creation; Forgejo repo read (CODES_GIT_INTERNAL_URL) for repo details; no URL from a person',
'server/domain/connections.js': 'OpenVibe.AI (CODES_AI_URL, configured) for a person\'s own keys: the credential routes and the two model endpoints, with the provider and the subject as path segments; no URL from a person',
'server/domain/connections.js': 'OpenVibe.AI (CODES_AI_URL, configured) for a person\'s own keys: the credential routes and the two model endpoints, with the provider and the subject as path segments; no URL from a person',
'server/domain/run-tokens.js': 'OpenVibe.Network\'s token endpoint (OV_NETWORK_INTERNAL_URL, configured) for a run token; no URL from a person',
'server/domain/run-tokens.js': 'OpenVibe.Network\'s token endpoint (OV_NETWORK_INTERNAL_URL, configured) for a run token; no URL from a person',