Re-send device config on e-stop changes; events skip the sensor window (#3)

ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 2aaff85..3004f84 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -42,7 +42,7 @@ At most 64 frames wait; any more are answered `error` `bot.not_ready`.
  |---|---|---|
  | `paired` | `device_id, credential, publish_key, whip_url, robot_ids, profile_id, profile` | the answer to `pair`; `credential`/`publish_key`/`whip_url` are shown **once** |
  | `hello` | `session_id, device_id, robot_ids, server_time` | on every authenticated connection |
--| `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 (or `allow` lists) and whenever Bot's latch is set or cleared (after the `estop` frame) |
  | `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` | `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 |
@@@ -54,7 +54,7 @@ At most 64 frames wait; any more are answered `error` `bot.not_ready`.
  |---|---|---|
  | `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` | `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 |
++| `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 and does not count against the samples' window |
  | `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 |
@@@ -288,7 +288,7 @@ shown; a service acts for `X-OV-Subject` / the body's `owner`.
  | `GET /robots` | user, or `bot.robot.read` + `?owner=` | `{ robots[] }` |
  | `POST /robots` | the owner, or `bot.robot.manage` | `201 { robot, pairing{code,expires_at,installer} }` |
  | `GET /robots/:id` | a member, or `bot.robot.read` | `{ robot, role }` |
--| `PATCH /robots/:id` | owner, or `bot.robot.manage` | `{ robot }`; new `limits` re-send `config` to the connected device |
++| `PATCH /robots/:id` | owner, or `bot.robot.manage` | `{ robot }`; new `limits` (including `allow`) re-send `config` to the connected device |
  | `DELETE /robots/:id` | owner, or `bot.robot.manage` | `204` |
  | `POST /robots/:id/pairing-code` | owner, or `bot.robot.manage` | `201 { code, expires_at, installer }` |
  | `GET /robots/:id/operators` | member, or `bot.robot.read` | `{ operators[] }` |
ddiff --git a/server/api/v1.js b/server/api/v1.js
iindex d0b3113..359a194 100644
---- a/server/api/v1.js
++++ b/server/api/v1.js
@@@ -107,7 +107,7 @@ function v1Router({ domain, apiAuth, limits, hub }) {
      r.patch('/robots/:id', manage, wrap(async (req, res) => {
          await owner(req, req.params.id);
          const updated = await domain.robots.update(req.params.id, { name: req.body?.name, access_policy: req.body?.access_policy, limits: req.body?.limits });
--        if (req.body?.limits !== undefined) await hub.refreshConfig(req.params.id);   // the device enforces the new limits at once
++        if (req.body?.limits !== undefined) await hub.refreshConfig(req.params.id);   // the device enforces the new limits and allowlist at once
          res.json({ robot: domain.present.robot(updated) });
      }));
      r.delete('/robots/:id', manage, wrap(async (req, res) => {
@@@ -165,14 +165,14 @@ function v1Router({ domain, apiAuth, limits, hub }) {
          const { robot } = await member(req, req.params.id, ['owner', 'operator']);
          if (req.principal.kind === 'service') requireControlCap(req);
          if (!robot.estop_latched) await domain.estop.set(req.params.id, { latched: true, by: me(req) || req.principal.sub, principalKind: req.principal.kind });
--        hub.sendToRobotDevices(req.params.id, { type: 'estop', latched: true, by: me(req) || req.principal.sub, at: new Date().toISOString() });
++        await hub.pushEstop(req.params.id, true, me(req) || req.principal.sub);
          hub.broadcast(req.params.id);
          res.json({ robot: domain.present.robot(await domain.robots.get(req.params.id)) });
      }));
      r.post('/robots/:id/estop/clear', control, wrap(async (req, res) => {
          const robot = await owner(req, req.params.id);
          await domain.estop.clear(req.params.id, robot.owner_subject);
--        hub.sendToRobotDevices(req.params.id, { type: 'estop', latched: false, by: robot.owner_subject, at: new Date().toISOString() });
++        await hub.pushEstop(req.params.id, false, robot.owner_subject);
          hub.broadcast(req.params.id);
          res.json({ robot: domain.present.robot(await domain.robots.get(req.params.id)) });
      }));
ddiff --git a/server/realtime.js b/server/realtime.js
iindex 85f24f0..59af240 100644
---- a/server/realtime.js
++++ b/server/realtime.js
@@@ -138,12 +138,17 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
              estop_latched: robot ? !!robot.estop_latched : false,
          });
      }
--    /** Re-send `config` to the robot's connected devices (the owner changed its limits). */
++    /** Re-send `config` to the robot's connected devices (the owner changed its limits or allowlist, or the latch moved). */
      async function refreshConfig(robotId) {
          for (const conn of [...deviceConns.values()]) {
              if (conn.online && deviceRobotIds(conn.device).includes(robotId)) await sendConfig(conn);
          }
      }
++    /** Tell the robot's devices the latch moved: the `estop` frame acts at once, then `config` carries the new state. */
++    async function pushEstop(robotId, latched, by) {
++        sendToRobotDevices(robotId, { type: 'estop', latched, by, at: iso(now()) });
++        await refreshConfig(robotId);
++    }
      async function bringOnline(conn) {
          await domain.devices.setOnline(conn.device, true).catch((e) => log.warn(`[Bot] online event: ${e.message}`));
          for (const robotId of deviceRobotIds(conn.device)) broadcast(robotId);
@@@ -202,10 +207,13 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
      }
  
      async function onTelemetry(conn, msg) {
--        // ≤ 2 Hz, except a frame carrying events (a fault, a bump, a low battery), which is never dropped
++        // Sensor samples ≤ 2 Hz. A frame carrying events (a fault, a bump, a low battery) is never dropped and
++        // does not use up the samples' window, so the next sample is judged against the last sample only.
          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();
++        if (!urgent) {
++            if (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);
      }
@@@ -237,7 +245,10 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
              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' });
++                if (robot && !robot.estop_latched) {
++                    await domain.estop.set(robotId, { latched: true, by: conn.device.id, principalKind: 'device' });
++                    await refreshConfig(robotId);
++                }
              }
              broadcast(robotId);
          }
@@@ -371,7 +382,7 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
                  await domain.estop.clear(robotId, conn.subject);   // owner only; throws otherwise
              }
          } catch (e) { return sendError(conn, e.code || 'bot.forbidden', e.detail || e.message); }
--        sendToRobotDevices(robotId, { type: 'estop', latched, by: conn.subject, at: iso(now()) });
++        await pushEstop(robotId, latched, conn.subject);
          broadcast(robotId);
      }
  
@@@ -437,7 +448,7 @@ function createRealtime({ config, keys, userAuth, log = console, now = () => Dat
      return {
          bindDomain(d) { domain = d; },
          handleUpgrade,
--        isOnline, deviceState, sendToDevice, sendToRobotDevices, broadcast, closeDevice, robotState, refreshConfig,
++        isOnline, deviceState, sendToDevice, sendToRobotDevices, broadcast, closeDevice, robotState, refreshConfig, pushEstop,
          onlineCount() { let n = 0; for (const c of deviceConns.values()) if (c.online) n++; return n; },
          devices() { return [...deviceConns.values()].map((c) => ({ device_id: c.device.id, online: c.online })); },
          async close() {
ddiff --git a/test/control.test.js b/test/control.test.js
iindex d624dc4..3c57246 100644
---- a/test/control.test.js
++++ b/test/control.test.js
@@@ -4,6 +4,8 @@
  // a viewer. Every decision is audited.
  const assert = require('assert');
  const { boot, check, done } = require('./helpers/app');
++const { loadProfiles } = require('../server/profiles');
++const plugins = require('./helpers/plugins');
  
  (async () => {
      const t = await boot();
@@@ -137,6 +139,63 @@ const { boot, check, done } = require('./helpers/app');
          a.close();
      });
  
++    // Over the socket, every Adeept (ordinary and mecanum) and Cozmo control reaches the device in the exact
++    // shape its Node plugin takes (test/helpers/plugins.js transcribes the plugins).
++    const PLUGIN_COMMANDS = {
++        'adeept.adr036': [
++            ['drive', { throttle: 0.5, steer: -2 }, { throttle: 0.5, steer: -1 }],
++            ['actuator', { name: 'pan', value: 0.4 }, { name: 'pan', value: 0.4 }],
++            ['actuator', { name: 'tilt', value: -3 }, { name: 'tilt', value: -1 }],
++            ['actuator', { name: 'buzzer', value: { note: 'A4' } }, { name: 'buzzer', value: { hz: 440 } }],
++            ['actuator', { name: 'buzzer', value: { hz: 5000 } }, { name: 'buzzer', value: { hz: 880 } }],
++            ['actuator', { name: 'buzzer', value: null }, { name: 'buzzer', value: null }],
++            ['actuator', { name: 'lights', value: { r: 300, g: 12.4, b: -5 } }, { name: 'lights', value: { r: 255, g: 12, b: 0 } }],
++            ['actuator', { name: 'lights', value: null }, { name: 'lights', value: null }],
++            ['halt', {}, {}],
++        ],
++        'adeept.adr036.mecanum': [
++            ['drive', { x: 0.2, y: -0.7, rotation: 0.3 }, { x: 0.2, y: -0.7, rotation: 0.3 }],
++            ['drive', { y: 1 }, { x: 0, y: 1, rotation: 0 }],
++            ['actuator', { name: 'pan', value: -0.25 }, { name: 'pan', value: -0.25 }],
++            ['actuator', { name: 'tilt', value: 1 }, { name: 'tilt', value: 1 }],
++            ['actuator', { name: 'buzzer', value: { note: 'C5' } }, { name: 'buzzer', value: { hz: 523.25 } }],
++            ['actuator', { name: 'lights', value: { r: 0, g: 128, b: 255 } }, { name: 'lights', value: { r: 0, g: 128, b: 255 } }],
++        ],
++        cozmo: [
++            ['drive', { throttle: -0.4, steer: 0.2 }, { throttle: -0.4, steer: 0.2 }],
++            ['actuator', { name: 'head', value: 0.5 }, { name: 'head', value: 0.5 }],
++            ['actuator', { name: 'lift', value: -0.5 }, { name: 'lift', value: 0 }],
++            ['actuator', { name: 'lift', value: 2 }, { name: 'lift', value: 1 }],
++            ['actuator', { name: 'backpack_lights', value: { r: 1, g: 2, b: 3 } }, { name: 'backpack_lights', value: { r: 1, g: 2, b: 3 } }],
++            ['say', { text: 'hello' }, { text: 'hello' }],
++            ['display', { face: 'happy' }, { face: 'happy' }],
++        ],
++    };
++    const shipped = loadProfiles();
++    for (const [profileId, cases] of Object.entries(PLUGIN_COMMANDS)) {
++        await check(`${profileId}: the commands the device gets over the socket are the ones its plugin accepts`, async () => {
++            const contract = plugins.contractFor(shipped.get(profileId));
++            const { robot, pairing } = await t.robot(alex, { profile_id: profileId });
++            const dev = await deviceFor(robot.id, pairing.code);
++            const a = await op(alex);
++            await joinResult(a, robot.id);
++            for (const [i, [kind, value, expected]] of cases.entries()) {
++                const id = `${profileId}-${i}`;
++                cmd(a, { id, kind, value });
++                const got = await Promise.race([
++                    dev.waitFor((m) => m.type === 'command' && m.ref === id),
++                    a.waitFor((m) => m.type === 'command_result' && m.id === id).then((r) => assert.fail(`${kind} ${JSON.stringify(value)}: ${r.result} ${r.code || ''} ${r.reason || ''}`)),
++                ]);
++                assert.strictEqual(got.kind, kind);
++                assert.deepStrictEqual(got.value, expected, `${kind} ${JSON.stringify(value)}`);
++                assert.strictEqual(contract.accepts(kind, got.value), null, `${contract.name} accepts ${kind} ${JSON.stringify(got.value)}`);
++                dev.send({ type: 'ack', id: got.id });
++                assert.strictEqual((await result(a, id)).result, 'ack');
++            }
++            a.close(); dev.close();
++        });
++    }
++
      await check('config carries the effective limits and is re-sent to the device when the owner changes them', async () => {
          const { robot, pairing } = await t.robot(alex, { profile_id: 'cozmo', limits: { max_speed: 0.9, max_turn: 0.5 } });
          const dev = await deviceFor(robot.id, pairing.code);
ddiff --git a/test/device-safety.test.js b/test/device-safety.test.js
iindex 62c19c0..1ee1405 100644
---- a/test/device-safety.test.js
++++ b/test/device-safety.test.js
@@@ -106,6 +106,56 @@ const { boot, check, done } = require('./helpers/app');
          op.close(); dev.close();
      });
  
++    await check('an event does not use up the sensor window: the next sample is judged against the last sample', async () => {
++        const { robot, dev } = await connected();
++        const op = await ownerJoined(robot.id);
++        const sample = (battery) => op.waitFor((m) => m.type === 'robot_state' && m.state.telemetry && m.state.telemetry.battery === battery);
++        // The injected clock moves the window; a heartbeat just before each step keeps the device online.
++        const beat = async (n) => { dev.send({ type: 'heartbeat', t: n }); await dev.waitFor((m) => m.type === 'heartbeat_ack' && m.echo === n); };
++        try {
++            dev.send({ type: 'telemetry', battery: 0.9 });
++            await sample(0.9);
++            await beat(1);
++            t.clock.offset += 300;
++            dev.send({ type: 'telemetry', battery: 0.8, events: [{ kind: 'bump' }] });
++            await sample(0.8);
++            dev.send({ type: 'telemetry', battery: 0.7 });   // 300 ms after the last sample: dropped
++            await beat(2);
++            t.clock.offset += 250;
++            dev.send({ type: 'telemetry', battery: 0.6 });   // 550 ms after the last sample, 250 after the event: delivered
++            await sample(0.6);
++            assert.ok(!op.messages.some((m) => m.type === 'robot_state' && m.state.telemetry && m.state.telemetry.battery === 0.7), 'a second sample inside the window is dropped');
++        } finally {
++            t.clock.offset -= 550;
++        }
++        op.close(); dev.close();
++    });
++
++    await check('setting or clearing the latch re-sends config with estop_latched, from the owner, REST and the device', async () => {
++        const { robot, dev } = await connected();
++        const first = dev.messages.find((m) => m.type === 'config');
++        assert.strictEqual(first.estop_latched, false);
++        const op = await ownerJoined(robot.id);
++        const nextConfig = (after, latched) => dev.waitFor((m) => m.type === 'config' && m.seq > after && m.estop_latched === latched);
++        op.send({ type: 'estop' });
++        const estop = await dev.waitFor((m) => m.type === 'estop' && m.latched === true);
++        const c1 = await nextConfig(first.seq, true);
++        assert.ok(c1.seq > estop.seq, 'the estop frame comes first');
++        assert.deepStrictEqual(c1.limits, first.limits);
++        assert.deepStrictEqual(c1.allowed_commands, first.allowed_commands);
++        op.send({ type: 'estop_clear' });
++        const c2 = await nextConfig(c1.seq, false);
++        const r = await t.call('POST', `/api/v1/robots/${robot.id}/estop`, { user: alex });
++        assert.strictEqual(r.status, 200, r.text);
++        const c3 = await nextConfig(c2.seq, true);
++        const cleared = await t.call('POST', `/api/v1/robots/${robot.id}/estop/clear`, { user: alex });
++        assert.strictEqual(cleared.status, 200, cleared.text);
++        const c4 = await nextConfig(c3.seq, false);
++        dev.send({ type: 'estop_state', latched: true });   // the device's own latch moves Bot's
++        await nextConfig(c4.seq, true);
++        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 });