openvibers/openvibe.link Commits

Align pairing and the device link with OpenVibe.Bot's contract (#4)

OP OpenVibers committed 2 Oct 2026 · 849644c1d2 · 32 files, +1041 −263

local checks ✗ 0/2 · CI ✓ 9/9 · review SHIP by DeepSeek harness · 1 fix round

https://github.com/OpenVibers/OpenVibe.Node/pull/4

Changes

ddiff --git a/README.md b/README.md
iindex 2aba329..87501ee 100644
---- a/README.md
++++ b/README.md
@@@ -5,10 +5,11 @@ like **OpenVibe.Bot** (robots: drive, cameras, sensors, with safety) and, later,
for AI agents) can use the device. It connects out, needs no open ports, and keeps a local kill switch.
```sh
--curl -fsSL https://openvibe.bot/install | sh -s -- ABCD-1234 --robot adeept
++curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_… --code ABCD-1234 --driver adeept
```
--Get the code on openvibe.bot. `--robot` is `adeept`, `adeept-mecanum`, `cozmo` or `none` (dry run). Details, manual
++openvibe.bot shows this command with the robot's id and a fresh code. `--driver` is `adeept`, `adeept-mecanum`, `cozmo`
++or `none` (dry run; the old `--robot adeept` form still works but is deprecated). Details, manual
install, Windows and configuration: [docs/install.md](docs/install.md).
```sh
ddiff --git a/cmd/openvibe-node/main.go b/cmd/openvibe-node/main.go
iindex c6ef81d..048decd 100644
---- a/cmd/openvibe-node/main.go
++++ b/cmd/openvibe-node/main.go
@@@ -43,7 +43,7 @@ var version = "dev"
const usage = `openvibe-node %s: connect this device to OpenVibe.
Usage:
-- openvibe-node pair <CODE> [--server URL] [--kind onboard|bridge] [--force]
++ openvibe-node pair <CODE> [--robot rob_…] [--name NAME] [--server URL] [--kind onboard|bridge] [--force]
openvibe-node run [--dry-run]
openvibe-node install [--user NAME]
openvibe-node uninstall
@@@ -83,6 +83,9 @@ func run(args []string, stdout, stderr io.Writer) int {
var (
server = fs.String("server", "", "OpenVibe.Bot origin (pair)")
kind = fs.String("kind", "", "onboard or bridge (pair)")
++ robot = fs.String("robot", "", "the rob_… id of the robot to pair with, from the installer command (pair)")
++ name = fs.String("name", "", "a name for this device on openvibe.bot; default: the hostname (pair)")
++ codeArg = fs.String("code", "", "the pairing code, instead of the argument (pair)")
force = fs.Bool("force", false, "pair again even if already paired")
dryRun = fs.Bool("dry-run", false, "run only the dry-run plugin and the test pattern (run)")
asJSON = fs.Bool("json", false, "print JSON (status, plugins)")
@@@ -106,11 +109,14 @@ func run(args []string, stdout, stderr io.Writer) int {
var err error
switch cmd {
case "pair":
++ if *codeArg != "" {
++ positional = append(positional, *codeArg)
++ }
if len(positional) != 1 {
-- fmt.Fprintln(stderr, "usage: openvibe-node pair <CODE>")
++ fmt.Fprintln(stderr, "usage: openvibe-node pair <CODE> [--robot rob_…] [--name NAME]")
return 2
}
-- err = cmdPair(g, positional[0], *server, *kind, *force, stdout)
++ err = cmdPair(g, pairArgs{code: positional[0], robot: *robot, name: *name, server: *server, kind: *kind, force: *force}, stdout)
case "run":
err = cmdRun(g, *dryRun)
case "install":
@@@ -190,11 +196,31 @@ func probeAll(cfg *config.Config, paths config.Paths, log *slog.Logger) (map[str
return out, failed
}
--func cmdPair(g *globals, code, server, kind string, force bool, stdout io.Writer) error {
-- code, err := link.NormalizeCode(code)
++// pairArgs are the flags of `openvibe-node pair`.
++type pairArgs struct {
++ code, robot, name, server, kind string
++ force bool
++}
++
++// maxNameLen is the longest device name OpenVibe.Bot stores.
++const maxNameLen = 80
++
++func cmdPair(g *globals, a pairArgs, stdout io.Writer) error {
++ code, err := link.NormalizeCode(a.code)
if err != nil {
return err
}
++ server, kind, force := a.server, a.kind, a.force
++ if a.robot != "" && !link.RobotRe.MatchString(a.robot) {
++ return fmt.Errorf("--robot must be a robot id like rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X (a driver kind such as %q goes in the config's plugins)", a.robot)
++ }
++ name := strings.TrimSpace(a.name)
++ if name == "" {
++ name, _ = os.Hostname()
++ }
++ if r := []rune(name); len(r) > maxNameLen {
++ name = string(r[:maxNameLen])
++ }
if err := g.paths.Ensure(); err != nil {
return fmt.Errorf("cannot create %s (run with sudo, or use --home): %w", g.paths.ConfigDir, err)
}
@@@ -216,9 +242,8 @@ func cmdPair(g *globals, code, server, kind string, force bool, stdout io.Writer
return err
}
descs, _ := probeAll(cfg, g.paths, g.log)
-- req := protocol.PairRequest{Code: code, AgentVersion: version, DeviceKind: cfg.DeviceKind, Drivers: []string{},
-- Capabilities: map[string]any{}, OS: runtime.GOOS, Arch: runtime.GOARCH}
-- req.Hostname, _ = os.Hostname()
++ req := protocol.PairRequest{Robot: a.robot, Code: code, AgentVersion: version, DeviceKind: cfg.DeviceKind, Drivers: []string{},
++ Capabilities: map[string]any{}, Name: name}
names := make([]string, 0, len(descs))
for n := range descs {
names = append(names, n)
@@@ -244,7 +269,7 @@ func cmdPair(g *globals, code, server, kind string, force bool, stdout io.Writer
}
chownLikeDir(g.paths.ConfigFile(), g.paths.ConfigDir)
}
-- fmt.Fprintf(stdout, "Paired as %s. Credential saved to %s (mode 600).\n", creds.DeviceID, g.paths.CredentialFile())
++ fmt.Fprintf(stdout, "Paired as %s (robot %s). Credential saved to %s (mode 600).\n", creds.DeviceID, creds.RobotID, g.paths.CredentialFile())
fmt.Fprintln(stdout, "Confirm the device on openvibe.bot, then start it: `openvibe-node run` or `sudo openvibe-node install`.")
if service.Status(serviceOptions(g, "")) == "running" {
_ = service.Control(serviceOptions(g, ""), "restart")
@@@ -396,8 +421,8 @@ func cmdStatus(g *globals, asJSON bool, stdout io.Writer) error {
}
fmt.Fprintln(stdout)
}
-- if b := s.Telemetry.Battery; b != nil && b.Percent != nil {
-- fmt.Fprintf(stdout, "Battery: %.0f%%\n", *b.Percent)
++ if b := s.Telemetry.Battery; b != nil {
++ fmt.Fprintf(stdout, "Battery: %.0f%%\n", *b*100)
}
return nil
}
ddiff --git a/cmd/openvibe-node/main_test.go b/cmd/openvibe-node/main_test.go
iindex 7bc1767..785a461 100644
---- a/cmd/openvibe-node/main_test.go
++++ b/cmd/openvibe-node/main_test.go
@@@ -12,6 +12,7 @@ import (
"testing"
"time"
++ "github.com/OpenVibers/OpenVibe.Node/internal/config"
"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
"github.com/OpenVibers/OpenVibe.Node/internal/fakebot"
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
@@@ -69,7 +70,11 @@ func TestPair(t *testing.T) {
if _, errOut, code := cli(t, "pair", "nope", "--home", home); code == 0 || !strings.Contains(errOut, "8 letters") {
t.Fatalf("bad code accepted: %s", errOut)
}
-- out, errOut, code := cli(t, "pair", "abcd-2345", "--home", home, "--server", srv.URL())
++ if _, errOut, code := cli(t, "pair", "abcd-2345", "--robot", "adeept", "--home", home, "--server", srv.URL()); code == 0 || !strings.Contains(errOut, "rob_") {
++ t.Fatalf("driver kind accepted as --robot: %s", errOut)
++ }
++ // The flags Bot's installer command uses: --robot rob_… --code CODE.
++ out, errOut, code := cli(t, "pair", "--code", "abcd-2345", "--robot", fakebot.RobotID, "--name", "Rover", "--home", home, "--server", srv.URL())
if code != 0 {
t.Fatalf("pair failed: %s", errOut)
}
@@@ -88,7 +93,7 @@ func TestPair(t *testing.T) {
t.Fatal("config not written")
}
reqs := srv.PairRequests()
-- if len(reqs) != 1 || reqs[0].Code != "ABCD2345" || reqs[0].AgentVersion != version {
++ if len(reqs) != 1 || reqs[0].Code != "ABCD2345" || reqs[0].AgentVersion != version || reqs[0].Robot != fakebot.RobotID || reqs[0].Name != "Rover" {
t.Fatalf("%+v", reqs)
}
if _, err := exec.LookPath("python3"); err == nil && len(reqs[0].Drivers) != 1 {
@@@ -141,6 +146,15 @@ func TestRunDryRun(t *testing.T) {
whip := &video.WHIPReceiver{PublishKey: c.PublishKey.Reveal()}
srv.SetWHIP(whip)
defer whip.Close()
++ // Bot hands out the publish key only; the WHIP endpoint is local config.
++ cfg, err := config.Load(filepath.Join(home, "config.json"))
++ if err != nil {
++ t.Fatal(err)
++ }
++ cfg.Video.WHIPURL = srv.URL() + "/whip/" + c.DeviceID
++ if err := config.Save(filepath.Join(home, "config.json"), cfg); err != nil {
++ t.Fatal(err)
++ }
exe, _ := os.Executable()
cmd := exec.Command(exe, "run", "--dry-run", "--home", home)
@@@ -170,7 +184,7 @@ func TestRunDryRun(t *testing.T) {
}
send := func(id, kind, value string) protocol.Message {
t.Helper()
-- conn.Send(protocol.Command{ID: id, Kind: kind, Value: json.RawMessage(value), DeadlineMS: 300})
++ conn.Command(protocol.Command{ID: id, Kind: kind, Value: json.RawMessage(value)}, 300)
r, err := conn.Reply(id, 5*time.Second)
if err != nil {
t.Fatal(err)
ddiff --git a/docs/install.md b/docs/install.md
iindex 5049f6b..a032ec1 100644
---- a/docs/install.md
++++ b/docs/install.md
@@@ -5,15 +5,19 @@ ports, works behind any home router), and a local kill switch stops everything i
## Quick start (Linux, Raspberry Pi, macOS)
--1. On openvibe.bot, add a robot. You get a code like `ABCD-1234` (valid 10 minutes).
--2. On the device:
++1. On openvibe.bot, add a robot. You get a code like `ABCD-1234` (valid 10 minutes) and the installer command with
++ the robot's id filled in.
++2. On the device, run that command and add the driver:
```sh
-- curl -fsSL https://openvibe.bot/install | sh -s -- ABCD-1234 --robot adeept
++ curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X --code ABCD-1234 --driver adeept
```
-- `--robot` is `adeept` (Adeept 4WD Smart Car, ordinary wheels), `adeept-mecanum`, `cozmo`, or `none` (the dry-run
-- plugin with a test-pattern camera, for trying the Node on a laptop).
++ `--robot` is the robot's `rob_…` id and `--code` the pairing code (both from openvibe.bot; the code may also be the
++ first argument). `--driver` is `adeept` (Adeept 4WD Smart Car, ordinary wheels), `adeept-mecanum`, `cozmo`, or
++ `none` (the default: the dry-run plugin with a test-pattern camera, for trying the Node on a laptop). `--name NAME`
++ names the device on openvibe.bot (default: the hostname). The old form `--robot adeept` (a driver kind after
++ `--robot`) still works but is deprecated: the installer warns and treats it as `--driver adeept`.
3. Confirm the new device on openvibe.bot.
The installer detects the OS and CPU (linux amd64/arm64/armv7, macOS amd64/arm64), downloads the binary and the plugin
@@@ -32,7 +36,7 @@ from plugins that send JPEG frames).
```sh
sudo install -m 755 openvibe-node-linux-arm64 /usr/local/bin/openvibe-node
--sudo openvibe-node pair ABCD-1234 # stores /etc/openvibe-node/credential.json (mode 600)
++sudo openvibe-node pair ABCD-1234 --robot rob_… # stores /etc/openvibe-node/credential.json (mode 600)
sudo openvibe-node install # systemd / launchd / Windows service, starts it
openvibe-node status
```
@@@ -53,7 +57,7 @@ need Python; set `"python"` in `config.json`.
| command | what it does |
|-------------------------------------------|----------------------------------------------------------------------------|
--| `openvibe-node pair <CODE> [--force]` | redeem a pairing code; `--server`, `--kind onboard\|bridge` override config |
++| `openvibe-node pair <CODE> [--force]` | redeem a pairing code (or `--code CODE`); `--robot rob_…` the robot it is for, `--name` the device name (default: hostname); `--server`, `--kind onboard\|bridge` override config |
| `openvibe-node run [--dry-run]` | run in the foreground (what the service runs) |
| `openvibe-node install [--user NAME]` | install and start the service |
| `openvibe-node uninstall` | stop and remove the service (config and credential are kept) |
@@@ -96,7 +100,9 @@ macOS: `/Library/Application Support/OpenVibe Node`. Windows: `%ProgramData%\Ope
test pattern if it asks for one), `test`, `command` (with `video.command`: a program writing H.264 Annex B to stdout,
e.g. `["ffmpeg", "-f", "v4l2", "-i", "/dev/video0", "-c:v", "libx264", "-preset", "ultrafast", "-tune",
"zerolatency", "-profile:v", "baseline", "-g", "30", "-f", "h264", "-"]`), `plugin` (with `video.plugin`), `off`.
-- `fps`, `width`, `height`, `ffmpeg` tune the rest.
++ `fps`, `width`, `height`, `ffmpeg` tune the rest. `whip_url`: where the camera publishes, with the publish key from
++ pairing. OpenVibe.Bot hands out the key but no endpoint, so until `video.whip_url` is set (an `https://` URL) video
++ stays off and the log says so.
- `limits`: local caps on top of the owner's limits from openvibe.bot; the stricter value wins.
- `python`: interpreter for bundled plugins (default `<state>/venv/bin/python`, then `python3`).
@@@ -108,7 +114,7 @@ Unknown keys are rejected, so a typo does not silently change behaviour.
Raspberry Pi OS Bookworm (64-bit recommended). Enable I²C and SPI: `sudo raspi-config` → Interface Options, or
`dtparam=i2c_arm=on` and `dtparam=spi=on` in `/boot/firmware/config.txt`, then reboot. Install with
--`--robot adeept` (or `adeept-mecanum`).
++`--driver adeept` (or `adeept-mecanum`).
The kit's own software (`Adeept_Robot.service`, `WebServer.py` on `0.0.0.0:8888` with the fixed login
`admin:123456`, MJPEG on `:5000`) must not run: anyone on the network could drive the car. The installer disables it;
@@@ -121,7 +127,7 @@ If a direction is wrong on your car, fix it in `config.json` → `plugins[0].con
Cozmo makes its own Wi-Fi network without internet. Use a Raspberry Pi or laptop with two network interfaces: one
joined to Cozmo's network (`Cozmo_…`; raise and lower the lift to see the password on its face), the other to the
--internet (Ethernet, or a second Wi-Fi adapter). Install with `--robot cozmo`; the device kind is `bridge`. See
++internet (Ethernet, or a second Wi-Fi adapter). Install with `--driver cozmo`; the device kind is `bridge`. See
[the plugin README](../plugins/cozmo/README.md).
## Troubleshooting
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 105d3e5..df46d4c 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -1,46 +1,50 @@
# Node ↔ OpenVibe.Bot protocol (bot.device-message@1, device side)
--This is the wire format the Node speaks today, from ADR-043 decisions 1, 2, 4, 5 and 6. OpenVibe.Bot's own
--`docs/protocol.md` is the source of truth; every field name below lives in one Go package, `internal/protocol/`, so
--aligning with it is a change to that package only. The ADR-043 text mirrored in this repo is updated to match (pairing
--field `code`, relative `deadline_ms`, plugin events in `telemetry.events`). Points still to be confirmed with the
--service are marked **(to align)**.
++OpenVibe.Bot's own `docs/protocol.md` (section 1, the `/device` socket) and its `POST /api/v1/pair` handler are the
++source of truth. Every field name below lives in one Go package, `internal/protocol/`, and the exact frames from Bot's
++doc are pinned verbatim in [`internal/protocol/testdata/bot`](../internal/protocol/testdata/bot) (its README names the
++Bot commit). `go test ./internal/protocol ./internal/link` decodes every one of them and drives the fake server
++(`internal/fakebot`) through pair → connect → command → ack → reconnect with those exact bytes.
## 1. Pairing
--The owner adds a robot on openvibe.bot and gets an 8-character code (Crockford base32, shown `XXXX-XXXX`, 10 minutes,
--single use, 5 wrong tries end it). The Node normalises it (upper case; dash and spaces removed; `I`/`L` → `1`, `O` → `0`)
--and redeems it:
++The owner adds a robot on openvibe.bot and gets an 8-character code (shown `XXXX-XXXX`, 10 minutes, single use, 5 wrong
++tries end it) and an installer command with the robot's id: `… | sh -s -- --robot rob_… --code XXXX-XXXX`. The Node
++normalises the code (upper case; dash and spaces removed; `I`/`L` → `1`, `O` → `0`, as Bot does) and redeems it over
++HTTPS, which Bot offers for agents that pair without a socket:
```http
POST https://openvibe.bot/api/v1/pair
Content-Type: application/json
--{"code": "ABCD1234", "agent_version": "v0.1.0", "device_kind": "onboard",
++{"robot": "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X", "code": "7Q2M4XZP", "agent_version": "v0.1.0", "device_kind": "onboard",
"drivers": ["adeept_adr036"],
"capabilities": {"adeept_adr036": {"drive": {"type": "differential"}, "ptz": {...}, ...}},
-- "os": "linux", "arch": "arm64", "hostname": "rover"}
++ "name": "rover"}
```
++The body has exactly the fields of Bot's `pair` frame without the envelope. `robot` (from `--robot`) attributes a wrong
++try to that robot's code; `name` (from `--name`, default the hostname, at most 80 characters) is what the owner sees.
`device_kind` is `onboard` (the Node runs on the robot) or `bridge` (it drives the robot over the robot's own link, as
with Cozmo). `capabilities` is keyed by plugin name and holds each plugin's `describe` capabilities
--([plugins.md](plugins.md)). The code field is `code`, as in Bot's `docs/protocol.md`.
++([plugins.md](plugins.md)).
--Answer `2xx`:
++Answer `201`:
```json
--{"device_id": "dev_…", "credential": "<32 random bytes, encoded>", "publish_key": "…",
-- "profile": {…bot.robot-profile@1…},
-- "whip_url": "https://…/whip/…", "device_url": "wss://openvibe.bot/device",
-- "ice_servers": [{"urls": ["turn:…"], "username": "…", "credential": "…"}]}
++{"device_id": "dev_…", "credential": "…", "publish_key": "…", "robot_id": "rob_…",
++ "profile": {"id": "adeept.adr036", "limits": {"max_command_ms": 300, "heartbeat_ms": 1000}, …}}
```
--`whip_url`, `device_url` and `ice_servers` are optional; without `device_url` the Node dials `wss://<origin>/device`
--**(to align: where the WHIP URL comes from)**. Errors: `404`/`410` or `{"error":"invalid_code"|"expired"}` → "code wrong,
--used or expired"; `429` → "too many tries". Plain `http://` is refused except for `localhost` (tests).
++This is Bot's `paired` frame with `robot_ids[0]` as `robot_id` (and no `profile_id`; it is `profile.id`). Errors are
++RFC 9457 problems; the Node turns their `code` into a sentence: `bot.pairing_code_invalid` (wrong code),
++`bot.pairing_code_locked` (5 wrong tries), `bot.pairing_code_used`, `bot.pairing_code_expired`, `bot.no_pairing_code`,
++`bot.invalid_pairing_code` (malformed); `429` → "wait a minute". Plain `http://` is refused except for `localhost`
++(tests).
--The credential and publish key are stored in `credential.json` (mode 0600) and never appear in a URL, a log line, an
--error, `openvibe-node status` or telemetry (the Go type `credentials.Secret` prints as `[redacted]` in every format).
++The credential and publish key are stored in `credential.json` (mode 0600, with the device and robot ids) and never
++appear in a URL, a log line, an error, `openvibe-node status` or telemetry (the Go type `credentials.Secret` prints as
++`[redacted]` in every format).
## 2. The control link
@@@ -52,7 +56,14 @@ Authorization: Bearer <credential>
User-Agent: openvibe-node/<version>
```
--`401`/`403` means the credential was revoked or rotated away: the Node logs "pair again" and keeps retrying slowly.
++Bot upgrades first and authenticates after, so the Node sends nothing and counts the link as up only once `hello`
++arrives (Bot answers any frame on a not-yet-authenticated socket with `error` `bot.not_paired`). Close codes:
++
++| code / answer | meaning | the Node |
++|------------------------------|------------------------------------------------------|--------------------------------------------------------|
++| `4000` | a second connection with this credential replaced it | waits 60 s (+ jitter) before dialing, so two agents with one credential do not kick each other off every second |
++| `4002`, or HTTP `401`/`403` | the credential is wrong or revoked | retries after ≥ 10 s; gives up after 3 in a row, "pair again" |
++| `4003` | the owner revoked the device while it was connected | stops the link at once, "pair again" |
### Frames
@@@ -65,34 +76,40 @@ Every frame is one JSON object in one text message with the envelope fields at t
| `ts` | int | sender's clock, Unix milliseconds |
| `type` | string | message type |
--Unknown `type`s are ignored (a newer server may send more). Frames are at most 1 MiB.
++Unknown `type`s are ignored (a newer server may send more). Frames are at most 1 MiB. `heartbeat` and `heartbeat_ack`
++carry their own `seq` (the echoed heartbeat number) in the same object as the envelope's; the Node sets a heartbeat's
++`seq` equal to its envelope `seq`, so either reading gives the same number.
### Server → device
| type | fields |
|-----------------|--------------------------------------------------------------------------------------------------|
--| `hello` | `device_id`, `robot_id`, `session`, `server_time` |
--| `config` | `limits` {`max_speed` 0..1, `max_turn` 0..1, `max_command_ms`}, `heartbeat_ms`, `deadman_ms`, `telemetry_ms` (≥ 500), `allowed` [kinds] |
--| `command` | `id` (idempotency key), `kind`, `value`, `deadline_ms`, `operator`, `role`, `target` (optional plugin name) |
--| `estop` | `reason`, `by` — latched |
--| `estop_clear` | `by` — the owner clears the remote e-stop |
--| `heartbeat_ack` | `t` (echo of the heartbeat's `t`; the Node measures RTT from it) |
++| `hello` | `session_id`, `device_id`, `robot_ids` [], `server_time` (RFC 3339) — on every authenticated connection |
++| `config` | `heartbeat_ms`, `limits` {`max_speed` 0..1, `max_turn` 0..1, `max_command_ms`, `heartbeat_ms`}, `allowed_commands` [kinds], `estop_latched` — right after `hello` |
++| `command` | `id` (idempotency key), `kind`, `value`, `deadline_ms` (absolute, motion kinds only), `operator` {`subject`, `role`}, `robot_id`; Node only: `target` (plugin name) |
++| `estop` | `latched` (true: the e-stop latched; false: the owner cleared it), `by`, `at` |
++| `heartbeat_ack` | `seq` (the heartbeat's), `server_time`; the Node measures RTT from its own send time |
++| `error` | `code`, `detail` — a frame the server refused; logged, never fatal |
++
++`config.estop_latched` is the robot's e-stop as the server holds it: an owner e-stop set while the device was offline
++latches the Node on reconnect, and a clear made while offline releases it.
### Device → server
| type | fields |
|---------------|-----------------------------------------------------------------------------------------------------|
--| `status` | `agent_version`, `device_kind`, `os`, `arch`, `drivers` [{`name`, `driver`, `version`, `state`, `capabilities`}], `faults` [{`code`, `driver`, `message`}], `estop_latched`, `local_stop`, `video` |
--| `telemetry` | `battery` {`volts`, `percent`}, `rssi`, `sensors` {…}, `events` [{`name`, `driver`, `ts`, `fields`}] |
--| `ack` | `id`, `latency_ms` (receipt → plugin ack) |
--| `nack` | `id`, `fault_code`, `message` |
--| `heartbeat` | `t` (device clock, ms) |
--| `estop_state` | `latched` (remote e-stop), `local_stop` (local kill switch), `reason` |
--
--`status` and `estop_state` are sent on every connect and whenever a plugin changes state, faults or the latch changes.
--Plugin events (`cliff`, `picked_up`, `low_battery`, `heartbeat_lost`, …) travel in `telemetry.events` and are sent
--within 100 ms; sensor telemetry is sent at most every `telemetry_ms` (default 500 ms, so ≤ 2 Hz). There is no
--separate `event` message type.
++| `status` | `firmware` (`openvibe-node-<version>`), `capabilities` {per plugin}, `faults` [{`code`, `driver`, `message`}], `estop_latched`; Node extras: `agent_version`, `device_kind`, `os`, `arch`, `drivers` [{`name`, `driver`, `version`, `state`, `capabilities`}], `local_stop`, `video` |
++| `telemetry` | `battery` (0..1), `voltage` (V), `rssi`, `sensors` {…}; Node extra: `events` [{`name`, `driver`, `ts`, `fields`}] |
++| `ack` | `id`; Node extra: `latency_ms` (receipt → plugin ack) |
++| `nack` | `id`, `fault_code`; Node extra: `message` |
++| `heartbeat` | `seq`, `rtt_ms` (the last measured round trip, once there is one) |
++| `estop_state` | `latched` (the remote e-stop), `by` (`device`), `at`; Node extras: `local_stop` (local kill switch), `reason` |
++
++`status` is sent when `hello` arrives and whenever a plugin changes state or faults; `estop_state` after every
++`config` and whenever the latch changes (never before `config`: Bot takes a device's `estop_state` as the robot's
++latch, so reporting "clear" before learning of an owner e-stop would clear it). Plugin events (`cliff`, `picked_up`,
++`low_battery`, `heartbeat_lost`, …) travel in `telemetry.events` and are sent within 100 ms; sensor telemetry is sent
++at most every 500 ms (≤ 2 Hz, Bot's cap). There is no separate `event` message type.
### Commands
@@@ -109,15 +126,18 @@ Sign conventions: `throttle`/`x` positive = forward; `steer` positive = turn rig
`rotation` positive = counter-clockwise (left); `pan` positive = right, `tilt` positive = up (plugins can invert a
servo in their config to make the hardware match).
--`deadline_ms` is **relative to receipt** (milliseconds the command stays valid), not an absolute timestamp: a device's
--clock may be minutes off, and a relative deadline cannot be stretched by clock skew. Default 300 ms; capped at
--`max_command_ms` (default 1000). A held control is re-sent by the panel every 150 ms, each with a new `id`.
++`deadline_ms` is an **absolute** instant in Unix milliseconds on the server's clock, present on `drive`, `actuator` and
++`ptz`. A device's clock may be minutes off, so the Node never compares it with its own clock: it takes
++`deadline_ms − ts` (both from the server's clock) as the time the command stays valid and starts that window at
++receipt. A deadline at or before `ts` → `nack bad_value`. No `deadline_ms` → 300 ms. Always capped at `max_command_ms`
++(default 1000). A held control is re-sent by the panel every 150 ms, each with a new `id`. Plugins still get a
++relative `deadline_ms` ([plugins.md](plugins.md)).
What the Node does with a command, in order:
1. A repeated `id` is answered with the first result and never executed again (the cache lives for 10 minutes and
across reconnects).
--2. Unknown kind → `nack unsupported`. Kind not in `config.allowed` → `nack not_allowed` (`halt` is always allowed).
++2. Unknown kind → `nack unsupported`. Kind not in `config.allowed_commands` → `nack not_allowed` (`halt` is always allowed).
3. `halt` → every plugin stops, `ack`.
4. `drive`, `ptz`, `actuator` while the local kill switch is latched → `nack local_stop`; while the remote e-stop is
latched → `nack estopped`.
@@@ -139,10 +159,11 @@ Commands are never queued while offline and never replayed after a reconnect.
### Heartbeat, deadman, reconnect
--- The device sends `heartbeat` every `heartbeat_ms` (default 1000). The server answers `heartbeat_ack`.
--- If nothing at all arrives from the server for `deadman_ms` (default 2 × `heartbeat_ms`: two missed heartbeats), the
-- Node closes the link. Every close stops every actuator at once.
--- Reconnect: exponential backoff from 0.5 s to 30 s with full jitter; at least 10 s after a `401`/`403`.
++- The device sends `heartbeat` every `config.heartbeat_ms` (default 1000). The server answers `heartbeat_ack`.
++- If nothing at all arrives from the server for 2 × `heartbeat_ms` (two missed heartbeats), the Node closes the link.
++ Every close stops every actuator at once. (Bot marks the device offline after `heartbeat_ms × 2 + 3000 ms`.)
++- Reconnect: exponential backoff from 0.5 s to 30 s with full jitter; see the close-code table for `4000`, `4002` and
++ `4003`.
## 3. Safety paths (ADR-043 decision 6)
@@@ -150,7 +171,7 @@ Commands are never queued while offline and never replayed after a reconnect.
|----------------------------------------------|----------------------------------------------------|-----------------------------------------|
| motion command's deadline passes | the plugin (its own timer) | no |
| 2 missed heartbeats / link closed | the core → `stop` to every plugin | no |
--| server `estop` | the core → `estop` to every plugin | yes, persisted; `estop_clear` or `openvibe-node resume` |
++| server `estop` | the core → `estop` to every plugin | yes, persisted; `estop` `latched:false` or `openvibe-node resume` |
| `openvibe-node stop` (local kill switch) | the core → `estop` to every plugin | yes, persisted; only `openvibe-node resume` |
| core shutdown | `stop`, then stdin closed | no |
| core killed / crashed | the plugin: stdin EOF, or 1 s without a heartbeat | no |
@@@ -161,7 +182,9 @@ The latch file is `latch.json` in the state directory. A corrupt latch file is r
## 4. Video (WHIP)
--The Node publishes one H.264 track (Constrained Baseline, `profile-level-id=42e01f`, packetization-mode 1) to
--`whip_url` with `Authorization: Bearer <publish_key>`: `POST` with `Content-Type: application/sdp` → `201` with the
++Bot hands out only the publish key (in the pairing answer); it does not own media and names no WHIP endpoint (the
++camera publishes to OpenRe). So the endpoint is local config, `video.whip_url` in `config.json`; without it video stays
++off and the log says so. When it is set, the Node publishes one H.264 track (Constrained Baseline,
++`profile-level-id=42e01f`, packetization-mode 1) to it with `Authorization: Bearer <publish_key>`: `POST` with `Content-Type: application/sdp` → `201` with the
answer and a `Location`; `DELETE <Location>` on shutdown. PLI/FIR from the far end request a keyframe. The session is
re-established with backoff; video runs on its own goroutines and never waits on control or vice versa.
ddiff --git a/install/install.sh b/install/install.sh
iindex 8bb20dd..9a365e2 100755
---- a/install/install.sh
++++ b/install/install.sh
@@@ -1,14 +1,19 @@
#!/bin/sh
# OpenVibe Node installer.
#
--# curl -fsSL https://openvibe.bot/install | sh -s -- <CODE> [--robot adeept|adeept-mecanum|cozmo|none]
++# curl -fsSL https://openvibe.bot/install | sh -s -- --robot <rob_…> --code <CODE> [--driver adeept|adeept-mecanum|cozmo|none]
#
# Detects the OS and CPU, downloads the openvibe-node binary and the plugin bundle (checking their SHA-256), creates
# a Python virtual environment for the plugins, disables the Adeept kit's stock control server if it finds it,
# installs the system service and pairs the device with <CODE>. Run it again to upgrade; the credential is kept.
++# openvibe.bot shows the first line with the robot id and a fresh code filled in.
#
# Options:
--# --robot KIND adeept (ordinary wheels), adeept-mecanum, cozmo (bridge), none (dry run). Default: none.
++# --robot ROBOT the rob_… id of the robot to pair with (from openvibe.bot). A driver kind here (adeept, …) is the
++# old meaning of --robot: still accepted, but deprecated; use --driver.
++# --code CODE the one-time pairing code (or give it as the first argument).
++# --name NAME the device's name on openvibe.bot (default: the hostname).
++# --driver KIND adeept (ordinary wheels), adeept-mecanum, cozmo (bridge), none (dry run). Default: none.
# --no-service install and pair, but do not install the service.
# --local DIR install from DIR (openvibe-node-<os>-<arch> and openvibe-node-plugins.tar.gz) instead of
# downloading.
@@@ -20,22 +25,38 @@ set -eu
REPO_URL="https://github.com/OpenVibers/OpenVibe.Node"
CODE=""
--ROBOT="none"
++DRIVER="none"
++ROBOT_ID=""
++NAME=""
SERVICE=1
LOCAL=""
TAG="latest"
usage() {
-- echo "usage: install.sh [CODE] [--robot adeept|adeept-mecanum|cozmo|none] [--no-service] [--local DIR] [--version TAG]"
-- echo " curl -fsSL https://openvibe.bot/install | sh -s -- <CODE> --robot adeept"
++ echo "usage: install.sh [--robot rob_…] [--code CODE | CODE] [--name NAME] [--driver adeept|adeept-mecanum|cozmo|none]"
++ echo " [--no-service] [--local DIR] [--version TAG]"
++ echo " curl -fsSL https://openvibe.bot/install | sh -s -- --robot rob_… --code ABCD-1234 --driver adeept"
}
say() { printf '%s\n' "openvibe-node: $*"; }
die() { printf '%s\n' "openvibe-node: error: $*" >&2; exit 1; }
++# --robot is the rob_… pairing target; it used to name the driver kind, which still works but is deprecated.
++robot_arg() {
++ case "$1" in
++ rob_*) ROBOT_ID="$1" ;;
++ *) say "warning: --robot $1 is deprecated; use --driver $1 (--robot now takes the rob_… id from openvibe.bot)"; DRIVER="$1" ;;
++ esac
++}
while [ $# -gt 0 ]; do
case "$1" in
-- --robot) ROBOT="${2:-}"; shift 2 ;;
-- --robot=*) ROBOT="${1#*=}"; shift ;;
++ --robot) robot_arg "${2:-}"; shift 2 ;;
++ --robot=*) robot_arg "${1#*=}"; shift ;;
++ --driver) DRIVER="${2:-}"; shift 2 ;;
++ --driver=*) DRIVER="${1#*=}"; shift ;;
++ --code) CODE="${2:-}"; shift 2 ;;
++ --code=*) CODE="${1#*=}"; shift ;;
++ --name) NAME="${2:-}"; shift 2 ;;
++ --name=*) NAME="${1#*=}"; shift ;;
--no-service) SERVICE=0; shift ;;
--local) LOCAL="${2:-}"; shift 2 ;;
--version) TAG="${2:-}"; shift 2 ;;
@@@ -45,7 +66,8 @@ while [ $# -gt 0 ]; do
esac
done
--case "$ROBOT" in adeept|adeept-mecanum|cozmo|none) ;; *) die "--robot must be adeept, adeept-mecanum, cozmo or none" ;; esac
++case "$DRIVER" in adeept|adeept-mecanum|cozmo|none) ;; *) die "--driver must be adeept, adeept-mecanum, cozmo or none" ;; esac
++case "$ROBOT_ID" in ""|rob_*) ;; *) die "--robot must be a rob_… id" ;; esac
# Root for the service, /usr/local/bin and /etc.
SUDO=""
@@@ -137,11 +159,11 @@ $SUDO "$PIP" install --quiet "$STATE/plugins/sdk" "$STATE/plugins/dryrun"
PLUGINS='[{"name": "dryrun"}]'
KIND=onboard
VIDEO='"source": "auto"'
--case "$ROBOT" in
++case "$DRIVER" in
adeept|adeept-mecanum)
[ $IS_PI = 1 ] || say "warning: this is not a Raspberry Pi; the Adeept plugin will report a hardware fault"
$SUDO "$PIP" install --quiet "$STATE/plugins/adeept_adr036[pi]"
-- WHEELS=ordinary; [ "$ROBOT" = adeept-mecanum ] && WHEELS=mecanum
++ WHEELS=ordinary; [ "$DRIVER" = adeept-mecanum ] && WHEELS=mecanum
PLUGINS="[{\"name\": \"adeept_adr036\", \"config\": {\"backend\": \"real\", \"wheels\": \"$WHEELS\"}}]"
;;
cozmo)
@@@ -193,7 +215,10 @@ if [ -n "$CODE" ]; then
if $SUDO "$BIN_DIR/openvibe-node" status --json 2>/dev/null | grep -q '"paired":true'; then
say "already paired; keeping the credential (to pair again: sudo openvibe-node pair --force <CODE>)"
else
-- $SUDO "$BIN_DIR/openvibe-node" pair "$CODE"
++ set -- pair "$CODE"
++ [ -z "$ROBOT_ID" ] || set -- "$@" --robot "$ROBOT_ID"
++ [ -z "$NAME" ] || set -- "$@" --name "$NAME"
++ $SUDO "$BIN_DIR/openvibe-node" "$@"
fi
else
say "not paired yet: get a code on openvibe.bot and run: sudo openvibe-node pair <CODE>"
ddiff --git a/internal/config/config.go b/internal/config/config.go
iindex 8c857af..2e6eeb2 100644
---- a/internal/config/config.go
++++ b/internal/config/config.go
@@@ -137,6 +137,9 @@ type VideoConfig struct {
FPS int `json:"fps,omitempty"`
Width int `json:"width,omitempty"`
Height int `json:"height,omitempty"`
++ // WHIPURL is where the camera publishes (with the pairing's publish key). OpenVibe.Bot does not hand one out,
++ // so video stays off until it is set here.
++ WHIPURL string `json:"whip_url,omitempty"`
}
// Bundled lists the plugins that ship with the Node.
@@@ -225,6 +228,9 @@ func (c *Config) Validate() error {
if c.Video.Source == "command" && len(c.Video.Command) == 0 {
return errors.New("config: video.source command needs video.command")
}
++ if c.Video.WHIPURL != "" && !strings.HasPrefix(c.Video.WHIPURL, "https://") && !strings.HasPrefix(c.Video.WHIPURL, "http://") {
++ return errors.New("config: video.whip_url must be an https:// URL")
++ }
for _, l := range []*float64{c.Limits.MaxSpeed, c.Limits.MaxTurn} {
if l != nil && (*l < 0 || *l > 1) {
return errors.New("config: limits must be between 0 and 1")
ddiff --git a/internal/credentials/credentials.go b/internal/credentials/credentials.go
iindex 26331e7..8d2b6f2 100644
---- a/internal/credentials/credentials.go
++++ b/internal/credentials/credentials.go
@@@ -44,8 +44,9 @@ func (s *Secret) UnmarshalJSON([]byte) error {
// Credentials is what `openvibe-node pair` stores.
type Credentials struct {
DeviceID string
++ RobotID string // rob_… the device was paired to
Credential Secret
-- PublishKey Secret
++ PublishKey Secret // WHIP publish key; secret like the credential
Server string
DeviceURL string
WHIPURL string
@@@ -58,6 +59,7 @@ type Credentials struct {
type file struct {
Version int `json:"version"`
DeviceID string `json:"device_id"`
++ RobotID string `json:"robot_id,omitempty"`
Credential string `json:"credential"`
PublishKey string `json:"publish_key"`
Server string `json:"server"`
@@@ -77,7 +79,7 @@ func Save(path string, c *Credentials) error {
return errors.New("credentials: refusing to save an empty credential")
}
b, err := json.MarshalIndent(file{
-- Version: 1, DeviceID: c.DeviceID, Credential: c.Credential.Reveal(), PublishKey: c.PublishKey.Reveal(),
++ Version: 1, DeviceID: c.DeviceID, RobotID: c.RobotID, Credential: c.Credential.Reveal(), PublishKey: c.PublishKey.Reveal(),
Server: c.Server, DeviceURL: c.DeviceURL, WHIPURL: c.WHIPURL, ICEServers: c.ICEServers, Profile: c.Profile,
PairedAt: c.PairedAt,
}, "", " ")
@@@ -144,7 +146,7 @@ func Load(path string, warn func(string)) (*Credentials, error) {
return nil, fmt.Errorf("credentials: %s is incomplete; pair again", path)
}
return &Credentials{
-- DeviceID: f.DeviceID, Credential: NewSecret(f.Credential), PublishKey: NewSecret(f.PublishKey),
++ DeviceID: f.DeviceID, RobotID: f.RobotID, Credential: NewSecret(f.Credential), PublishKey: NewSecret(f.PublishKey),
Server: f.Server, DeviceURL: f.DeviceURL, WHIPURL: f.WHIPURL, ICEServers: f.ICEServers, Profile: f.Profile,
PairedAt: f.PairedAt,
}, nil
ddiff --git a/internal/fakebot/fakebot.go b/internal/fakebot/fakebot.go
iindex 3a06f83..531e65e 100644
---- a/internal/fakebot/fakebot.go
++++ b/internal/fakebot/fakebot.go
@@@ -1,6 +1,8 @@
--// Package fakebot is a stand-in for OpenVibe.Bot used by tests: it redeems pairing codes, accepts device WebSockets
--// (credential in the Authorization header only), speaks the ADR-043 message set, and can mount a WHIP endpoint. Tests
--// drive it to send commands, e-stops and silence, and read back what the device sent.
++// Package fakebot is a stand-in for OpenVibe.Bot used by tests. It speaks Bot's real device contract (docs/protocol.md
++// section 1 and POST /api/v1/pair, pinned in internal/protocol/testdata/bot): it redeems pairing codes, upgrades every
++// /device request and then authenticates the Authorization header (a bad credential closes with 4002, a revoked
++// device with 4003, a second connection replaces the first with 4000), sends hello and config, and answers
++// heartbeats. Tests drive it to send commands, e-stops and silence, and read back what the device sent.
package fakebot
import (
@@@ -26,12 +28,20 @@ type Server struct {
// Config is sent to every device right after hello.
Config protocol.Config
++ // Greeting, when set, replaces hello and config with these exact frames (fixtures).
++ Greeting [][]byte
++ // PairBody, when set, is the exact 201 body of a successful POST /api/v1/pair (fixtures). The credential in it is
++ // registered for its device_id.
++ PairBody []byte
// WHIP, when set, serves POST/DELETE under /whip/.
WHIP http.Handler
mu sync.Mutex
-- codes map[string]bool
++ codes map[string]string // code → robot id
creds map[string]string // credential → device id
++ robots map[string]string // device id → robot id
++ live map[string]*Conn // device id → its current connection
++ estop map[string]bool // robot id → the robot's e-stop latch, as Bot keeps it
pairRequests []protocol.PairRequest
credInURL bool
@@@ -39,12 +49,16 @@ type Server struct {
count atomic.Int64
}
++// RobotID is the robot AddCode and AddCredential attach devices to.
++const RobotID = "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X"
++
// New starts the fake server.
func New() *Server {
-- s := &Server{
-- codes: map[string]bool{}, creds: map[string]string{}, conns: make(chan *Conn, 16),
-- Config: protocol.Config{HeartbeatMS: 1000, Limits: protocol.Limits{MaxCommandMS: 1000}},
-- }
++ s := &Server{codes: map[string]string{}, creds: map[string]string{}, robots: map[string]string{}, live: map[string]*Conn{}}
++ s.estop = map[string]bool{}
++ s.conns = make(chan *Conn, 16)
++ s.Config = protocol.Config{HeartbeatMS: 1000, Limits: protocol.Limits{MaxCommandMS: 1000, HeartbeatMS: 1000}}
++ s.Config.AllowedCommands = append([]string(nil), protocol.Kinds...)
mux := http.NewServeMux()
mux.HandleFunc(protocol.PairPath, s.pair)
mux.HandleFunc(protocol.DevicePath, s.device)
@@@ -74,10 +88,10 @@ func (s *Server) SetWHIP(h http.Handler) {
s.mu.Unlock()
}
--// AddCode makes a one-time pairing code valid.
++// AddCode makes a one-time pairing code (normalized, without the dash) valid for RobotID.
func (s *Server) AddCode(code string) {
s.mu.Lock()
-- s.codes[code] = true
++ s.codes[code] = RobotID
s.mu.Unlock()
}
@@@ -85,14 +99,21 @@ func (s *Server) AddCode(code string) {
func (s *Server) AddCredential(cred, deviceID string) {
s.mu.Lock()
s.creds[cred] = deviceID
++ s.robots[deviceID] = RobotID
s.mu.Unlock()
}
--// Revoke invalidates a credential for future connections.
++// Revoke invalidates a credential and closes the device's live connection with 4003, as Bot's revoke does.
func (s *Server) Revoke(cred string) {
s.mu.Lock()
++ dev := s.creds[cred]
delete(s.creds, cred)
++ c := s.live[dev]
++ delete(s.live, dev)
s.mu.Unlock()
++ if c != nil {
++ c.CloseWith(protocol.CloseRevoked, "revoked")
++ }
}
// PairRequests returns what devices sent to /api/v1/pair.
@@@ -109,7 +130,7 @@ func (s *Server) CredentialSeenInURL() bool {
return s.credInURL
}
--// Connections counts accepted device connections.
++// Connections counts accepted (authenticated) device connections.
func (s *Server) Connections() int { return int(s.count.Load()) }
func token() string {
@@@ -118,36 +139,54 @@ func token() string {
return hex.EncodeToString(b)
}
++// problem writes an RFC 9457 body the way Bot's API does.
++func problem(w http.ResponseWriter, status int, code, detail string) {
++ w.Header().Set("Content-Type", "application/problem+json")
++ w.WriteHeader(status)
++ _ = json.NewEncoder(w).Encode(map[string]any{"type": "https://openvibe.network/problems/" + code,
++ "title": http.StatusText(status), "status": status, "code": code, "detail": detail, "error": detail})
++}
++
func (s *Server) pair(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
-- http.Error(w, "method", http.StatusMethodNotAllowed)
++ problem(w, http.StatusMethodNotAllowed, "bot.method_not_allowed", "POST only")
return
}
var req protocol.PairRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
-- w.WriteHeader(http.StatusBadRequest)
++ problem(w, http.StatusBadRequest, "bot.bad_json", "the body must be JSON")
return
}
++ code := strings.ToUpper(strings.ReplaceAll(req.Code, "-", ""))
s.mu.Lock()
s.pairRequests = append(s.pairRequests, req)
-- ok := s.codes[req.Code]
-- delete(s.codes, req.Code)
-- var cred, dev string
++ robot, ok := s.codes[code]
++ if ok && req.Robot != "" && req.Robot != robot {
++ ok = false
++ }
if ok {
-- cred, dev = token(), "dev_"+token()[:12]
-- s.creds[cred] = dev
++ delete(s.codes, code)
}
s.mu.Unlock()
-- w.Header().Set("Content-Type", "application/json")
if !ok {
-- w.WriteHeader(http.StatusNotFound)
-- _ = json.NewEncoder(w).Encode(protocol.PairError{Error: "invalid_code"})
++ problem(w, http.StatusForbidden, "bot.pairing_code_invalid", "that is not a live pairing code")
return
}
-- _ = json.NewEncoder(w).Encode(protocol.PairResponse{
-- DeviceID: dev, Credential: cred, PublishKey: "pk_" + token()[:16],
-- WHIPURL: s.HTTP.URL + "/whip/" + dev, Profile: json.RawMessage(`{"profile":"sim.rover","capabilities":["drive"]}`),
-- })
++ body := s.PairBody
++ if body == nil {
++ cred, dev := token(), "dev_"+token()[:12]
++ body, _ = json.Marshal(protocol.PairResponse{DeviceID: dev, Credential: cred, PublishKey: "pk_" + token()[:16],
++ RobotID: robot, Profile: json.RawMessage(`{"id":"sim.rover","limits":{"max_command_ms":300,"heartbeat_ms":1000}}`)})
++ }
++ var pr protocol.PairResponse
++ _ = json.Unmarshal(body, &pr)
++ s.mu.Lock()
++ s.creds[pr.Credential] = pr.DeviceID
++ s.robots[pr.DeviceID] = pr.RobotID
++ s.mu.Unlock()
++ w.Header().Set("Content-Type", "application/json")
++ w.WriteHeader(http.StatusCreated)
++ _, _ = w.Write(body)
}
var upgrader = websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
@@@ -160,21 +199,49 @@ func (s *Server) device(w http.ResponseWriter, r *http.Request) {
}
}
auth := r.Header.Get("Authorization")
-- dev, ok := s.creds[strings.TrimPrefix(auth, "Bearer ")]
s.mu.Unlock()
-- if !strings.HasPrefix(auth, "Bearer ") || !ok {
-- http.Error(w, "unauthorized", http.StatusUnauthorized)
-- return
-- }
ws, err := upgrader.Upgrade(w, r, nil)
if err != nil {
return
}
++ c := &Conn{ws: ws, recv: make(chan protocol.Frame, 1024), done: make(chan struct{})}
++ if auth == "" {
++ // Bot: a socket without a credential may only pair; everything else is error bot.not_paired.
++ go c.unpaired()
++ return
++ }
++ s.mu.Lock()
++ dev, ok := s.creds[strings.TrimPrefix(auth, "Bearer ")]
++ if ok && strings.HasPrefix(auth, "Bearer ") {
++ c.DeviceID, c.RobotID = dev, s.robots[dev]
++ }
++ old := s.live[dev]
++ if c.DeviceID != "" {
++ s.live[dev] = c
++ }
++ greeting := s.Greeting
++ cfg := s.Config
++ cfg.EstopLatched = s.estop[c.RobotID]
++ c.srv = s
++ s.mu.Unlock()
++ if c.DeviceID == "" {
++ c.CloseWith(protocol.CloseInvalid, "invalid credential")
++ return
++ }
++ if old != nil {
++ old.CloseWith(protocol.CloseReplaced, "replaced by a newer connection")
++ }
s.count.Add(1)
-- c := &Conn{ws: ws, DeviceID: dev, recv: make(chan protocol.Frame, 1024), done: make(chan struct{})}
c.autoAck.Store(true)
-- _ = c.Send(protocol.Hello{DeviceID: dev, Session: token()[:8], ServerTime: time.Now().UnixMilli()})
-- _ = c.Send(s.Config)
++ if greeting != nil {
++ for _, b := range greeting {
++ _ = c.SendRaw(b)
++ }
++ } else {
++ _ = c.Send(protocol.Hello{SessionID: "sess_" + token()[:12], DeviceID: dev, RobotIDs: []string{c.RobotID},
++ ServerTime: time.Now().UTC().Format(time.RFC3339Nano)})
++ _ = c.Send(cfg)
++ }
go c.read()
select {
case s.conns <- c:
@@@ -195,6 +262,8 @@ func (s *Server) NextConn(timeout time.Duration) (*Conn, error) {
// Conn is one device connection as the server sees it.
type Conn struct {
DeviceID string
++ RobotID string
++ srv *Server
ws *websocket.Conn
wmu sync.Mutex
seq uint64
@@@ -207,17 +276,66 @@ type Conn struct {
}
// Send writes one frame to the device.
--func (c *Conn) Send(m protocol.Message) error {
++func (c *Conn) Send(m protocol.Message) error { return c.sendAt(time.Now().UnixMilli(), m) }
++
++// Command sends a command the way Bot does: deadline_ms is the absolute instant ts + deadlineMS (none when 0), and
++// the operator is usr_test unless cmd names one.
++func (c *Conn) Command(cmd protocol.Command, deadlineMS int) error {
++ ts := time.Now().UnixMilli()
++ if deadlineMS > 0 {
++ cmd.DeadlineMS = ts + int64(deadlineMS)
++ }
++ if cmd.Operator == nil {
++ cmd.Operator = &protocol.Operator{Subject: "usr_test", Role: "operator"}
++ }
++ if cmd.RobotID == "" {
++ cmd.RobotID = c.RobotID
++ }
++ return c.sendAt(ts, cmd)
++}
++
++// setEstop records the robot's latch: Bot keeps it from estop frames it sends and estop_state frames it receives.
++func (c *Conn) setEstop(latched bool) {
++ if c.srv == nil {
++ return
++ }
++ c.srv.mu.Lock()
++ c.srv.estop[c.RobotID] = latched
++ c.srv.mu.Unlock()
++}
++
++func (c *Conn) sendAt(ts int64, m protocol.Message) error {
++ if e, ok := m.(protocol.Estop); ok {
++ c.setEstop(e.Latched)
++ }
c.wmu.Lock()
defer c.wmu.Unlock()
c.seq++
-- b, err := protocol.Encode(c.seq, time.Now().UnixMilli(), m)
++ b, err := protocol.Encode(c.seq, ts, m)
if err != nil {
return err
}
return c.ws.WriteMessage(websocket.TextMessage, b)
}
++// CloseWith ends the connection with a WebSocket close code, as Bot does for 4000, 4002 and 4003.
++func (c *Conn) CloseWith(code int, reason string) {
++ c.wmu.Lock()
++ _ = c.ws.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(code, reason), time.Now().Add(time.Second))
++ c.wmu.Unlock()
++ _ = c.ws.Close()
++}
++
++func (c *Conn) unpaired() {
++ defer c.ws.Close()
++ for {
++ if _, _, err := c.ws.ReadMessage(); err != nil {
++ return
++ }
++ _ = c.Send(protocol.Error{Code: protocol.ErrNotPaired, Detail: "send pair first"})
++ }
++}
++
// SendRaw writes arbitrary bytes (for malformed-frame tests).
func (c *Conn) SendRaw(b []byte) error {
c.wmu.Lock()
@@@ -245,8 +363,11 @@ func (c *Conn) read() {
if err != nil {
continue
}
++ if es, ok := f.Msg.(protocol.EstopState); ok {
++ c.setEstop(es.Latched)
++ }
if hb, ok := f.Msg.(protocol.Heartbeat); ok && c.autoAck.Load() && !c.muted.Load() {
-- _ = c.Send(protocol.HeartbeatAck{T: hb.T})
++ _ = c.Send(protocol.HeartbeatAck{Seq: hb.Seq, ServerTime: time.Now().UTC().Format(time.RFC3339Nano)})
}
c.fmu.Lock()
c.frames = append(c.frames, f)
ddiff --git a/internal/link/botfixture_test.go b/internal/link/botfixture_test.go
nnew file mode 100644
iindex 0000000..41870d5
---- /dev/null
++++ b/internal/link/botfixture_test.go
@@@ -0,0 +1,124 @@
++package link
++
++import (
++ "bytes"
++ "context"
++ "encoding/json"
++ "io"
++ "log/slog"
++ "os"
++ "path/filepath"
++ "reflect"
++ "testing"
++ "time"
++
++ "github.com/OpenVibers/OpenVibe.Node/internal/fakebot"
++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++func botFixture(t *testing.T, name string) []byte {
++ t.Helper()
++ b, err := os.ReadFile(filepath.Join("..", "protocol", "testdata", "bot", name+".json"))
++ if err != nil {
++ t.Fatal(err)
++ }
++ return bytes.TrimSpace(b)
++}
++
++// TestBotFixturesEndToEnd drives the fake server with the exact frames of OpenVibe.Bot's docs/protocol.md
++// (internal/protocol/testdata/bot): pair → connect (hello + config) → command → ack → reconnect.
++func TestBotFixturesEndToEnd(t *testing.T) {
++ srv := fakebot.New()
++ defer srv.Close()
++ srv.PairBody = protocol.PairBodyFromPaired(botFixture(t, "paired"))
++ srv.Greeting = [][]byte{botFixture(t, "hello"), botFixture(t, "config")}
++
++ // Pair with the fields of Bot's pair frame (the code normalized, as `openvibe-node pair` does).
++ var req protocol.PairRequest
++ if err := json.Unmarshal(botFixture(t, "pair"), &req); err != nil {
++ t.Fatal(err)
++ }
++ code, err := NormalizeCode(req.Code)
++ if err != nil {
++ t.Fatal(err)
++ }
++ srv.AddCode(code)
++ req.Code = code
++ creds, err := Pair(context.Background(), nil, srv.URL(), req)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if creds.DeviceID != "dev_01J8Z4…" || creds.Credential.Reveal() != "Xb3…" || creds.PublishKey.Reveal() != "Vt9…" ||
++ creds.RobotID != "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" || len(creds.Profile) == 0 {
++ t.Fatalf("credentials %+v", creds)
++ }
++ if got := srv.PairRequests(); len(got) != 1 || !reflect.DeepEqual(got[0], req) {
++ t.Fatalf("pair request %+v, want %+v", got, req)
++ }
++
++ // Connect: hello and config arrive as Bot sends them.
++ u, _ := DeviceURL(srv.URL())
++ rec := newRecorder()
++ l := New(Options{URL: u, Credential: creds.Credential, HeartbeatInterval: 100 * time.Millisecond, BackoffMin: 20 * time.Millisecond,
++ BackoffMax: 100 * time.Millisecond, Log: slog.New(slog.NewTextHandler(io.Discard, nil))}, rec)
++ ctx, cancel := context.WithCancel(context.Background())
++ done := make(chan struct{})
++ go func() { l.Run(ctx); close(done) }()
++ defer func() { cancel(); <-done }()
++
++ greet := func() *fakebot.Conn {
++ t.Helper()
++ c, err := srv.NextConn(3 * time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if h := waitFrame(t, rec, protocol.TypeHello).Msg.(protocol.Hello); h.DeviceID != creds.DeviceID || h.SessionID != "sess_01J8Z4…" {
++ t.Fatalf("hello %+v", h)
++ }
++ if cf := waitFrame(t, rec, protocol.TypeConfig).Msg.(protocol.Config); cf.HeartbeatMS != 1000 || cf.Limits.MaxCommandMS != 300 ||
++ len(cf.AllowedCommands) != 6 || cf.EstopLatched {
++ t.Fatalf("config %+v", cf)
++ }
++ if !l.Connected() {
++ t.Fatal("not connected after hello")
++ }
++ return c
++ }
++ c := greet()
++
++ // Command → ack, both in Bot's shapes.
++ if err := c.SendRaw(botFixture(t, "command_drive")); err != nil {
++ t.Fatal(err)
++ }
++ cmd := waitFrame(t, rec, protocol.TypeCommand).Msg.(protocol.Command)
++ if cmd.ID != "cmd_01J8Z4F…" || cmd.Kind != protocol.KindDrive || cmd.DeadlineMS != 1738065600300 || cmd.Operator == nil ||
++ cmd.Operator.Subject != "usr_01J8…" || cmd.RobotID != creds.RobotID {
++ t.Fatalf("command %+v", cmd)
++ }
++ if err := l.Send(protocol.Ack{ID: cmd.ID}); err != nil {
++ t.Fatal(err)
++ }
++ r, err := c.Reply(cmd.ID, 3*time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ want, err := protocol.Decode(botFixture(t, "ack"))
++ if err != nil {
++ t.Fatal(err)
++ }
++ if !reflect.DeepEqual(r, want.Msg) {
++ t.Fatalf("ack %+v, want %+v", r, want.Msg)
++ }
++ // Heartbeats carry Bot's seq and are acked by it.
++ if f, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, nil); err != nil || f.Msg.(protocol.Heartbeat).Seq != f.Seq {
++ t.Fatalf("heartbeat %+v %v", f, err)
++ }
++
++ // Reconnect: the link comes back with the same credential and gets hello + config again.
++ c.Close()
++ <-rec.gotDown
++ greet()
++ if n := srv.Connections(); n != 2 {
++ t.Fatalf("connections %d", n)
++ }
++}
ddiff --git a/internal/link/link.go b/internal/link/link.go
iindex 1713c6c..95e1454 100644
---- a/internal/link/link.go
++++ b/internal/link/link.go
@@@ -2,7 +2,12 @@
// authenticated with the device credential in the Authorization header of the upgrade (never in the URL). It sends a
// heartbeat every interval, declares the link lost when nothing has arrived for the deadman interval (two heartbeats
// by default), and reconnects with exponential backoff and full jitter. Nothing is queued while offline: Send fails
--// and the caller decides.
++// and the caller decides. The connection counts as up only once the server's hello arrives: OpenVibe.Bot
++// authenticates after the upgrade and answers anything sent earlier with error bot.not_paired.
++//
++// Close codes: 4000 (another connection with this credential replaced this one) waits ReplacedBackoff before trying
++// again, so two agents sharing a credential do not take turns kicking each other off; 4003 (revoked) ends Run;
++// 4002 (credential refused) ends Run after MaxRefused refusals in a row, so a server blip is survived.
package link
import (
@@@ -29,9 +34,10 @@ var ErrOffline = errors.New("link offline")
// Handler receives link events. Calls are made from the link's goroutines, one at a time per connection.
type Handler interface {
-- // Connected runs after the upgrade, before any frame is read. The handler sends status and estop_state here.
++ // Connected runs when the server's hello arrives, before the hello itself reaches Frame. The handler sends status
++ // and estop_state here.
Connected()
-- // Frame is every decoded server frame except heartbeat_ack.
++ // Frame is every decoded server frame except heartbeat_ack and error.
Frame(f protocol.Frame)
// Disconnected runs when a connection ends for any reason. The handler must stop every actuator.
Disconnected(err error)
@@@ -49,6 +55,8 @@ type Options struct {
Deadman time.Duration // default 2 × heartbeat
BackoffMin time.Duration // default 500 ms
BackoffMax time.Duration // default 30 s
++ ReplacedBackoff time.Duration // wait after close 4000; default 60 s
++ MaxRefused int // consecutive 4002 / HTTP 401 before giving up; default 3
}
// Link is the reconnecting control connection.
@@@ -60,6 +68,7 @@ type Link struct {
mu sync.Mutex
out chan []byte
seq uint64
++ hbSent map[uint64]time.Time // heartbeat seq → send time, for the RTT
heartbeat time.Duration
deadman time.Duration
@@@ -83,6 +92,12 @@ func New(opt Options, h Handler) *Link {
if opt.BackoffMax <= 0 {
opt.BackoffMax = 30 * time.Second
}
++ if opt.ReplacedBackoff <= 0 {
++ opt.ReplacedBackoff = 60 * time.Second
++ }
++ if opt.MaxRefused <= 0 {
++ opt.MaxRefused = 3
++ }
if opt.Log == nil {
opt.Log = slog.Default()
}
@@@ -94,17 +109,15 @@ func New(opt Options, h Handler) *Link {
return l
}
--// SetTiming applies the server's heartbeat_ms / deadman_ms (from config) to the current and later connections.
--func (l *Link) SetTiming(heartbeatMS, deadmanMS int) {
++// SetTiming applies the server's config.heartbeat_ms to the current and later connections; the link is declared
++// lost after two intervals without a frame.
++func (l *Link) SetTiming(heartbeatMS int) {
l.mu.Lock()
defer l.mu.Unlock()
if heartbeatMS > 0 {
l.heartbeat = time.Duration(heartbeatMS) * time.Millisecond
l.deadman = 2 * l.heartbeat
}
-- if deadmanMS > 0 {
-- l.deadman = time.Duration(deadmanMS) * time.Millisecond
-- }
}
func (l *Link) timing() (time.Duration, time.Duration) {
@@@ -142,7 +155,23 @@ func (l *Link) Send(m protocol.Message) error {
return ErrOffline
}
l.seq++
-- b, err := protocol.Encode(l.seq, time.Now().UnixMilli(), m)
++ now := time.Now()
++ if hb, ok := m.(protocol.Heartbeat); ok {
++ // The body seq equals the envelope seq, so heartbeat_ack's echo is the same whichever the server reads.
++ hb.Seq = l.seq
++ if rtt := l.lastRTT.Load(); rtt > 0 {
++ ms := time.Duration(rtt).Milliseconds()
++ hb.RTTMS = &ms
++ }
++ m = hb
++ for s := range l.hbSent {
++ if s+8 < l.seq {
++ delete(l.hbSent, s)
++ }
++ }
++ l.hbSent[l.seq] = now
++ }
++ b, err := protocol.Encode(l.seq, now.UnixMilli(), m)
if err != nil {
return err
}
@@@ -154,9 +183,10 @@ func (l *Link) Send(m protocol.Message) error {
}
}
--// Run connects and reconnects until ctx ends.
++// Run connects and reconnects until ctx ends, the credential is revoked (close 4003), or the server has refused the
++// credential MaxRefused times in a row.
func (l *Link) Run(ctx context.Context) {
-- attempt := 0
++ attempt, refused := 0, 0
for ctx.Err() == nil {
start := time.Now()
err := l.session(ctx)
@@@ -165,6 +195,18 @@ func (l *Link) Run(ctx context.Context) {
}
msg := errMsg(err)
l.lastErr.Store(msg)
++ if errors.Is(err, errRevoked) {
++ l.log.Error("link stopped", "err", msg)
++ return
++ }
++ if errors.Is(err, errUnauthorized) {
++ if refused++; refused >= l.opt.MaxRefused {
++ l.log.Error("link stopped: the server refused the credential", "times", refused, "err", msg)
++ return
++ }
++ } else {
++ refused = 0
++ }
if time.Since(start) > 30*time.Second {
attempt = 0
}
@@@ -177,6 +219,10 @@ func (l *Link) Run(ctx context.Context) {
}
// Full jitter, with a floor so a flapping server is not hammered.
wait := l.opt.BackoffMin/2 + time.Duration(rand.Int63n(int64(max)))
++ if errors.Is(err, errReplaced) {
++ // Another agent holds this credential now. Fighting it would flap both; come back much later.
++ wait = l.opt.ReplacedBackoff + time.Duration(rand.Int63n(int64(l.opt.ReplacedBackoff)/4+1))
++ }
attempt++
l.reconnects.Add(1)
l.log.Warn("link down; reconnecting", "err", msg, "in", wait.Round(time.Millisecond))
@@@ -188,7 +234,28 @@ func (l *Link) Run(ctx context.Context) {
}
}
--var errUnauthorized = errors.New("the server refused the device credential (revoked or rotated?); pair again with `openvibe-node pair <CODE>`")
++var (
++ errUnauthorized = errors.New("the server refused the device credential (revoked or rotated?); pair again with `openvibe-node pair <CODE>`")
++ errRevoked = errors.New("the owner revoked this device; pair again with `openvibe-node pair <CODE>`")
++ errReplaced = errors.New("another connection with this device's credential replaced this one (is a second agent running with the same credential?)")
++)
++
++// closeErr maps the server's close codes to the errors Run acts on.
++func closeErr(err error) error {
++ var ce *websocket.CloseError
++ if !errors.As(err, &ce) {
++ return nil
++ }
++ switch ce.Code {
++ case protocol.CloseReplaced:
++ return errReplaced
++ case protocol.CloseInvalid:
++ return errUnauthorized
++ case protocol.CloseRevoked:
++ return errRevoked
++ }
++ return nil
++}
func errMsg(err error) string {
if err == nil {
@@@ -216,13 +283,11 @@ func (l *Link) session(ctx context.Context) error {
defer conn.Close()
conn.SetReadLimit(1 << 20)
++ // Nothing is sent until the server's hello: until then the socket is not authenticated.
out := make(chan []byte, 256)
l.mu.Lock()
-- l.out, l.seq = out, 0
++ l.out, l.seq, l.hbSent = nil, 0, map[uint64]time.Time{}
l.mu.Unlock()
-- l.connected.Store(true)
-- l.connectedAt.Store(time.Now().UnixMilli())
-- l.log.Info("link up", "url", redactURL(l.opt.URL))
sessCtx, cancel := context.WithCancel(ctx)
var endErr error
@@@ -285,7 +350,7 @@ func (l *Link) session(ctx context.Context) error {
}
if !now.Before(next) {
next = now.Add(hb)
-- _ = l.Send(protocol.Heartbeat{T: now.UnixMilli()})
++ _ = l.Send(protocol.Heartbeat{})
}
if now.Sub(time.Unix(0, lastRx.Load())) > deadman {
end(fmt.Errorf("heartbeat lost: nothing from the server for %s", deadman))
@@@ -295,13 +360,16 @@ func (l *Link) session(ctx context.Context) error {
}
}()
-- l.handler.Connected()
--
++ up := false
for {
_, data, err := conn.ReadMessage()
if err != nil {
if sessCtx.Err() == nil {
-- end(fmt.Errorf("read: %s", scrub(err.Error(), l.opt.Credential)))
++ if ce := closeErr(err); ce != nil {
++ end(ce)
++ } else {
++ end(fmt.Errorf("read: %s", scrub(err.Error(), l.opt.Credential)))
++ }
}
return endErr
}
@@@ -315,11 +383,39 @@ func (l *Link) session(ctx context.Context) error {
}
continue
}
-- if ack, ok := f.Msg.(protocol.HeartbeatAck); ok {
-- if ack.T > 0 {
-- l.lastRTT.Store(int64(time.Since(time.UnixMilli(ack.T))))
++ switch m := f.Msg.(type) {
++ case protocol.HeartbeatAck:
++ l.mu.Lock()
++ sent, ok := l.hbSent[m.Seq]
++ delete(l.hbSent, m.Seq)
++ l.mu.Unlock()
++ if ok {
++ // A coarse clock (Windows) can read 0 on a fast loopback; 0 means "not measured yet", so floor at 1 ns.
++ l.lastRTT.Store(int64(max(time.Since(sent), 1)))
}
continue
++ case protocol.Error:
++ if m.Code == protocol.ErrNotPaired {
++ l.log.Warn("server: this connection is not authenticated yet", "code", m.Code, "detail", m.Detail)
++ } else {
++ l.log.Warn("server refused a frame", "code", m.Code, "detail", m.Detail)
++ }
++ continue
++ case protocol.Hello:
++ if !up {
++ up = true
++ l.mu.Lock()
++ l.out = out
++ l.mu.Unlock()
++ l.connected.Store(true)
++ l.connectedAt.Store(time.Now().UnixMilli())
++ l.log.Info("link up", "url", redactURL(l.opt.URL), "device", m.DeviceID, "session", m.SessionID)
++ l.handler.Connected()
++ }
++ }
++ if !up {
++ l.log.Warn("ignoring a frame before hello", "type", f.Type)
++ continue
}
l.handler.Frame(f)
}
ddiff --git a/internal/link/link_test.go b/internal/link/link_test.go
iindex 563ec27..611cc02 100644
---- a/internal/link/link_test.go
++++ b/internal/link/link_test.go
@@@ -4,6 +4,7 @@ import (
"bytes"
"context"
"errors"
++ "io"
"log/slog"
"strings"
"sync"
@@@ -169,7 +170,7 @@ func TestReconnectAfterDrop(t *testing.T) {
waitFrame(t, rec, protocol.TypeConfig)
c.Close()
<-rec.gotDown
-- if err := l.Send(protocol.Heartbeat{T: 1}); !errors.Is(err, ErrOffline) && l.Connected() == false {
++ if err := l.Send(protocol.Heartbeat{}); !errors.Is(err, ErrOffline) && l.Connected() == false {
t.Fatalf("send while offline: %v", err)
}
if _, err := srv.NextConn(3 * time.Second); err != nil {
@@@ -187,12 +188,11 @@ func TestSendOffline(t *testing.T) {
}
}
--func TestUnauthorizedNeverLogsCredential(t *testing.T) {
++// A revoke closes the live socket with 4003: the link stops for good and never logs the credential.
++func TestRevokedStopsAndNeverLogsCredential(t *testing.T) {
srv, l, rec, logs, cancel := setup(t, 100*time.Millisecond)
-- c, _ := srv.NextConn(3 * time.Second)
waitFrame(t, rec, protocol.TypeConfig)
srv.Revoke("cred-0123456789abcdef")
-- c.Close()
<-rec.gotDown
deadline := time.Now().Add(3 * time.Second)
for !strings.Contains(l.Stats().LastError, "pair again") {
@@@ -201,12 +201,88 @@ func TestUnauthorizedNeverLogsCredential(t *testing.T) {
}
time.Sleep(10 * time.Millisecond)
}
++ time.Sleep(200 * time.Millisecond) // well past BackoffMax: a running loop would have dialed again
++ if n := srv.Connections(); n != 1 {
++ t.Fatalf("reconnected after a revoke: %d connections", n)
++ }
cancel()
if strings.Contains(logs.String(), "cred-0123456789abcdef") {
t.Fatalf("credential in logs: %s", logs.String())
}
}
++// A credential the server does not know is closed with 4002 after the upgrade; the link gives up after MaxRefused.
++func TestRefusedCredentialGivesUp(t *testing.T) {
++ srv := fakebot.New()
++ defer srv.Close()
++ u, _ := DeviceURL(srv.URL())
++ rec := newRecorder()
++ l := New(Options{URL: u, Credential: credentials.NewSecret("cred-unknown"), BackoffMin: 10 * time.Millisecond, MaxRefused: 1,
++ Log: slog.New(slog.NewTextHandler(io.Discard, nil))}, rec)
++ done := make(chan struct{})
++ go func() { l.Run(context.Background()); close(done) }()
++ select {
++ case <-done:
++ case <-time.After(5 * time.Second):
++ t.Fatal("link kept dialing with a refused credential")
++ }
++ if !strings.Contains(l.Stats().LastError, "pair again") || rec.connected != 0 {
++ t.Fatalf("last error %q, connected %d", l.Stats().LastError, rec.connected)
++ }
++}
++
++// A second connection with the same credential replaces the first (4000); the replaced link waits ReplacedBackoff
++// instead of dialing straight back and kicking the newcomer off.
++func TestReplacedWaitsBeforeReconnecting(t *testing.T) {
++ srv := fakebot.New()
++ defer srv.Close()
++ srv.AddCredential("cred-x", "dev_x")
++ u, _ := DeviceURL(srv.URL())
++ quiet := slog.New(slog.NewTextHandler(io.Discard, nil))
++ ctx, cancel := context.WithCancel(context.Background())
++ defer cancel()
++ rec := newRecorder()
++ first := New(Options{URL: u, Credential: credentials.NewSecret("cred-x"), BackoffMin: 10 * time.Millisecond,
++ ReplacedBackoff: time.Hour, Log: quiet}, rec)
++ go first.Run(ctx)
++ waitFrame(t, rec, protocol.TypeConfig)
++ second := New(Options{URL: u, Credential: credentials.NewSecret("cred-x"), Log: quiet}, newRecorder())
++ go second.Run(ctx)
++ if err := <-rec.gotDown; err == nil || !strings.Contains(err.Error(), "replaced") {
++ t.Fatalf("err %v", err)
++ }
++ time.Sleep(300 * time.Millisecond)
++ if n := srv.Connections(); n != 2 || !second.Connected() {
++ t.Fatalf("connections %d, newcomer connected %v", n, second.Connected())
++ }
++}
++
++// Before hello the socket is not authenticated: nothing is sent and Connected has not run.
++func TestNothingSentBeforeHello(t *testing.T) {
++ srv := fakebot.New()
++ defer srv.Close()
++ srv.AddCredential("cred-x", "dev_x")
++ srv.Greeting = [][]byte{} // upgrade, authenticate, but stay silent
++ u, _ := DeviceURL(srv.URL())
++ rec := newRecorder()
++ l := New(Options{URL: u, Credential: credentials.NewSecret("cred-x"), HeartbeatInterval: 20 * time.Millisecond,
++ Log: slog.New(slog.NewTextHandler(io.Discard, nil))}, rec)
++ ctx, cancel := context.WithCancel(context.Background())
++ defer cancel()
++ go l.Run(ctx)
++ c, err := srv.NextConn(3 * time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ time.Sleep(30 * time.Millisecond)
++ if len(c.Frames()) != 0 || rec.connected != 0 || l.Connected() {
++ t.Fatalf("sent %d frames before hello", len(c.Frames()))
++ }
++ if err := l.Send(protocol.Heartbeat{}); !errors.Is(err, ErrOffline) {
++ t.Fatalf("send before hello: %v", err)
++ }
++}
++
func TestDeviceURL(t *testing.T) {
for in, want := range map[string]string{
"https://openvibe.bot": "wss://openvibe.bot/device",
@@@ -228,18 +304,18 @@ func TestPair(t *testing.T) {
srv := fakebot.New()
defer srv.Close()
srv.AddCode("ABCD1234")
-- req := protocol.PairRequest{Code: "ABCD1234", AgentVersion: "t", DeviceKind: "onboard", Drivers: []string{"dryrun"}}
++ req := protocol.PairRequest{Robot: fakebot.RobotID, Code: "ABCD1234", AgentVersion: "t", DeviceKind: "onboard", Drivers: []string{"dryrun"}}
c, err := Pair(context.Background(), nil, srv.URL(), req)
if err != nil {
t.Fatal(err)
}
-- if c.DeviceID == "" || c.Credential.IsZero() || c.PublishKey.IsZero() || !strings.HasSuffix(c.WHIPURL, c.DeviceID) {
++ if c.DeviceID == "" || c.Credential.IsZero() || c.PublishKey.IsZero() || c.RobotID != fakebot.RobotID {
t.Fatalf("%+v", c)
}
-- if _, err := Pair(context.Background(), nil, srv.URL(), req); err == nil || !strings.Contains(err.Error(), "wrong, used or expired") {
++ if _, err := Pair(context.Background(), nil, srv.URL(), req); err == nil || !strings.Contains(err.Error(), "not the pairing code") {
t.Fatalf("code reused: %v", err)
}
-- if got := srv.PairRequests(); len(got) != 2 || got[0].Drivers[0] != "dryrun" || got[0].DeviceKind != "onboard" {
++ if got := srv.PairRequests(); len(got) != 2 || got[0].Drivers[0] != "dryrun" || got[0].DeviceKind != "onboard" || got[0].Robot != fakebot.RobotID {
t.Fatalf("%+v", got)
}
if _, err := Pair(context.Background(), nil, "http://openvibe.example", req); err == nil {
ddiff --git a/internal/link/pair.go b/internal/link/pair.go
iindex c95e84d..2d19cf1 100644
---- a/internal/link/pair.go
++++ b/internal/link/pair.go
@@@ -30,7 +30,20 @@ func NormalizeCode(code string) (string, error) {
return c, nil
}
--// Pair redeems a one-time code at <server>/api/v1/pair.
++// RobotRe matches an OpenVibe.Bot robot id (rob_ + a ULID).
++var RobotRe = regexp.MustCompile(`^rob_[0-9A-Za-z]{6,64}$`)
++
++// pairMessages turns Bot's problem codes into what the person at the terminal should do next.
++var pairMessages = map[string]string{
++ "bot.invalid_pairing_code": "a pairing code is 8 letters and digits, like ABCD-1234",
++ "bot.pairing_code_invalid": "that is not the pairing code for this robot; check it, or ask for a new one on openvibe.bot",
++ "bot.pairing_code_locked": "too many wrong tries: the code is dead; ask for a new one on openvibe.bot",
++ "bot.pairing_code_used": "the pairing code was already used; ask for a new one on openvibe.bot",
++ "bot.pairing_code_expired": "the pairing code expired (codes last 10 minutes); ask for a new one on openvibe.bot",
++ "bot.no_pairing_code": "this robot has no live pairing code; ask for a new one on openvibe.bot",
++}
++
++// Pair redeems a one-time code with POST <server>/api/v1/pair.
func Pair(ctx context.Context, client *http.Client, server string, req protocol.PairRequest) (*credentials.Credentials, error) {
u, err := url.Parse(server)
if err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Host == "" {
@@@ -63,19 +76,19 @@ func Pair(ctx context.Context, client *http.Client, server string, req protocol.
return nil, fmt.Errorf("pairing response: %w", err)
}
if resp.StatusCode/100 != 2 {
-- var pe protocol.PairError
++ var pe protocol.Problem
_ = json.Unmarshal(raw, &pe)
-- switch {
-- case resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone || pe.Error == "invalid_code" || pe.Error == "expired":
-- return nil, errors.New("the pairing code is wrong, used or expired; ask for a new one on openvibe.bot")
-- case resp.StatusCode == http.StatusTooManyRequests:
-- return nil, errors.New("too many wrong tries for this code; ask for a new one on openvibe.bot")
++ if m, ok := pairMessages[pe.Code]; ok {
++ return nil, errors.New(m)
++ }
++ if resp.StatusCode == http.StatusTooManyRequests {
++ return nil, errors.New("too many pairing attempts from here; wait a minute and try again")
}
-- msg := pe.Message
++ msg := pe.Detail
if msg == "" {
-- msg = pe.Error
++ msg = pe.Title
}
-- return nil, fmt.Errorf("pairing refused: HTTP %d %s", resp.StatusCode, msg)
++ return nil, fmt.Errorf("pairing refused: HTTP %d %s %s", resp.StatusCode, pe.Code, msg)
}
var pr protocol.PairResponse
if err := json.Unmarshal(raw, &pr); err != nil {
@@@ -85,9 +98,9 @@ func Pair(ctx context.Context, client *http.Client, server string, req protocol.
return nil, errors.New("pairing response has no device id or credential")
}
return &credentials.Credentials{
-- DeviceID: pr.DeviceID, Credential: credentials.NewSecret(pr.Credential), PublishKey: credentials.NewSecret(pr.PublishKey),
-- Server: strings.TrimSuffix(server, "/"), DeviceURL: pr.DeviceURL, WHIPURL: pr.WHIPURL, ICEServers: pr.ICEServers,
-- Profile: pr.Profile, PairedAt: time.Now().UTC(),
++ DeviceID: pr.DeviceID, RobotID: pr.RobotID, Credential: credentials.NewSecret(pr.Credential),
++ PublishKey: credentials.NewSecret(pr.PublishKey), Server: strings.TrimSuffix(server, "/"), Profile: pr.Profile,
++ PairedAt: time.Now().UTC(),
}, nil
}
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex 2f64c6d..96ae687 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -15,6 +15,7 @@ import (
"errors"
"fmt"
"log/slog"
++ "math"
"os"
"os/exec"
"runtime"
@@@ -124,8 +125,16 @@ func New(opt Options) (*Node, error) {
}
n.link = link.New(link.Options{URL: u, Credential: opt.Creds.Credential, UserAgent: "openvibe-node/" + opt.Version,
Log: opt.Log, HeartbeatInterval: opt.HeartbeatInterval, BackoffMin: opt.LinkBackoffMin}, n)
-- if opt.Creds.WHIPURL != "" && opt.Config.Video.Source != "off" {
-- n.pub = video.NewPublisher(video.Options{WHIPURL: opt.Creds.WHIPURL, PublishKey: opt.Creds.PublishKey,
++ // Bot hands out only the publish key; the WHIP endpoint is local config (or an older pairing's stored URL).
++ whip := opt.Config.Video.WHIPURL
++ if whip == "" {
++ whip = opt.Creds.WHIPURL
++ }
++ if whip == "" && opt.Config.Video.Source != "off" {
++ opt.Log.Info("video off: no video.whip_url in the config")
++ }
++ if whip != "" && opt.Config.Video.Source != "off" {
++ n.pub = video.NewPublisher(video.Options{WHIPURL: whip, PublishKey: opt.Creds.PublishKey,
ICEServers: opt.Creds.ICEServers, Log: opt.Log, IncludeLoopback: opt.VideoLoopback})
}
}
@@@ -176,9 +185,10 @@ func (n *Node) Run(ctx context.Context) error {
// ---- link.Handler ----
++// Connected sends status. estop_state waits for config: Bot treats a device's estop_state as the robot's latch, so
++// reporting "clear" before config told us about an owner e-stop set while we were offline would clear it.
func (n *Node) Connected() {
n.sendStatus()
-- n.sendEstopState()
}
func (n *Node) Disconnected(err error) {
@@@ -189,51 +199,69 @@ func (n *Node) Disconnected(err error) {
func (n *Node) Frame(f protocol.Frame) {
switch m := f.Msg.(type) {
case protocol.Hello:
-- n.log.Info("server hello", "device", m.DeviceID)
++ n.log.Info("server hello", "device", m.DeviceID, "robots", m.RobotIDs, "session", m.SessionID)
case protocol.Config:
n.applyConfig(m)
case protocol.Command:
-- n.command(m, time.Now())
++ n.command(m, f.TS, time.Now())
case protocol.Estop:
-- n.log.Warn("e-stop from the server", "reason", m.Reason, "by", m.By)
-- n.mgr.EstopAll() // at once, before the disk write
-- if _, err := n.latch.SetRemote(m.Reason); err != nil {
-- n.log.Error("could not persist the e-stop (it is held in memory)", "err", err)
-- }
-- n.sendEstopState()
-- case protocol.EstopClear:
-- n.log.Warn("e-stop cleared by the server", "by", m.By)
-- if _, err := n.latch.ClearRemote(); err != nil {
-- n.log.Error("could not persist the e-stop clear", "err", err)
++ if m.Latched {
++ n.remoteEstop("by "+m.By, m.By)
++ } else {
++ n.remoteClear(m.By)
}
n.sendEstopState()
}
}
++func (n *Node) remoteEstop(reason, by string) {
++ n.log.Warn("e-stop from the server", "by", by)
++ n.mgr.EstopAll() // at once, before the disk write
++ if _, err := n.latch.SetRemote(reason); err != nil {
++ n.log.Error("could not persist the e-stop (it is held in memory)", "err", err)
++ }
++}
++
++func (n *Node) remoteClear(by string) {
++ n.log.Warn("e-stop cleared by the server", "by", by)
++ if _, err := n.latch.ClearRemote(); err != nil {
++ n.log.Error("could not persist the e-stop clear", "err", err)
++ }
++}
++
func (n *Node) applyConfig(c protocol.Config) {
++ // The server's estop_latched is the owner's latch: it may have been set or cleared while the link was down.
++ if st := n.latch.State(); c.EstopLatched && !st.Remote {
++ n.remoteEstop("latched on the server", "server")
++ } else if !c.EstopLatched && st.Remote {
++ n.remoteClear("server")
++ }
++ n.sendEstopState()
l := safety.Merge(c.Limits, n.opt.Config.Limits.MaxSpeed, n.opt.Config.Limits.MaxTurn, n.opt.Config.Limits.MaxCommandMS)
n.mu.Lock()
n.limits = l
n.allowed = nil
-- if len(c.Allowed) > 0 {
++ if len(c.AllowedCommands) > 0 {
n.allowed = map[string]bool{protocol.KindHalt: true}
-- for _, k := range c.Allowed {
++ for _, k := range c.AllowedCommands {
n.allowed[k] = true
}
}
-- if c.TelemetryMS >= 500 {
-- n.telemMS = c.TelemetryMS
-- }
n.mu.Unlock()
if n.link != nil {
-- n.link.SetTiming(c.HeartbeatMS, c.DeadmanMS)
++ hb := c.HeartbeatMS
++ if hb <= 0 {
++ hb = c.Limits.HeartbeatMS
++ }
++ n.link.SetTiming(hb)
}
n.log.Info("config", "max_speed", l.MaxSpeed, "max_turn", l.MaxTurn, "max_command_ms", l.MaxCommandMS)
}
// command validates and routes one command. The synchronous part (checks, clamping, queueing to the plugin) keeps
--// commands in arrival order; the reply is awaited on its own goroutine.
--func (n *Node) command(c protocol.Command, received time.Time) {
++// commands in arrival order; the reply is awaited on its own goroutine. sentMS is the frame's ts: the absolute
++// deadline_ms minus it is the time the server allowed, on the server's own clock, so clock skew does not matter.
++func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
if c.ID == "" {
n.log.Warn("command without id ignored", "kind", c.Kind)
return
@@@ -310,7 +338,19 @@ func (n *Node) command(c protocol.Command, received time.Time) {
nack(protocol.FaultUnsupported, "no driver on this device handles "+c.Kind)
return
}
-- deadline := limits.Deadline(c.DeadlineMS)
++ deadline := limits.Deadline(0)
++ if c.DeadlineMS > 0 {
++ ref := sentMS
++ if ref <= 0 {
++ ref = received.UnixMilli()
++ }
++ left := c.DeadlineMS - ref
++ if left <= 0 {
++ nack(protocol.FaultBadValue, "the command's deadline had already passed")
++ return
++ }
++ deadline = limits.Deadline(int(min(left, int64(limits.MaxCommandMS)+1)))
++ }
wait := p.Begin(n.ctx, c.ID, c.Kind, value, deadline)
go func() {
r := wait()
@@@ -365,7 +405,8 @@ func (n *Node) pollLatch(ctx context.Context) {
func (n *Node) sendEstopState() {
st := n.latch.State()
-- n.send(protocol.EstopState{Latched: st.Remote, LocalStop: st.Local, Reason: st.RemoteReason})
++ n.send(protocol.EstopState{Latched: st.Remote, By: "device", At: time.Now().UTC().Format(time.RFC3339Nano),
++ LocalStop: st.Local, Reason: st.RemoteReason})
}
// ---- plugin output ----
@@@ -441,7 +482,7 @@ func (n *Node) flushTelemetry(eventsOnly bool) {
n.lastTelem = time.Now()
}
n.mu.Unlock()
-- if t.Battery == nil && len(t.Sensors) == 0 && t.RSSI == nil && len(t.Events) == 0 {
++ if t.Battery == nil && t.Voltage == nil && len(t.Sensors) == 0 && t.RSSI == nil && len(t.Events) == 0 {
return
}
n.send(t)
@@@ -456,8 +497,13 @@ func mergeTelemetry(per map[string]map[string]any, events []protocol.Event) prot
sort.Strings(names)
for _, name := range names {
d := per[name]
-- if b, ok := d["battery"].(map[string]any); ok && t.Battery == nil {
-- t.Battery = &protocol.Battery{Volts: num(b["volts"]), Percent: num(b["percent"])}
++ // Plugins report battery {percent, volts}; Bot wants battery as a 0..1 fraction and voltage in volts.
++ if b, ok := d["battery"].(map[string]any); ok && t.Battery == nil && t.Voltage == nil {
++ if p := num(b["percent"]); p != nil {
++ frac := math.Max(0, math.Min(1, *p/100))
++ t.Battery = &frac
++ }
++ t.Voltage = num(b["volts"])
}
if r := num(d["rssi"]); r != nil && t.RSSI == nil {
v := int(*r)
@@@ -651,12 +697,14 @@ func (n *Node) Status() Status {
func (n *Node) sendStatus() {
st := n.latch.State()
-- s := protocol.Status{AgentVersion: n.opt.Version, DeviceKind: n.opt.Config.DeviceKind, OS: runtime.GOOS, Arch: runtime.GOARCH,
-- EstopLatched: st.Remote, LocalStop: st.Local, Drivers: []protocol.DriverStatus{}, Faults: []protocol.Fault{}}
++ s := protocol.Status{Firmware: "openvibe-node-" + n.opt.Version, Capabilities: map[string]any{}, AgentVersion: n.opt.Version,
++ DeviceKind: n.opt.Config.DeviceKind, OS: runtime.GOOS, Arch: runtime.GOARCH, EstopLatched: st.Remote, LocalStop: st.Local,
++ Drivers: []protocol.DriverStatus{}, Faults: []protocol.Fault{}}
for _, i := range n.mgr.Infos() {
d := protocol.DriverStatus{Name: i.Name, State: i.State}
if i.Describe != nil {
d.Driver, d.Version, d.Capabilities = i.Describe.Driver, i.Describe.Version, i.Describe.Capabilities
++ s.Capabilities[i.Name] = i.Describe.Capabilities
}
s.Drivers = append(s.Drivers, d)
if i.Fault != nil {
ddiff --git a/internal/node/node_test.go b/internal/node/node_test.go
iindex cdbaea1..cb707fb 100644
---- a/internal/node/node_test.go
++++ b/internal/node/node_test.go
@@@ -83,7 +83,7 @@ func start(t *testing.T) *env {
cfg.Python = py
cfg.Plugins = []config.PluginConfig{{Name: "dryrun", Config: map[string]any{"record": record, "log_commands": false},
Env: map[string]string{"PYTHONPATH": filepath.Join(root, "plugins/sdk") + string(os.PathListSeparator) + filepath.Join(root, "plugins/dryrun")}}}
-- cfg.Video = config.VideoConfig{Source: "auto", Width: 64, Height: 48, FPS: 10}
++ cfg.Video = config.VideoConfig{Source: "auto", Width: 64, Height: 48, FPS: 10, WHIPURL: srv.URL() + "/whip/" + creds.DeviceID}
n, err := New(Options{Config: cfg, Paths: paths, Creds: creds, Version: "test", HeartbeatInterval: 100 * time.Millisecond,
LinkBackoffMin: 50 * time.Millisecond, VideoLoopback: true, LatchPoll: 50 * time.Millisecond,
@@@ -141,7 +141,7 @@ func (e *env) cmd(kind, value string, deadline int) protocol.Message {
func (e *env) cmdID(id, kind, value string, deadline int) protocol.Message {
e.t.Helper()
-- if err := e.conn.Send(protocol.Command{ID: id, Kind: kind, Value: json.RawMessage(value), DeadlineMS: deadline, Operator: "usr_1", Role: "operator"}); err != nil {
++ if err := e.conn.Command(protocol.Command{ID: id, Kind: kind, Value: json.RawMessage(value)}, deadline); err != nil {
e.t.Fatal(err)
}
r, err := e.conn.Reply(id, 5*time.Second)
@@@ -320,7 +320,7 @@ func TestEstopLatchedUntilOwnerClears(t *testing.T) {
e := start(t)
n := len(e.records())
e.cmd("drive", `{"throttle":0.3}`, 1000)
-- e.conn.Send(protocol.Estop{Reason: "owner", By: "usr_owner"})
++ e.conn.Send(protocol.Estop{Latched: true, By: "usr_owner"})
e.waitStop(n, time.Second)
if _, err := e.conn.Expect(protocol.TypeEstopState, 2*time.Second, func(m protocol.Message) bool { return m.(protocol.EstopState).Latched }); err != nil {
t.Fatal(err)
@@@ -342,7 +342,7 @@ func TestEstopLatchedUntilOwnerClears(t *testing.T) {
if _, err := e.conn.Expect(protocol.TypeEstopState, 5*time.Second, func(m protocol.Message) bool { return m.(protocol.EstopState).Latched }); err != nil {
t.Fatal(err)
}
-- e.conn.Send(protocol.EstopClear{By: "usr_owner"})
++ e.conn.Send(protocol.Estop{Latched: false, By: "usr_owner"})
if _, err := e.conn.Expect(protocol.TypeEstopState, 2*time.Second, func(m protocol.Message) bool { return !m.(protocol.EstopState).Latched }); err != nil {
t.Fatal(err)
}
@@@ -363,7 +363,7 @@ func TestLocalKillSwitch(t *testing.T) {
t.Fatalf("%q", c)
}
// The owner's estop_clear does not clear the local kill switch.
-- e.conn.Send(protocol.EstopClear{By: "usr_owner"})
++ e.conn.Send(protocol.Estop{Latched: false, By: "usr_owner"})
time.Sleep(100 * time.Millisecond)
if c := nackCode(e.cmd("drive", `{"throttle":0.3}`, 300)); c != protocol.FaultLocalStop {
t.Fatalf("after server clear: %q", c)
@@@ -442,7 +442,7 @@ func TestVideoAndTelemetry(t *testing.T) {
if err != nil {
t.Fatal(err)
}
-- if tl := f.Msg.(protocol.Telemetry); tl.Battery.Percent == nil || tl.Sensors["distance_cm"] == nil {
++ if tl := f.Msg.(protocol.Telemetry); tl.Battery == nil || tl.Sensors["distance_cm"] == nil {
t.Fatalf("%+v", tl)
}
// ≤ 2 Hz.
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex 7b15c18..d2f8118 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -2,7 +2,8 @@
// device WebSocket's frame envelope, each message type, fault codes and the endpoint paths. Nothing outside this
// package knows a JSON field name, so aligning with the service's docs/protocol.md is a change to this package only.
//
--// Source of truth: ADR-043 decisions 1, 2, 4, 5 and 6 (bot.device-message@1).
++// Source of truth: OpenVibe.Bot docs/protocol.md section 1 (the /device socket) and its POST /pair; the exact frames
++// are pinned in testdata/bot (see its README for the Bot commit).
package protocol
import (
@@@ -26,11 +27,21 @@ const (
TypeHello = "hello"
TypeConfig = "config"
TypeCommand = "command"
-- TypeEstop = "estop"
-- TypeEstopClear = "estop_clear"
++ TypeEstop = "estop" // latched true sets the e-stop, false is the owner clearing it
TypeHeartbeatAck = "heartbeat_ack"
++ TypeError = "error"
)
++// Close codes the server ends the device socket with.
++const (
++ CloseReplaced = 4000 // a second connection with the same credential replaced this one
++ CloseInvalid = 4002 // the credential is wrong or revoked (sent right after the upgrade)
++ CloseRevoked = 4003 // the owner revoked the device while it was connected
++)
++
++// Error codes in error.code that the Node acts on.
++const ErrNotPaired = "bot.not_paired" // a frame arrived before the server authenticated the socket
++
// Device → server message types.
const (
TypeStatus = "status"
@@@ -105,10 +116,10 @@ type Frame struct {
// ---- server → device ----
type Hello struct {
-- DeviceID string `json:"device_id,omitempty"`
-- RobotID string `json:"robot_id,omitempty"`
-- Session string `json:"session,omitempty"`
-- ServerTime int64 `json:"server_time,omitempty"`
++ SessionID string `json:"session_id"`
++ DeviceID string `json:"device_id"`
++ RobotIDs []string `json:"robot_ids"`
++ ServerTime string `json:"server_time"` // RFC 3339
}
type Limits struct {
@@@ -116,37 +127,51 @@ type Limits struct {
MaxSpeed *float64 `json:"max_speed,omitempty"`
MaxTurn *float64 `json:"max_turn,omitempty"`
MaxCommandMS int `json:"max_command_ms,omitempty"`
++ HeartbeatMS int `json:"heartbeat_ms,omitempty"`
}
type Config struct {
-- Limits Limits `json:"limits"`
-- HeartbeatMS int `json:"heartbeat_ms,omitempty"`
-- DeadmanMS int `json:"deadman_ms,omitempty"`
-- TelemetryMS int `json:"telemetry_ms,omitempty"`
-- Allowed []string `json:"allowed,omitempty"` // command kinds the current operator may send; empty = all
++ HeartbeatMS int `json:"heartbeat_ms"`
++ Limits Limits `json:"limits"`
++ AllowedCommands []string `json:"allowed_commands"` // command kinds operators may send; empty = all
++ EstopLatched bool `json:"estop_latched"`
++}
++
++// Operator is who sent a command, as the server's gate saw them.
++type Operator struct {
++ Subject string `json:"subject"`
++ Role string `json:"role"`
}
type Command struct {
-- ID string `json:"id"`
-- Kind string `json:"kind"`
-- Value json.RawMessage `json:"value,omitempty"`
-- DeadlineMS int `json:"deadline_ms,omitempty"` // milliseconds from receipt; see docs/protocol.md
-- Operator string `json:"operator,omitempty"`
-- Role string `json:"role,omitempty"`
-- Target string `json:"target,omitempty"` // optional driver name; default: the first driver with the capability
++ ID string `json:"id"`
++ Kind string `json:"kind"`
++ Value json.RawMessage `json:"value,omitempty"`
++ // DeadlineMS is an absolute instant in epoch milliseconds on the server's clock, present on motion kinds only.
++ DeadlineMS int64 `json:"deadline_ms,omitempty"`
++ Operator *Operator `json:"operator,omitempty"`
++ RobotID string `json:"robot_id,omitempty"`
++ Target string `json:"target,omitempty"` // Node only: a driver name; default: the first driver with the capability
}
++// Estop is sent when the e-stop latches (Latched true) or the owner clears it (Latched false).
type Estop struct {
-- Reason string `json:"reason,omitempty"`
-- By string `json:"by,omitempty"`
++ Latched bool `json:"latched"`
++ By string `json:"by,omitempty"`
++ At string `json:"at,omitempty"` // RFC 3339
}
--type EstopClear struct {
-- By string `json:"by,omitempty"`
++// HeartbeatAck echoes the heartbeat's seq. The server writes it over the envelope's seq, so the decoded frame's
++// Seq is the echoed value too.
++type HeartbeatAck struct {
++ Seq uint64 `json:"seq"`
++ ServerTime string `json:"server_time,omitempty"`
}
--type HeartbeatAck struct {
-- T int64 `json:"t"`
++// Error is a frame the server refused.
++type Error struct {
++ Code string `json:"code"`
++ Detail string `json:"detail,omitempty"`
}
// ---- device → server ----
@@@ -166,6 +191,8 @@ type Fault struct {
}
type Status struct {
++ Firmware string `json:"firmware"` // Bot: the agent's own version string
++ Capabilities map[string]any `json:"capabilities"` // Bot: per driver, as in the pairing request
AgentVersion string `json:"agent_version"`
DeviceKind string `json:"device_kind,omitempty"`
OS string `json:"os,omitempty"`
@@@ -177,11 +204,6 @@ type Status struct {
Video string `json:"video,omitempty"`
}
--type Battery struct {
-- Volts *float64 `json:"volts,omitempty"`
-- Percent *float64 `json:"percent,omitempty"`
--}
--
type Event struct {
Name string `json:"name"`
Driver string `json:"driver,omitempty"`
@@@ -190,7 +212,8 @@ type Event struct {
}
type Telemetry struct {
-- Battery *Battery `json:"battery,omitempty"`
++ Battery *float64 `json:"battery,omitempty"` // charge, 0..1
++ Voltage *float64 `json:"voltage,omitempty"`
RSSI *int `json:"rssi,omitempty"`
Sensors map[string]any `json:"sensors,omitempty"`
Events []Event `json:"events,omitempty"`
@@@ -207,12 +230,17 @@ type Nack struct {
Message string `json:"message,omitempty"`
}
++// Heartbeat carries the same seq as its envelope (the link sets it), so the server's echo is unambiguous.
type Heartbeat struct {
-- T int64 `json:"t"`
++ Seq uint64 `json:"seq"`
++ RTTMS *int64 `json:"rtt_ms,omitempty"`
}
type EstopState struct {
Latched bool `json:"latched"`
++ By string `json:"by,omitempty"`
++ At string `json:"at,omitempty"` // RFC 3339
++ RobotID string `json:"robot_id,omitempty"`
LocalStop bool `json:"local_stop"`
Reason string `json:"reason,omitempty"`
}
@@@ -221,8 +249,8 @@ func (Hello) MessageType() string { return TypeHello }
func (Config) MessageType() string { return TypeConfig }
func (Command) MessageType() string { return TypeCommand }
func (Estop) MessageType() string { return TypeEstop }
--func (EstopClear) MessageType() string { return TypeEstopClear }
func (HeartbeatAck) MessageType() string { return TypeHeartbeatAck }
++func (Error) MessageType() string { return TypeError }
func (Status) MessageType() string { return TypeStatus }
func (Telemetry) MessageType() string { return TypeTelemetry }
func (Ack) MessageType() string { return TypeAck }
@@@ -235,8 +263,8 @@ var registry = map[string]func() Message{
TypeConfig: func() Message { return &Config{} },
TypeCommand: func() Message { return &Command{} },
TypeEstop: func() Message { return &Estop{} },
-- TypeEstopClear: func() Message { return &EstopClear{} },
TypeHeartbeatAck: func() Message { return &HeartbeatAck{} },
++ TypeError: func() Message { return &Error{} },
TypeStatus: func() Message { return &Status{} },
TypeTelemetry: func() Message { return &Telemetry{} },
TypeAck: func() Message { return &Ack{} },
@@@ -305,7 +333,7 @@ func derefMessage(m Message) Message {
return *v
case *Estop:
return *v
-- case *EstopClear:
++ case *Error:
return *v
case *HeartbeatAck:
return *v
@@@ -327,21 +355,42 @@ func derefMessage(m Message) Message {
// ---- pairing (HTTPS) ----
++// PairBodyFromPaired turns Bot's `paired` frame (pairing over the socket) into the body its POST /api/v1/pair answers
++// with: the same device id, credential, publish key and profile, and robot_ids[0] as robot_id (server/api/v1.js).
++func PairBodyFromPaired(frame []byte) []byte {
++ var p struct {
++ DeviceID string `json:"device_id"`
++ Credential string `json:"credential"`
++ PublishKey string `json:"publish_key"`
++ RobotIDs []string `json:"robot_ids"`
++ Profile json.RawMessage `json:"profile"`
++ }
++ if json.Unmarshal(frame, &p) != nil {
++ return nil
++ }
++ r := PairResponse{DeviceID: p.DeviceID, Credential: p.Credential, PublishKey: p.PublishKey, Profile: p.Profile}
++ if len(p.RobotIDs) > 0 {
++ r.RobotID = p.RobotIDs[0]
++ }
++ b, _ := json.Marshal(r)
++ return b
++}
++
// DeviceKind values for PairRequest.DeviceKind.
const (
DeviceOnboard = "onboard"
DeviceBridge = "bridge"
)
++// PairRequest is the body of POST /api/v1/pair (the same fields as Bot's `pair` frame, without the envelope).
type PairRequest struct {
++ Robot string `json:"robot,omitempty"` // rob_… from the installer command; attributes a wrong try to that robot's code
Code string `json:"code"`
AgentVersion string `json:"agent_version"`
DeviceKind string `json:"device_kind"`
Drivers []string `json:"drivers"`
Capabilities map[string]any `json:"capabilities"`
-- OS string `json:"os,omitempty"`
-- Arch string `json:"arch,omitempty"`
-- Hostname string `json:"hostname,omitempty"`
++ Name string `json:"name,omitempty"`
}
type ICEServer struct {
@@@ -350,18 +399,19 @@ type ICEServer struct {
Credential string `json:"credential,omitempty"`
}
++// PairResponse is the 201 answer of POST /api/v1/pair. The credential and publish key appear only here.
type PairResponse struct {
DeviceID string `json:"device_id"`
Credential string `json:"credential"`
PublishKey string `json:"publish_key"`
++ RobotID string `json:"robot_id"`
Profile json.RawMessage `json:"profile,omitempty"`
-- WHIPURL string `json:"whip_url,omitempty"`
-- DeviceURL string `json:"device_url,omitempty"` // optional override of wss://<origin>/device
-- ICEServers []ICEServer `json:"ice_servers,omitempty"`
}
--// PairError is the body of a non-2xx pairing answer.
--type PairError struct {
-- Error string `json:"error"`
-- Message string `json:"message,omitempty"`
++// Problem is the RFC 9457 body of a non-2xx answer from Bot's REST API.
++type Problem struct {
++ Status int `json:"status"`
++ Code string `json:"code"` // e.g. bot.pairing_code_invalid
++ Title string `json:"title,omitempty"`
++ Detail string `json:"detail,omitempty"`
}
ddiff --git a/internal/protocol/protocol_test.go b/internal/protocol/protocol_test.go
iindex acd8692..4024103 100644
---- a/internal/protocol/protocol_test.go
++++ b/internal/protocol/protocol_test.go
@@@ -3,23 +3,25 @@ package protocol
import (
"encoding/json"
"errors"
++ "os"
++ "path/filepath"
"testing"
)
func TestEncodeDecodeRoundTrip(t *testing.T) {
msgs := []Message{
-- Hello{DeviceID: "dev_1"},
-- Config{Limits: Limits{MaxCommandMS: 500}, HeartbeatMS: 1000},
-- Command{ID: "c1", Kind: KindDrive, Value: json.RawMessage(`{"throttle":0.5}`), DeadlineMS: 300},
-- Estop{Reason: "owner"},
-- EstopClear{By: "owner"},
-- HeartbeatAck{T: 5},
++ Hello{SessionID: "sess_1", DeviceID: "dev_1", RobotIDs: []string{"rob_1"}},
++ Config{Limits: Limits{MaxCommandMS: 500}, HeartbeatMS: 1000, AllowedCommands: []string{KindDrive}},
++ Command{ID: "c1", Kind: KindDrive, Value: json.RawMessage(`{"throttle":0.5}`), DeadlineMS: 1738065600300},
++ Estop{Latched: true, By: "usr_1"},
++ HeartbeatAck{Seq: 5},
Status{AgentVersion: "1", Drivers: []DriverStatus{}, Faults: []Fault{}},
Telemetry{Sensors: map[string]any{"a": 1.0}},
Ack{ID: "c1"},
Nack{ID: "c1", FaultCode: FaultEstopped},
-- Heartbeat{T: 7},
++ Heartbeat{Seq: 10},
EstopState{Latched: true},
++ Error{Code: ErrNotPaired},
}
for i, m := range msgs {
b, err := Encode(uint64(i+1), 1234, m)
@@@ -30,7 +32,12 @@ func TestEncodeDecodeRoundTrip(t *testing.T) {
if err := json.Unmarshal(b, &raw); err != nil {
t.Fatalf("%s: %v: %s", m.MessageType(), err, b)
}
-- if raw["v"] != float64(1) || raw["seq"] != float64(i+1) || raw["ts"] != float64(1234) || raw["type"] != m.MessageType() {
++ // heartbeat and heartbeat_ack carry their own seq, which wins over the envelope's (as on Bot's wire).
++ _, ownSeq := m.(Heartbeat)
++ if _, ok := m.(HeartbeatAck); ok {
++ ownSeq = true
++ }
++ if raw["v"] != float64(1) || (!ownSeq && raw["seq"] != float64(i+1)) || raw["ts"] != float64(1234) || raw["type"] != m.MessageType() {
t.Fatalf("envelope wrong: %s", b)
}
f, err := Decode(b)
@@@ -44,16 +51,112 @@ func TestEncodeDecodeRoundTrip(t *testing.T) {
}
func TestCommandFields(t *testing.T) {
-- f, err := Decode([]byte(`{"v":1,"seq":9,"ts":1,"type":"command","id":"x","kind":"drive","value":{"throttle":1},"deadline_ms":250,"operator":"usr_1","role":"operator"}`))
++ f, err := Decode([]byte(`{"v":1,"seq":9,"ts":1,"type":"command","id":"x","ref":"op-1","kind":"drive","value":{"throttle":1},"deadline_ms":1738065600300,"operator":{"subject":"usr_1","role":"operator"},"robot_id":"rob_1"}`))
if err != nil {
t.Fatal(err)
}
c := f.Msg.(Command)
-- if c.ID != "x" || c.DeadlineMS != 250 || c.Operator != "usr_1" || string(c.Value) != `{"throttle":1}` {
++ if c.ID != "x" || c.DeadlineMS != 1738065600300 || c.Operator == nil || c.Operator.Subject != "usr_1" || c.Operator.Role != "operator" ||
++ c.RobotID != "rob_1" || string(c.Value) != `{"throttle":1}` {
t.Fatalf("%+v", c)
}
}
++func fixture(t *testing.T, name string) []byte {
++ t.Helper()
++ b, err := os.ReadFile(filepath.Join("testdata", "bot", name+".json"))
++ if err != nil {
++ t.Fatal(err)
++ }
++ return b
++}
++
++// TestBotFixtures decodes every frame Bot's docs/protocol.md shows for the device socket, copied verbatim into
++// testdata/bot, and checks the fields the Node reads.
++func TestBotFixtures(t *testing.T) {
++ decode := func(name string) Frame {
++ t.Helper()
++ f, err := Decode(fixture(t, name))
++ if err != nil {
++ t.Fatalf("%s: %v", name, err)
++ }
++ return f
++ }
++ if h := decode("hello").Msg.(Hello); h.SessionID != "sess_01J8Z4…" || h.DeviceID != "dev_01J8Z4…" ||
++ len(h.RobotIDs) != 1 || h.RobotIDs[0] != "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" || h.ServerTime != "2026-09-29T19:20:00.000Z" {
++ t.Errorf("hello %+v", h)
++ }
++ if c := decode("config").Msg.(Config); c.HeartbeatMS != 1000 || c.Limits.MaxCommandMS != 300 || c.Limits.HeartbeatMS != 1000 ||
++ c.Limits.MaxSpeed == nil || *c.Limits.MaxSpeed != 1 || len(c.AllowedCommands) != 6 || c.EstopLatched {
++ t.Errorf("config %+v", c)
++ }
++ f := decode("command_drive")
++ if c := f.Msg.(Command); c.ID != "cmd_01J8Z4F…" || c.Kind != KindDrive || c.DeadlineMS-f.TS != 300 || c.Operator == nil ||
++ c.Operator.Role != "operator" || c.RobotID != "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" || string(c.Value) != `{ "throttle": 0.6, "steer": -0.25 }` {
++ t.Errorf("command %+v", c)
++ }
++ if f := decode("heartbeat_ack"); f.Seq != 41 || f.Msg.(HeartbeatAck).Seq != 41 {
++ t.Errorf("heartbeat_ack %+v", f)
++ }
++ if e := decode("estop").Msg.(Estop); e.Latched || e.By != "usr_01J8…" {
++ t.Errorf("estop %+v", e)
++ }
++ if e := decode("error").Msg.(Error); e.Code != "bot.forbidden" || e.Detail == "" {
++ t.Errorf("error %+v", e)
++ }
++ // What the device sends decodes too, so the Node's types carry Bot's field names.
++ if a := decode("ack").Msg.(Ack); a.ID != "cmd_01J8Z4F…" {
++ t.Errorf("ack %+v", a)
++ }
++ if n := decode("nack").Msg.(Nack); n.FaultCode != "obstacle" {
++ t.Errorf("nack %+v", n)
++ }
++ if h := decode("heartbeat").Msg.(Heartbeat); h.Seq != 41 || h.RTTMS == nil || *h.RTTMS != 63 {
++ t.Errorf("heartbeat %+v", h)
++ }
++ if tl := decode("telemetry").Msg.(Telemetry); tl.Battery == nil || *tl.Battery != 0.72 || tl.Voltage == nil || tl.RSSI == nil {
++ t.Errorf("telemetry %+v", tl)
++ }
++ if s := decode("estop_state").Msg.(EstopState); !s.Latched || s.By != "device" || s.At == "" {
++ t.Errorf("estop_state %+v", s)
++ }
++ if s := decode("status").Msg.(Status); s.EstopLatched || s.Faults == nil {
++ t.Errorf("status %+v", s)
++ }
++}
++
++// TestBotPairShapes checks the pairing bodies against Bot's pair frame: the request carries exactly its fields, and
++// the REST answer (v1.js: device_id, credential, publish_key, robot_id = robot_ids[0], profile) decodes.
++func TestBotPairShapes(t *testing.T) {
++ var frame map[string]any
++ if err := json.Unmarshal(fixture(t, "pair"), &frame); err != nil {
++ t.Fatal(err)
++ }
++ for _, k := range []string{"v", "seq", "ts", "type"} {
++ delete(frame, k)
++ }
++ b, _ := json.Marshal(PairRequest{Robot: "rob_1", Code: "7Q2M4XZP", AgentVersion: "0.1.0", DeviceKind: DeviceOnboard,
++ Drivers: []string{}, Capabilities: map[string]any{}, Name: "Rover"})
++ var ours map[string]any
++ _ = json.Unmarshal(b, &ours)
++ if len(ours) != len(frame) {
++ t.Fatalf("request fields %v, Bot's pair frame has %v", ours, frame)
++ }
++ for k := range frame {
++ if _, ok := ours[k]; !ok {
++ t.Errorf("request lacks %q", k)
++ }
++ }
++ var r PairResponse
++ if err := json.Unmarshal(PairBodyFromPaired(fixture(t, "paired")), &r); err != nil {
++ t.Fatal(err)
++ }
++ if r.DeviceID != "dev_01J8Z4…" || r.Credential != "Xb3…" || r.PublishKey != "Vt9…" || r.RobotID != "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" ||
++ len(r.Profile) == 0 {
++ t.Fatalf("%+v", r)
++ }
++}
++
func TestDecodeRejects(t *testing.T) {
if _, err := Decode([]byte(`{"v":2,"type":"hello"}`)); err == nil {
t.Fatal("version 2 accepted")
ddiff --git a/internal/protocol/testdata/bot/README.md b/internal/protocol/testdata/bot/README.md
nnew file mode 100644
iindex 0000000..951e9f3
---- /dev/null
++++ b/internal/protocol/testdata/bot/README.md
@@@ -0,0 +1,20 @@
++# OpenVibe.Bot device-contract fixtures
++
++Pinned to **OpenVibe.Bot `d398445479edefa845f0eb32198537434fa11bbd`**.
++
++Each `.json` file is one frame copied verbatim (whitespace and `…` included) from the JSON examples in Bot's
++`docs/protocol.md` at that commit:
++
++| file | Bot `docs/protocol.md` |
++|----------------------------------------------------------------------------|-----------------------------------------------|
++| `pair`, `paired`, `hello`, `config`, `command_drive`, `ack`, `nack`, | section 1 (`/device`), "Examples" |
++| `heartbeat`, `heartbeat_ack`, `telemetry`, `status`, `estop_state`, `estop` | |
++| `error` | section 2 (`/control`): the same `error` frame shape Bot sends on `/device` |
++
++The REST pairing answer has no example in the doc; tests derive it from `paired.json` the way Bot's
++`POST /api/v1/pair` handler (`server/api/v1.js`) builds it: `device_id, credential, publish_key, robot_id`
++(`robot_ids[0]`), `profile` (see `protocol.PairBodyFromPaired`).
++
++Used by `internal/protocol` (every frame decodes, with the fields the Node reads) and `internal/link`
++(`TestBotFixturesEndToEnd`: pair → hello + config → command → ack → reconnect through `internal/fakebot`). When Bot's
++contract changes, copy the new examples here, update the SHA above, and let the tests say what moved.
ddiff --git a/internal/protocol/testdata/bot/ack.json b/internal/protocol/testdata/bot/ack.json
nnew file mode 100644
iindex 0000000..74b0b7b
---- /dev/null
++++ b/internal/protocol/testdata/bot/ack.json
@@@ -0,0 +1 @@
++{ "v": 1, "seq": 10, "ts": 1738065600005, "type": "ack", "id": "cmd_01J8Z4F…" }
ddiff --git a/internal/protocol/testdata/bot/command_drive.json b/internal/protocol/testdata/bot/command_drive.json
nnew file mode 100644
iindex 0000000..cde60b4
---- /dev/null
++++ b/internal/protocol/testdata/bot/command_drive.json
@@@ -0,0 +1,3 @@
++{ "v": 1, "seq": 9, "ts": 1738065600000, "type": "command", "id": "cmd_01J8Z4F…", "kind": "drive",
++ "value": { "throttle": 0.6, "steer": -0.25 }, "deadline_ms": 1738065600300,
++ "operator": { "subject": "usr_01J8…", "role": "operator" }, "robot_id": "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X" }
ddiff --git a/internal/protocol/testdata/bot/config.json b/internal/protocol/testdata/bot/config.json
nnew file mode 100644
iindex 0000000..d70d059
---- /dev/null
++++ b/internal/protocol/testdata/bot/config.json
@@@ -0,0 +1,3 @@
++{ "v": 1, "seq": 3, "ts": 1738065600000, "type": "config",
++ "heartbeat_ms": 1000, "limits": { "max_speed": 1, "max_turn": 1, "max_command_ms": 300, "heartbeat_ms": 1000 },
++ "allowed_commands": ["drive", "actuator", "ptz", "say", "display", "halt"], "estop_latched": false }
ddiff --git a/internal/protocol/testdata/bot/error.json b/internal/protocol/testdata/bot/error.json
nnew file mode 100644
iindex 0000000..96cef4b
---- /dev/null
++++ b/internal/protocol/testdata/bot/error.json
@@@ -0,0 +1 @@
++{ "v": 1, "seq": 4, "ts": 1738065601001, "type": "error", "code": "bot.forbidden", "detail": "only the owner clears the e-stop" }
ddiff --git a/internal/protocol/testdata/bot/estop.json b/internal/protocol/testdata/bot/estop.json
nnew file mode 100644
iindex 0000000..10ca5d4
---- /dev/null
++++ b/internal/protocol/testdata/bot/estop.json
@@@ -0,0 +1 @@
++{ "v": 1, "seq": 6, "ts": 1738065601201, "type": "estop", "latched": false, "by": "usr_01J8…", "at": "2026-09-29T19:20:01.201Z" }
ddiff --git a/internal/protocol/testdata/bot/estop_state.json b/internal/protocol/testdata/bot/estop_state.json
nnew file mode 100644
iindex 0000000..269151f
---- /dev/null
++++ b/internal/protocol/testdata/bot/estop_state.json
@@@ -0,0 +1 @@
++{ "v": 1, "seq": 15, "ts": 1738065601200, "type": "estop_state", "latched": true, "by": "device", "at": "2026-09-29T19:20:01.200Z" }
ddiff --git a/internal/protocol/testdata/bot/heartbeat.json b/internal/protocol/testdata/bot/heartbeat.json
nnew file mode 100644
iindex 0000000..3096c2c
---- /dev/null
++++ b/internal/protocol/testdata/bot/heartbeat.json
@@@ -0,0 +1 @@
++{ "v": 1, "seq": 12, "ts": 1738065600500, "type": "heartbeat", "seq": 41, "rtt_ms": 63 }
ddiff --git a/internal/protocol/testdata/bot/heartbeat_ack.json b/internal/protocol/testdata/bot/heartbeat_ack.json
nnew file mode 100644
iindex 0000000..0ba3e5a
---- /dev/null
++++ b/internal/protocol/testdata/bot/heartbeat_ack.json
@@@ -0,0 +1 @@
++{ "v": 1, "seq": 4, "ts": 1738065600501, "type": "heartbeat_ack", "seq": 41, "server_time": "2026-09-29T19:20:00.501Z" }
ddiff --git a/internal/protocol/testdata/bot/hello.json b/internal/protocol/testdata/bot/hello.json
nnew file mode 100644
iindex 0000000..54144a8
---- /dev/null
++++ b/internal/protocol/testdata/bot/hello.json
@@@ -0,0 +1,2 @@
++{ "v": 1, "seq": 2, "ts": 1738065600000, "type": "hello", "session_id": "sess_01J8Z4…",
++ "device_id": "dev_01J8Z4…", "robot_ids": ["rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X"], "server_time": "2026-09-29T19:20:00.000Z" }
ddiff --git a/internal/protocol/testdata/bot/nack.json b/internal/protocol/testdata/bot/nack.json
nnew file mode 100644
iindex 0000000..08e1029
---- /dev/null
++++ b/internal/protocol/testdata/bot/nack.json
@@@ -0,0 +1 @@
++{ "v": 1, "seq": 11, "ts": 1738065600010, "type": "nack", "id": "cmd_01J8Z4F…", "fault_code": "obstacle" }
ddiff --git a/internal/protocol/testdata/bot/pair.json b/internal/protocol/testdata/bot/pair.json
nnew file mode 100644
iindex 0000000..a76964b
---- /dev/null
++++ b/internal/protocol/testdata/bot/pair.json
@@@ -0,0 +1,3 @@
++{ "v": 1, "seq": 1, "ts": 1738065600000, "type": "pair", "robot": "rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X",
++ "code": "7Q2M-4XZP", "agent_version": "0.1.0", "device_kind": "onboard",
++ "drivers": ["pca9685", "ads7830"], "capabilities": { "camera": { "resolution": "640x480" } }, "name": "Rover" }
ddiff --git a/internal/protocol/testdata/bot/paired.json b/internal/protocol/testdata/bot/paired.json
nnew file mode 100644
iindex 0000000..165a7e2
---- /dev/null
++++ b/internal/protocol/testdata/bot/paired.json
@@@ -0,0 +1,3 @@
++{ "v": 1, "seq": 1, "ts": 1738065600000, "type": "paired", "device_id": "dev_01J8Z4…",
++ "credential": "Xb3…", "publish_key": "Vt9…", "robot_ids": ["rob_01J8Z4M2Q0R7T9YV3K6N8P1W2X"],
++ "profile_id": "adeept.adr036", "profile": { "id": "adeept.adr036", "limits": { "max_command_ms": 300, "heartbeat_ms": 1000 } } }
ddiff --git a/internal/protocol/testdata/bot/status.json b/internal/protocol/testdata/bot/status.json
nnew file mode 100644
iindex 0000000..a24dd1a
---- /dev/null
++++ b/internal/protocol/testdata/bot/status.json
@@@ -0,0 +1,2 @@
++{ "v": 1, "seq": 14, "ts": 1738065600900, "type": "status", "firmware": "adeept-0.1.0",
++ "capabilities": { "drive": { "type": "differential", "axes": ["throttle", "steer"] } }, "faults": [], "estop_latched": false }
ddiff --git a/internal/protocol/testdata/bot/telemetry.json b/internal/protocol/testdata/bot/telemetry.json
nnew file mode 100644
iindex 0000000..36dc6a2
---- /dev/null
++++ b/internal/protocol/testdata/bot/telemetry.json
@@@ -0,0 +1,2 @@
++{ "v": 1, "seq": 13, "ts": 1738065600800, "type": "telemetry", "battery": 0.72, "voltage": 7.41,
++ "sensors": { "ultrasonic": 118 }, "rssi": -57 }