openvibers/openvibe.bot Commits

Add a service-token command route for a bound channel's chat (#28)

OP OpenVibers committed 5 Oct 2026 · 8779aa6dd2 · 10 files, +324 −10

* Add a service-token command route for a bound channel's chat

**Files changed**

- `server/realtime.js`: added and exported `operatorCommand(robotId, principal, subject, msg)` — builds a synthetic operator connection whose fake socket captures the `command_result` frame, then runs the existing `onOperatorCommand` path, so the HTTP caller shares the same gate, audit and per-subject idempotency (`onAck`/`expireCommand`/refusals all answer it).
- `server/api/v1.js`: added `POST /robots/:id/commands` (rate-limited with the `control` limiter, `requireControlCap`, `actedFor` for `X-OV-Subject`), which awaits `hub.operatorCommand` and returns `{ robot_id, id, result, code?, reason?, latency_ms?, cached? }`; refusals become RFC 9457 problems via `commandRefusalStatus`. Updated the route table comment.
- `test/control.test.js`: two checks — a `bot.robot.control` service acting for an operator forwards a command, the device acks, the HTTP response is `result: 'ack'` and the audit records it (plus cached idempotent replay); a service acting for a chat viewer with no role gets `403 bot.not_an_operator`, an invited viewer gets `403 bot.read_only`, and a service without the capability gets `capability.denied`.
- `README.md`: capability table row for `bot.robot.control` now names the HTTP command route.
- `STATUS.json`: the control-gate feature line now names `POST /robots/:id/commands`.
- `docs/protocol.md`: §2 notes the HTTP gate, §3 adds the route to the unauthenticated-route list and the endpoint table with its refusal-status mapping.

**Tests run**

- `ov test test/control.test.js` → 1 passed (7.1 s), after the final realtime.js change.
- `ov test test/docs.test.js test/control.test.js` → docs passed; control skipped as unchanged-since-green at that point.

**Not done**

- Nothing from the brief was skipped. I did not commit/push/deploy/install. There is no CHANGELOG in the repo, so none was updated.

One interpretation worth flagging: the brief's "a viewer is refused `bot.not_an_operator`" — a subject with no role gets `bot.not_an_operator`, while an invited `viewer`-role member gets `bot.read_only` (the gate's established behaviour, since `prepare` maps viewer → read-only). Both are tested.

* Per-person command limits, 409 for in-flight replays, settle pending on close, more gate tests (review)

**Changes (file:line)**

