openvibers/openvibe.bot Commits

Agent Streams S7: robot commands as steps and the kill switch's reach (#315)

OP OpenVibers committed 6 days ago · 08a830ab65 · 5 files, +213 −0

Changes

ddiff --git a/migrations/0032_robot_agent_steps.sql b/migrations/0032_robot_agent_steps.sql
nnew file mode 100644
iindex 0000000..7e7c86e
---- /dev/null
++++ b/migrations/0032_robot_agent_steps.sql
@@@ -0,0 +1,22 @@
++-- phase: expand
++-- Agent Streams S7 (design 2026-11-06, AGENT-STREAMS-S7-REACH-DESIGN §6): the robot commands an agent drives through
++-- Bot, as steps. When an agent stream's agent drives a robot, each command appears on the stream beside the desk
++-- ("Drive forward 1 m": passed). Bot keeps the command itself and the stream the agent named, so the harness (or a
++-- reader of the steps) can see what the robot did; chat never writes here through S5 or S6, and Bot's gate, audit and
++-- kill switch are unchanged.
++--
++-- Expand only: every table is new and an older release ignores it.
++CREATE TABLE IF NOT EXISTS robot_agent_steps (
++ id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
++ robot_id text COLLATE "C" NOT NULL,
++ subject_id text COLLATE "C",
++ agent_stream_id text COLLATE "C",
++ kind text COLLATE "C" NOT NULL,
++ title text COLLATE "C" NOT NULL,
++ status text COLLATE "C" NOT NULL CHECK (status IN ('passed', 'failed', 'refused')),
++ reason text COLLATE "C",
++ value jsonb,
++ at timestamptz NOT NULL DEFAULT now()
++);
++CREATE INDEX IF NOT EXISTS robot_agent_steps_robot ON robot_agent_steps (robot_id, at DESC, id DESC);
++CREATE INDEX IF NOT EXISTS robot_agent_steps_stream ON robot_agent_steps (agent_stream_id, at DESC) WHERE agent_stream_id IS NOT NULL;
ddiff --git a/server/agent-steps.js b/server/agent-steps.js
nnew file mode 100644
iindex 0000000..c061b39
---- /dev/null
++++ b/server/agent-steps.js
@@@ -0,0 +1,73 @@
++'use strict';
++/**
++ * A robot command as an agent-stream step (Agent Streams S7, design 2026-11-06 §6).
++ *
++ * When an agent drives a robot through Bot, every command is also a step on the agent's stream: "Drive forward 1 m":
++ * passed. The row keeps the command's kind and value, who drove (the owner the agent acted for), the stream the caller
++ * named and Bot's own result. The gate, the audit and the kill switch stay Bot's: this is a record, never a second way
++ * to command the robot.
++ *
++ * Chat never writes a step through S5 or S6: the command route refuses the agent-stream surface for a chat token
++ * before the robot moves, and this table is written only after Bot's own gate has answered.
++ */
++const MAX_TITLE = 120;
++
++/** "Drive forward (throttle 1) for 1000 ms" from a command Bot accepted (or refused). */
++function titleOf(kind, value = {}) {
++ const k = String(kind || 'command');
++ const v = value && typeof value === 'object' ? value : {};
++ let text;
++ if (k === 'drive') {
++ const bits = Object.entries(v).slice(0, 3).map(([key, val]) => `${key} ${typeof val === 'number' ? Math.round(val * 1000) / 1000 : String(val)}`);
++ text = `Drive ${bits.length ? `(${bits.join(', ')})` : ''}`;
++ } else if (k === 'say') {
++ text = `Say ${String(v.text || '').slice(0, 80)}`;
++ } else if (k === 'stop') {
++ text = 'Stop';
++ } else {
++ const bits = Object.entries(v).slice(0, 2).map(([key, val]) => `${key} ${typeof val === 'number' ? Math.round(val * 1000) / 1000 : String(val)}`);
++ text = `${k.charAt(0).toUpperCase()}${k.slice(1)}${bits.length ? ` (${bits.join(', ')})` : ''}`;
++ }
++ return Array.from(text.replace(/\s+/g, ' ').trim()).slice(0, MAX_TITLE).join('');
++}
++
++/** Record one command after Bot's gate answered. A failure to record never changes the command's answer. */
++async function record(database, { robot, subject = null, streamId = null, body = {}, result = {} } = {}) {
++ if (!robot || !robot.id) return null;
++ const status = result.result === 'ack' ? 'passed' : (result.result === 'refused' ? 'refused' : 'failed');
++ const title = titleOf(body.kind, body.value);
++ try {
++ const inserted = await database.query(
++ `INSERT INTO robot_agent_steps (robot_id, subject_id, agent_stream_id, kind, title, status, reason, value)
++ VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb) RETURNING id, at`,
++ [robot.id, subject, /^ags_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(streamId || '')) ? String(streamId) : null,
++ String(body.kind || 'command').slice(0, 40), title, status, result.code || result.reason || null,
++ JSON.stringify(body.value && typeof body.value === 'object' ? body.value : {})]
++ );
++ const row = inserted && inserted.rows && inserted.rows[0];
++ return row ? { id: Number(row.id), title, status } : { title, status };
++ } catch (err) {
++ console.warn('[Bot] the agent step was not recorded:', err && err.message);
++ return null;
++ }
++}
++
++/** The recent commands of a robot as steps, newest last (the overlay reads them in order). */
++async function recent(database, robotId, { streamId = null, limit = 20 } = {}) {
++ const cap = Math.min(50, Math.max(1, Number(limit) || 20));
++ const result = streamId
++ ? await database.query('SELECT * FROM robot_agent_steps WHERE robot_id = $1 AND agent_stream_id = $2 ORDER BY id DESC LIMIT $3', [String(robotId), String(streamId), cap])
++ : await database.query('SELECT * FROM robot_agent_steps WHERE robot_id = $1 ORDER BY id DESC LIMIT $2', [String(robotId), cap]);
++ const rows = (result && result.rows) || [];
++ return rows.reverse().map((r) => ({
++ id: Number(r.id),
++ at: r.at instanceof Date ? r.at.toISOString() : String(r.at),
++ title: r.title,
++ status: r.status,
++ kind: r.kind,
++ agent_stream_id: r.agent_stream_id || null,
++ ...(r.reason ? { reason: r.reason } : {}),
++ }));
++}
++
++module.exports = { record, recent, titleOf, MAX_TITLE };
ddiff --git a/server/api/v1.js b/server/api/v1.js
iindex eec3ebe..9fd82a8 100644
---- a/server/api/v1.js
++++ b/server/api/v1.js
@@@ -55,6 +55,7 @@ const resourceIndex = require('./resource-index');
const resourceControl = require('./resource-control');
const { aiOf } = require('../machine-ai');
const deskLeases = require('./desk-leases');
++const agentSteps = require('../agent-steps');
const CAP = {
read: 'bot.robot.read',
@@@ -481,6 +482,12 @@ function v1Router({ domain, apiAuth, limits, hub, jobs: jobService, config, subm
r.post('/robots/:id/commands', commandControl, wrap(async (req, res) => {
requireControlCap(req);
const subject = actedFor(req);
++ // S7: the agent stream's own chat never drives a robot. Chat's bound-channel commands keep their existing
++ // path (a viewer's role); an S5/S6 call that names the agent-stream surface is refused before the gate.
++ const surface = String((req.body && req.body.surface) || req.get('x-ov-surface') || '');
++ if (surface === 'agent-stream' && /(^|:)chat$/.test(String(req.principal.sub || ''))) {
++ fail(403, 'bot.chat_cannot_drive', 'Only the owner\'s agent drives a robot from an agent stream; chat cannot');
++ }
// The owner's agent (OpenVibe.Actor) drives only a robot its owner opened to it, and only as that owner.
if (req.principal.kind === 'service' && AGENT_SERVICES.has(req.principal.sub)) {
const robot = await domain.robots.get(req.params.id);
@@@ -492,6 +499,15 @@ function v1Router({ domain, apiAuth, limits, hub, jobs: jobService, config, subm
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,
});
++ // S7: after Bot's own gate answered, the command is also a step on the agent's stream. A recording failure
++ // never changes the command's answer.
++ try {
++ const robot = await domain.robots.get(req.params.id);
++ await agentSteps.record(domain.db, {
++ robot, subject, streamId: (req.body && req.body.agent_stream_id) || req.get('x-ov-agent-stream') || null,
++ body: req.body || {}, result,
++ });
++ } catch (err) { console.warn('[Bot] agent step:', err && err.message); }
if (result.result === 'refused') fail(commandRefusalStatus(result.code), result.code, result.reason || null);
res.json({ robot_id: req.params.id, ...result });
}));
@@@ -816,6 +832,9 @@ function v1Router({ domain, apiAuth, limits, hub, jobs: jobService, config, subm
res.json({
robot_id: robot.id, name: robot.name, profile_id: robot.profile_id, online: st.online, estop: st.estop, battery: st.battery, sensors,
steps: stepHelp(profile), max_steps: AGENT_MAX_STEPS, max_seconds: AGENT_MAX_S,
++ // S7: the commands an agent drove through Bot, as steps ("Drive forward 1 m": passed). A command that
++ // named its stream is filterable with ?agent_stream=ags_…; the list never carries a second way to command.
++ commands: await agentSteps.recent(domain.db, robot.id, { streamId: req.query.agent_stream || null, limit: req.query.limit }),
});
}));
r.post('/robots/:id/steps', commandControl, wrap(async (req, res) => {
ddiff --git a/server/app.js b/server/app.js
iindex cd7fe37..041f867 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -165,6 +165,31 @@ function createApp(opts = {}) {
};
app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets, send: accountSend, onEvent, log }));
++ // ── OpenVibe.Live's kill switch reaches the robot (Agent Streams S7) ──
++ // When an agent stream ends, Live e-stops the robot its channel is bound to. The call is loopback-only (nginx
++ // answers 404 for /internal/) and carries the deployment's shared secret; it can only latch the e-stop, never
++ // clear it. No person token, no chat token, and no route in the public API reaches this.
++ app.post('/internal/robots/:id/estop', express.json({ limit: '4kb' }), async (req, res) => {
++ const secret = String(process.env.BOT_KILL_SECRET || '');
++ const given = String(req.get('x-ov-bot-kill') || '');
++ const a = Buffer.from(given);
++ const b = Buffer.from(secret);
++ if (!secret || a.length !== b.length || !require('crypto').timingSafeEqual(a, b)) {
++ return res.status(401).json({ error: 'kill.auth', detail: 'the kill switch secret is not valid' });
++ }
++ const id = String(req.params.id || '');
++ if (!/^rob_[A-Za-z0-9_-]{4,60}$/.test(id)) return res.status(404).json({ error: 'bot.robot_not_found' });
++ try {
++ const robot = await domain.robots.get(id);
++ if (!robot) return res.status(404).json({ error: 'bot.robot_not_found' });
++ if (!robot.estop_latched) await domain.estop.set(id, { latched: true, by: 'openvibe.live', principalKind: 'service' });
++ return res.json({ ok: true, robot_id: id, estop: { latched: true } });
++ } catch (err) {
++ log.error('[Bot] the kill switch failed:', err && err.message);
++ return res.status(500).json({ error: 'bot.kill_failed' });
++ }
++ });
++
// The OpenVibe Frame's scripts and the showcase stylesheet the front page links (content-hashed URLs).
app.use('/shared', require('openvibe-shared/serve').handler());
ddiff --git a/test/agent-stream-steps.test.js b/test/agent-stream-steps.test.js
nnew file mode 100644
iindex 0000000..6581358
---- /dev/null
++++ b/test/agent-stream-steps.test.js
@@@ -0,0 +1,74 @@
++'use strict';
++// Agent Streams S7, Bot's part: an agent's commands are steps on its stream, chat cannot drive through the agent
++// stream, and Live's kill switch latches the robot's e-stop through Bot.
++const assert = require('assert');
++const { boot, check, done } = require('./helpers/app');
++
++const AGS = 'ags_01JAB2C3D4E5F6G7H8J9K0MNPS';
++
++(async () => {
++ process.env.BOT_KILL_SECRET = 'test-kill-secret';
++ const t = await boot({ env: { BOT_ROUTINE_TICK_MS: '0', BOT_WEBHOOK_TICK_MS: '0', BOT_KILL_SECRET: 'test-kill-secret' } });
++ const alex = t.network.newUser('alex');
++ const { robot } = await t.robot(alex, { name: 'Stream rover', profile_id: 'sim.rover' });
++ assert.ok(await t.app.locals.sim.attach(robot.id));
++ const poll = async (fn, ms = 8000) => { const end = Date.now() + ms; while (Date.now() < end) { const v = await fn(); if (v) return v; await t.wait(50); } return null; };
++ await poll(() => t.hub.isOnline(`dev_sim_${robot.id}`));
++ await t.domain.robots.setAgentControl(robot.id, true);
++ const actor = t.network.signService({ sub: 'svc:actor', cap: ['bot.robot.read', 'bot.robot.control'] });
++ const asAgent = (method, path, body) => t.call(method, `/api/v1/robots/${robot.id}${path}`, { token: actor, headers: { 'X-OV-Subject': alex.subject }, body });
++
++ await check('an agent command is a step on the stream', async () => {
++ const r = await asAgent('POST', '/commands', { id: 's7-1', kind: 'drive', value: { throttle: 0.5 }, ms: 500, agent_stream_id: AGS });
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.json.result, 'ack');
++ const steps = await asAgent('GET', `/steps?agent_stream=${AGS}`);
++ assert.strictEqual(steps.status, 200, steps.text);
++ const step = steps.json.commands.find((c) => c.agent_stream_id === AGS);
++ assert.ok(step, JSON.stringify(steps.json.commands));
++ assert.match(step.title, /^Drive/);
++ assert.strictEqual(step.status, 'passed');
++ assert.strictEqual(step.kind, 'drive');
++ const other = await asAgent('GET', '/steps?agent_stream=ags_01JAB2C3D4E5F6G7H8J9K0MNPV');
++ assert.ok(!other.json.commands.some((c) => c.agent_stream_id === AGS), 'a stream sees only its own steps');
++ });
++
++ await check('chat cannot drive a robot through the agent-stream surface', async () => {
++ const chat = t.network.signService({ sub: 'svc:chat', cap: ['bot.robot.control'] });
++ const r = await t.call('POST', `/api/v1/robots/${robot.id}/commands`, {
++ token: chat, headers: { 'X-OV-Subject': alex.subject },
++ body: { id: 's7-chat', kind: 'drive', value: { throttle: 0.2 }, surface: 'agent-stream' },
++ });
++ assert.strictEqual(r.status, 403, r.text);
++ assert.strictEqual(r.json.code, 'bot.chat_cannot_drive');
++ const state = await asAgent('GET', '/steps');
++ assert.strictEqual(state.json.estop.latched, false, 'the refused call did not reach the robot');
++ });
++
++ await check('Live\'s kill switch latches the e-stop through Bot, and the next command is refused', async () => {
++ const wrong = await t.call('POST', `/internal/robots/${robot.id}/estop`, { headers: { 'X-OV-Bot-Kill': 'not-the-secret' }, body: {} });
++ assert.strictEqual(wrong.status, 401, wrong.text);
++ const ok = await t.call('POST', `/internal/robots/${robot.id}/estop`, { headers: { 'X-OV-Bot-Kill': 'test-kill-secret' }, body: { reason: 'agent_stream_ended' } });
++ assert.strictEqual(ok.status, 200, ok.text);
++ assert.strictEqual(ok.json.estop.latched, true);
++ const state = await asAgent('GET', '/steps');
++ assert.strictEqual(state.json.estop.latched, true, 'the robot is stopped');
++ const refused = await asAgent('POST', '/commands', { id: 's7-2', kind: 'drive', value: { throttle: 0.5 }, ms: 500, agent_stream_id: AGS });
++ assert.strictEqual(refused.status, 409, refused.text);
++ assert.strictEqual(refused.json.code, 'bot.estop_latched');
++ const steps = await asAgent('GET', `/steps?agent_stream=${AGS}`);
++ const last = steps.json.commands[steps.json.commands.length - 1];
++ assert.strictEqual(last.status, 'refused', 'the refused command is still a step, with the reason');
++ assert.strictEqual(last.reason, 'bot.estop_latched');
++ });
++
++ await check('the step title is the command, never markup or unbounded text', async () => {
++ const { titleOf } = require('../server/agent-steps');
++ assert.strictEqual(titleOf('drive', { throttle: 1 }), 'Drive (throttle 1)');
++ assert.strictEqual(titleOf('say', { text: 'hello' }), 'Say hello');
++ assert.ok(Array.from(titleOf('drive', { note: 'x'.repeat(500) })).length <= 120);
++ });
++
++ await t.close();
++ done();
++})().catch((e) => { console.error(e.stack || String(e)); process.exit(1); });