openvibers/openvibe.bot Commits

Serve the internal bot.job.dispatch jobs API for Run

OP OpenVibers committed 4 Oct 2026 · 824b4d2c33 · 6 files, +111 −8

Asked: Follow-up of https://github.com/OpenVibers/OpenVibe.Contracts/pull/37 (Add internal bot.job.dispatch capability for Run-to-Bot job dispatch): its review asked for a change in OpenVibe.Bot that the PR cannot make from OpenVibe.Contracts. The review said: # Review: OpenVibe.Contracts uncommitted chan…
Harness job bot-follow-up-of-https-github-com-openvi-234631.

Changes

ddiff --git a/README.md b/README.md
iindex f715552..6108da1 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`, no HTTP route) 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/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
@@@ -45,6 +45,7 @@ OpenVibe.Bot pairs a robot's machines, keeps their state and gates every operato
| `bot.robot.manage` | create, patch and delete a robot, pairing codes, operators |
| `bot.robot.control` | the e-stop set and clear (clear is owner-only) and the control gate |
| `bot.device.connect` | rotate and revoke a device credential |
++| `bot.job.dispatch` | the internal Run → Bot jobs API (`POST /jobs`, `POST /jobs/:id/cancel`, `GET /jobs/:id`); services only |
A person acts on their own robots with a Network user token; a service acts with a service token plus the capability, for the subject it names. The REST routes are in `docs/protocol.md` §3 and the two WebSockets in §§1–2.
ddiff --git a/STATUS.json b/STATUS.json
iindex c0d290c..00a8bed 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -15,7 +15,8 @@
"bot.robot.read",
"bot.robot.manage",
"bot.robot.control",
-- "bot.device.connect"
++ "bot.device.connect",
++ "bot.job.dispatch"
],
"events": [
"bot.robot.online",
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex b869fed..a46117c 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -243,9 +243,10 @@ heartbeat or any frame brings it back. Commands are never replayed after a recon
### 1.2 Jobs (platform.job-frame@1)
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. There is no HTTP route; the Run service (which owns `run.job.*`) calls
--`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
++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
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).
@@@ -421,7 +422,9 @@ and `POST /devices/bind` answers `401 bot.sign_in`: `GET|POST /robots`, `GET|PAT
`POST /robots/:id/pairing-code`, `GET|POST /robots/:id/operators`, `DELETE /robots/:id/operators/:subject`,
`GET /robots/:id/devices`, `POST /devices/:id/rotate|revoke`, `POST /robots/:id/estop`,
`POST /robots/:id/estop/clear` and `GET /robots/:id/audit`. A node token gets `403 bot.forbidden` on the same
--routes. Both come before any lookup, so a real robot or device id answers exactly as a made-up one.
++routes. Both come before any lookup, so a real robot or device id answers exactly as a made-up one. The three
++jobs routes are service-only: the same `401 bot.sign_in` without a token, `403 bot.forbidden` for a person or a
++node token, and a service token is judged on `bot.job.dispatch` alone.
| Method & path | Auth (capability) | Answer |
|---|---|---|
@@@ -442,6 +445,7 @@ routes. Both come before any lookup, so a real robot or device id answers exactl
| `POST /robots/:id/estop` | owner/operator, `bot.robot.control` | `{ robot }` |
| `POST /robots/:id/estop/clear` | **owner only**, `bot.robot.control` | `{ robot }` |
| `GET /robots/:id/audit?limit=&before=` | owner, or `bot.robot.read` | `{ audit[], next_before }` (newest first) |
++| `POST /jobs`, `POST /jobs/:id/cancel`, `GET /jobs/:id` | **a service only**, with `bot.job.dispatch` (Run → Bot; a person or a node token is `403 bot.forbidden`) | `201 { job, sent }` / `{ job, sent }` / `{ job, stdout }`; `422 bot.invalid_input` without `project_id`, `409 bot.class_unadvertised` when the device does not advertise the job's class, `404 bot.job_not_found` |
| `POST /pair` | the one-time code is the credential | `201 { device_id, credential, publish_key, whip_url, robot_id, profile }`; with `BOT_PAIRING_AUTHORITY=network` always `410 bot.pairing_moved` (the `detail` names the Network URL) |
Not in `/api/v1`: `GET /api/health`, `GET /api/ready`, `GET /metrics`, `GET /release.json`, the
ddiff --git a/server/api/v1.js b/server/api/v1.js
iindex 4ad7048..b3adbe9 100644
---- a/server/api/v1.js
++++ b/server/api/v1.js
@@@ -23,6 +23,7 @@
* | POST /robots/:id/estop | bot.robot.control | owner or operator |
* | POST /robots/:id/estop/clear | bot.robot.control | the owner only |
* | GET /robots/:id/audit | bot.robot.read | owner (paged) |
++ * | POST /jobs, /jobs/:id/cancel, GET /jobs/:id | bot.job.dispatch | — (services only) |
* | POST /pair | — (one-time code) | the agent (code is the credential) |
*
* With BOT_PAIRING_AUTHORITY=network the pairing codes are Network's and POST /pair answers 410 bot.pairing_moved.
@@@ -31,12 +32,15 @@ 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',
manage: 'bot.robot.manage',
control: 'bot.robot.control',
connect: 'bot.device.connect',
++ dispatch: 'bot.job.dispatch',
};
function v1Router({ domain, apiAuth, limits, hub }) {
@@@ -45,6 +49,7 @@ function v1Router({ domain, apiAuth, limits, hub }) {
const manage = limits('bot.robot.manage', { minute: 30, hour: 300 });
const control = limits('bot.control', { minute: 120, hour: 1200 });
const pair = limits('bot.pair', { minute: 20, hour: 200 });
++ const jobs = limits('bot.jobs', { minute: 120, hour: 1200 });
const wrap = (fn) => (req, res, next) => Promise.resolve(fn(req, res, next)).catch(next);
const me = (req) => (req.principal.kind === 'user' ? req.principal.subject : null);
@@@ -70,6 +75,12 @@ function v1Router({ domain, apiAuth, limits, hub }) {
const requireManage = (req, extra) => { if (req.principal.kind === 'service' && !apiAuth.granted(req.principal, CAP.manage)) fail(403, 'capability.denied', `${CAP.manage} not granted`); return actedFor(req, extra); };
const requireRead = (req) => { fence(req); if (req.principal.kind === 'service' && !apiAuth.granted(req.principal, CAP.read)) fail(403, 'capability.denied', `${CAP.read} not granted`); };
const requireControlCap = (req) => { if (req.principal.kind === 'service' && !apiAuth.granted(req.principal, CAP.control)) fail(403, 'capability.denied', `${CAP.control} not granted`); };
++ /** Run → Bot job dispatch is a service-to-service call: only a service token holding bot.job.dispatch. */
++ const requireJobDispatch = (req) => {
++ if (req.principal.kind === 'anonymous') fail(401, 'bot.sign_in', 'sign in to do that');
++ if (req.principal.kind !== 'service') fail(403, 'bot.forbidden', 'job dispatch is a service-to-service call');
++ if (!apiAuth.granted(req.principal, CAP.dispatch)) fail(403, 'capability.denied', `${CAP.dispatch} not granted`);
++ };
// Both fence before the lookup: without a token (or with a node token) a real robot id and a made-up one
// get the same answer.
@@@ -233,6 +244,29 @@ 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) ─────────
++ // 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 });
++ }));
++ 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 });
++ }));
++ 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) });
++ }));
++
// ── Pairing without a prior socket (the agent may POST instead of using the WS) ───────────────
r.post('/pair', pair, wrap(async (req, res) => {
const result = await domain.pairing.redeem({
@@@ -257,6 +291,19 @@ 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/jobs/dispatch.js b/server/jobs/dispatch.js
iindex 2166f63..0851f21 100644
---- a/server/jobs/dispatch.js
++++ b/server/jobs/dispatch.js
@@@ -2,8 +2,9 @@
/**
* 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"). There is no HTTP route: the Run service, which
-- * owns run.job.*, calls this module later.
++ * 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).
*
* dispatch(db, nodeId, job, { link, project, subject, provider }) validate, store, send (`job`)
* cancel(db, jobId, { link }) `job_cancel`
ddiff --git a/test/run-jobs.test.js b/test/run-jobs.test.js
iindex 39a58fb..85d3174 100644
---- a/test/run-jobs.test.js
++++ b/test/run-jobs.test.js
@@@ -267,10 +267,59 @@ const PROJECT = `prj_${ids.ulid()}`;
await relay.stop();
});
++ await check('the jobs API serves bot.job.dispatch (dispatch, state with stdout, cancel) to a service only', async () => {
++ const n = await node();
++ const job = newJob();
++ const body = { node_id: n.deviceId, job, project_id: PROJECT, subject: SUBJECT };
++ const call = (method, p, opts = {}) => t.call(method, p, { cap: ['bot.job.dispatch'], ...opts });
++ // A person, a service without the capability, a node token and no token at all are all refused.
++ refused(await call('POST', '/api/v1/jobs', { user: alex, body }), 403, 'bot.forbidden');
++ refused(await call('POST', '/api/v1/jobs', { cap: ['bot.robot.read'], body }), 403, 'capability.denied');
++ refused(await call('POST', '/api/v1/jobs', { token: t.network.signNode(n.principal), body }), 403, 'bot.forbidden');
++ refused(await call('POST', '/api/v1/jobs', { token: null, body }), 401, 'bot.sign_in');
++ // project_id is required: Run is the payer.
++ refused(await call('POST', '/api/v1/jobs', { body: { ...body, project_id: undefined } }), 422, 'bot.invalid_input');
++ // The dispatch reaches the Node once; the same job again is idempotent.
++ const r = await call('POST', '/api/v1/jobs', { body });
++ assert.strictEqual(r.status, 201, r.text);
++ assert.strictEqual(r.json.sent, true);
++ assert.strictEqual(r.json.job.state, 'queued');
++ await n.c.waitFor((m) => m.type === 'job' && m.job.id === job.id);
++ n.say('ack', { id: job.id });
++ await poll(async () => (await row(job.id)).state === 'placed');
++ // The same job id again is idempotent: the same row, not sent again once it is placed.
++ const again = await call('POST', '/api/v1/jobs', { body });
++ assert.strictEqual(again.json.job.id, job.id);
++ assert.deepStrictEqual(again.json.job.job, job);
++ assert.strictEqual(again.json.sent, false);
++ // A class the device does not advertise is refused and never stored.
++ const code = newJob({ class: 'code' });
++ refused(await call('POST', '/api/v1/jobs', { body: { ...body, job: code } }), 409, 'bot.class_unadvertised');
++ assert.strictEqual(await row(code.id), null);
++ // GET answers the state and the captured stdout.
++ n.say('job_stdout', { id: job.id, chunk_seq: 1, chunk: 'hello\n' });
++ await poll(async () => (await call('GET', `/api/v1/jobs/${job.id}`)).json.stdout);
++ const got = await call('GET', `/api/v1/jobs/${job.id}`);
++ assert.strictEqual(got.status, 200, got.text);
++ assert.strictEqual(got.json.job.state, 'placed');
++ assert.strictEqual(got.json.stdout.text, 'hello\n');
++ refused(await call('GET', '/api/v1/jobs/job_00000000000000000000000000'), 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);
++ assert.strictEqual(c.json.job.state, 'cancelled');
++ await n.c.waitFor((m) => m.type === 'job_cancel' && m.id === job.id);
++ });
++
await t.close();
done();
})().catch((e) => { console.error(e); process.exit(1); });
++function refused(res, status, code) {
++ assert.strictEqual(res.status, status, `${res.status} ${res.text}`);
++ assert.strictEqual(res.json && res.json.code, code, JSON.stringify(res.json));
++}
++
async function poll(fn, ms = 20000) {
const until = Date.now() + ms;
for (;;) {