1. **Per-person rate limit** — `server/api/actor-limits.js:14` `actor()` now returns `user:<subject>` when `req.limitSubject` is set; `:44-52` new `limiter.actedFor(name, own)` validates a service's `X-OV-Subject`/body `owner` and sets it (a person's own token is unchanged). `server/api/v1.js:52` adds `commandControl = limits.actedFor('bot.control', …)`; `:240` the route uses it. Test `test/control.test.js:407` (A exhausts 120/min, B still gets 409 not 429).
2. **In-flight replay → 409 `bot.command_pending`** — `server/realtime.js:440` answers `refused`/`bot.command_pending` instead of `pending`; `server/api/v1.js:313` maps the code to 409; documented `docs/protocol.md:455` and `:375`. Test `test/control.test.js:395` (concurrent same id: second 409, first ack); the shared-path expectation in `test/device.test.js:137` updated.
3. **`hub.close()` settles pending commands** — `server/realtime.js:716-722` sends `expired` to each pending command's connection and clears `inflight` before `pendingCmds.clear()`, so each HTTP request gets exactly one answer. Test `test/control.test.js:438`.
4. **Queue-join difference documented** — `docs/protocol.md:300`: the HTTP gate does not queue a no-role caller, so a chat viewer needs an explicit owner/operator role.
5. **New route tests** — `test/control.test.js:341` (own user token + mismatched `X-OV-Subject` 403), `:356` (service without subject 422 / anonymous 401), `:365` (device offline 409 + latched e-stop 409), `:380` (different subject same id isn't served the cached result), `:429` (never-acked command expires).

**Test results**
- `ov test test/control.test.js` → 1 passed (9.4 s)
- `ov test test/docs.test.js` → 1 passed (0.5 s)
- `node test/device.test.js` → all passed; `node test/api.test.js` → all passed (run directly for the shared realtime/limiter paths; both re-verified)

* Cap a service's command rate overall, not only per acted-for subject (review)

- `server/api/actor-limits.js:50` — `actedFor(name, own, service)` now builds a second limiter (`<name>.service`) and, for a service principal, runs it first (while `req.limitSubject` is still unset, so it keys on `principal.sub`), then the per-subject limiter; a person's own token takes the same path as before.
- `server/api/v1.js:55` — the command route's `actedFor` passes the per-service cap; `server/api/v1.js:45` takes `config`.
- `server/config.js:152` — configurable `controlServiceMinute`/`controlServiceHour` (defaults 1200/12000, i.e. 10× the 120/1200 per-person bucket) via `BOT_LIMITS_CONTROL_SERVICE_MINUTE`/`_HOUR`.
- `server/app.js:93` — pass `config` into `v1Router`.
- `test/control.test.js:453` — new check on an isolated `boot` with a small cap: a service rotates 30 well-formed non-member `usr_…` subjects (all 403), the 31st gets 429 `rate_limited`; the existing per-subject test (A exhausted, B allowed) is kept.
- `docs/protocol.md:455` — the command-route note now mentions the 1200/min + 12000/hour per-service ceiling (`bot.control.service`).

Test results: `ov test test/control.test.js test/api.test.js` → 2 passed (`api.test.js` 5.5 s, `control.test.js` 25.9 s). Final re-run: `control.test.js` ✓ 22.0 s, `api.test.js` skipped (unchanged since green). I also confirmed the new check fails without the service bucket. No commit or push was made.

Changes

ddiff --git a/README.md b/README.md
iindex d70e04c..2a0bf78 100644
---- a/README.md
++++ b/README.md
@@@ -43,7 +43,7 @@ OpenVibe.Bot pairs a robot's machines, keeps their state and gates every operato
|---|---|
| `bot.robot.read` | `GET /robots`, `GET /robots/:id`, operators, devices and the audit, for a service acting for an owner |
| `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.robot.control` | the e-stop set and clear (clear is owner-only), the control gate and `POST /robots/:id/commands` (so a bound channel's chat can forward a command over HTTP instead of a hardware socket) |
| `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 |
ddiff --git a/STATUS.json b/STATUS.json
iindex 8eb1e09..fd1090c 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -29,7 +29,7 @@
"deployment": "Production at https://openvibe.bot on openvibe-ovh since 2026-10-02: release layout /opt/openvibe.bot, unit openvibe-bot.service (WorkingDirectory /opt/openvibe.bot, EnvironmentFile /etc/openvibe/bot.env, PORT 4630, TimeoutStopSec 10), behind the nginx vhost deploy/nginx/openvibe.bot.conf, deployed with `ov deploy bot`. Liveness /api/health and readiness /api/ready are loopback-only.",
"features": [
"device WebSocket /device: one outbound socket per machine, credential or Network node token, heartbeat and telemetry, commands never replayed after a reconnect",
-- "operator WebSocket /control and the REST API (/api/v1) over one command gate: roles, per-robot limits and allowlists, cooldowns and the turn budget",
++ "operator WebSocket /control and the REST API (/api/v1, including POST /robots/:id/commands so a bound channel's chat can forward a command over HTTP) over one command gate: roles, per-robot limits and allowlists, cooldowns and the turn budget",
"pairing with one-time codes, minted by Bot or delegated to OpenVibe.Network (BOT_PAIRING_AUTHORITY=network); a node machine binds on first use and reauths every 300 s",
"device credentials hashed at rest and rotatable; revoke closes the socket at once and revokes the Network principal for a node machine",
"profiles instead of per-robot rows: adeept.adr036 (+ mecanum variant), cozmo, camera.onvif, sim.rover, seeded from server/profiles/*.json",
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 9f16b91..a7df725 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -296,6 +296,10 @@ twice.
**service token** with capability `bot.robot.control` (audience `openvibe.bot`) plus
`X-OV-Subject: usr_…` naming the person it acts for. Without one the socket closes with code **4002**.
The gate then applies the **acted-for subject's** role — a service acting for a viewer is a viewer.
++The same gate is reachable over HTTP as `POST /robots/:id/commands` (§3), so a bound channel's chat can
++forward commands without holding a socket. The HTTP gate is the same gate but not the same join: a caller
++with no role is not put into a `queue` robot's queue (as a socket `join` would), so over HTTP a chat viewer
++needs an explicit owner or operator role — a no-role caller is refused `bot.not_an_operator`.
### Client → server
@@@ -367,6 +371,8 @@ Refusal codes (the gate, in order): `bot.robot_not_found`, `bot.unknown_command`
(the profile takes no such kind), `bot.estop_latched`, `bot.not_an_operator`, `bot.sign_in`,
`bot.read_only`, `bot.command_not_allowed` (the role's allowlist), `bot.device_offline`, `bot.cooldown`,
`bot.not_your_turn`, `bot.turn_budget`, `bot.unknown_actuator`, `bot.invalid_input`, `bot.text_too_long`.
++A `command` whose `id` is still in flight (the first one has not been acked) is answered `refused`
++`bot.command_pending` rather than a second device command or a silent wait.
`halt` is never refused for the e-stop, the allowlist, a cooldown or the turn budget; it still needs a
role that may drive (a queue robot's turn holder, not someone waiting) and an online device.
@@@ -423,7 +429,7 @@ the subject it acts for). Without a token, every route but `GET /profiles`, `GET
and `POST /devices/bind` answers `401 bot.sign_in`: `GET|POST /robots`, `GET|PATCH|DELETE /robots/:id`,
`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
++`POST /robots/:id/estop/clear`, `POST /robots/:id/commands` 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. 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.
@@@ -446,6 +452,7 @@ node token, and a service token is judged on `bot.job.dispatch` alone.
| `POST /devices/bind` | a Network node token (audience `openvibe.bot`), no body | `201 { device_id, publish_key, whip_url, robot_id, profile }` — `POST /pair`'s answer without `credential`; again → the same device and a new publish key (the old one stops working); without OpenRe `video: "not_configured"` instead of the key. Refused: `401 bot.node_token_required`, `403 bot.node_not_bound` |
| `POST /robots/:id/estop` | owner/operator, `bot.robot.control` | `{ robot }` |
| `POST /robots/:id/estop/clear` | **owner only**, `bot.robot.control` | `{ robot }` |
++| `POST /robots/:id/commands` | owner/operator, `bot.robot.control` | `{ robot_id, id, result, code?, reason?, latency_ms?, cached? }` — one command through the same gate, audit and per-subject `id` idempotency key as `/control` (a bound channel's chat forwards here instead of a hardware socket). `result` is `ack`, `nack` or `expired` (a repeated `id` answers `cached: true`); a gate refusal is its code as an RFC 9457 problem: `403 bot.not_an_operator` / `bot.read_only` / `bot.not_your_turn`, `404 bot.robot_not_found`, `409 bot.device_offline` / `bot.estop_latched` / `bot.command_pending` (the same `id` is still in flight), `422` the command-shape codes, `429 bot.cooldown` / `bot.turn_budget`. A service acting for a subject is rate-limited as that person (`user:<subject>`) and also, as a service, at 1200 a minute and 12000 an hour on its own token (`bot.control.service`), so naming a fresh well-formed subject on every request never lifts every ceiling |
| `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) |
ddiff --git a/server/api/actor-limits.js b/server/api/actor-limits.js
iindex ffa3525..c0ee1e8 100644
---- a/server/api/actor-limits.js
++++ b/server/api/actor-limits.js
@@@ -6,8 +6,12 @@
* Never limited: /api/health, /api/ready, /release.json, /metrics and the one-line GET /.
*/
const { createActorLimiter, createValkeyLimitStore, defaultActor } = require('openvibe-sdk/limits');
++const { userSubject } = require('../util');
function actor(req) {
++ // A route a service may call on a person's behalf (limiter.actedFor below) counts the call against that
++ // person, exactly as their own token would — never against the shared service principal.
++ if (req.limitSubject) return `user:${req.limitSubject}`;
const p = req.principal;
if (p && (p.kind === 'service' || p.kind === 'node') && p.sub) return p.sub;
if (p && p.kind === 'user' && p.subject) return `user:${p.subject}`;
@@@ -30,6 +34,38 @@ function createActorLimits({ config, valkey = null, now = () => Date.now(), regi
const limit = limiter(name);
return (req, res, next) => (req.method === 'GET' || req.method === 'HEAD' ? limit(req, res, next) : next());
};
++ /**
++ * A route a service may call on a person's behalf (the command route: a bound channel's chat forwards a
++ * viewer's command). When the caller is a service naming a subject (`X-OV-Subject` / body `owner`), the
++ * call is counted against `user:<subject>` exactly as that person's own token is, so one shared service
++ * token never puts every viewer in one bucket. A person's own token is keyed as before; an invalid subject
++ * is left for the route to refuse.
++ *
++ * Pass `service` to also count a service principal in its own bucket (`<name>.service`, keyed on its
++ * `principal.sub`): `userSubject` only checks the `usr_…` shape, so a service naming a fresh well-formed
++ * subject on every request would otherwise mint a new `user:<subject>` bucket each time and never reach a
++ * ceiling. That bucket is counted first, while `req.limitSubject` is still unset, so it can never be
++ * attributed to the person.
++ */
++ limiter.actedFor = (name, own = {}, service = null) => {
++ const perSubject = limiter(name, own);
++ const perService = service ? limiter(`${name}.service`, service) : null;
++ const namedSubject = (req) => {
++ const s = req.headers['x-ov-subject'] || req.body?.owner;
++ if (s == null || s === '') return null;
++ try { return userSubject(s, 'subject'); } catch { return null; } // the route refuses it
++ };
++ return (req, res, next) => {
++ if (!req.principal || req.principal.kind !== 'service') return perSubject(req, res, next);
++ const countSubject = () => {
++ const s = namedSubject(req);
++ if (s != null) req.limitSubject = s;
++ return perSubject(req, res, next);
++ };
++ if (!perService) return countSubject();
++ return perService(req, res, (err) => (err ? next(err) : countSubject()));
++ };
++ };
return limiter;
}
ddiff --git a/server/api/v1.js b/server/api/v1.js
iindex a1b7127..8baed7c 100644
---- a/server/api/v1.js
++++ b/server/api/v1.js
@@@ -22,6 +22,7 @@
* | POST /devices/bind | — (a node token) | a machine paired for Bot on Network |
* | POST /robots/:id/estop | bot.robot.control | owner or operator |
* | POST /robots/:id/estop/clear | bot.robot.control | the owner only |
++ * | POST /robots/:id/commands | bot.robot.control | owner or operator |
* | 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) |
@@@ -41,11 +42,18 @@ const CAP = {
dispatch: 'bot.job.dispatch',
};
--function v1Router({ domain, apiAuth, limits, hub, jobs: jobService }) {
++function v1Router({ domain, apiAuth, limits, hub, jobs: jobService, config }) {
const r = express.Router();
r.use(limits.reads('bot.read'));
const manage = limits('bot.robot.manage', { minute: 30, hour: 300 });
const control = limits('bot.control', { minute: 120, hour: 1200 });
++ // The command route counts against the acted-for person, not the service token: a bound channel's chat
++ // forwards many viewers' commands through one service, and each viewer gets their own bucket. A service
++ // is also capped by its own token (bot.control.service, 10× the per-person bucket): the subject is only
++ // shape-checked here, so a service naming a fresh well-formed viewer on every request would otherwise
++ // never reach a ceiling.
++ const commandControl = limits.actedFor('bot.control', { minute: 120, hour: 1200 },
++ { minute: config.actorLimits.controlServiceMinute, hour: config.actorLimits.controlServiceHour });
const pair = limits('bot.pair', { minute: 20, hour: 200 });
const jobs = limits('bot.jobs', { minute: 120, hour: 1200 });
@@@ -230,6 +238,20 @@ function v1Router({ domain, apiAuth, limits, hub, jobs: jobService }) {
res.json({ robot: domain.present.robot(await domain.robots.get(req.params.id)) });
}));
++ // ── Commands (the /control gate over HTTP: a service forwards a bound channel's chat commands) ─
++ // The same onOperatorCommand path as the /control socket — the gate, the audit and the per-subject
++ // idempotency key are shared. `id` is that idempotency key; a repeated id is answered from the cache.
++ r.post('/robots/:id/commands', commandControl, wrap(async (req, res) => {
++ requireControlCap(req);
++ const subject = actedFor(req);
++ const principal = req.principal.kind === 'service' ? { ...req.principal, subject } : req.principal;
++ const result = await hub.operatorCommand(req.params.id, principal, subject, {
++ id: req.body?.id, kind: req.body?.kind, value: req.body?.value, ms: req.body?.ms,
++ });
++ if (result.result === 'refused') fail(commandRefusalStatus(result.code), result.code, result.reason || null);
++ res.json({ robot_id: req.params.id, ...result });
++ }));
++
// ── Audit ─────────────────────────────────────────────────────────────────────────────────────
r.get('/robots/:id/audit', wrap(async (req, res) => {
await ownerOrRead(req, req.params.id);
@@@ -287,6 +309,15 @@ function v1Router({ domain, apiAuth, limits, hub, jobs: jobService }) {
return r;
}
++/** The HTTP status a refused command's gate code answers with (the /control refusal codes, docs/protocol.md §2). */
++function commandRefusalStatus(code) {
++ if (code === 'bot.robot_not_found') return 404;
++ if (code === 'bot.not_an_operator' || code === 'bot.read_only' || code === 'bot.forbidden' || code === 'bot.not_your_turn') return 403;
++ if (code === 'bot.cooldown' || code === 'bot.turn_budget') return 429;
++ if (code === 'bot.device_offline' || code === 'bot.estop_latched' || code === 'bot.command_pending') return 409;
++ return 422; // unknown_command, command_not_allowed, unknown_actuator, invalid_input, text_too_long
++}
++
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 623960f..3ada7f3 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -90,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, jobs }));
++ 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, config }));
app.use('/auth', createSessionRoutes(config, userAuth, { fetchImpl }));
app.use(viewerMiddleware(userAuth));
app.use(createWebRoutes(config, { domain, sim, limits, log }));
ddiff --git a/server/config.js b/server/config.js
iindex bceb086..8dfa682 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -147,6 +147,10 @@ function loadConfig(env = process.env) {
actorLimits: {
minute: Math.max(1, int(env.BOT_LIMITS_MINUTE, 120)),
hour: Math.max(1, int(env.BOT_LIMITS_HOUR, 3000)),
++ // The command route also caps a service by its own token (server/api/v1.js; a service may name a
++ // different subject on every request, so the per-person bucket alone has no overall ceiling).
++ controlServiceMinute: Math.max(1, int(env.BOT_LIMITS_CONTROL_SERVICE_MINUTE, 1200)),
++ controlServiceHour: Math.max(1, int(env.BOT_LIMITS_CONTROL_SERVICE_HOUR, 12000)),
},
jobs: {
ddiff --git a/server/realtime.js b/server/realtime.js
iindex 2611332..668ddd5 100644
---- a/server/realtime.js
++++ b/server/realtime.js
@@@ -434,7 +434,10 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
const key = `${conn.subject || `${conn.principal.kind}:${conn.principal.id || ''}`}|${id}`;
const cached = results.get(key);
if (cached) return sendFrame(conn, 'command_result', { id, result: cached.result, reason: cached.reason, cached: true });
-- if (inflight.has(key)) return sendFrame(conn, 'command_result', { id, result: 'pending', cached: true });
++ // The same id is still awaiting its ack/nack: never a second device command, and never a silent wait.
++ // A refusal code (not `result: pending`) so the HTTP route answers 409 bot.command_pending like the
++ // other refusals; the first request still gets its own result.
++ if (inflight.has(key)) return sendFrame(conn, 'command_result', { id, result: 'refused', code: 'bot.command_pending', reason: 'the same command id is still in flight' });
// Reserved before the first await: two frames with one id are handled concurrently, and only the first may
// reach the robot. Every path below either hands the reservation to a pending command or releases it.
inflight.set(key, null);
@@@ -479,6 +482,39 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
function cacheResult(key, r) { results.set(key, { ...r, at: now() }); if (results.size > 4096) results.delete(results.keys().next().value); }
++ /**
++ * One operator command off the REST route POST /robots/:id/commands (api/v1.js): the same gate, audit
++ * and per-operator idempotency as the /control socket, but the command_result is returned instead of
++ * written to a socket. `principal` is the caller ({ kind, sub, cap } for a service), `subject` who it
++ * acts for. Resolves with { id, result, code?, reason?, latency_ms?, cached? }.
++ */
++ function operatorCommand(robotId, principal, subject, msg) {
++ return new Promise((resolve, reject) => {
++ let settled = false;
++ const finish = (result) => { if (!settled) { settled = true; resolve(result); } };
++ // Every path through onOperatorCommand answers with a command_result: a cache hit or a gate
++ // refusal at once, and a device command on its ack/nack or on expireCommand (which is always
++ // scheduled). So the HTTP request is answered without a timeout of its own.
++ const conn = {
++ ws: {
++ readyState: OPEN,
++ send(raw) {
++ let frame; try { frame = JSON.parse(String(raw)); } catch { return; }
++ if (frame.type !== 'command_result') return;
++ const r = { id: frame.id != null ? String(frame.id) : null, result: frame.result };
++ if (frame.code != null) r.code = frame.code;
++ if (frame.reason != null) r.reason = frame.reason;
++ if (frame.latency_ms != null) r.latency_ms = frame.latency_ms;
++ if (frame.cached) r.cached = true;
++ finish(r);
++ },
++ },
++ seq: 0, principal, subject, robotId, role: null,
++ };
++ onOperatorCommand(conn, msg).catch((e) => { if (!settled) { settled = true; reject(e); } });
++ });
++ }
++
async function onOperatorEstop(conn, latched) {
if (!conn.robotId) return sendError(conn, 'bot.not_joined', 'join a robot first');
const robotId = conn.robotId;
@@@ -669,14 +705,21 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
bindDomain(d) { domain = d; },
bindJobs(j) { jobs = j; },
handleUpgrade, attachSim, detachSim,
-- isOnline, deviceState, sendToDevice, sendToRobotDevices, broadcast, closeDevice, robotState, refreshConfig, pushEstop, closeWatchers,
++ isOnline, deviceState, sendToDevice, sendToRobotDevices, broadcast, closeDevice, robotState, refreshConfig, pushEstop, closeWatchers, operatorCommand,
// 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 })); },
async close() {
closed = true;
for (const t of timers) clearInterval(t);
-- for (const p of pendingCmds.values()) if (p.timer) clearTimeout(p.timer);
++ // A command the device never acked settles as `expired` before the hub forgets it, so every HTTP
++ // request waiting on operatorCommand gets exactly one answer instead of hanging until the process dies.
++ for (const p of pendingCmds.values()) {
++ if (p.timer) clearTimeout(p.timer);
++ inflight.delete(p.key);
++ cacheResult(p.key, { result: 'expired', reason: 'the server closed before the device acknowledged' });
++ sendFrame(p.conn, 'command_result', { id: p.opId, result: 'expired', reason: 'the server closed before the device acknowledged' });
++ }
pendingCmds.clear();
inflight.clear();
for (const conn of [...deviceConns.values()]) { try { conn.ws.close(1001, 'server closing'); } catch { /* gone */ } }
ddiff --git a/test/control.test.js b/test/control.test.js
iindex 3c57246..05636e2 100644
---- a/test/control.test.js
++++ b/test/control.test.js
@@@ -3,6 +3,7 @@
// operator, cooldowns, the latched e-stop (owner-cleared only), the turn queue and a service acting for
// a viewer. Every decision is audited.
const assert = require('assert');
++const { ids } = require('openvibe-contracts');
const { boot, check, done } = require('./helpers/app');
const { loadProfiles } = require('../server/profiles');
const plugins = require('./helpers/plugins');
@@@ -275,6 +276,198 @@ const plugins = require('./helpers/plugins');
assert.strictEqual(await anon.waitForClose(), 4002, 'an anonymous socket is refused');
});
++ await check('a service token forwards a command over HTTP and the device\'s ack comes back', async () => {
++ const { robot, pairing } = await t.robot(alex);
++ await invite(robot.id, bob.subject, 'operator');
++ const dev = await deviceFor(robot.id, pairing.code);
++ const svc = t.network.signService({ sub: 'svc:live', cap: ['bot.robot.control'], aud: ['openvibe.bot'] });
++ const call = (id) => t.call('POST', `/api/v1/robots/${robot.id}/commands`, {
++ token: svc, headers: { 'X-OV-Subject': bob.subject }, body: { id, kind: 'drive', value: { throttle: 0.2 } },
++ });
++ const pending = call('chat-1'); // the route waits for the command_result, so ack before awaiting it
++ const c = await dev.waitFor((m) => m.type === 'command' && m.ref === 'chat-1');
++ assert.strictEqual(c.kind, 'drive');
++ dev.send({ type: 'ack', id: c.id });
++ const r = await pending;
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.json.result, 'ack');
++ assert.strictEqual(r.json.id, 'chat-1');
++ const audit = await t.domain.audit.list(robot.id, { limit: 5 });
++ assert.ok(audit.some((x) => x.kind === 'drive' && x.result === 'ack' && x.operator_kind === 'service'), 'the HTTP command is audited like a socket one');
++ // The id is the same per-subject idempotency key as the socket: a repeated id is answered from the cache.
++ const again = await call('chat-1');
++ assert.strictEqual(again.json.cached, true);
++ assert.strictEqual(again.json.result, 'ack');
++ dev.close();
++ });
++
++ await check('the HTTP command route refuses a service acting for a viewer, and a service without bot.robot.control', async () => {
++ const { robot, pairing } = await t.robot(alex);
++ await deviceFor(robot.id, pairing.code);
++ const svc = t.network.signService({ sub: 'svc:live', cap: ['bot.robot.control'], aud: ['openvibe.bot'] });
++ const command = (id, subject, token = svc) => t.call('POST', `/api/v1/robots/${robot.id}/commands`, {
++ token, headers: { 'X-OV-Subject': subject }, body: { id, kind: 'drive', value: { throttle: 0.2 } },
++ });
++ // A chat viewer with no role on the robot is refused exactly as a /control join would refuse them.
++ const stranger = t.network.newUser('chatviewer');
++ const r = await command('s1', stranger.subject);
++ assert.strictEqual(r.status, 403, r.text);
++ assert.strictEqual(r.json.code, 'bot.not_an_operator');
++ // An invited viewer has a role, but the gate keeps them read-only (the socket's answer, over HTTP).
++ const invited = t.network.newUser('memberviewer');
++ await invite(robot.id, invited.subject, 'viewer');
++ const ro = await command('s2', invited.subject);
++ assert.strictEqual(ro.status, 403, ro.text);
++ assert.strictEqual(ro.json.code, 'bot.read_only');
++ // Without bot.robot.control a service cannot use the route at all.
++ const weak = t.network.signService({ sub: 'svc:live', cap: ['bot.robot.read'], aud: ['openvibe.bot'] });
++ const denied = await command('s3', bob.subject, weak);
++ assert.strictEqual(denied.status, 403);
++ assert.strictEqual(denied.json.code, 'capability.denied');
++ });
++
++ // ── The HTTP command route (POST /robots/:id/commands) ────────────────────────────────────────
++ // Fixtures go through the domain so these checks do not spend the per-owner `bot.robot.manage` limit the
++ // rest of the file already uses; one robot + device (kept online with heartbeats) serves the device checks.
++ const hxSetup = await t.domain.robots.create({ owner: alex.subject, name: 'HTTP route rover', profile_id: 'sim.rover' });
++ const hx = hxSetup.robot;
++ const hdev = await deviceFor(hx.id, hxSetup.pairing.code);
++ const svc = t.network.signService({ sub: 'svc:live', cap: ['bot.robot.control'], aud: ['openvibe.bot'] });
++ const hcmd = (o) => t.call('POST', `/api/v1/robots/${hx.id}/commands`, { token: svc, headers: { 'X-OV-Subject': o.subject }, body: { id: o.id, kind: o.kind || 'drive', value: o.value || { throttle: 0.2 } } });
++ await t.domain.members.add(hx.id, bob.subject, 'operator', alex.subject);
++ await t.domain.members.add(hx.id, carol.subject, 'operator', alex.subject);
++ // Device liveness here is 400 ms; a heartbeat every 100 ms keeps the shared link online across checks.
++ const heartbeat = setInterval(() => { try { hdev.send({ type: 'heartbeat', t: Date.now() }); } catch { /* closing */ } }, 100);
++
++ await check('a person\'s own user token drives the HTTP route, and a mismatched X-OV-Subject is refused', async () => {
++ // bob's token with no X-OV-Subject acts as himself, exactly like a WebSocket operator.
++ const own = t.call('POST', `/api/v1/robots/${hx.id}/commands`, { user: bob, body: { id: 'me-1', kind: 'drive', value: { throttle: 0.2 } } });
++ const c = await hdev.waitFor((m) => m.type === 'command' && m.ref === 'me-1');
++ assert.strictEqual(c.operator.subject, bob.subject);
++ hdev.send({ type: 'ack', id: c.id });
++ const r = await own;
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.json.result, 'ack');
++ // Naming someone else with a user token is refused, never ignored.
++ const mismatch = await t.call('POST', `/api/v1/robots/${hx.id}/commands`, { user: bob, headers: { 'X-OV-Subject': carol.subject }, body: { id: 'me-2', kind: 'drive', value: { throttle: 0.2 } } });
++ assert.strictEqual(mismatch.status, 403, mismatch.text);
++ assert.strictEqual(mismatch.json.code, 'bot.forbidden');
++ });
++
++ await check('the HTTP command route needs a subject from a service (422) and refuses an anonymous caller (401)', async () => {
++ const noSubject = await t.call('POST', `/api/v1/robots/${hx.id}/commands`, { token: svc, body: { id: 'nos-1', kind: 'drive', value: { throttle: 0.2 } } });
++ assert.strictEqual(noSubject.status, 422, noSubject.text);
++ assert.strictEqual(noSubject.json.code, 'bot.invalid_input');
++ const anon = await t.call('POST', `/api/v1/robots/${hx.id}/commands`, { token: null, body: { id: 'anon-1', kind: 'drive', value: { throttle: 0.2 } } });
++ assert.strictEqual(anon.status, 401, anon.text);
++ assert.strictEqual(anon.json.code, 'bot.sign_in');
++ });
++
++ await check('a latched e-stop and an offline device answer 409 over the HTTP route', async () => {
++ // A device-less robot of its own: offline first, then latched (the e-stop is checked before liveness).
++ const setup = await t.domain.robots.create({ owner: alex.subject, name: 'No device', profile_id: 'sim.rover' });
++ const robot = setup.robot;
++ await t.domain.members.add(robot.id, bob.subject, 'operator', alex.subject);
++ const send = (id) => t.call('POST', `/api/v1/robots/${robot.id}/commands`, { token: svc, headers: { 'X-OV-Subject': bob.subject }, body: { id, kind: 'drive', value: { throttle: 0.2 } } });
++ const offline = await send('off-1');
++ assert.strictEqual(offline.status, 409, offline.text);
++ assert.strictEqual(offline.json.code, 'bot.device_offline');
++ await t.domain.estop.set(robot.id, { latched: true, by: alex.subject, principalKind: 'user' });
++ const latched = await send('estop-1');
++ assert.strictEqual(latched.status, 409, latched.text);
++ assert.strictEqual(latched.json.code, 'bot.estop_latched');
++ });
++
++ await check('a different subject reusing an id is never served another subject\'s cached result', async () => {
++ const first = hcmd({ id: 'same-1', subject: bob.subject });
++ const c1 = await hdev.waitFor((m) => m.type === 'command' && m.ref === 'same-1');
++ hdev.send({ type: 'ack', id: c1.id });
++ assert.strictEqual((await first).json.result, 'ack');
++ // carol's id is her own key: it reaches the device again rather than answering bob's cached ack.
++ const second = hcmd({ id: 'same-1', subject: carol.subject });
++ const c2 = await hdev.waitFor((m) => m.type === 'command' && m.ref === 'same-1' && m.id !== c1.id);
++ hdev.send({ type: 'ack', id: c2.id });
++ const r2 = await second;
++ assert.strictEqual(r2.status, 200, r2.text);
++ assert.strictEqual(r2.json.result, 'ack');
++ assert.ok(!r2.json.cached, 'carol was served bob\'s cached result');
++ });
++
++ await check('the same id while the first command is in flight answers 409 bot.command_pending', async () => {
++ const first = hcmd({ id: 'race-1', subject: bob.subject });
++ const c = await hdev.waitFor((m) => m.type === 'command' && m.ref === 'race-1'); // reserved, awaiting the ack
++ const second = await hcmd({ id: 'race-1', subject: bob.subject });
++ assert.strictEqual(second.status, 409, second.text);
++ assert.strictEqual(second.json.code, 'bot.command_pending');
++ hdev.send({ type: 'ack', id: c.id });
++ const r1 = await first;
++ assert.strictEqual(r1.status, 200, r1.text);
++ assert.strictEqual(r1.json.result, 'ack');
++ });
++
++ await check('the command route rate-limits per acted-for person, not per service token', async () => {
++ const setup = await t.domain.robots.create({ owner: alex.subject, name: 'Flood rover', profile_id: 'sim.rover' });
++ const robot = setup.robot;
++ const floodA = t.network.newUser('floodA');
++ const floodB = t.network.newUser('floodB');
++ await t.domain.members.add(robot.id, floodA.subject, 'operator', alex.subject);
++ await t.domain.members.add(robot.id, floodB.subject, 'operator', alex.subject);
++ // No device: every counted call is refused 409 device_offline. The route allows 120/min per person.
++ const send = (subject) => t.call('POST', `/api/v1/robots/${robot.id}/commands`, { token: svc, headers: { 'X-OV-Subject': subject }, body: { kind: 'drive', value: { throttle: 0.2 } } });
++ // Start well inside a fresh 60 s window so the burst cannot straddle a window boundary.
++ const into = Date.now() % 60000;
++ if (into > 30000) await t.wait(60000 - into + 100);
++ const burst = await Promise.all(Array.from({ length: 120 }, () => send(floodA.subject)));
++ assert.ok(burst.every((r) => r.status === 409), 'the first 120 of A\'s calls are counted but not limited');
++ const overA = await send(floodA.subject);
++ assert.strictEqual(overA.status, 429, 'A exhausts their own bucket');
++ assert.strictEqual(overA.json.code, 'rate_limited');
++ const overB = await send(floodB.subject);
++ assert.strictEqual(overB.status, 409, overB.text); // B's own bucket is untouched
++ assert.notStrictEqual(overB.json.code, 'rate_limited');
++ });
++
++ await check('a command the device never acks expires and answers the HTTP route', async () => {
++ const pending = hcmd({ id: 'never-1', subject: bob.subject });
++ await hdev.waitFor((m) => m.type === 'command' && m.ref === 'never-1'); // the device never acks
++ const r = await pending; // resolves when the deadline passes (expireCommand)
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.json.result, 'expired');
++ });
++
++ // Last: hub.close() stops the hub for good, so this runs after every other check.
++ await check('hub.close() settles a pending command as expired so its HTTP request never hangs', async () => {
++ const pending = hcmd({ id: 'closing-1', subject: bob.subject });
++ await hdev.waitFor((m) => m.type === 'command' && m.ref === 'closing-1'); // in flight; the device never acks
++ await t.hub.close();
++ clearInterval(heartbeat);
++ const r = await pending;
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.json.result, 'expired');
++ });
++
++ // After the shared hub is closed, and in an app of its own so the extra boot's work can disturb no other
++ // check. A small per-service cap (the default is 1200/min) tests the rotation without a 1200-request burst:
++ // a service names a different well-formed subject on every request, so its per-person buckets never trip
++ // and only its own token's ceiling holds.
++ await check('a service cannot escape the command route\'s cap by rotating through fresh subjects', async () => {
++ const u = await boot({ env: { BOT_LIMITS_CONTROL_SERVICE_MINUTE: '30', BOT_LIMITS_CONTROL_SERVICE_HOUR: '300' } });
++ try {
++ const owner = u.network.newUser('rotator');
++ const { robot } = await u.domain.robots.create({ owner: owner.subject, name: 'Rotating rover', profile_id: 'sim.rover' });
++ const rotating = u.network.signService({ sub: 'svc:rotating', cap: ['bot.robot.control'], aud: ['openvibe.bot'] });
++ const send = () => u.call('POST', `/api/v1/robots/${robot.id}/commands`, {
++ token: rotating, headers: { 'X-OV-Subject': ids.newId('user') }, // usr_…, but no member
++ body: { id: ids.newId('user'), kind: 'drive', value: { throttle: 0.2 } },
++ });
++ const counted = await Promise.all(Array.from({ length: 30 }, send));
++ assert.ok(counted.every((r) => r.status === 403), `the first 30 rotated subjects are counted but not limited: ${counted.filter((r) => r.status !== 403).map((r) => r.status).join(',')}`);
++ const over = await send();
++ assert.strictEqual(over.status, 429, over.text);
++ assert.strictEqual(over.json.code, 'rate_limited');
++ } finally { await u.close(); }
++ });
++
await t.close();
done();
})();
ddiff --git a/test/device.test.js b/test/device.test.js
iindex 73a8eba..9a66805 100644
---- a/test/device.test.js
++++ b/test/device.test.js
@@@ -134,8 +134,8 @@ const { boot, check, done } = require('./helpers/app');
dev.waitFor(counter, 800).catch(() => {});
op.send({ type: 'command', id: 'p1', kind: 'drive', value: { throttle: 0.2, steer: 0 } });
op.send({ type: 'command', id: 'p1', kind: 'drive', value: { throttle: 0.2, steer: 0 } });
-- const dup = await op.waitFor((m) => m.type === 'command_result' && m.id === 'p1' && m.result === 'pending');
-- assert.strictEqual(dup.cached, true);
++ const dup = await op.waitFor((m) => m.type === 'command_result' && m.id === 'p1' && m.result === 'refused');
++ assert.strictEqual(dup.code, 'bot.command_pending');
await new Promise((r) => setTimeout(r, 300));
assert.strictEqual(sentToDevice, 1, 'an in-flight duplicate is not sent to the robot');
assert.strictEqual(dev.messages.filter((m) => m.type === 'command' && m.ref === 'c1').length, 1, 'only one command frame');