openvibers/openvibe.bot Commits

A queue robot is halted when a turn ends or its holder leaves, before the next person drives: Bot's own halt (operator role system), audited (#153)

OP OpenVibers committed 20 days ago · 808f452af6 · 4 files, +82 −6

Changes

ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 7104e6a..9f0c551 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -95,6 +95,12 @@ kinds its robot's profile declares (§1.1), with values in the shapes and ranges
on every disconnect or crash path, without asking the network. `halt` (value `{}`, no deadline) stops
everything at once and passes the gate even while the e-stop is latched.
++Bot sends a `halt` of its own when a `queue` robot's turn ends — the turn ran out, or its holder left —
++before the next person's first command can pass the gate, so a held button, a drive still inside its
++deadline or firmware that keeps going on its own never carries into the next turn. Its `operator` is
++`{ "subject": null, "role": "system" }`; the device acks it like any command. A waiting person leaving
++sends none.
++
`status` of a Network-paired machine also says what it is: `device_kind` (`onboard` \| `bridge` \|
`server`), `drivers` (an array of strings), `capabilities` (an object) and `agent_version` (a string of at
most 40 characters), each optional. Bot stores them on the device row when they differ from it (the row
@@@ -344,7 +350,8 @@ needs an explicit owner or operator role — a no-role caller is refused `bot.no
`robot_state.queue` (only for a `queue` robot) is
`{ robot_id, subject, active, turn_ends_at, position, budget, used }` — `subject` is the reader's own,
never another person's, so a waiting person's `position` is 1-based (0 while they hold the turn) and no
--member ever learns the active driver's subject.
++member ever learns the active driver's subject. When a turn ends Bot halts the robot before the next
++turn starts (§1).
### Examples
@@@ -499,7 +506,8 @@ the panel's owner-only forms `POST /robots/:id/operators` (body `username`, `rol
Every command — allowed or refused — is written to `command_audit` (robot, device, the operator
principal and its kind, role, kind, the clamped value summary, result `ack|nack|refused|expired`,
--reason, latency, time), kept 30 days. Changing a robot's streaming toggles writes one row per real change
++reason, latency, time), kept 30 days. Bot's own halt at the end of a queue turn (§1) is a row with
++`operator_kind` `system`, no subject and `value` `{ reason: "turn_ended" | "turn_left", turn_of }`. Changing a robot's streaming toggles writes one row per real change
(`kind` `streaming.media` or `streaming.live`, `value` `{ on }`, role `owner`, result `ack`); an
idempotent repeat writes none. The outbox publishes `bot.robot.online`, `bot.robot.offline`,
`bot.estop.set`, `bot.estop.cleared` and `bot.command.refused` — small payloads of ids, kinds and
ddiff --git a/server/domain/index.js b/server/domain/index.js
iindex 86b080b..8a68811 100644
---- a/server/domain/index.js
++++ b/server/domain/index.js
@@@ -845,13 +845,22 @@ function createDomain({ db, config, outbox, link = null, nodes = null, openre =
// ── Turn queue (ADR-043 decision 8) ───────────────────────────────────────────────────────────
const turnMs = (robot) => { const l = json(robot.limits, {}); return Number(l.turn_ms) > 0 ? Number(l.turn_ms) : config.control.queueTurnMs; };
const turnBudget = (robot) => { const l = json(robot.limits, {}); return Number(l.turn_budget) > 0 ? Number(l.turn_budget) : config.control.queueTurnBudget; };
++ // Listeners told when a turn ends (it ran out, or its holder left) before the next one starts: the hub halts the
++ // robot (server/realtime.js), so nothing the last driver sent carries on into the next person's turn.
++ const handoffListeners = new Set();
++ const onHandoff = (fn) => { handoffListeners.add(fn); return () => handoffListeners.delete(fn); };
++ const handedOff = (robotId, from, reason) => { for (const fn of handoffListeners) { try { fn(robotId, { from, reason }); } catch { /* a listener never breaks the caller */ } } };
/** The active turn for a robot, expiring it and promoting the oldest waiting row when it has ended. */
async function currentTurn(robotId, q = db) {
const robot = await q.maybe('SELECT * FROM robots WHERE id = $1', [robotId]);
if (!robot) return null;
let active = await q.maybe(`SELECT * FROM robot_queue WHERE robot_id = $1 AND state = 'active'`, [robotId]);
if (active && new Date(active.turn_ends_at).getTime() > now()) return active;
-- if (active) await q.query(`UPDATE robot_queue SET state = 'done' WHERE robot_id = $1 AND subject = $2`, [robotId, active.subject]);
++ if (active) {
++ // Only the caller that ends the turn reports it (two at once both read it active).
++ const ended = await q.maybe(`UPDATE robot_queue SET state = 'done' WHERE robot_id = $1 AND subject = $2 AND state = 'active' RETURNING subject`, [robotId, active.subject]);
++ if (ended) handedOff(robotId, active.subject, 'turn_ended');
++ }
const next = await q.maybe(`SELECT * FROM robot_queue WHERE robot_id = $1 AND state = 'waiting' ORDER BY joined_at LIMIT 1 FOR UPDATE`, [robotId]);
if (!next) return null;
const at = now();
@@@ -873,7 +882,8 @@ function createDomain({ db, config, outbox, link = null, nodes = null, openre =
async function leaveQueue(robotId, subject) {
const robot = await getRobot(robotId);
if (!robot) fail(404, 'bot.robot_not_found', 'no such robot');
-- await db.query('DELETE FROM robot_queue WHERE robot_id = $1 AND subject = $2', [robotId, subject]);
++ const gone = await db.maybe('DELETE FROM robot_queue WHERE robot_id = $1 AND subject = $2 RETURNING state', [robotId, subject]);
++ if (gone && gone.state === 'active') handedOff(robotId, subject, 'turn_left');
await currentTurn(robotId);
return queueState(robotId, subject);
}
@@@ -1177,7 +1187,7 @@ function createDomain({ db, config, outbox, link = null, nodes = null, openre =
streaming: { get: streaming, set: setStreaming },
video: { live: liveVideo },
estop: { set: setEstop, clear: clearEstop },
-- queue: { join: joinQueue, leave: leaveQueue, state: queueState, currentTurn, consume: consumeTurn, sweep: sweepQueues },
++ queue: { join: joinQueue, leave: leaveQueue, state: queueState, currentTurn, consume: consumeTurn, sweep: sweepQueues, onHandoff },
audit: { record: auditCommand, list: listAudit, listPage: listAuditPage, prune: pruneAudit },
control: { prepare, allowedFor, effectiveLimits, deviceLimits, buildValue, DEFAULT_ALLOW, KINDS },
};
ddiff --git a/server/realtime.js b/server/realtime.js
iindex a15756e..d1f669c 100644
---- a/server/realtime.js
++++ b/server/realtime.js
@@@ -576,6 +576,22 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
sendFrame(p.conn, 'command_result', { id: p.opId, result: 'expired', reason: 'no acknowledgement before the deadline' });
}
++ // Bot's own halt when a queue turn ends or its holder leaves (domain.queue.onHandoff), sent before the next person's
++ // first command can pass the gate: a hold button, a drive still inside its deadline or firmware that keeps going
++ // on its own never runs into the next turn. Its operator is { subject: null, role: 'system' }; its ack or expiry is
++ // audited as operator_kind `system`, value { reason, turn_of }, like any command, and goes to no one.
++ const SYSTEM_CONN = { ws: { readyState: 3, send() {} }, seq: 0 };
++ function haltForTurn(robotId, { from, reason }) {
++ const device = deviceForRobot(robotId);
++ if (!device) return;
++ const id = prefixedId('cmd', now());
++ if (!send(device.ws, { v: 1, type: 'command', seq: nextSeq(device), ts: now(), id, kind: 'halt', value: {}, operator: { subject: null, role: 'system' }, robot_id: robotId })) return;
++ const p = { id, opId: id, key: `system|${id}`, deviceId: device.device.id, conn: SYSTEM_CONN, robotId, subject: null, operatorKind: 'system', role: null, kind: 'halt', value: { reason, turn_of: from || null }, at: now(), timer: null };
++ pendingCmds.set(id, p);
++ p.timer = setTimeout(() => expireCommand(id), 2500);
++ if (p.timer.unref) p.timer.unref();
++ }
++
function cacheResult(key, r) { results.set(key, { ...r, at: now() }); if (results.size > 4096) results.delete(results.keys().next().value); }
/**
@@@ -917,7 +933,7 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
for (const t of timers) if (t.unref) t.unref();
return {
-- bindDomain(d) { domain = d; },
++ bindDomain(d) { domain = d; d.queue.onHandoff(haltForTurn); },
bindJobs(j) { jobs = j; },
handleUpgrade, attachSim, detachSim,
/** A one-time ticket for the embedded panel's Drive (POST /panel/:id/ticket), or null. */
ddiff --git a/test/control.test.js b/test/control.test.js
iindex 08b9f6a..a7244b2 100644
---- a/test/control.test.js
++++ b/test/control.test.js
@@@ -244,12 +244,54 @@ const plugins = require('./helpers/plugins');
assert.strictEqual(rc.result, 'refused', 'the expired turn no longer drives');
cmd(d, { id: 'q3', kind: 'drive', value: { throttle: 0.2 } });
assert.ok(await dev.waitFor((m) => m.type === 'command' && m.ref === 'q3'), 'the next waiting person is promoted');
++ const order = dev.messages.filter((m) => m.type === 'command').map((m) => (m.kind === 'halt' ? 'halt' : m.ref));
++ assert.deepStrictEqual(order, ['halt', 'q3'], 'the ended turn halts the robot before the next person drives');
++ const halt = dev.messages.find((m) => m.type === 'command' && m.kind === 'halt');
++ assert.deepStrictEqual(halt.operator, { subject: null, role: 'system' });
++ assert.deepStrictEqual(halt.value, {});
++ assert.strictEqual(halt.deadline_ms, undefined);
} finally {
t.clock.offset -= 120000;
}
c.close(); d.close();
});
++ await check('the turn holder leaving halts the robot once, before the next person drives; the halt is audited as system', async () => {
++ const { robot, pairing } = await t.robot(alex, { access_policy: 'queue', limits: { turn_ms: 60000, turn_budget: 5 } });
++ const dev = await deviceFor(robot.id, pairing.code);
++ const c = await op(carol);
++ await joinResult(c, robot.id);
++ const d = await op(dave);
++ await joinResult(d, robot.id);
++ cmd(c, { id: 'l1', kind: 'drive', value: { throttle: 0.3 } });
++ const l1 = await dev.waitFor((m) => m.type === 'command' && m.ref === 'l1');
++ dev.send({ type: 'ack', id: l1.id });
++ await result(c, 'l1');
++ await t.domain.queue.leave(robot.id, dave.subject); // a waiting person leaving changes no turn
++ await t.wait(50);
++ assert.ok(!dev.messages.some((m) => m.type === 'command' && m.kind === 'halt'), 'no halt while the turn holder stays');
++ await t.domain.queue.join(robot.id, dave.subject);
++ await t.domain.queue.leave(robot.id, carol.subject);
++ const halt = await dev.waitFor((m) => m.type === 'command' && m.kind === 'halt');
++ assert.deepStrictEqual(halt.operator, { subject: null, role: 'system' });
++ assert.strictEqual(halt.robot_id, robot.id);
++ dev.send({ type: 'ack', id: halt.id });
++ cmd(d, { id: 'l2', kind: 'drive', value: { throttle: 0.3 } });
++ const l2 = await dev.waitFor((m) => m.type === 'command' && m.ref === 'l2');
++ dev.send({ type: 'ack', id: l2.id });
++ await result(d, 'l2');
++ const order = dev.messages.filter((m) => m.type === 'command').map((m) => (m.kind === 'halt' ? 'halt' : m.ref));
++ assert.deepStrictEqual(order, ['l1', 'halt', 'l2']);
++ const audit = await t.domain.audit.list(robot.id, { limit: 5 });
++ const row = audit.find((x) => x.kind === 'halt');
++ assert.ok(row, 'the halt is audited');
++ assert.strictEqual(row.result, 'ack');
++ assert.strictEqual(row.operator_kind, 'system');
++ assert.strictEqual(row.operator_subject, null);
++ assert.deepStrictEqual(row.value, { reason: 'turn_left', turn_of: carol.subject });
++ c.close(); d.close();
++ });
++
await check('a service token with bot.robot.control acts for a viewer and is held to the viewer role', async () => {
const { robot, pairing } = await t.robot(alex);
const viewer = t.network.newUser('viewer');