openvibers/openvibe.bot Commits
Treat a device's estopstate as a report and queue frames during auth (#1)
Changes
@@@ -29,6 +29,13 @@ header; such a socket may only send `pair` until it is answered (anything else`bot.not_paired`). One socket per device: a second connection replaces the first (close code **4000**),so a reconnecting agent cannot be shadowed by a stale socket.++Frames sent while authentication is still in progress — right after an upgrade that carries the++header, or after `pair` until it is answered — are **kept** and handled in arrival order once it++completes (after `hello` and `config`), so a device may send its first `status` and `estop_state`++straight after the upgrade. They are never answered `bot.not_paired`. If the credential is refused++they are discarded with the socket; after a refused `pair` they are answered as on an unpaired socket.++At most 64 frames wait; any more are answered `error` `bot.not_ready`.++### Server → device| type | fields (besides v, seq, ts) | when |@@@ -38,7 +45,7 @@ so a reconnecting agent cannot be shadowed by a stale socket.| `config` | `heartbeat_ms, limits, allowed_commands, estop_latched` | right after `hello`, and again whenever the owner changes the robot's limits || `command` | `id, kind, value, deadline_ms, operator{subject,role}, robot_id` | an operator's command passed the gate || `estop` | `latched, by, at` | the e-stop latched or the owner cleared it |--| `heartbeat_ack` | `seq, server_time` | the answer to `heartbeat`; the device computes RTT |++| `heartbeat_ack` | `echo, server_time` | the answer to `heartbeat`; `echo` is the heartbeat's `t` (or `null`), so the device computes RTT. The envelope `seq` is Bot's own counter and never echoes the device's || `error` | `code, detail` | a frame the server refused |### Device → server@@@ -46,12 +53,12 @@ so a reconnecting agent cannot be shadowed by a stale socket.| type | fields (besides v, seq, ts) | notes ||---|---|---|| `pair` | `robot, code, agent_version, device_kind, drivers[], capabilities{}, name` | the one-time pairing code; `robot` (from the installer command / QR) attributes a wrong try to that robot's code |--| `heartbeat` | `seq, rtt_ms` | every `heartbeat_ms` (1 s); `rtt_ms` is the device's own measured latency |--| `telemetry` | `battery, voltage, sensors{}, …` | at most 2 Hz; extra frames are dropped |++| `heartbeat` | `t, rtt_ms` | every `heartbeat_ms` (1 s); `t` is any correlation value (e.g. the device's send time), returned as `heartbeat_ack.echo`; `rtt_ms` is the device's own measured latency |++| `telemetry` | `battery, voltage, sensors{}, events[]?, …` | at most 2 Hz; extra frames are dropped — except a frame with a non-empty `events` array (a fault, a bump, low battery), which is always delivered || `status` | `firmware, capabilities, faults[], estop_latched` | on connect and on change || `ack` | `id` | a command ran || `nack` | `id, fault_code` | a command was refused on the device |--| `estop_state` | `latched, by, at, robot_id?` | a hardware latch; `robot_id` omitted = every robot the device serves |++| `estop_state` | `latched, by, at, robot_id?` | the device's own latch, a **report**; `robot_id` omitted = every robot the device serves. See below |`kind` is one of `drive`, `actuator`, `ptz`, `say`, `display`, `halt`, and a device only ever gets thekinds its robot's profile declares (§1.1), with values in the shapes and ranges declared there.@@@ -99,6 +106,13 @@ against these):| `camera.onvif` | server driver `onvif` | `ptz {pan, tilt, zoom}` −1..1 || `sim.rover` | server driver `sim` | `drive {throttle, steer}` |++**A device's `estop_state` is a report, never a clear.** `latched: true` latches each robot the device is++attached to (if it is not already latched). `latched: false` is recorded as the device's reported state++only (`robot_state.state.device_estop`); it never clears Bot's latch — clearing is the owner's alone++(`estop_clear` on `/control`, `POST /robots/:id/estop/clear`), after which Bot sends `estop` with++`latched: false`. A `robot_id` the device is not attached to is refused with `error` `bot.forbidden`++and audited (`kind: estop_state`, `result: refused`); nothing changes on that robot.++### ExamplesPair (unauthenticated socket):@@@ -151,13 +165,14 @@ A drive command (a held control re-sends every 150 ms with a fresh `id`):``````json--{ "v": 1, "seq": 12, "ts": 1738065600500, "type": "heartbeat", "seq": 41, "rtt_ms": 63 }--{ "v": 1, "seq": 4, "ts": 1738065600501, "type": "heartbeat_ack", "seq": 41, "server_time": "2026-09-29T19:20:00.501Z" }++{ "v": 1, "seq": 12, "ts": 1738065600500, "type": "heartbeat", "t": 1738065600500, "rtt_ms": 63 }++{ "v": 1, "seq": 4, "ts": 1738065600501, "type": "heartbeat_ack", "echo": 1738065600500, "server_time": "2026-09-29T19:20:00.501Z" }``````json{ "v": 1, "seq": 13, "ts": 1738065600800, "type": "telemetry", "battery": 0.72, "voltage": 7.41,"sensors": { "ultrasonic": 118 }, "rssi": -57 }++{ "v": 1, "seq": 14, "ts": 1738065600850, "type": "telemetry", "battery": 0.72, "events": [{ "kind": "bump" }] }``````json@@@ -201,7 +216,7 @@ The gate then applies the **acted-for subject's** role — a service acting for| type | fields | notes ||---|---|---|| `joined` | `robot, role, profile, allowed_commands, state` | the answer to `join`; `allowed_commands` is the role's allowlist cut to the profile's kinds, with `halt` for every role that may drive |--| `robot_state` | `state{robot_id, online, estop{latched,by,at}, latency_ms, battery, telemetry, status, queue}` | on every change and on telemetry, ≤ 2 Hz |++| `robot_state` | `state{robot_id, online, estop{latched,by,at}, latency_ms, battery, telemetry, status, device_estop{latched,at}, queue}` | on every change and on telemetry, ≤ 2 Hz (a telemetry frame with `events` always); `estop` is Bot's latch, `device_estop` the device's last report (or `null`) || `command_result` | `id, result, code?, reason?, latency_ms?, cached?` | `result` is `ack`, `nack`, `refused` or `expired`; a repeated `id` is answered `cached: true` || `error` | `code, detail` | a frame or join the server refused |@@@ -243,7 +258,7 @@ person's `position` is 1-based (0 while they hold the turn).{ "v": 1, "seq": 3, "ts": 1738065600800, "type": "robot_state","state": { "robot_id": "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X", "online": true,"estop": { "latched": false, "by": null, "at": null }, "latency_ms": 63, "battery": 0.71,-- "telemetry": { "battery": 0.71, "voltage": 7.4 }, "status": null,++ "telemetry": { "battery": 0.71, "voltage": 7.4 }, "status": null, "device_estop": null,"queue": { "active": false, "position": 3, "turn_ends_at": null, "budget": 50, "used": 0 } } }```@@@ -274,7 +274,13 @@ function createDomain({ db, config, outbox, link = null, now = () => Date.now(),}// ── E-stop (ADR-043 decision 6) ───────────────────────────────────────────────────────────────++ // `estop.set` only latches: a clear goes through clearEstop (owner only), so no device report and no other++ // caller can ever lift an owner's latch.async function setEstop(robotId, { latched, by, principalKind = 'device' }) {++ if (!latched) fail(403, 'bot.forbidden', 'only the owner clears the e-stop');++ return writeEstop(robotId, { latched: true, by, principalKind });++ }++ async function writeEstop(robotId, { latched, by, principalKind }) {const robot = await getRobot(robotId);if (!robot) fail(404, 'bot.robot_not_found', 'no such robot');const at = iso(now());@@@ -291,7 +297,7 @@ function createDomain({ db, config, outbox, link = null, now = () => Date.now(),const robot = await getRobot(robotId);if (!robot) fail(404, 'bot.robot_not_found', 'no such robot');if (robot.owner_subject !== owner) fail(403, 'bot.forbidden', 'only the owner clears the e-stop');-- return setEstop(robotId, { latched: false, by: owner, principalKind: 'user' });++ return writeEstop(robotId, { latched: false, by: owner, principalKind: 'user' });}// ── Turn queue (ADR-043 decision 8) ───────────────────────────────────────────────────────────@@@ -21,6 +21,7 @@ const { userPrincipal, verifyService, PRINCIPAL_SUB, decodePayload } = require('const { getProfile } = require('./profiles');const OPEN = 1;++const MAX_BACKLOG = 64; // frames a device may send before its authentication completesconst ACCESS_COOKIE = 'ov_token';function parseCookies(header) {@@@ -66,22 +67,45 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat// ── Device side ───────────────────────────────────────────────────────────────────────────────async function onDeviceSocket(ws, req) {-- const conn = { ws, seq: 0, pending: true, device: null, online: false, rttMs: null, telemetry: null, status: null, lastHeartbeat: now(), lastTelemetry: 0, sessionId: prefixedId('sess', now()) };++ // backlog: while authentication (credential or pair) is in progress, frames wait here and are handled++ // in order once it resolves, so a device's first status/estop_state is never lost or answered not_paired.++ // deviceEstop: robot_id → the device's own last estop_state report (a report only, never a clear).++ const conn = { ws, seq: 0, pending: true, device: null, online: false, rttMs: null, telemetry: null, status: null, deviceEstop: new Map(), backlog: null, lastHeartbeat: now(), lastTelemetry: 0, sessionId: prefixedId('sess', now()) };ws.conn = conn;-- ws.on('message', (raw) => handleDeviceMessage(conn, raw).catch((e) => { log.warn(`[Bot] device ${conn.device ? conn.device.id : '?'}: ${e.message}`); sendError(conn, e.code || 'bot.internal', e.detail || e.message); }));-- ws.on('close', () => dropDevice(conn));++ ws.on('message', (raw) => onDeviceFrame(conn, raw));++ ws.on('close', () => { conn.backlog = null; dropDevice(conn); });ws.on('error', () => { /* the close event does the work */ });const cred = bearer(req); // the query string is never readif (cred) {++ conn.backlog = [];try {await authenticateDevice(conn, cred);sendHello(conn);await sendConfig(conn);await bringOnline(conn);-- } catch (e) { log.warn(`[Bot] device auth: ${e.message}`); ws.close(4002, 'invalid credential'); }++ } catch (e) { log.warn(`[Bot] device auth: ${e.message}`); conn.backlog = null; ws.close(4002, 'invalid credential'); return; }++ await drainBacklog(conn);}}++ function deviceFailed(conn, e) {++ log.warn(`[Bot] device ${conn.device ? conn.device.id : '?'}: ${e.message}`);++ sendError(conn, e.code || 'bot.internal', e.detail || e.message);++ }++ function onDeviceFrame(conn, raw) {++ if (!conn.backlog) return handleDeviceMessage(conn, raw).catch((e) => deviceFailed(conn, e));++ if (conn.backlog.length < MAX_BACKLOG) return conn.backlog.push(raw);++ return sendError(conn, 'bot.not_ready', 'too many frames before authentication completed');++ }++ /** Handles the frames that arrived during authentication, in arrival order (later ones join the queue). */++ async function drainBacklog(conn) {++ while (conn.backlog && conn.backlog.length) {++ const raw = conn.backlog.shift();++ await handleDeviceMessage(conn, raw).catch((e) => deviceFailed(conn, e));++ }++ conn.backlog = null;++ }++async function authenticateDevice(conn, credential) {const device = await domain.devices.byCredential(credential);if (!device) throw new Error('unknown or revoked credential');@@@ -147,7 +171,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Datswitch (msg.type) {case 'heartbeat':if (Number.isFinite(msg.rtt_ms)) conn.rttMs = Math.round(msg.rtt_ms);-- return sendFrame(conn, 'heartbeat_ack', { seq: msg.seq, server_time: iso(now()) });++ // `echo` carries the device's `t` back for its RTT; the envelope `seq` stays Bot's own counter.++ return sendFrame(conn, 'heartbeat_ack', { echo: msg.t !== undefined ? msg.t : null, server_time: iso(now()) });case 'telemetry': return onTelemetry(conn, msg);case 'status': conn.status = msg; for (const r of deviceRobotIds(conn.device)) broadcast(r); return undefined;case 'ack': case 'nack': return onAck(conn, msg);@@@ -157,23 +182,29 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat}async function handlePair(conn, msg) {-- const r = await domain.pairing.redeem({-- robot: msg.robot || null, code: msg.code, agent_version: msg.agent_version || null,-- device_kind: msg.device_kind || 'onboard', drivers: Array.isArray(msg.drivers) ? msg.drivers : [],-- capabilities: msg.capabilities && typeof msg.capabilities === 'object' ? msg.capabilities : {}, name: msg.name || null,-- });-- attachDevice(conn, r.device);-- sendFrame(conn, 'paired', {-- device_id: r.device.id, credential: r.credential, publish_key: r.publish_key, whip_url: r.whip_url,-- robot_ids: deviceRobotIds(r.device), profile_id: r.profile ? r.profile.id : null, profile: r.profile,-- });-- sendHello(conn);-- await sendConfig(conn);-- await bringOnline(conn);++ conn.backlog = [];++ try {++ const r = await domain.pairing.redeem({++ robot: msg.robot || null, code: msg.code, agent_version: msg.agent_version || null,++ device_kind: msg.device_kind || 'onboard', drivers: Array.isArray(msg.drivers) ? msg.drivers : [],++ capabilities: msg.capabilities && typeof msg.capabilities === 'object' ? msg.capabilities : {}, name: msg.name || null,++ });++ attachDevice(conn, r.device);++ sendFrame(conn, 'paired', {++ device_id: r.device.id, credential: r.credential, publish_key: r.publish_key, whip_url: r.whip_url,++ robot_ids: deviceRobotIds(r.device), profile_id: r.profile ? r.profile.id : null, profile: r.profile,++ });++ sendHello(conn);++ await sendConfig(conn);++ await bringOnline(conn);++ } catch (e) { deviceFailed(conn, e); }++ await drainBacklog(conn); // after a failed pair the socket is still unpaired and the queue is answered as such}async function onTelemetry(conn, msg) {-- if (now() - conn.lastTelemetry < Math.floor(1000 / config.control.telemetryHz)) return; // ≤ 2 Hz++ // ≤ 2 Hz, except a frame carrying events (a fault, a bump, a low battery), which is never dropped++ const urgent = Array.isArray(msg.events) && msg.events.length > 0;++ if (!urgent && now() - conn.lastTelemetry < Math.floor(1000 / config.control.telemetryHz)) return;conn.lastTelemetry = now();conn.telemetry = msg;for (const robotId of deviceRobotIds(conn.device)) broadcast(robotId);@@@ -193,12 +224,21 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => DatsendFrame(p.conn, 'command_result', { id: p.opId, result, reason: msg.fault_code || null, latency_ms: latency });}++ // A device reports its own latch. It may latch a robot it is attached to; it never clears one — latched:false++ // is recorded as the device's report only, and the owner's latch stands until the owner clears it.async function onEstopState(conn, msg) {-- const robotIds = msg.robot_id ? [msg.robot_id] : deviceRobotIds(conn.device);-- for (const robotId of robotIds) {-- const robot = await domain.robots.get(robotId);-- if (!robot || !!robot.estop_latched === !!msg.latched) continue;-- await domain.estop.set(robotId, { latched: !!msg.latched, by: 'device', principalKind: 'device' });++ const attached = deviceRobotIds(conn.device);++ const latched = !!msg.latched;++ if (msg.robot_id != null && !attached.includes(msg.robot_id)) {++ await domain.audit.record({ robotId: String(msg.robot_id).slice(0, 64), deviceId: conn.device.id, operatorKind: 'device', kind: 'estop_state', value: { latched }, result: 'refused', reason: 'bot.forbidden' });++ return sendError(conn, 'bot.forbidden', 'this device is not attached to that robot');++ }++ for (const robotId of msg.robot_id != null ? [msg.robot_id] : attached) {++ conn.deviceEstop.set(robotId, { latched, at: iso(now()) });++ if (latched) {++ const robot = await domain.robots.get(robotId);++ if (robot && !robot.estop_latched) await domain.estop.set(robotId, { latched: true, by: conn.device.id, principalKind: 'device' });++ }broadcast(robotId);}}@@@ -351,16 +391,23 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Datfunction deviceState(deviceId) { const c = deviceConns.get(deviceId); return c ? { online: c.online, rtt_ms: c.rttMs, telemetry: c.telemetry, status: c.status } : { online: false, rtt_ms: null, telemetry: null, status: null }; }async function robotState(robotId, subjectForQueue = null) {-- const robot = await domain.robots.get(robotId);++ // The device's state is read before the first await, so each broadcast carries the frame that caused it++ // (two telemetry frames with events back to back are both delivered, not the second one twice).const conn = deviceForRobot(robotId);const online = !!(conn && conn.online);++ const telemetry = online && conn.telemetry ? conn.telemetry : null;++ const live = {++ latency_ms: online ? (conn.rttMs != null ? conn.rttMs : null) : null,++ battery: telemetry ? (telemetry.battery != null ? telemetry.battery : null) : null,++ telemetry,++ status: online && conn.status ? conn.status : null,++ device_estop: online ? (conn.deviceEstop.get(robotId) || null) : null,++ };++ const robot = await domain.robots.get(robotId);const state = {robot_id: robotId, online,estop: robot ? { latched: !!robot.estop_latched, by: robot.estop_by || null, at: robot.estop_at ? iso(new Date(robot.estop_at).getTime()) : null } : null,-- latency_ms: online ? (conn.rttMs != null ? conn.rttMs : null) : null,-- battery: online && conn.telemetry ? (conn.telemetry.battery != null ? conn.telemetry.battery : null) : null,-- telemetry: online && conn.telemetry ? conn.telemetry : null,-- status: online && conn.status ? conn.status : null,++ ...live,queue: null,};if (robot && robot.access_policy === 'queue') state.queue = await domain.queue.state(robotId, subjectForQueue);nnew file mode 100644@@@ -0,0 +1,132 @@++'use strict';++// Device-socket safety (plan T15 follow-up A): a device's estop_state is a report that may latch a robot++// it is attached to and never clears the owner's latch; frames sent during authentication are kept and++// handled in order; telemetry carrying events passes the 2-Hz rule; heartbeat_ack correlates through++// `echo` while Bot's envelope seq stays strictly increasing.++const assert = require('assert');++const { boot, check, done } = require('./helpers/app');++++(async () => {++ const t = await boot();++ const alex = t.network.newUser('alex');++++ async function ownerJoined(robotId) {++ const op = await t.ws('/control', { headers: { Authorization: `Bearer ${t.network.signUser(alex)}` } });++ op.send({ type: 'join', robot_id: robotId });++ await op.waitFor((m) => m.type === 'joined');++ return op;++ }++ /** A credential from the REST pair, so the socket authenticates by header on the upgrade. */++ async function credentialFor() {++ const { robot, pairing } = await t.robot(alex);++ const paired = await t.call('POST', '/api/v1/pair', { token: null, body: { robot: robot.id, code: pairing.code } });++ assert.strictEqual(paired.status, 201, paired.text);++ return { robot, credential: paired.json.credential, deviceId: paired.json.device_id };++ }++ async function connected() {++ const c = await credentialFor();++ const dev = await t.ws('/device', { headers: { Authorization: `Bearer ${c.credential}` } });++ await dev.waitFor((m) => m.type === 'config');++ return { ...c, dev };++ }++++ await check('a device estop_state latched:false never clears the owner latch (a report only)', async () => {++ const { robot, dev } = await connected();++ const op = await ownerJoined(robot.id);++ op.send({ type: 'estop' });++ await dev.waitFor((m) => m.type === 'estop' && m.latched === true);++ dev.send({ type: 'estop_state', latched: false, robot_id: robot.id });++ const st = await op.waitFor((m) => m.type === 'robot_state' && m.state.device_estop && m.state.device_estop.latched === false);++ assert.strictEqual(st.state.estop.latched, true, 'the owner latch stands');++ assert.strictEqual(!!(await t.domain.robots.get(robot.id)).estop_latched, true);++ dev.send({ type: 'estop_state', latched: false }); // robot_id omitted = every attached robot++ await t.wait(100);++ assert.strictEqual(!!(await t.domain.robots.get(robot.id)).estop_latched, true);++ assert.ok(!dev.messages.some((m) => m.type === 'error'), JSON.stringify(dev.messages.filter((m) => m.type === 'error')));++ await assert.rejects(t.domain.estop.set(robot.id, { latched: false, by: 'device', principalKind: 'device' }), (e) => e.code === 'bot.forbidden');++ // the owner clear path still works and tells the device++ op.send({ type: 'estop_clear' });++ await dev.waitFor((m) => m.type === 'estop' && m.latched === false);++ assert.strictEqual(!!(await t.domain.robots.get(robot.id)).estop_latched, false);++ // and a device's latched:true still latches++ dev.send({ type: 'estop_state', latched: true });++ await op.waitFor((m) => m.type === 'robot_state' && m.state.estop.latched === true);++ op.close(); dev.close();++ });++++ await check('a device cannot latch or report for a robot it is not attached to (bot.forbidden, audited)', async () => {++ const { dev, deviceId } = await connected();++ const { robot: other } = await t.robot(alex);++ dev.send({ type: 'estop_state', latched: true, robot_id: other.id });++ const e1 = await dev.waitFor((m) => m.type === 'error');++ assert.strictEqual(e1.code, 'bot.forbidden');++ dev.send({ type: 'estop_state', latched: false, robot_id: other.id });++ await dev.waitFor((m) => m.type === 'error' && m !== e1);++ assert.strictEqual(!!(await t.domain.robots.get(other.id)).estop_latched, false, 'the other robot is untouched');++ const audit = await t.domain.audit.list(other.id, { limit: 5 });++ const refused = audit.filter((a) => a.kind === 'estop_state' && a.result === 'refused' && a.reason === 'bot.forbidden' && a.device_id === deviceId);++ assert.strictEqual(refused.length, 2, JSON.stringify(audit));++ dev.close();++ });++++ await check('status and estop_state sent right after the upgrade are retained, in order', async () => {++ const { robot, credential, deviceId } = await credentialFor();++ // A slow credential lookup, so the frames below certainly arrive while authentication is in progress.++ const byCredential = t.domain.devices.byCredential;++ t.domain.devices.byCredential = async (c) => { await t.wait(200); return byCredential(c); };++ let dev;++ try {++ dev = await t.ws('/device', { headers: { Authorization: `Bearer ${credential}` } });++ dev.send({ type: 'status', firmware: 'adeept-0.1.0', faults: [] });++ dev.send({ type: 'estop_state', latched: true });++ dev.send({ type: 'status', firmware: 'adeept-0.1.1', faults: [] });++ await dev.waitFor((m) => m.type === 'config');++ } finally { t.domain.devices.byCredential = byCredential; }++ const latched = await poll(async () => !!(await t.domain.robots.get(robot.id)).estop_latched, 2000);++ assert.ok(latched, 'the startup estop_state latched the robot');++ assert.ok(!dev.messages.some((m) => m.type === 'error'), JSON.stringify(dev.messages.filter((m) => m.type === 'error')));++ const state = t.hub.deviceState(deviceId);++ assert.strictEqual(state.status && state.status.firmware, 'adeept-0.1.1', 'both status frames handled, in order');++ const types = dev.messages.map((m) => m.type);++ assert.ok(types.indexOf('hello') < types.indexOf('config'), types.join());++ dev.close();++ });++++ await check('a telemetry frame with events inside the 500-ms window is delivered; plain ones are still throttled', async () => {++ const { robot, dev } = await connected();++ const op = await ownerJoined(robot.id);++ dev.send({ type: 'telemetry', battery: 0.5 });++ dev.send({ type: 'telemetry', battery: 0.4, events: [{ kind: 'bump', at: Date.now() }] });++ dev.send({ type: 'telemetry', battery: 0.39, events: [{ kind: 'low_battery' }] });++ dev.send({ type: 'telemetry', battery: 0.3 });++ await op.waitFor((m) => m.type === 'robot_state' && m.state.telemetry && m.state.telemetry.battery === 0.4);++ await op.waitFor((m) => m.type === 'robot_state' && m.state.telemetry && m.state.telemetry.battery === 0.39);++ await t.wait(150);++ assert.ok(!op.messages.some((m) => m.type === 'robot_state' && m.state.telemetry && m.state.telemetry.battery === 0.3), 'a plain frame inside the window is dropped');++ op.close(); dev.close();++ });++++ await check('heartbeat_ack echoes t; Bot\'s envelope seq stays strictly increasing', async () => {++ const { dev } = await connected();++ for (let i = 1; i <= 4; i++) dev.send({ type: 'heartbeat', seq: 1, t: 1000 + i, rtt_ms: 10 });++ dev.send({ type: 'heartbeat', rtt_ms: 10 });++ await dev.waitFor(() => dev.messages.filter((m) => m.type === 'heartbeat_ack').length === 5);++ const acks = dev.messages.filter((m) => m.type === 'heartbeat_ack');++ assert.deepStrictEqual(acks.map((a) => a.echo), [1001, 1002, 1003, 1004, null]);++ const seqs = dev.messages.map((m) => m.seq);++ for (let i = 1; i < seqs.length; i++) assert.ok(seqs[i] > seqs[i - 1], `seq ${seqs.join(',')}`);++ dev.close();++ });++++ await t.close();++ done();++})();++++async function poll(fn, ms) {++ const end = Date.now() + ms;++ for (;;) {++ if (await fn()) return true;++ if (Date.now() > end) return false;++ await new Promise((r) => setTimeout(r, 25));++ }++}@@@ -43,9 +43,9 @@ const { boot, check, done } = require('./helpers/app');await check('a paired device gets hello + config; heartbeats are acknowledged with the RTT', async () => {const { robot, pairing } = await t.robot(alex);const { dev, paired } = await pairOverWs(robot.id, pairing.code);-- dev.send({ type: 'heartbeat', seq: 1, rtt_ms: 42 });++ dev.send({ type: 'heartbeat', t: 1738065600500, rtt_ms: 42 });const ack = await dev.waitFor((m) => m.type === 'heartbeat_ack');-- assert.strictEqual(ack.seq, 1);++ assert.strictEqual(ack.echo, 1738065600500);assert.ok(ack.server_time);assert.strictEqual(t.hub.deviceState(paired.device_id).rtt_ms, 42);dev.close();
@@ -29,6 +29,13 @@ header; such a socket may only send `pair` until it is answered (anything else
@@ -29,6 +29,13 @@ header; such a socket may only send `pair` until it is answered (anything else
`bot.not_paired`). One socket per device: a second connection replaces the first (close code **4000**),
`bot.not_paired`). One socket per device: a second connection replaces the first (close code **4000**),
so a reconnecting agent cannot be shadowed by a stale socket.
so a reconnecting agent cannot be shadowed by a stale socket.
+Frames sent while authentication is still in progress — right after an upgrade that carries the
+header, or after `pair` until it is answered — are **kept** and handled in arrival order once it
+completes (after `hello` and `config`), so a device may send its first `status` and `estop_state`
+straight after the upgrade. They are never answered `bot.not_paired`. If the credential is refused
+they are discarded with the socket; after a refused `pair` they are answered as on an unpaired socket.
+At most 64 frames wait; any more are answered `error` `bot.not_ready`.
+
### Server → device
### Server → device
| type | fields (besides v, seq, ts) | when |
| type | fields (besides v, seq, ts) | when |
@@ -38,7 +45,7 @@ so a reconnecting agent cannot be shadowed by a stale socket.
@@ -38,7 +45,7 @@ so a reconnecting agent cannot be shadowed by a stale socket.
| `config` | `heartbeat_ms, limits, allowed_commands, estop_latched` | right after `hello`, and again whenever the owner changes the robot's limits |
| `config` | `heartbeat_ms, limits, allowed_commands, estop_latched` | right after `hello`, and again whenever the owner changes the robot's limits |
| `command` | `id, kind, value, deadline_ms, operator{subject,role}, robot_id` | an operator's command passed the gate |
| `command` | `id, kind, value, deadline_ms, operator{subject,role}, robot_id` | an operator's command passed the gate |
| `estop` | `latched, by, at` | the e-stop latched or the owner cleared it |
| `estop` | `latched, by, at` | the e-stop latched or the owner cleared it |
-| `heartbeat_ack` | `seq, server_time` | the answer to `heartbeat`; the device computes RTT |
+| `heartbeat_ack` | `echo, server_time` | the answer to `heartbeat`; `echo` is the heartbeat's `t` (or `null`), so the device computes RTT. The envelope `seq` is Bot's own counter and never echoes the device's |
| `error` | `code, detail` | a frame the server refused |
| `error` | `code, detail` | a frame the server refused |
### Device → server
### Device → server
@@ -46,12 +53,12 @@ so a reconnecting agent cannot be shadowed by a stale socket.
@@ -46,12 +53,12 @@ so a reconnecting agent cannot be shadowed by a stale socket.
| type | fields (besides v, seq, ts) | notes |
| type | fields (besides v, seq, ts) | notes |
|---|---|---|
|---|---|---|
| `pair` | `robot, code, agent_version, device_kind, drivers[], capabilities{}, name` | the one-time pairing code; `robot` (from the installer command / QR) attributes a wrong try to that robot's code |
| `pair` | `robot, code, agent_version, device_kind, drivers[], capabilities{}, name` | the one-time pairing code; `robot` (from the installer command / QR) attributes a wrong try to that robot's code |
-| `heartbeat` | `seq, rtt_ms` | every `heartbeat_ms` (1 s); `rtt_ms` is the device's own measured latency |
-| `telemetry` | `battery, voltage, sensors{}, …` | at most 2 Hz; extra frames are dropped |
+| `heartbeat` | `t, rtt_ms` | every `heartbeat_ms` (1 s); `t` is any correlation value (e.g. the device's send time), returned as `heartbeat_ack.echo`; `rtt_ms` is the device's own measured latency |
+| `telemetry` | `battery, voltage, sensors{}, events[]?, …` | at most 2 Hz; extra frames are dropped — except a frame with a non-empty `events` array (a fault, a bump, low battery), which is always delivered |
| `status` | `firmware, capabilities, faults[], estop_latched` | on connect and on change |
| `status` | `firmware, capabilities, faults[], estop_latched` | on connect and on change |
| `ack` | `id` | a command ran |
| `ack` | `id` | a command ran |
| `nack` | `id, fault_code` | a command was refused on the device |
| `nack` | `id, fault_code` | a command was refused on the device |
-| `estop_state` | `latched, by, at, robot_id?` | a hardware latch; `robot_id` omitted = every robot the device serves |
+| `estop_state` | `latched, by, at, robot_id?` | the device's own latch, a **report**; `robot_id` omitted = every robot the device serves. See below |
`kind` is one of `drive`, `actuator`, `ptz`, `say`, `display`, `halt`, and a device only ever gets the
`kind` is one of `drive`, `actuator`, `ptz`, `say`, `display`, `halt`, and a device only ever gets the
kinds its robot's profile declares (§1.1), with values in the shapes and ranges declared there.
kinds its robot's profile declares (§1.1), with values in the shapes and ranges declared there.
@@ -99,6 +106,13 @@ against these):
@@ -99,6 +106,13 @@ against these):
| `camera.onvif` | server driver `onvif` | `ptz {pan, tilt, zoom}` −1..1 |
| `camera.onvif` | server driver `onvif` | `ptz {pan, tilt, zoom}` −1..1 |
| `sim.rover` | server driver `sim` | `drive {throttle, steer}` |
| `sim.rover` | server driver `sim` | `drive {throttle, steer}` |
+**A device's `estop_state` is a report, never a clear.** `latched: true` latches each robot the device is
+attached to (if it is not already latched). `latched: false` is recorded as the device's reported state
+only (`robot_state.state.device_estop`); it never clears Bot's latch — clearing is the owner's alone
+(`estop_clear` on `/control`, `POST /robots/:id/estop/clear`), after which Bot sends `estop` with
+`latched: false`. A `robot_id` the device is not attached to is refused with `error` `bot.forbidden`
+and audited (`kind: estop_state`, `result: refused`); nothing changes on that robot.
+
### Examples
### Examples
Pair (unauthenticated socket):
Pair (unauthenticated socket):
@@ -151,13 +165,14 @@ A drive command (a held control re-sends every 150 ms with a fresh `id`):
@@ -151,13 +165,14 @@ A drive command (a held control re-sends every 150 ms with a fresh `id`):
```
```
```json
```json
-{ "v": 1, "seq": 12, "ts": 1738065600500, "type": "heartbeat", "seq": 41, "rtt_ms": 63 }
-{ "v": 1, "seq": 4, "ts": 1738065600501, "type": "heartbeat_ack", "seq": 41, "server_time": "2026-09-29T19:20:00.501Z" }
+{ "v": 1, "seq": 12, "ts": 1738065600500, "type": "heartbeat", "t": 1738065600500, "rtt_ms": 63 }
+{ "v": 1, "seq": 4, "ts": 1738065600501, "type": "heartbeat_ack", "echo": 1738065600500, "server_time": "2026-09-29T19:20:00.501Z" }
```
```
```json
```json
{ "v": 1, "seq": 13, "ts": 1738065600800, "type": "telemetry", "battery": 0.72, "voltage": 7.41,
{ "v": 1, "seq": 13, "ts": 1738065600800, "type": "telemetry", "battery": 0.72, "voltage": 7.41,
"sensors": { "ultrasonic": 118 }, "rssi": -57 }
"sensors": { "ultrasonic": 118 }, "rssi": -57 }
+{ "v": 1, "seq": 14, "ts": 1738065600850, "type": "telemetry", "battery": 0.72, "events": [{ "kind": "bump" }] }
```
```
```json
```json
@@ -201,7 +216,7 @@ The gate then applies the **acted-for subject's** role — a service acting for
@@ -201,7 +216,7 @@ The gate then applies the **acted-for subject's** role — a service acting for
| type | fields | notes |
| type | fields | notes |
|---|---|---|
|---|---|---|
| `joined` | `robot, role, profile, allowed_commands, state` | the answer to `join`; `allowed_commands` is the role's allowlist cut to the profile's kinds, with `halt` for every role that may drive |
| `joined` | `robot, role, profile, allowed_commands, state` | the answer to `join`; `allowed_commands` is the role's allowlist cut to the profile's kinds, with `halt` for every role that may drive |
-| `robot_state` | `state{robot_id, online, estop{latched,by,at}, latency_ms, battery, telemetry, status, queue}` | on every change and on telemetry, ≤ 2 Hz |
+| `robot_state` | `state{robot_id, online, estop{latched,by,at}, latency_ms, battery, telemetry, status, device_estop{latched,at}, queue}` | on every change and on telemetry, ≤ 2 Hz (a telemetry frame with `events` always); `estop` is Bot's latch, `device_estop` the device's last report (or `null`) |
| `command_result` | `id, result, code?, reason?, latency_ms?, cached?` | `result` is `ack`, `nack`, `refused` or `expired`; a repeated `id` is answered `cached: true` |
| `command_result` | `id, result, code?, reason?, latency_ms?, cached?` | `result` is `ack`, `nack`, `refused` or `expired`; a repeated `id` is answered `cached: true` |
| `error` | `code, detail` | a frame or join the server refused |
| `error` | `code, detail` | a frame or join the server refused |
@@ -243,7 +258,7 @@ person's `position` is 1-based (0 while they hold the turn).
@@ -243,7 +258,7 @@ person's `position` is 1-based (0 while they hold the turn).
{ "v": 1, "seq": 3, "ts": 1738065600800, "type": "robot_state",
{ "v": 1, "seq": 3, "ts": 1738065600800, "type": "robot_state",
"state": { "robot_id": "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X", "online": true,
"state": { "robot_id": "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X", "online": true,
"estop": { "latched": false, "by": null, "at": null }, "latency_ms": 63, "battery": 0.71,
"estop": { "latched": false, "by": null, "at": null }, "latency_ms": 63, "battery": 0.71,
- "telemetry": { "battery": 0.71, "voltage": 7.4 }, "status": null,
+ "telemetry": { "battery": 0.71, "voltage": 7.4 }, "status": null, "device_estop": null,
"queue": { "active": false, "position": 3, "turn_ends_at": null, "budget": 50, "used": 0 } } }
"queue": { "active": false, "position": 3, "turn_ends_at": null, "budget": 50, "used": 0 } } }
```
```
@@ -274,7 +274,13 @@ function createDomain({ db, config, outbox, link = null, now = () => Date.now(),
@@ -274,7 +274,13 @@ function createDomain({ db, config, outbox, link = null, now = () => Date.now(),
}
}
// ── E-stop (ADR-043 decision 6) ───────────────────────────────────────────────────────────────
// ── E-stop (ADR-043 decision 6) ───────────────────────────────────────────────────────────────
+ // `estop.set` only latches: a clear goes through clearEstop (owner only), so no device report and no other
+ // caller can ever lift an owner's latch.
async function setEstop(robotId, { latched, by, principalKind = 'device' }) {
async function setEstop(robotId, { latched, by, principalKind = 'device' }) {
+ if (!latched) fail(403, 'bot.forbidden', 'only the owner clears the e-stop');
+ return writeEstop(robotId, { latched: true, by, principalKind });
+ }
+ async function writeEstop(robotId, { latched, by, principalKind }) {
const robot = await getRobot(robotId);
const robot = await getRobot(robotId);
if (!robot) fail(404, 'bot.robot_not_found', 'no such robot');
if (!robot) fail(404, 'bot.robot_not_found', 'no such robot');
const at = iso(now());
const at = iso(now());
@@ -291,7 +297,7 @@ function createDomain({ db, config, outbox, link = null, now = () => Date.now(),
@@ -291,7 +297,7 @@ function createDomain({ db, config, outbox, link = null, now = () => Date.now(),
const robot = await getRobot(robotId);
const robot = await getRobot(robotId);
if (!robot) fail(404, 'bot.robot_not_found', 'no such robot');
if (!robot) fail(404, 'bot.robot_not_found', 'no such robot');
if (robot.owner_subject !== owner) fail(403, 'bot.forbidden', 'only the owner clears the e-stop');
if (robot.owner_subject !== owner) fail(403, 'bot.forbidden', 'only the owner clears the e-stop');
- return setEstop(robotId, { latched: false, by: owner, principalKind: 'user' });
+ return writeEstop(robotId, { latched: false, by: owner, principalKind: 'user' });
}
}
// ── Turn queue (ADR-043 decision 8) ───────────────────────────────────────────────────────────
// ── Turn queue (ADR-043 decision 8) ───────────────────────────────────────────────────────────
@@ -21,6 +21,7 @@ const { userPrincipal, verifyService, PRINCIPAL_SUB, decodePayload } = require('
@@ -21,6 +21,7 @@ const { userPrincipal, verifyService, PRINCIPAL_SUB, decodePayload } = require('
const { getProfile } = require('./profiles');
const { getProfile } = require('./profiles');
const OPEN = 1;
const OPEN = 1;
+const MAX_BACKLOG = 64; // frames a device may send before its authentication completes
const ACCESS_COOKIE = 'ov_token';
const ACCESS_COOKIE = 'ov_token';
function parseCookies(header) {
function parseCookies(header) {
@@ -66,22 +67,45 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -66,22 +67,45 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
// ── Device side ───────────────────────────────────────────────────────────────────────────────
// ── Device side ───────────────────────────────────────────────────────────────────────────────
async function onDeviceSocket(ws, req) {
async function onDeviceSocket(ws, req) {
- const conn = { ws, seq: 0, pending: true, device: null, online: false, rttMs: null, telemetry: null, status: null, lastHeartbeat: now(), lastTelemetry: 0, sessionId: prefixedId('sess', now()) };
+ // backlog: while authentication (credential or pair) is in progress, frames wait here and are handled
+ // in order once it resolves, so a device's first status/estop_state is never lost or answered not_paired.
+ // deviceEstop: robot_id → the device's own last estop_state report (a report only, never a clear).
+ const conn = { ws, seq: 0, pending: true, device: null, online: false, rttMs: null, telemetry: null, status: null, deviceEstop: new Map(), backlog: null, lastHeartbeat: now(), lastTelemetry: 0, sessionId: prefixedId('sess', now()) };
ws.conn = conn;
ws.conn = conn;
- ws.on('message', (raw) => handleDeviceMessage(conn, raw).catch((e) => { log.warn(`[Bot] device ${conn.device ? conn.device.id : '?'}: ${e.message}`); sendError(conn, e.code || 'bot.internal', e.detail || e.message); }));
- ws.on('close', () => dropDevice(conn));
+ ws.on('message', (raw) => onDeviceFrame(conn, raw));
+ ws.on('close', () => { conn.backlog = null; dropDevice(conn); });
ws.on('error', () => { /* the close event does the work */ });
ws.on('error', () => { /* the close event does the work */ });
const cred = bearer(req); // the query string is never read
const cred = bearer(req); // the query string is never read
if (cred) {
if (cred) {
+ conn.backlog = [];
try {
try {
await authenticateDevice(conn, cred);
await authenticateDevice(conn, cred);
sendHello(conn);
sendHello(conn);
await sendConfig(conn);
await sendConfig(conn);
await bringOnline(conn);
await bringOnline(conn);
- } catch (e) { log.warn(`[Bot] device auth: ${e.message}`); ws.close(4002, 'invalid credential'); }
+ } catch (e) { log.warn(`[Bot] device auth: ${e.message}`); conn.backlog = null; ws.close(4002, 'invalid credential'); return; }
+ await drainBacklog(conn);
}
}
}
}
+ function deviceFailed(conn, e) {
+ log.warn(`[Bot] device ${conn.device ? conn.device.id : '?'}: ${e.message}`);
+ sendError(conn, e.code || 'bot.internal', e.detail || e.message);
+ }
+ function onDeviceFrame(conn, raw) {
+ if (!conn.backlog) return handleDeviceMessage(conn, raw).catch((e) => deviceFailed(conn, e));
+ if (conn.backlog.length < MAX_BACKLOG) return conn.backlog.push(raw);
+ return sendError(conn, 'bot.not_ready', 'too many frames before authentication completed');
+ }
+ /** Handles the frames that arrived during authentication, in arrival order (later ones join the queue). */
+ async function drainBacklog(conn) {
+ while (conn.backlog && conn.backlog.length) {
+ const raw = conn.backlog.shift();
+ await handleDeviceMessage(conn, raw).catch((e) => deviceFailed(conn, e));
+ }
+ conn.backlog = null;
+ }
+
async function authenticateDevice(conn, credential) {
async function authenticateDevice(conn, credential) {
const device = await domain.devices.byCredential(credential);
const device = await domain.devices.byCredential(credential);
if (!device) throw new Error('unknown or revoked credential');
if (!device) throw new Error('unknown or revoked credential');
@@ -147,7 +171,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -147,7 +171,8 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
switch (msg.type) {
switch (msg.type) {
case 'heartbeat':
case 'heartbeat':
if (Number.isFinite(msg.rtt_ms)) conn.rttMs = Math.round(msg.rtt_ms);
if (Number.isFinite(msg.rtt_ms)) conn.rttMs = Math.round(msg.rtt_ms);
- return sendFrame(conn, 'heartbeat_ack', { seq: msg.seq, server_time: iso(now()) });
+ // `echo` carries the device's `t` back for its RTT; the envelope `seq` stays Bot's own counter.
+ return sendFrame(conn, 'heartbeat_ack', { echo: msg.t !== undefined ? msg.t : null, server_time: iso(now()) });
case 'telemetry': return onTelemetry(conn, msg);
case 'telemetry': return onTelemetry(conn, msg);
case 'status': conn.status = msg; for (const r of deviceRobotIds(conn.device)) broadcast(r); return undefined;
case 'status': conn.status = msg; for (const r of deviceRobotIds(conn.device)) broadcast(r); return undefined;
case 'ack': case 'nack': return onAck(conn, msg);
case 'ack': case 'nack': return onAck(conn, msg);
@@ -157,23 +182,29 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -157,23 +182,29 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
}
}
async function handlePair(conn, msg) {
async function handlePair(conn, msg) {
- const r = await domain.pairing.redeem({
- robot: msg.robot || null, code: msg.code, agent_version: msg.agent_version || null,
- device_kind: msg.device_kind || 'onboard', drivers: Array.isArray(msg.drivers) ? msg.drivers : [],
- capabilities: msg.capabilities && typeof msg.capabilities === 'object' ? msg.capabilities : {}, name: msg.name || null,
- });
- attachDevice(conn, r.device);
- sendFrame(conn, 'paired', {
- device_id: r.device.id, credential: r.credential, publish_key: r.publish_key, whip_url: r.whip_url,
- robot_ids: deviceRobotIds(r.device), profile_id: r.profile ? r.profile.id : null, profile: r.profile,
- });
- sendHello(conn);
- await sendConfig(conn);
- await bringOnline(conn);
+ conn.backlog = [];
+ try {
+ const r = await domain.pairing.redeem({
+ robot: msg.robot || null, code: msg.code, agent_version: msg.agent_version || null,
+ device_kind: msg.device_kind || 'onboard', drivers: Array.isArray(msg.drivers) ? msg.drivers : [],
+ capabilities: msg.capabilities && typeof msg.capabilities === 'object' ? msg.capabilities : {}, name: msg.name || null,
+ });
+ attachDevice(conn, r.device);
+ sendFrame(conn, 'paired', {
+ device_id: r.device.id, credential: r.credential, publish_key: r.publish_key, whip_url: r.whip_url,
+ robot_ids: deviceRobotIds(r.device), profile_id: r.profile ? r.profile.id : null, profile: r.profile,
+ });
+ sendHello(conn);
+ await sendConfig(conn);
+ await bringOnline(conn);
+ } catch (e) { deviceFailed(conn, e); }
+ await drainBacklog(conn); // after a failed pair the socket is still unpaired and the queue is answered as such
}
}
async function onTelemetry(conn, msg) {
async function onTelemetry(conn, msg) {
- if (now() - conn.lastTelemetry < Math.floor(1000 / config.control.telemetryHz)) return; // ≤ 2 Hz
+ // ≤ 2 Hz, except a frame carrying events (a fault, a bump, a low battery), which is never dropped
+ const urgent = Array.isArray(msg.events) && msg.events.length > 0;
+ if (!urgent && now() - conn.lastTelemetry < Math.floor(1000 / config.control.telemetryHz)) return;
conn.lastTelemetry = now();
conn.lastTelemetry = now();
conn.telemetry = msg;
conn.telemetry = msg;
for (const robotId of deviceRobotIds(conn.device)) broadcast(robotId);
for (const robotId of deviceRobotIds(conn.device)) broadcast(robotId);
@@ -193,12 +224,21 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -193,12 +224,21 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
sendFrame(p.conn, 'command_result', { id: p.opId, result, reason: msg.fault_code || null, latency_ms: latency });
sendFrame(p.conn, 'command_result', { id: p.opId, result, reason: msg.fault_code || null, latency_ms: latency });
}
}
+ // A device reports its own latch. It may latch a robot it is attached to; it never clears one — latched:false
+ // is recorded as the device's report only, and the owner's latch stands until the owner clears it.
async function onEstopState(conn, msg) {
async function onEstopState(conn, msg) {
- const robotIds = msg.robot_id ? [msg.robot_id] : deviceRobotIds(conn.device);
- for (const robotId of robotIds) {
- const robot = await domain.robots.get(robotId);
- if (!robot || !!robot.estop_latched === !!msg.latched) continue;
- await domain.estop.set(robotId, { latched: !!msg.latched, by: 'device', principalKind: 'device' });
+ const attached = deviceRobotIds(conn.device);
+ const latched = !!msg.latched;
+ if (msg.robot_id != null && !attached.includes(msg.robot_id)) {
+ await domain.audit.record({ robotId: String(msg.robot_id).slice(0, 64), deviceId: conn.device.id, operatorKind: 'device', kind: 'estop_state', value: { latched }, result: 'refused', reason: 'bot.forbidden' });
+ return sendError(conn, 'bot.forbidden', 'this device is not attached to that robot');
+ }
+ for (const robotId of msg.robot_id != null ? [msg.robot_id] : attached) {
+ conn.deviceEstop.set(robotId, { latched, at: iso(now()) });
+ if (latched) {
+ const robot = await domain.robots.get(robotId);
+ if (robot && !robot.estop_latched) await domain.estop.set(robotId, { latched: true, by: conn.device.id, principalKind: 'device' });
+ }
broadcast(robotId);
broadcast(robotId);
}
}
}
}
@@ -351,16 +391,23 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
@@ -351,16 +391,23 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
function deviceState(deviceId) { const c = deviceConns.get(deviceId); return c ? { online: c.online, rtt_ms: c.rttMs, telemetry: c.telemetry, status: c.status } : { online: false, rtt_ms: null, telemetry: null, status: null }; }
function deviceState(deviceId) { const c = deviceConns.get(deviceId); return c ? { online: c.online, rtt_ms: c.rttMs, telemetry: c.telemetry, status: c.status } : { online: false, rtt_ms: null, telemetry: null, status: null }; }
async function robotState(robotId, subjectForQueue = null) {
async function robotState(robotId, subjectForQueue = null) {
- const robot = await domain.robots.get(robotId);
+ // The device's state is read before the first await, so each broadcast carries the frame that caused it
+ // (two telemetry frames with events back to back are both delivered, not the second one twice).
const conn = deviceForRobot(robotId);
const conn = deviceForRobot(robotId);
const online = !!(conn && conn.online);
const online = !!(conn && conn.online);
+ const telemetry = online && conn.telemetry ? conn.telemetry : null;
+ const live = {
+ latency_ms: online ? (conn.rttMs != null ? conn.rttMs : null) : null,
+ battery: telemetry ? (telemetry.battery != null ? telemetry.battery : null) : null,
+ telemetry,
+ status: online && conn.status ? conn.status : null,
+ device_estop: online ? (conn.deviceEstop.get(robotId) || null) : null,
+ };
+ const robot = await domain.robots.get(robotId);
const state = {
const state = {
robot_id: robotId, online,
robot_id: robotId, online,
estop: robot ? { latched: !!robot.estop_latched, by: robot.estop_by || null, at: robot.estop_at ? iso(new Date(robot.estop_at).getTime()) : null } : null,
estop: robot ? { latched: !!robot.estop_latched, by: robot.estop_by || null, at: robot.estop_at ? iso(new Date(robot.estop_at).getTime()) : null } : null,
- latency_ms: online ? (conn.rttMs != null ? conn.rttMs : null) : null,
- battery: online && conn.telemetry ? (conn.telemetry.battery != null ? conn.telemetry.battery : null) : null,
- telemetry: online && conn.telemetry ? conn.telemetry : null,
- status: online && conn.status ? conn.status : null,
+ ...live,
queue: null,
queue: null,
};
};
if (robot && robot.access_policy === 'queue') state.queue = await domain.queue.state(robotId, subjectForQueue);
if (robot && robot.access_policy === 'queue') state.queue = await domain.queue.state(robotId, subjectForQueue);
new file mode 100644
new file mode 100644
@@ -0,0 +1,132 @@
@@ -0,0 +1,132 @@
+'use strict';
+// Device-socket safety (plan T15 follow-up A): a device's estop_state is a report that may latch a robot
+// it is attached to and never clears the owner's latch; frames sent during authentication are kept and
+// handled in order; telemetry carrying events passes the 2-Hz rule; heartbeat_ack correlates through
+// `echo` while Bot's envelope seq stays strictly increasing.
+const assert = require('assert');
+const { boot, check, done } = require('./helpers/app');
+
+(async () => {
+ const t = await boot();
+ const alex = t.network.newUser('alex');
+
+ async function ownerJoined(robotId) {
+ const op = await t.ws('/control', { headers: { Authorization: `Bearer ${t.network.signUser(alex)}` } });
+ op.send({ type: 'join', robot_id: robotId });
+ await op.waitFor((m) => m.type === 'joined');
+ return op;
+ }
+ /** A credential from the REST pair, so the socket authenticates by header on the upgrade. */
+ async function credentialFor() {
+ const { robot, pairing } = await t.robot(alex);
+ const paired = await t.call('POST', '/api/v1/pair', { token: null, body: { robot: robot.id, code: pairing.code } });
+ assert.strictEqual(paired.status, 201, paired.text);
+ return { robot, credential: paired.json.credential, deviceId: paired.json.device_id };
+ }
+ async function connected() {
+ const c = await credentialFor();
+ const dev = await t.ws('/device', { headers: { Authorization: `Bearer ${c.credential}` } });
+ await dev.waitFor((m) => m.type === 'config');
+ return { ...c, dev };
+ }
+
+ await check('a device estop_state latched:false never clears the owner latch (a report only)', async () => {
+ const { robot, dev } = await connected();
+ const op = await ownerJoined(robot.id);
+ op.send({ type: 'estop' });
+ await dev.waitFor((m) => m.type === 'estop' && m.latched === true);
+ dev.send({ type: 'estop_state', latched: false, robot_id: robot.id });
+ const st = await op.waitFor((m) => m.type === 'robot_state' && m.state.device_estop && m.state.device_estop.latched === false);
+ assert.strictEqual(st.state.estop.latched, true, 'the owner latch stands');
+ assert.strictEqual(!!(await t.domain.robots.get(robot.id)).estop_latched, true);
+ dev.send({ type: 'estop_state', latched: false }); // robot_id omitted = every attached robot
+ await t.wait(100);
+ assert.strictEqual(!!(await t.domain.robots.get(robot.id)).estop_latched, true);
+ assert.ok(!dev.messages.some((m) => m.type === 'error'), JSON.stringify(dev.messages.filter((m) => m.type === 'error')));
+ await assert.rejects(t.domain.estop.set(robot.id, { latched: false, by: 'device', principalKind: 'device' }), (e) => e.code === 'bot.forbidden');
+ // the owner clear path still works and tells the device
+ op.send({ type: 'estop_clear' });
+ await dev.waitFor((m) => m.type === 'estop' && m.latched === false);
+ assert.strictEqual(!!(await t.domain.robots.get(robot.id)).estop_latched, false);
+ // and a device's latched:true still latches
+ dev.send({ type: 'estop_state', latched: true });
+ await op.waitFor((m) => m.type === 'robot_state' && m.state.estop.latched === true);
+ op.close(); dev.close();
+ });
+
+ await check('a device cannot latch or report for a robot it is not attached to (bot.forbidden, audited)', async () => {
+ const { dev, deviceId } = await connected();
+ const { robot: other } = await t.robot(alex);
+ dev.send({ type: 'estop_state', latched: true, robot_id: other.id });
+ const e1 = await dev.waitFor((m) => m.type === 'error');
+ assert.strictEqual(e1.code, 'bot.forbidden');
+ dev.send({ type: 'estop_state', latched: false, robot_id: other.id });
+ await dev.waitFor((m) => m.type === 'error' && m !== e1);
+ assert.strictEqual(!!(await t.domain.robots.get(other.id)).estop_latched, false, 'the other robot is untouched');
+ const audit = await t.domain.audit.list(other.id, { limit: 5 });
+ const refused = audit.filter((a) => a.kind === 'estop_state' && a.result === 'refused' && a.reason === 'bot.forbidden' && a.device_id === deviceId);
+ assert.strictEqual(refused.length, 2, JSON.stringify(audit));
+ dev.close();
+ });
+
+ await check('status and estop_state sent right after the upgrade are retained, in order', async () => {
+ const { robot, credential, deviceId } = await credentialFor();
+ // A slow credential lookup, so the frames below certainly arrive while authentication is in progress.
+ const byCredential = t.domain.devices.byCredential;
+ t.domain.devices.byCredential = async (c) => { await t.wait(200); return byCredential(c); };
+ let dev;
+ try {
+ dev = await t.ws('/device', { headers: { Authorization: `Bearer ${credential}` } });
+ dev.send({ type: 'status', firmware: 'adeept-0.1.0', faults: [] });
+ dev.send({ type: 'estop_state', latched: true });
+ dev.send({ type: 'status', firmware: 'adeept-0.1.1', faults: [] });
+ await dev.waitFor((m) => m.type === 'config');
+ } finally { t.domain.devices.byCredential = byCredential; }
+ const latched = await poll(async () => !!(await t.domain.robots.get(robot.id)).estop_latched, 2000);
+ assert.ok(latched, 'the startup estop_state latched the robot');
+ assert.ok(!dev.messages.some((m) => m.type === 'error'), JSON.stringify(dev.messages.filter((m) => m.type === 'error')));
+ const state = t.hub.deviceState(deviceId);
+ assert.strictEqual(state.status && state.status.firmware, 'adeept-0.1.1', 'both status frames handled, in order');
+ const types = dev.messages.map((m) => m.type);
+ assert.ok(types.indexOf('hello') < types.indexOf('config'), types.join());
+ dev.close();
+ });
+
+ await check('a telemetry frame with events inside the 500-ms window is delivered; plain ones are still throttled', async () => {
+ const { robot, dev } = await connected();
+ const op = await ownerJoined(robot.id);
+ dev.send({ type: 'telemetry', battery: 0.5 });
+ dev.send({ type: 'telemetry', battery: 0.4, events: [{ kind: 'bump', at: Date.now() }] });
+ dev.send({ type: 'telemetry', battery: 0.39, events: [{ kind: 'low_battery' }] });
+ dev.send({ type: 'telemetry', battery: 0.3 });
+ await op.waitFor((m) => m.type === 'robot_state' && m.state.telemetry && m.state.telemetry.battery === 0.4);
+ await op.waitFor((m) => m.type === 'robot_state' && m.state.telemetry && m.state.telemetry.battery === 0.39);
+ await t.wait(150);
+ assert.ok(!op.messages.some((m) => m.type === 'robot_state' && m.state.telemetry && m.state.telemetry.battery === 0.3), 'a plain frame inside the window is dropped');
+ op.close(); dev.close();
+ });
+
+ await check('heartbeat_ack echoes t; Bot\'s envelope seq stays strictly increasing', async () => {
+ const { dev } = await connected();
+ for (let i = 1; i <= 4; i++) dev.send({ type: 'heartbeat', seq: 1, t: 1000 + i, rtt_ms: 10 });
+ dev.send({ type: 'heartbeat', rtt_ms: 10 });
+ await dev.waitFor(() => dev.messages.filter((m) => m.type === 'heartbeat_ack').length === 5);
+ const acks = dev.messages.filter((m) => m.type === 'heartbeat_ack');
+ assert.deepStrictEqual(acks.map((a) => a.echo), [1001, 1002, 1003, 1004, null]);
+ const seqs = dev.messages.map((m) => m.seq);
+ for (let i = 1; i < seqs.length; i++) assert.ok(seqs[i] > seqs[i - 1], `seq ${seqs.join(',')}`);
+ dev.close();
+ });
+
+ await t.close();
+ done();
+})();
+
+async function poll(fn, ms) {
+ const end = Date.now() + ms;
+ for (;;) {
+ if (await fn()) return true;
+ if (Date.now() > end) return false;
+ await new Promise((r) => setTimeout(r, 25));
+ }
+}
@@ -43,9 +43,9 @@ const { boot, check, done } = require('./helpers/app');
@@ -43,9 +43,9 @@ const { boot, check, done } = require('./helpers/app');
await check('a paired device gets hello + config; heartbeats are acknowledged with the RTT', async () => {
await check('a paired device gets hello + config; heartbeats are acknowledged with the RTT', async () => {
const { robot, pairing } = await t.robot(alex);
const { robot, pairing } = await t.robot(alex);
const { dev, paired } = await pairOverWs(robot.id, pairing.code);
const { dev, paired } = await pairOverWs(robot.id, pairing.code);
- dev.send({ type: 'heartbeat', seq: 1, rtt_ms: 42 });
+ dev.send({ type: 'heartbeat', t: 1738065600500, rtt_ms: 42 });
const ack = await dev.waitFor((m) => m.type === 'heartbeat_ack');
const ack = await dev.waitFor((m) => m.type === 'heartbeat_ack');
- assert.strictEqual(ack.seq, 1);
+ assert.strictEqual(ack.echo, 1738065600500);
assert.ok(ack.server_time);
assert.ok(ack.server_time);
assert.strictEqual(t.hub.deviceState(paired.device_id).rtt_ms, 42);
assert.strictEqual(t.hub.deviceState(paired.device_id).rtt_ms, 42);
dev.close();
dev.close();