openvibers/openvibe.run Commits

A6: Run publishes the agent's step (agent.step.started|ended) (#188)

OP OpenVibers committed 6 days ago · 728ec0f2ee · 4 files, +168 −5

A person's job is one step of the agent's timeline: a generated title, its state and
its page. One outbox row per job change, with the requester as the subject; never its
args, inputs, output or a result body. A service's own job publishes nothing.

Changes

ddiff --git a/server/app.js b/server/app.js
iindex 2b5845d..1e746ec 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -51,7 +51,7 @@ function createApp(opts = {}) {
const keys = opts.keys || createKeyProvider(config, { fetchImpl, log });
const userAuth = createUserAuth(config, keys);
const outbox = opts.outbox || createRunOutbox({ db, config, fetchImpl: opts.eventsFetch, now, log });
-- const events = opts.events || createRunEvents({ outbox, now, log });
++ const events = opts.events || createRunEvents({ outbox, now, log, config });
const stream = opts.stream || createJobStream({ config, now, log });
const tickets = opts.tickets || createStreamTickets({ config, now, log });
// The dispatcher bridge to Bot (plan T14 step 6, server/dispatch/index.js): placement from the Fabric
ddiff --git a/server/events/outbox.js b/server/events/outbox.js
iindex f907476..12c433b 100644
---- a/server/events/outbox.js
++++ b/server/events/outbox.js
@@@ -24,7 +24,8 @@ const { createServiceOutbox } = require('openvibe-sdk/events');
const TABLE = 'run_events_outbox'; // migrations/0001_jobs.sql
const ACTOR = { type: 'service', id: 'run' };
const EVENT_TYPES = ['run.job.queued', 'run.job.started', 'run.job.succeeded', 'run.job.failed', 'run.job.cancelled', 'run.job.expired',
-- 'run.session.started', 'run.session.ended']; // server/sessions/service.js (contracts 0.223.0)
++ 'run.session.started', 'run.session.ended',
++ 'agent.step.started', 'agent.step.updated', 'agent.step.ended']; // server/sessions/service.js (contracts 0.223.0)
function createRunOutbox({ db, config, fetchImpl, now, log = console }) {
return createServiceOutbox({
ddiff --git a/server/jobs/events.js b/server/jobs/events.js
iindex 506127e..44801b5 100644
---- a/server/jobs/events.js
++++ b/server/jobs/events.js
@@@ -68,12 +68,55 @@ function payloadOf(row) {
};
}
++/** The agent's step (design 2026-11-07): the three events every mode publishes into the network's agent namespace. */
++const AGENT_TOPICS = Object.freeze({
++ started: 'agent.step.started',
++ updated: 'agent.step.updated',
++ ended: 'agent.step.ended',
++});
++const AGENT_TOPIC_OF_STATE = Object.freeze({
++ queued: AGENT_TOPICS.started,
++ running: AGENT_TOPICS.started,
++ succeeded: AGENT_TOPICS.ended,
++ failed: AGENT_TOPICS.ended,
++ cancelled: AGENT_TOPICS.ended,
++ expired: AGENT_TOPICS.ended,
++});
++const AGENT_STATUS_OF_STATE = Object.freeze({
++ queued: 'waiting', running: 'running', succeeded: 'done', failed: 'failed', cancelled: 'stopped', expired: 'failed',
++});
++const AGENT_CLASS_WORDS = Object.freeze({
++ function: 'function', code: 'code', browser: 'browser', linux: 'Linux', desktop: 'desktop', gpu: 'GPU',
++});
++
++/** The job's step: what it is, where it runs, how it ended and where to see it — never args, inputs or output. */
++function agentStepOf(row, state, { baseUrl = null } = {}) {
++ const job = toJob(row);
++ const status = AGENT_STATUS_OF_STATE[state] || 'running';
++ const what = AGENT_CLASS_WORDS[job.class] || 'a job';
++ const base = String(baseUrl || 'https://openvibe.run').replace(/\/+$/, '');
++ return {
++ step_id: job.id,
++ mode: 'run',
++ product: 'run',
++ ref: job.id,
++ title: `A ${what} job`,
++ status,
++ started_at: job.timings.started_at || job.timings.created_at,
++ ended_at: status === 'waiting' || status === 'running' ? null : (job.timings.finished_at || new Date().toISOString()),
++ placement: { where: job.placement ? 'On a machine in the network' : 'On the network', why: 'it runs once where the tools it needs are, and stops when it ends' },
++ cost_usd: null,
++ link: `${base}/api/v1/jobs/${encodeURIComponent(job.id)}`,
++ reason: status === 'failed' ? (job.error && (job.error.detail || job.error.code) ? String(job.error.detail || job.error.code).slice(0, 240) : 'it did not finish') : null,
++ };
++}
++
/**
* createRunEvents({ outbox, now }) → one emitter per state plus emit(t, row, state). Every emitter
* validates the payload against its released contract before it reaches the outbox, so a projection bug
* is an error here rather than a malformed event on the bus.
*/
--function createRunEvents({ outbox, now = () => Date.now(), log = console }) {
++function createRunEvents({ outbox, now = () => Date.now(), log = console, config = null }) {
/**
* The outbox row for one (job, state), written in the caller's transaction. The row's key is
* `run:<job id>:<state>`; a row that is already there is left alone (ON CONFLICT DO NOTHING), which is
@@@ -108,7 +151,36 @@ function createRunEvents({ outbox, now = () => Date.now(), log = console }) {
// The payload contract is the event's own (run.job.started@1 for the running state: a state and
// its event name are not always the same word, and run.job.running@1 does not exist).
contracts.assertValid(spec.contract, payload);
-- return enqueueOnce(t, row, spec, payload, traceparent);
++ const out = await enqueueOnce(t, row, spec, payload, traceparent);
++ // The agent's step (design 2026-11-07): one row per job change, with the person's subject, so the one
++ // timeline behind the Dock shows it. A service's job has no person to show it to. Never args, inputs,
++ // output or a result body: a title, an outcome and a link.
++ if (out.enqueued) await enqueueAgentStep(t, row, state);
++ return out;
++ }
++
++ /** agent.step.started for queued/running and agent.step.ended for the four end states, as the same row. */
++ async function enqueueAgentStep(t, row, state) {
++ const person = toJob(row).requester;
++ if (!person || person.type !== 'user' || !/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(person.id || ''))) return null;
++ const topic = AGENT_TOPIC_OF_STATE[state];
++ if (!topic) return null;
++ const prepared = outbox.events.prepare({
++ event_type: topic,
++ version: 1,
++ source: 'run',
++ actor: { type: 'user', id: person.id },
++ timestamp: new Date(now()).toISOString(),
++ subject: { type: 'user', id: person.id },
++ visibility: 'internal',
++ payload: agentStepOf(row, state, { baseUrl: config && config.baseUrl }),
++ }, { now: now() });
++ contracts.assertValid('events.event-envelope@1', prepared);
++ // A row per (job, state), like the run.job.* row: a replay is the row already there, never a second event.
++ await t.maybe(
++ `INSERT INTO ${TABLE} (event_id, envelope, traceparent, created_at) VALUES ($1, $2, NULL, $3) ON CONFLICT (event_id) DO NOTHING`,
++ [`agent:${row.id}:${state}`, JSON.stringify(prepared), now()]);
++ return prepared;
}
return {
@@@ -124,4 +196,4 @@ function createRunEvents({ outbox, now = () => Date.now(), log = console }) {
};
}
--module.exports = { createRunEvents, payloadOf, EVENT_OF_STATE, TRACEPARENT_RE };
++module.exports = { createRunEvents, agentStepOf, AGENT_TOPICS, AGENT_TOPIC_OF_STATE, payloadOf, EVENT_OF_STATE, TRACEPARENT_RE };
ddiff --git a/test/agent-step.test.js b/test/agent-step.test.js
nnew file mode 100644
iindex 0000000..bf0a921
---- /dev/null
++++ b/test/agent-step.test.js
@@@ -0,0 +1,90 @@
++'use strict';
++/**
++ * The agent's step (design 2026-11-07): OpenVibe.Run publishes what one of a person's jobs is doing as
++ * agent.step.started|ended, so the one timeline behind the Dock shows it. The payload is a title, an outcome and a
++ * link: never args, inputs, output or a result body. A service's job has no person, so it publishes no step.
++ *
++ * node test/agent-step.test.js
++ */
++const assert = require('assert');
++const { boot, check, done } = require('./helpers/app');
++const { agentStepOf, AGENT_TOPICS } = require('../server/jobs/events');
++
++async function checks(t) {
++ const P = t.idsStub ? t.idsStub.newId('project') : require('openvibe-contracts').ids.newId('project');
++ const ada = require('openvibe-contracts').ids.newId('user');
++ t.network.members.set(`${P} ${ada}`, { role: 'admin' });
++ const as = (subject, project) => ({
++ token: t.network.signService({ sub: 'svc:watch', cap: ['run.job.submit', 'run.job.read', 'run.job.cancel'] }),
++ headers: { 'X-OV-Subject': subject, 'X-OV-Project': project },
++ });
++ const stepRows = async () => (await t.db.many('SELECT event_id, envelope FROM run_events_outbox ORDER BY id'))
++ .map((r) => ({ key: r.event_id, ...r.envelope }))
++ .filter((e) => String(e.event_type || '').startsWith('agent.step.'));
++
++ await check('the payload is a title, an outcome and a link: never args, inputs or the result', () => {
++ const row = { id: 'job_01JAB2C3D4E5F6G7H8J9K0MNPS', project_id: P, requester: `user:${ada}`, via: 'service:watch', class: 'code',
++ state: 'running', job: { args: { secret: 'the args a person typed' }, inputs: [{ url: 'https://example.test/private' }], limits: { wall_ms: 30000 } },
++ ttl_ms: 630000, created_at: 1762000000000, updated_at: 1762000001000, started_ms: 1762000000500, finished_ms: null,
++ wall_ms: 0, exit_reason: null, exit_code: null, node_id: 'dev_01JAB2C3D4E5F6G7H8J9K0MNPS', provider: null, region: null,
++ result: { answer: 'the result body never leaves Run' }, error_code: null, error_detail: null, idempotency_key: 'k' };
++ const step = agentStepOf(row, 'running', { baseUrl: 'https://openvibe.run' });
++ assert.strictEqual(step.step_id, row.id);
++ assert.strictEqual(step.mode, 'run');
++ assert.strictEqual(step.product, 'run');
++ assert.strictEqual(step.status, 'running');
++ assert.strictEqual(step.title, 'A code job');
++ assert.strictEqual(step.link, `https://openvibe.run/api/v1/jobs/${row.id}`);
++ assert.ok(step.started_at && !Number.isNaN(Date.parse(step.started_at)), 'the start is a time');
++ assert.strictEqual(step.ended_at, null, 'a running job has not ended');
++ const raw = JSON.stringify(step);
++ assert.ok(!raw.includes('the args a person typed'), 'no args');
++ assert.ok(!raw.includes('example.test/private'), 'no inputs');
++ assert.ok(!raw.includes('the result body never leaves Run'), 'no result');
++ assert.ok(!('args' in step) && !('inputs' in step) && !('result' in step) && !('output' in step), 'no such field');
++ assert.strictEqual(AGENT_TOPICS.started, 'agent.step.started');
++ assert.strictEqual(AGENT_TOPICS.ended, 'agent.step.ended');
++ });
++
++ await check('a person\'s job publishes started (waiting) and ended (stopped), one row per change', async () => {
++ const made = await t.call('POST', '/api/v1/jobs', { body: t.request(), ...as(ada, P) });
++ assert.strictEqual(made.status, 201, made.text);
++ const id = made.json.job.id;
++ let rows = (await stepRows()).filter((e) => e.payload && e.payload.step_id === id);
++ assert.strictEqual(rows.length, 1, JSON.stringify(rows));
++ assert.strictEqual(rows[0].event_type, 'agent.step.started');
++ assert.strictEqual(rows[0].key, `agent:${id}:queued`);
++ assert.deepStrictEqual(rows[0].subject, { type: 'user', id: ada });
++ assert.strictEqual(rows[0].payload.status, 'waiting');
++ assert.strictEqual(rows[0].payload.mode, 'run');
++ assert.ok(!JSON.stringify(rows[0].payload).includes('args'), 'no args field');
++
++ const cancelled = await t.call('POST', `/api/v1/jobs/${id}/cancel`, as(ada, P));
++ assert.strictEqual(cancelled.status, 200, cancelled.text);
++ rows = (await stepRows()).filter((e) => e.payload && e.payload.step_id === id);
++ const ended = rows.filter((e) => e.event_type === 'agent.step.ended');
++ assert.strictEqual(ended.length, 1, JSON.stringify(rows));
++ assert.strictEqual(ended[0].key, `agent:${id}:cancelled`);
++ assert.strictEqual(ended[0].payload.status, 'stopped');
++ assert.ok(ended[0].payload.ended_at, 'an ended step says when it ended');
++ assert.strictEqual(rows.filter((e) => e.event_type === 'agent.step.started').length, 1, 'a repeat is never a second row');
++ });
++
++ await check('a service\'s own job publishes no step: there is no person to show it to', async () => {
++ const before = (await stepRows()).length;
++ const made = await t.call('POST', '/api/v1/jobs', {
++ body: t.request(),
++ token: t.network.signService({ sub: 'svc:builder', cap: ['run.job.submit'], project_id: P }),
++ });
++ assert.strictEqual(made.status, 201, made.text);
++ assert.strictEqual((await stepRows()).length, before, 'no agent.step row for a service job');
++ });
++}
++
++async function main() {
++ const t = await boot({ env: { OV_OAUTH_CLIENT_SECRET: 'run-test-secret', BASE_URL: 'http://run.test' } });
++ try { await checks(t); } finally { await t.close(); }
++ done();
++}
++
++main().catch((e) => { console.error(e); process.exit(1); });