openvibers/openvibe.codes Commits
Codes: publish benchmark runs after start failures (#177)
Changes
@@@ -78,7 +78,7 @@ async function createApp(opts = {}) {// 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 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 tasks = createRunTasks({ store, run, config, log, sessions, outbox, runTokens, connections, attempts, benchAttemptEnded: opts.benchAttemptEnded || benchPublisher.onTaskTerminal });const ctx = { config, store, keys, sso, harnesses, log, sessions, outbox, run, tasks, connections, runTokens, attempts, benchPublisher };ctx.forgejo = createForgejoClient({ config, log });@@@ -316,6 +316,9 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ouconst failed = await db.one(`UPDATE run_tasks SET state = 'failed', error = $2, outcome = 'unfinished', outcome_note = $3, finished_at = $4::timestamptz, updated_at = $5WHERE id = $1 RETURNING *`, [id, why, `the task could not start: ${why}`, store.iso(), store.now()]);await taskOutbox.emitTask(failed, 'queued');++ if (benchAttemptEnded && actorRef && asObject(actorRef).bench) {++ await benchAttemptEnded(failed).catch((publishError) => log.warn(`[Codes] benchmark publish failed for task ${failed.id}: ${String(publishError && publishError.message).slice(0, 200)}`));++ }return failed;});}nnew file mode 100644@@@ -0,0 +1,30 @@++'use strict';++++const assert = require('assert');++const { boot, check, done } = require('./helpers/boot');++const { parseArgs } = require('../harness/job');++++(async () => {++ const published = [];++ const t = await boot({++ run: { async submit() { throw new Error('machine unavailable'); } },++ benchAttemptEnded: async (row) => published.push(row),++ });++ try {++ const person = t.network.addUser('bench-person');++ await check('a benchmark task rejected by Run is sent to the publisher as a terminal failure', async () => {++ const row = await t.ctx.tasks.start({++ subject: person.subject,++ project: 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1',++ args: parseArgs(JSON.stringify({ kind: 'bench', prompt: 'Return one short answer.', agent: 'deepseek', model: 'deepseek/test' })),++ actorRef: { bench: { run_id: 'brn_01J8Z3Q4R5S6T7V8W9X0Y1Z2R1', task_count: 1 } },++ });++ assert.strictEqual(row.state, 'failed');++ assert.strictEqual(published.length, 1);++ assert.strictEqual(published[0].id, row.id);++ });++ } finally {++ await t.close();++ }++ done();++})().catch((e) => { console.error(e); process.exit(1); });@@@ -37,7 +37,7 @@ async function boot(opts = {}) {const { createStore } = require('../../server/db');// One database per boot (PGlite, or CODES_TEST_STORE=pg: the containers), dropped when the boot closes.const testdb = await require('./db').testDb();-- const built = await createApp({ config, store: createStore(testdb.db), log, limitsNow: opts.limitsNow, actorLimits: opts.actorLimits === true, ...(opts.run ? { run: opts.run } : {}), ...(opts.accountSend ? { accountSend: opts.accountSend } : {}) });++ const built = await createApp({ config, store: createStore(testdb.db), log, limitsNow: opts.limitsNow, actorLimits: opts.actorLimits === true, ...(opts.run ? { run: opts.run } : {}), ...(opts.benchAttemptEnded ? { benchAttemptEnded: opts.benchAttemptEnded } : {}), ...(opts.accountSend ? { accountSend: opts.accountSend } : {}) });await built.ctx.keys.ensure();const server = await new Promise((resolve) => { const s = http.createServer(built.app); s.listen(0, '127.0.0.1', () => resolve(s)); });const base = `http://127.0.0.1:${server.address().port}`;
@@ -78,7 +78,7 @@ async function createApp(opts = {}) {
@@ -78,7 +78,7 @@ async function createApp(opts = {}) {
// 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 benchPublisher = createBenchPublisher({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });
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 tasks = createRunTasks({ store, run, config, log, sessions, outbox, runTokens, connections, attempts, benchAttemptEnded: opts.benchAttemptEnded || benchPublisher.onTaskTerminal });
const ctx = { config, store, keys, sso, harnesses, log, sessions, outbox, run, tasks, connections, runTokens, attempts, benchPublisher };
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 });
@@ -316,6 +316,9 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
@@ -316,6 +316,9 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
const failed = await db.one(`UPDATE run_tasks SET state = 'failed', error = $2, outcome = 'unfinished', outcome_note = $3, finished_at = $4::timestamptz, updated_at = $5
const failed = await db.one(`UPDATE run_tasks SET state = 'failed', error = $2, outcome = 'unfinished', outcome_note = $3, finished_at = $4::timestamptz, updated_at = $5
WHERE id = $1 RETURNING *`, [id, why, `the task could not start: ${why}`, store.iso(), store.now()]);
WHERE id = $1 RETURNING *`, [id, why, `the task could not start: ${why}`, store.iso(), store.now()]);
await taskOutbox.emitTask(failed, 'queued');
await taskOutbox.emitTask(failed, 'queued');
+ if (benchAttemptEnded && actorRef && asObject(actorRef).bench) {
+ await benchAttemptEnded(failed).catch((publishError) => log.warn(`[Codes] benchmark publish failed for task ${failed.id}: ${String(publishError && publishError.message).slice(0, 200)}`));
+ }
return failed;
return failed;
});
});
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,30 @@
@@ -0,0 +1,30 @@
+'use strict';
+
+const assert = require('assert');
+const { boot, check, done } = require('./helpers/boot');
+const { parseArgs } = require('../harness/job');
+
+(async () => {
+ const published = [];
+ const t = await boot({
+ run: { async submit() { throw new Error('machine unavailable'); } },
+ benchAttemptEnded: async (row) => published.push(row),
+ });
+ try {
+ const person = t.network.addUser('bench-person');
+ await check('a benchmark task rejected by Run is sent to the publisher as a terminal failure', async () => {
+ const row = await t.ctx.tasks.start({
+ subject: person.subject,
+ project: 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1',
+ args: parseArgs(JSON.stringify({ kind: 'bench', prompt: 'Return one short answer.', agent: 'deepseek', model: 'deepseek/test' })),
+ actorRef: { bench: { run_id: 'brn_01J8Z3Q4R5S6T7V8W9X0Y1Z2R1', task_count: 1 } },
+ });
+ assert.strictEqual(row.state, 'failed');
+ assert.strictEqual(published.length, 1);
+ assert.strictEqual(published[0].id, row.id);
+ });
+ } finally {
+ await t.close();
+ }
+ done();
+})().catch((e) => { console.error(e); process.exit(1); });
@@ -37,7 +37,7 @@ async function boot(opts = {}) {
@@ -37,7 +37,7 @@ async function boot(opts = {}) {
const { createStore } = require('../../server/db');
const { createStore } = require('../../server/db');
// One database per boot (PGlite, or CODES_TEST_STORE=pg: the containers), dropped when the boot closes.
// One database per boot (PGlite, or CODES_TEST_STORE=pg: the containers), dropped when the boot closes.
const testdb = await require('./db').testDb();
const testdb = await require('./db').testDb();
- const built = await createApp({ config, store: createStore(testdb.db), log, limitsNow: opts.limitsNow, actorLimits: opts.actorLimits === true, ...(opts.run ? { run: opts.run } : {}), ...(opts.accountSend ? { accountSend: opts.accountSend } : {}) });
+ const built = await createApp({ config, store: createStore(testdb.db), log, limitsNow: opts.limitsNow, actorLimits: opts.actorLimits === true, ...(opts.run ? { run: opts.run } : {}), ...(opts.benchAttemptEnded ? { benchAttemptEnded: opts.benchAttemptEnded } : {}), ...(opts.accountSend ? { accountSend: opts.accountSend } : {}) });
await built.ctx.keys.ensure();
await built.ctx.keys.ensure();
const server = await new Promise((resolve) => { const s = http.createServer(built.app); s.listen(0, '127.0.0.1', () => resolve(s)); });
const server = await new Promise((resolve) => { const s = http.createServer(built.app); s.listen(0, '127.0.0.1', () => resolve(s)); });
const base = `http://127.0.0.1:${server.address().port}`;
const base = `http://127.0.0.1:${server.address().port}`;