openvibers/openvibe.bot Commits

Pin contracts v0.96.0 and share one jobs service for socket and API (#27)

OP OpenVibers committed 5 Oct 2026 · de6273603a · 11 files, +90 −47

Asked: Part of a cross-repository initiative (init-t14-step-6-first-half-r1c-di-1004T2328): T14 step 6, first half (R1c dispatcher bridge, Run → Bot; dossier /home/workstation/openvibe/agents/ds/prep/T14.md section 'T14 step 6', items 1-3). Security-critical (cross-service auth, money attribution): every …
Harness job init-t14-step-6-first-2328-s2-002048 (plan T14).

Changes

ddiff --git a/README.md b/README.md
iindex 6108da1..d70e04c 100644
---- a/README.md
++++ b/README.md
@@@ -14,7 +14,7 @@ OpenVibe.Bot pairs a robot's machines, keeps their state and gates every operato
- The `/device` and `/control` WebSockets, the command gate and the latched e-stop.
- The signed-in pages (`server/web/routes.js`): `/robots` to list and add robots, `/pair/:id` for a pairing code and the installer command, and `/panel/:id`, the panel rendered from the profile. A `sim.rover` robot is driven by an in-process simulator (`server/sim`), so the panel works before any hardware exists. It gets no jobs, and it is not counted as a device or reported online in the outbox. The panel (`public/panel.js`, no build step) drives with a touch joystick, the keyboard or a gamepad, shows a latency meter and one camera tile per camera the profile lists; the camera stays a placeholder until OpenRe.Stream can play a WHIP-published stream back to a browser.
- The command audit (`command_audit`, kept 30 days) and the `bot.*` outbox events.
--- Dispatching `platform.job@1` jobs to a paired Node over `/device` (`server/jobs/dispatch.js`; the internal `bot.job.dispatch` API is `POST /api/v1/jobs`, `POST /jobs/:id/cancel`, `GET /jobs/:id`) and metering them per wall-clock second (`run_jobs`, `run_usage_outbox`; docs/protocol.md §1.2).
++- Dispatching `platform.job@1` jobs to a paired Node over `/device` (`server/jobs/index.js` over `server/jobs/dispatch.js`; the internal `bot.job.dispatch` API is `POST /api/v1/jobs`, `POST /jobs/:id/cancel`, `GET /jobs/:id`) and metering them per wall-clock second (`run_jobs`, `run_usage_outbox`; docs/protocol.md §1.2).
## Does not own
@@@ -35,7 +35,7 @@ OpenVibe.Bot pairs a robot's machines, keeps their state and gates every operato
- OpenVibe.Network: user and service tokens, node tokens, and pairing when the authority is `network`.
- OpenRe.Stream: the WHIP ingest base (`BOT_WHIP_BASE`) devices publish to.
- OpenVibe.Billing: job usage readings go to `billing.usage.record` (`BOT_BILLING_URL`, `BOT_BILLING_TOKEN`); unset, they wait in `run_usage_outbox`.
--- `openvibe-contracts` v0.92.0, `openvibe-sdk` v0.26.0 and `openvibe-shared` v2.5.0 (package.json).
++- `openvibe-contracts` v0.96.0, `openvibe-sdk` v0.26.0 and `openvibe-shared` v2.5.0 (package.json).
## Capabilities
ddiff --git a/STATUS.json b/STATUS.json
iindex 00a8bed..8eb1e09 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -5,9 +5,9 @@
"runtime": "node 22, express 4, ws; PostgreSQL 18 via openvibe-sdk/db (pg, or embedded PGlite in tests); Valkey optional; no build step",
"port": 4630,
"domain": "openvibe.bot (www -> apex; /device and /control WebSockets; /install)",
-- "contracts": "openvibe-contracts v0.92.0",
++ "contracts": "openvibe-contracts v0.96.0",
"packages": [
-- "openvibe-contracts v0.92.0",
++ "openvibe-contracts v0.96.0",
"openvibe-sdk v0.26.0",
"openvibe-shared v2.5.0"
],
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex a46117c..9f16b91 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -245,8 +245,10 @@ heartbeat or any frame brings it back. Commands are never replayed after a recon
Bot is the dispatcher of plan T14: it hands a `platform.job@1` job to a paired OpenVibe.Node over this socket
and meters it. The Run service (which owns `run.job.*`) calls it over the internal HTTP API under the
`bot.job.dispatch` capability: `POST /api/v1/jobs`, `POST /api/v1/jobs/:id/cancel` and `GET /api/v1/jobs/:id`
--(`server/api/v1.js`) wrap `server/jobs/dispatch.js` `dispatch(db, nodeId, job, { link, project, subject, provider })`
--and `cancel(db, jobId, { link })`. Each frame carries the envelope (`v`, `seq`, `ts`, `type`) and is validated
++(`server/api/v1.js`) call the jobs service `server/jobs/index.js` `createJobs({ db, hub, usage, log, now })`, which
++wraps `server/jobs/dispatch.js` `dispatch(db, nodeId, job, { link, project, subject, provider })` and
++`cancel(db, jobId, { link })` and holds the one `createJobFrames` instance (and its stdout rings) that both this
++socket and `GET /api/v1/jobs/:id` use. Each frame carries the envelope (`v`, `seq`, `ts`, `type`) and is validated
against `platform.job-frame@1` from the pinned openvibe-contracts; an invalid one is answered `error`
`bot.bad_frame`, and one naming a job of another device `error` `bot.unknown_job` (nothing changes).
ddiff --git a/package-lock.json b/package-lock.json
iindex 75af0a8..7875412 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -14,7 +14,7 @@
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.96.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"pg": "^8.23.0",
@@@ -844,8 +844,8 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.92.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
++ "version": "0.96.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.96.0",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex feff7c5..048de41 100644
---- a/package.json
++++ b/package.json
@@@ -26,7 +26,7 @@
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"jsonwebtoken": "^9.0.2",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.92.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.96.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"pg": "^8.23.0",
ddiff --git a/server/api/v1.js b/server/api/v1.js
iindex b3adbe9..a1b7127 100644
---- a/server/api/v1.js
++++ b/server/api/v1.js
@@@ -32,8 +32,6 @@ const express = require('express');
const { http } = require('openvibe-contracts');
const { fail, userSubject, isRobotId, json } = require('../util');
const { getProfile, listProfiles } = require('../profiles');
--const { dispatch, cancel } = require('../jobs/dispatch');
--const { get: getJob } = require('../jobs/store');
const CAP = {
read: 'bot.robot.read',
@@@ -43,7 +41,7 @@ const CAP = {
dispatch: 'bot.job.dispatch',
};
--function v1Router({ domain, apiAuth, limits, hub }) {
++function v1Router({ domain, apiAuth, limits, hub, jobs: jobService }) {
const r = express.Router();
r.use(limits.reads('bot.read'));
const manage = limits('bot.robot.manage', { minute: 30, hour: 300 });
@@@ -244,27 +242,25 @@ function v1Router({ domain, apiAuth, limits, hub }) {
res.json({ audit: rows.map(auditPresent), next_before: rows.length ? rows[rows.length - 1].id : null });
}));
-- // ── Jobs (Run → Bot: hand a platform.job@1 to a paired Node, server/jobs/dispatch.js) ─────────
++ // ── Jobs (Run → Bot: hand a platform.job@1 to a paired Node, server/jobs/index.js) ────────────
// Internal and service-to-service only: OpenVibe.Run's Network service token, never a person's.
r.post('/jobs', jobs, wrap(async (req, res) => {
requireJobDispatch(req);
const project = req.body?.project_id;
if (project == null || project === '') fail(422, 'bot.invalid_input', 'project_id is required (Run is the payer)');
-- const { job, sent } = await dispatch(domain.db, req.body?.node_id, req.body?.job, {
-- link: hub, project, subject: req.body?.subject ?? null, provider: req.body?.provider ?? null,
-- });
-- res.status(201).json({ job: presentJob(job), sent });
++ res.status(201).json(await jobService.dispatch(req.body?.node_id, req.body?.job, {
++ project, subject: req.body?.subject ?? null, provider: req.body?.provider ?? null,
++ }));
}));
r.post('/jobs/:id/cancel', jobs, wrap(async (req, res) => {
requireJobDispatch(req);
-- const { job, sent } = await cancel(domain.db, req.params.id, { link: hub });
-- res.json({ job: presentJob(job), sent });
++ res.json(await jobService.cancel(req.params.id));
}));
r.get('/jobs/:id', wrap(async (req, res) => {
requireJobDispatch(req);
-- const row = await getJob(domain.db, req.params.id);
-- if (!row) fail(404, 'bot.job_not_found', 'no such job');
-- res.json({ job: presentJob(row), stdout: hub.jobStdout(row.id) });
++ const state = await jobService.state(req.params.id);
++ if (!state) fail(404, 'bot.job_not_found', 'no such job');
++ res.json(state);
}));
// ── Pairing without a prior socket (the agent may POST instead of using the WS) ───────────────
@@@ -291,19 +287,6 @@ function v1Router({ domain, apiAuth, limits, hub }) {
return r;
}
--/** The job as the dispatch API answers it: the run_jobs row (server/jobs/store.js) plus its original platform.job@1. */
--function presentJob(row) {
-- return {
-- id: row.id, node_id: row.node_id, class: row.class, state: row.state,
-- project_id: row.project_id, subject: row.subject, provider: row.provider,
-- cancel_requested: row.cancel_requested, fault_code: row.fault_code,
-- sent_at: row.sent_at, started_ms: row.started_ms, finished_at: row.finished_at,
-- exit_reason: row.exit_reason, exit_code: row.exit_code, wall_ms: row.wall_ms,
-- usage_read: row.usage_read, result: json(row.result, null), job: json(row.job, null),
-- created_at: row.created_at, updated_at: row.updated_at,
-- };
--}
--
function auditPresent(a) {
return { id: a.id, robot_id: a.robot_id, device_id: a.device_id, operator_subject: a.operator_subject, operator_kind: a.operator_kind, role: a.role, kind: a.kind, value: a.value, result: a.result, reason: a.reason, latency_ms: a.latency_ms, at: a.at };
}
ddiff --git a/server/app.js b/server/app.js
iindex a41a73b..623960f 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -31,6 +31,7 @@ const { createRealtime } = require('./realtime');
const { createSimulator } = require('./sim');
const { createApiAuth } = require('./api/auth');
const { v1Router } = require('./api/v1');
++const { createJobs } = require('./jobs');
const { createActorLimits } = require('./api/actor-limits');
const { createSessionRoutes, viewerMiddleware } = require('./web/session');
const { createWebRoutes } = require('./web/routes');
@@@ -51,11 +52,14 @@ function createApp(opts = {}) {
const outbox = opts.outbox || createBotOutbox({ db, config, fetchImpl: opts.eventsFetch, now, log });
// Job usage readings → Billing (server/jobs/metering.js); started by server/index.js with the other jobs.
const usage = opts.usage || createUsageRelay({ db, config, fetchImpl: opts.billingFetch || fetchImpl, now, log });
-- const hub = opts.hub || createRealtime({ config, keys, userAuth, usage, log, now });
++ const hub = opts.hub || createRealtime({ config, keys, userAuth, log, now });
const nodes = opts.nodes || createNodePrincipals(config, { fetchImpl });
const openre = opts.openre !== undefined ? opts.openre : createOpenRe(config, { fetchImpl });
const domain = createDomain({ db, config, outbox, link: hub, nodes, openre, now, log });
hub.bindDomain(domain);
++ // One jobs service for the device socket and the jobs API, so both see the same stdout rings.
++ const jobs = createJobs({ db: () => domain.db, hub, usage, log, now });
++ hub.bindJobs(jobs);
const sim = opts.sim || createSimulator({ config, domain, hub, now, log });
const apiAuth = createApiAuth({ config, keys, userAuth });
const release = require('openvibe-shared/release').createRelease({ service: 'bot', root: path.join(__dirname, '..') });
@@@ -86,7 +90,7 @@ function createApp(opts = {}) {
release.mount(app, { registry: metrics.registry });
const limits = createActorLimits({ config, valkey, now: opts.limitsNow || now, registry: metrics.registry, log });
-- app.use('/api/v1', express.json({ limit: '64kb' }), (req, res, next) => { res.setHeader('Cache-Control', 'no-store'); next(); }, apiAuth.middleware, v1Router({ domain, apiAuth, limits, hub }));
++ app.use('/api/v1', express.json({ limit: '64kb' }), (req, res, next) => { res.setHeader('Cache-Control', 'no-store'); next(); }, apiAuth.middleware, v1Router({ domain, apiAuth, limits, hub, jobs }));
app.use('/auth', createSessionRoutes(config, userAuth, { fetchImpl }));
app.use(viewerMiddleware(userAuth));
app.use(createWebRoutes(config, { domain, sim, limits, log }));
@@@ -108,7 +112,7 @@ function createApp(opts = {}) {
return res.status(500).type('text/plain').send('Something went wrong\n');
});
-- Object.assign(app.locals, { config, db, valkey, domain, keys, outbox, usage, hub, sim, userAuth, metrics });
++ Object.assign(app.locals, { config, db, valkey, domain, keys, outbox, usage, hub, jobs, sim, userAuth, metrics });
return app;
}
ddiff --git a/server/jobs/dispatch.js b/server/jobs/dispatch.js
iindex 0851f21..c0228af 100644
---- a/server/jobs/dispatch.js
++++ b/server/jobs/dispatch.js
@@@ -4,7 +4,8 @@
* The dispatcher (plan T14 step L1): Bot hands platform.job@1 jobs to a paired Node over the device link and
* meters them (platform.job-frame@1, docs/protocol.md "Jobs"). The Run service, which owns run.job.*, calls it
* over the internal HTTP API (bot.job.dispatch): POST /api/v1/jobs, POST /api/v1/jobs/:id/cancel and
-- * GET /api/v1/jobs/:id (server/api/v1.js).
++ * GET /api/v1/jobs/:id (server/api/v1.js), through the jobs service server/jobs/index.js, which also holds the one
++ * createJobFrames instance the device socket (server/realtime.js) uses.
*
* dispatch(db, nodeId, job, { link, project, subject, provider }) validate, store, send (`job`)
* cancel(db, jobId, { link }) `job_cancel`
ddiff --git a/server/jobs/index.js b/server/jobs/index.js
nnew file mode 100644
iindex 0000000..f279ced
---- /dev/null
++++ b/server/jobs/index.js
@@@ -0,0 +1,51 @@
++'use strict';
++
++/**
++ * The jobs service (plan T14 step 6): one instance holds the device side of the dispatcher (createJobFrames,
++ * with the stdout rings) for both the device socket (server/realtime.js, bound with hub.bindJobs) and the
++ * internal jobs API (server/api/v1.js, bot.job.dispatch).
++ *
++ * createJobs({ db, hub, usage, log, now }) → {
++ * dispatch(nodeId, job, { project, subject, provider }) → { job, sent } validate, store, send (`job`)
++ * cancel(jobId) → { job, sent } `job_cancel`
++ * state(jobId) → { job, stdout } or null when unknown
++ * frames the device side (onAck, onConnect, onFrame)
++ * stdout(jobId) what is held of a job's stdout, or null
++ * }
++ *
++ * `db` is a getter (the domain's handle is bound after the hub exists); `hub` is the realtime hub, the link a
++ * job and its cancel go out on. `job` in the answers is presentJob's view of the run_jobs row.
++ */
++const { dispatch, cancel, createJobFrames } = require('./dispatch');
++const store = require('./store');
++const { json } = require('../util');
++
++function createJobs({ db: getDb, hub, usage = null, log = console, now = () => Date.now() }) {
++ const frames = createJobFrames({ db: getDb, usage, log, now });
++ const present = ({ job, sent }) => ({ job: presentJob(job), sent });
++ return {
++ frames,
++ stdout: (jobId) => frames.stdout(jobId),
++ dispatch: async (nodeId, job, opts = {}) => present(await dispatch(getDb(), nodeId, job, { link: hub, now, ...opts })),
++ cancel: async (jobId, opts = {}) => present(await cancel(getDb(), jobId, { link: hub, now, ...opts })),
++ async state(jobId) {
++ const row = await store.get(getDb(), jobId);
++ return row ? { job: presentJob(row), stdout: frames.stdout(row.id) } : null;
++ },
++ };
++}
++
++/** The job as the jobs API answers it: the run_jobs row (server/jobs/store.js) plus its original platform.job@1. */
++function presentJob(row) {
++ return {
++ id: row.id, node_id: row.node_id, class: row.class, state: row.state,
++ project_id: row.project_id, subject: row.subject, provider: row.provider,
++ cancel_requested: row.cancel_requested, fault_code: row.fault_code,
++ sent_at: row.sent_at, started_ms: row.started_ms, finished_at: row.finished_at,
++ exit_reason: row.exit_reason, exit_code: row.exit_code, wall_ms: row.wall_ms,
++ usage_read: row.usage_read, result: json(row.result, null), job: json(row.job, null),
++ created_at: row.created_at, updated_at: row.updated_at,
++ };
++}
++
++module.exports = { createJobs, presentJob };
ddiff --git a/server/realtime.js b/server/realtime.js
iindex 05ddb3d..2611332 100644
---- a/server/realtime.js
++++ b/server/realtime.js
@@@ -26,7 +26,6 @@ const { capabilities, ids } = require('openvibe-contracts');
const { json, iso, prefixedId } = require('./util');
const { userPrincipal, verifyService, verifyNode, isNodeToken, PRINCIPAL_SUB, decodePayload } = require('./api/auth');
const { getProfile } = require('./profiles');
--const { createJobFrames } = require('./jobs/dispatch');
const OPEN = 1;
const MAX_BACKLOG = 64; // frames a device may send before its authentication completes
@@@ -46,7 +45,7 @@ function parseCookies(header) {
const bearer = (req) => { const h = String(req.headers.authorization || ''); return h.startsWith('Bearer ') ? h.slice(7).trim() : null; };
const isSubject = (v) => typeof v === 'string' && ids.isSubjectId('user', v);
--function createRealtime({ config, keys, userAuth, usage = null, log = console, now = () => Date.now() }) {
++function createRealtime({ config, keys, userAuth, log = console, now = () => Date.now() }) {
let domain = null;
const deviceConns = new Map(); // device_id → conn
const subsByRobot = new Map(); // robot_id → Set<conn>
@@@ -62,7 +61,7 @@ function createRealtime({ config, keys, userAuth, usage = null, log = console, n
const watchConns = new Set(); // every /watch socket, joined or not
const watchersByAddr = new Map(); // client address → open /watch sockets
const offlineAfterMs = config.device.heartbeatMs * config.device.offlineMisses + config.device.offlineGraceMs;
-- const jobs = createJobFrames({ db: () => domain.db, usage, log, now });
++ let jobs = null; // the jobs service (server/jobs/index.js), bound by server/app.js
const nextSeq = (conn) => ++conn.seq;
function send(ws, obj) { if (ws.readyState === OPEN) { ws.send(JSON.stringify(obj)); return true; } return false; }
@@@ -73,7 +72,7 @@ function createRealtime({ config, keys, userAuth, usage = null, log = console, n
const jobLink = (conn) => ({ deviceId: conn.device.id, send: (type, fields) => sendFrame(conn, type, fields), error: (code, detail) => sendError(conn, code, detail) });
/** A socket's job frames are handled one at a time in arrival order: second n is queued before n+1, job_exit last. */
function jobFrame(conn, msg) {
-- const run = (conn.jobFrames || Promise.resolve()).then(() => jobs.onFrame(jobLink(conn), msg));
++ const run = (conn.jobFrames || Promise.resolve()).then(() => jobs.frames.onFrame(jobLink(conn), msg));
conn.jobFrames = run.catch(() => {});
return run;
}
@@@ -135,7 +134,7 @@ function createRealtime({ config, keys, userAuth, usage = null, log = console, n
await sendConfig(conn);
await bringOnline(conn);
} catch (e) { log.warn(`[Bot] device auth: ${e.message}`); conn.backlog = null; ws.close(4002, 'invalid credential'); return; }
-- await jobs.onConnect(jobLink(conn)).catch((e) => log.warn(`[Bot] job resend to ${conn.device.id}: ${e.message}`));
++ await jobs.frames.onConnect(jobLink(conn)).catch((e) => log.warn(`[Bot] job resend to ${conn.device.id}: ${e.message}`));
await drainBacklog(conn);
}
}
@@@ -329,7 +328,7 @@ function createRealtime({ config, keys, userAuth, usage = null, log = console, n
async function onAck(conn, msg) {
const id = String(msg.id || '');
-- if (id.startsWith('job_')) return jobs.onAck(jobLink(conn), msg); // a job's ack/nack, never a command's
++ if (id.startsWith('job_')) return jobs.frames.onAck(jobLink(conn), msg); // a job's ack/nack, never a command's
const p = pendingCmds.get(id);
if (!p || p.deviceId !== conn.device.id) return; // unknown, already answered, or another device's command
pendingCmds.delete(id);
@@@ -668,9 +667,9 @@ function createRealtime({ config, keys, userAuth, usage = null, log = console, n
return {
bindDomain(d) { domain = d; },
++ bindJobs(j) { jobs = j; },
handleUpgrade, attachSim, detachSim,
isOnline, deviceState, sendToDevice, sendToRobotDevices, broadcast, closeDevice, robotState, refreshConfig, pushEstop, closeWatchers,
-- jobStdout: (jobId) => jobs.stdout(jobId),
// Real devices only: a simulator is not a machine anyone runs.
onlineCount() { let n = 0; for (const c of deviceConns.values()) if (c.online && !c.sim) n++; return n; },
devices() { return [...deviceConns.values()].filter((c) => !c.sim).map((c) => ({ device_id: c.device.id, online: c.online })); },
ddiff --git a/test/run-jobs.test.js b/test/run-jobs.test.js
iindex 85d3174..1415363 100644
---- a/test/run-jobs.test.js
++++ b/test/run-jobs.test.js
@@@ -303,7 +303,10 @@ const PROJECT = `prj_${ids.ulid()}`;
assert.strictEqual(got.status, 200, got.text);
assert.strictEqual(got.json.job.state, 'placed');
assert.strictEqual(got.json.stdout.text, 'hello\n');
++ // One jobs service (server/jobs/index.js) holds the rings for the device socket and the API.
++ assert.deepStrictEqual(t.app.locals.jobs.stdout(job.id), got.json.stdout);
refused(await call('GET', '/api/v1/jobs/job_00000000000000000000000000'), 404, 'bot.job_not_found');
++ refused(await call('POST', '/api/v1/jobs/job_00000000000000000000000000/cancel'), 404, 'bot.job_not_found');
// Cancel asks the Node to stop it.
const c = await call('POST', `/api/v1/jobs/${job.id}/cancel`);
assert.strictEqual(c.status, 200, c.text);