openvibers/openvibe.link Commits

Core: wire protocol, credential store, config and directories

OP OpenVibers committed 29 Sept 2026 · 575dd06ffc · 7 files, +1002 −0

internal/protocol keeps every wire detail (envelope v/seq/ts/type, the
ADR-043 message set, fault codes, pairing request/response).
internal/credentials stores the device credential 0600 behind a Secret type
that prints as [redacted] in every format. internal/config finds the
system or per-user directories and validates config.json.

Changes

ddiff --git a/go.mod b/go.mod
nnew file mode 100644
iindex 0000000..a2f951d
---- /dev/null
++++ b/go.mod
@@@ -0,0 +1,3 @@
++module github.com/OpenVibers/OpenVibe.Node
++
++go 1.24.7
ddiff --git a/internal/config/config.go b/internal/config/config.go
nnew file mode 100644
iindex 0000000..8c857af
---- /dev/null
++++ b/internal/config/config.go
@@@ -0,0 +1,260 @@
++// Package config finds the Node's directories and reads its config file.
++//
++// Layout (system install):
++//
++// Linux /etc/openvibe-node/{config.json,credential.json} /var/lib/openvibe-node/{latch.json,node.sock,venv}
++// macOS /Library/Application Support/OpenVibe Node/ (both)
++// Windows %ProgramData%\OpenVibe Node\ (both)
++//
++// OPENVIBE_NODE_HOME (or --home) puts everything in one directory, which is how tests and people trying the Node
++// without root run it. A non-root user on a machine without a system install gets the per-user config directory.
++package config
++
++import (
++ "encoding/json"
++ "errors"
++ "fmt"
++ "io/fs"
++ "os"
++ "path/filepath"
++ "runtime"
++ "strings"
++)
++
++const (
++ DefaultServer = "https://openvibe.bot"
++ EnvHome = "OPENVIBE_NODE_HOME"
++)
++
++// Paths are the Node's directories.
++type Paths struct {
++ ConfigDir string
++ StateDir string
++}
++
++func (p Paths) ConfigFile() string { return filepath.Join(p.ConfigDir, "config.json") }
++func (p Paths) CredentialFile() string { return filepath.Join(p.ConfigDir, "credential.json") }
++func (p Paths) LatchFile() string { return filepath.Join(p.StateDir, "latch.json") }
++func (p Paths) SocketPath() string { return filepath.Join(p.StateDir, "node.sock") }
++func (p Paths) PluginsDir() string { return filepath.Join(p.StateDir, "plugins") }
++func (p Paths) VenvDir() string { return filepath.Join(p.StateDir, "venv") }
++func (p Paths) FramesDir() string { return filepath.Join(p.StateDir, "frames") }
++
++// Ensure creates the directories with owner-only permissions.
++func (p Paths) Ensure() error {
++ for _, d := range []string{p.ConfigDir, p.StateDir} {
++ if err := os.MkdirAll(d, 0o700); err != nil {
++ return err
++ }
++ }
++ return nil
++}
++
++// SystemPaths are the directories of a system-wide install on this OS.
++func SystemPaths() Paths {
++ switch runtime.GOOS {
++ case "windows":
++ base := os.Getenv("ProgramData")
++ if base == "" {
++ base = `C:\ProgramData`
++ }
++ d := filepath.Join(base, "OpenVibe Node")
++ return Paths{ConfigDir: d, StateDir: d}
++ case "darwin":
++ d := "/Library/Application Support/OpenVibe Node"
++ return Paths{ConfigDir: d, StateDir: d}
++ default:
++ return Paths{ConfigDir: "/etc/openvibe-node", StateDir: "/var/lib/openvibe-node"}
++ }
++}
++
++// HomePaths puts config and state in one directory.
++func HomePaths(dir string) Paths { return Paths{ConfigDir: dir, StateDir: dir} }
++
++// DefaultPaths picks the directories: --home / OPENVIBE_NODE_HOME, else the system install when running as root (or
++// when one exists), else the per-user config directory.
++func DefaultPaths(home string) Paths {
++ if home == "" {
++ home = os.Getenv(EnvHome)
++ }
++ if home != "" {
++ return HomePaths(home)
++ }
++ sys := SystemPaths()
++ if runtime.GOOS == "windows" || os.Geteuid() == 0 {
++ return sys
++ }
++ if _, err := os.Stat(sys.ConfigDir); err == nil {
++ return sys
++ }
++ if d, err := os.UserConfigDir(); err == nil {
++ return HomePaths(filepath.Join(d, "openvibe-node"))
++ }
++ return sys
++}
++
++// Config is config.json.
++type Config struct {
++ // Server is the OpenVibe.Bot origin used for pairing (and the device link unless the pairing returns one).
++ Server string `json:"server,omitempty"`
++ // DeviceKind is "onboard" (the Node runs on the robot) or "bridge" (it drives a robot over the robot's own link).
++ DeviceKind string `json:"device_kind,omitempty"`
++ Plugins []PluginConfig `json:"plugins"`
++ Video VideoConfig `json:"video"`
++ // Limits are the local owner's caps, applied on top of the server's (the stricter value wins).
++ Limits LocalLimits `json:"limits"`
++ // Python is the interpreter for plugins without an explicit command (default: <state>/venv/bin/python, then python3).
++ Python string `json:"python,omitempty"`
++ LogLevel string `json:"log_level,omitempty"`
++}
++
++type LocalLimits struct {
++ MaxSpeed *float64 `json:"max_speed,omitempty"`
++ MaxTurn *float64 `json:"max_turn,omitempty"`
++ MaxCommandMS int `json:"max_command_ms,omitempty"`
++}
++
++// PluginConfig is one driver.
++type PluginConfig struct {
++ // Name identifies the plugin in commands (command.target), logs and status.
++ Name string `json:"name"`
++ // Command is the executable and arguments. Empty for a bundled plugin: <python> -m openvibe_<name>.
++ Command []string `json:"command,omitempty"`
++ Env map[string]string `json:"env,omitempty"`
++ Dir string `json:"dir,omitempty"`
++ Config map[string]any `json:"config,omitempty"` // sent to the plugin in hello
++ Disabled bool `json:"disabled,omitempty"`
++}
++
++// VideoConfig selects the camera source published over WHIP.
++type VideoConfig struct {
++ // Source: "auto" (default: a plugin's h264_command if its program exists, else a plugin's JPEG frames, else the
++ // test pattern when a plugin asks for it), "test", "command", "plugin" or "off".
++ Source string `json:"source,omitempty"`
++ Command []string `json:"command,omitempty"` // for "command": a program writing H.264 Annex B to stdout
++ Plugin string `json:"plugin,omitempty"` // for "plugin": which plugin's frames (default: the first with a camera)
++ FFmpeg string `json:"ffmpeg,omitempty"` // path to ffmpeg for JPEG → H.264 (default: ffmpeg on PATH)
++ FPS int `json:"fps,omitempty"`
++ Width int `json:"width,omitempty"`
++ Height int `json:"height,omitempty"`
++}
++
++// Bundled lists the plugins that ship with the Node.
++var Bundled = map[string]string{
++ "dryrun": "openvibe_dryrun",
++ "adeept_adr036": "openvibe_adeept_adr036",
++ "cozmo": "openvibe_cozmo",
++}
++
++// Default is the config written by `openvibe-node pair` when none exists: the dry-run plugin and its test pattern.
++func Default() *Config {
++ return &Config{
++ Server: DefaultServer,
++ DeviceKind: "onboard",
++ Plugins: []PluginConfig{{Name: "dryrun"}},
++ Video: VideoConfig{Source: "auto"},
++ }
++}
++
++// Load reads config.json; a missing file yields Default().
++func Load(path string) (*Config, error) {
++ b, err := os.ReadFile(path)
++ if errors.Is(err, fs.ErrNotExist) {
++ return Default(), nil
++ }
++ if err != nil {
++ return nil, err
++ }
++ c := &Config{}
++ dec := json.NewDecoder(strings.NewReader(string(b)))
++ dec.DisallowUnknownFields()
++ if err := dec.Decode(c); err != nil {
++ return nil, fmt.Errorf("config %s: %w", path, err)
++ }
++ c.fill()
++ return c, c.Validate()
++}
++
++// Save writes config.json (mode 0600: it can hold plugin settings such as Wi-Fi names).
++func Save(path string, c *Config) error {
++ b, err := json.MarshalIndent(c, "", " ")
++ if err != nil {
++ return err
++ }
++ if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
++ return err
++ }
++ return os.WriteFile(path, append(b, '\n'), 0o600)
++}
++
++func (c *Config) fill() {
++ if c.Server == "" {
++ c.Server = DefaultServer
++ }
++ if c.DeviceKind == "" {
++ c.DeviceKind = "onboard"
++ }
++ if c.Video.Source == "" {
++ c.Video.Source = "auto"
++ }
++}
++
++// Validate checks names and values.
++func (c *Config) Validate() error {
++ if c.DeviceKind != "onboard" && c.DeviceKind != "bridge" {
++ return fmt.Errorf("config: device_kind must be onboard or bridge, not %q", c.DeviceKind)
++ }
++ seen := map[string]bool{}
++ for _, p := range c.Plugins {
++ if p.Name == "" || strings.ContainsAny(p.Name, " /\\") {
++ return fmt.Errorf("config: plugin name %q is empty or has spaces or slashes", p.Name)
++ }
++ if seen[p.Name] {
++ return fmt.Errorf("config: plugin %q listed twice", p.Name)
++ }
++ seen[p.Name] = true
++ if len(p.Command) == 0 && Bundled[p.Name] == "" {
++ return fmt.Errorf("config: plugin %q needs a command (it is not a bundled plugin)", p.Name)
++ }
++ }
++ switch c.Video.Source {
++ case "auto", "test", "command", "plugin", "off":
++ default:
++ return fmt.Errorf("config: video.source %q is not auto, test, command, plugin or off", c.Video.Source)
++ }
++ if c.Video.Source == "command" && len(c.Video.Command) == 0 {
++ return errors.New("config: video.source command needs video.command")
++ }
++ 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")
++ }
++ }
++ return nil
++}
++
++// PluginCommand resolves a plugin's argv.
++func (c *Config) PluginCommand(p PluginConfig, paths Paths) []string {
++ if len(p.Command) > 0 {
++ return p.Command
++ }
++ return []string{c.PythonPath(paths), "-m", Bundled[p.Name]}
++}
++
++// PythonPath is the interpreter for bundled plugins.
++func (c *Config) PythonPath(paths Paths) string {
++ if c.Python != "" {
++ return c.Python
++ }
++ venv := filepath.Join(paths.VenvDir(), "bin", "python")
++ if runtime.GOOS == "windows" {
++ venv = filepath.Join(paths.VenvDir(), "Scripts", "python.exe")
++ }
++ if _, err := os.Stat(venv); err == nil {
++ return venv
++ }
++ if runtime.GOOS == "windows" {
++ return "python"
++ }
++ return "python3"
++}
ddiff --git a/internal/config/config_test.go b/internal/config/config_test.go
nnew file mode 100644
iindex 0000000..8abd258
---- /dev/null
++++ b/internal/config/config_test.go
@@@ -0,0 +1,67 @@
++package config
++
++import (
++ "os"
++ "path/filepath"
++ "testing"
++)
++
++func TestDefaultWhenMissing(t *testing.T) {
++ c, err := Load(filepath.Join(t.TempDir(), "none.json"))
++ if err != nil {
++ t.Fatal(err)
++ }
++ if c.Server != DefaultServer || len(c.Plugins) != 1 || c.Plugins[0].Name != "dryrun" {
++ t.Fatalf("%+v", c)
++ }
++}
++
++func TestSaveLoadValidate(t *testing.T) {
++ p := filepath.Join(t.TempDir(), "config.json")
++ c := Default()
++ c.Plugins = append(c.Plugins, PluginConfig{Name: "custom", Command: []string{"/bin/true"}})
++ if err := Save(p, c); err != nil {
++ t.Fatal(err)
++ }
++ got, err := Load(p)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if len(got.Plugins) != 2 {
++ t.Fatalf("%+v", got)
++ }
++ bad := []string{
++ `{"device_kind":"robot"}`,
++ `{"plugins":[{"name":"x"}]}`,
++ `{"plugins":[{"name":"dryrun"},{"name":"dryrun"}]}`,
++ `{"video":{"source":"webcam"}}`,
++ `{"limits":{"max_speed":2}}`,
++ `{"typo":1}`,
++ }
++ for _, b := range bad {
++ os.WriteFile(p, []byte(b), 0o600)
++ if _, err := Load(p); err == nil {
++ t.Errorf("accepted %s", b)
++ }
++ }
++}
++
++func TestPaths(t *testing.T) {
++ p := DefaultPaths("/tmp/x")
++ if p.ConfigDir != "/tmp/x" || p.CredentialFile() != filepath.Join("/tmp/x", "credential.json") {
++ t.Fatalf("%+v", p)
++ }
++ t.Setenv(EnvHome, "/tmp/y")
++ if DefaultPaths("").StateDir != "/tmp/y" {
++ t.Fatal("env ignored")
++ }
++}
++
++func TestPluginCommand(t *testing.T) {
++ c := Default()
++ c.Python = "/usr/bin/python3"
++ got := c.PluginCommand(PluginConfig{Name: "cozmo"}, HomePaths(t.TempDir()))
++ if len(got) != 3 || got[2] != "openvibe_cozmo" {
++ t.Fatalf("%v", got)
++ }
++}
ddiff --git a/internal/credentials/credentials.go b/internal/credentials/credentials.go
nnew file mode 100644
iindex 0000000..26331e7
---- /dev/null
++++ b/internal/credentials/credentials.go
@@@ -0,0 +1,151 @@
++// Package credentials stores the device credential and publish key the pairing returned. The file is mode 0600 and
++// the secret values are wrapped in Secret, whose every string form is "[redacted]", so they cannot reach a log line,
++// an error or `openvibe-node status` by accident.
++package credentials
++
++import (
++ "encoding/json"
++ "errors"
++ "fmt"
++ "io/fs"
++ "os"
++ "path/filepath"
++ "runtime"
++ "time"
++
++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++const redacted = "[redacted]"
++
++// Secret holds a value that must never be printed. Use Reveal only where the value goes on the wire.
++type Secret struct{ v string }
++
++func NewSecret(s string) Secret { return Secret{v: s} }
++
++// Reveal returns the secret. Call it only to put the value in an Authorization header or the credential file.
++func (s Secret) Reveal() string { return s.v }
++
++func (s Secret) IsZero() bool { return s.v == "" }
++
++func (s Secret) String() string { return redacted }
++func (s Secret) GoString() string { return redacted }
++
++// Format makes %v, %+v, %#v, %s and %q all print "[redacted]".
++func (s Secret) Format(f fmt.State, verb rune) { _, _ = f.Write([]byte(redacted)) }
++
++func (s Secret) MarshalJSON() ([]byte, error) { return json.Marshal(redacted) }
++
++// UnmarshalJSON refuses: secrets are read only through Load.
++func (s *Secret) UnmarshalJSON([]byte) error {
++ return errors.New("credentials: secrets are not decoded from JSON")
++}
++
++// Credentials is what `openvibe-node pair` stores.
++type Credentials struct {
++ DeviceID string
++ Credential Secret
++ PublishKey Secret
++ Server string
++ DeviceURL string
++ WHIPURL string
++ ICEServers []protocol.ICEServer
++ Profile json.RawMessage
++ PairedAt time.Time
++}
++
++// file is the on-disk form; only Save and Load see plain secrets.
++type file struct {
++ Version int `json:"version"`
++ DeviceID string `json:"device_id"`
++ Credential string `json:"credential"`
++ PublishKey string `json:"publish_key"`
++ Server string `json:"server"`
++ DeviceURL string `json:"device_url,omitempty"`
++ WHIPURL string `json:"whip_url,omitempty"`
++ ICEServers []protocol.ICEServer `json:"ice_servers,omitempty"`
++ Profile json.RawMessage `json:"profile,omitempty"`
++ PairedAt time.Time `json:"paired_at"`
++}
++
++// ErrNotPaired is returned by Load when there is no credential file.
++var ErrNotPaired = errors.New("this device is not paired; run `openvibe-node pair <CODE>`")
++
++// Save writes the credentials atomically with mode 0600.
++func Save(path string, c *Credentials) error {
++ if c.Credential.IsZero() || c.DeviceID == "" {
++ 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(),
++ Server: c.Server, DeviceURL: c.DeviceURL, WHIPURL: c.WHIPURL, ICEServers: c.ICEServers, Profile: c.Profile,
++ PairedAt: c.PairedAt,
++ }, "", " ")
++ if err != nil {
++ return err
++ }
++ dir := filepath.Dir(path)
++ if err := os.MkdirAll(dir, 0o700); err != nil {
++ return fmt.Errorf("credentials: %w", err)
++ }
++ tmp, err := os.CreateTemp(dir, ".credential-*")
++ if err != nil {
++ return fmt.Errorf("credentials: %w", err)
++ }
++ defer os.Remove(tmp.Name())
++ if err := tmp.Chmod(0o600); err != nil && runtime.GOOS != "windows" {
++ tmp.Close()
++ return fmt.Errorf("credentials: %w", err)
++ }
++ if _, err := tmp.Write(append(b, '\n')); err != nil {
++ tmp.Close()
++ return fmt.Errorf("credentials: write failed")
++ }
++ if err := tmp.Sync(); err != nil {
++ tmp.Close()
++ return fmt.Errorf("credentials: %w", err)
++ }
++ if err := tmp.Close(); err != nil {
++ return fmt.Errorf("credentials: %w", err)
++ }
++ if err := os.Rename(tmp.Name(), path); err != nil {
++ return fmt.Errorf("credentials: %w", err)
++ }
++ return os.Chmod(path, 0o600)
++}
++
++// Load reads the credentials. A file readable by group or others is tightened to 0600 and reported through warn.
++func Load(path string, warn func(string)) (*Credentials, error) {
++ st, err := os.Stat(path)
++ if errors.Is(err, fs.ErrNotExist) {
++ return nil, ErrNotPaired
++ }
++ if err != nil {
++ return nil, fmt.Errorf("credentials: %w", err)
++ }
++ if runtime.GOOS != "windows" && st.Mode().Perm()&0o077 != 0 {
++ if err := os.Chmod(path, 0o600); err != nil {
++ return nil, fmt.Errorf("credentials: %s is readable by others (mode %o) and could not be fixed: %w", path, st.Mode().Perm(), err)
++ }
++ if warn != nil {
++ warn(fmt.Sprintf("credential file %s was mode %o; set it to 600", path, st.Mode().Perm()))
++ }
++ }
++ b, err := os.ReadFile(path)
++ if err != nil {
++ return nil, fmt.Errorf("credentials: %w", err)
++ }
++ var f file
++ if err := json.Unmarshal(b, &f); err != nil {
++ // Never wrap the decoder error: it can quote file contents.
++ return nil, fmt.Errorf("credentials: %s is not valid JSON", path)
++ }
++ if f.DeviceID == "" || f.Credential == "" {
++ return nil, fmt.Errorf("credentials: %s is incomplete; pair again", path)
++ }
++ return &Credentials{
++ DeviceID: f.DeviceID, 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/credentials/credentials_test.go b/internal/credentials/credentials_test.go
nnew file mode 100644
iindex 0000000..e3ac3a5
---- /dev/null
++++ b/internal/credentials/credentials_test.go
@@@ -0,0 +1,85 @@
++package credentials
++
++import (
++ "encoding/json"
++ "errors"
++ "fmt"
++ "os"
++ "path/filepath"
++ "runtime"
++ "strings"
++ "testing"
++)
++
++const secret = "s3cr3t-credential-value"
++
++func TestSecretNeverPrints(t *testing.T) {
++ c := Credentials{DeviceID: "dev_1", Credential: NewSecret(secret), PublishKey: NewSecret("pk-" + secret)}
++ for _, s := range []string{
++ fmt.Sprint(c), fmt.Sprintf("%v %+v %#v %s %q", c, c, c, c.Credential, c.Credential),
++ fmt.Sprintf("%v", &c), c.Credential.String(),
++ } {
++ if strings.Contains(s, secret) {
++ t.Fatalf("secret printed: %s", s)
++ }
++ }
++ b, _ := json.Marshal(c)
++ if strings.Contains(string(b), secret) {
++ t.Fatalf("secret in JSON: %s", b)
++ }
++ if c.Credential.Reveal() != secret {
++ t.Fatal("Reveal broken")
++ }
++}
++
++func TestSaveLoad(t *testing.T) {
++ p := filepath.Join(t.TempDir(), "sub", "credential.json")
++ if _, err := Load(p, nil); !errors.Is(err, ErrNotPaired) {
++ t.Fatalf("want ErrNotPaired, got %v", err)
++ }
++ in := &Credentials{DeviceID: "dev_1", Credential: NewSecret(secret), PublishKey: NewSecret("pk"), Server: "https://x"}
++ if err := Save(p, in); err != nil {
++ t.Fatal(err)
++ }
++ if runtime.GOOS != "windows" {
++ st, _ := os.Stat(p)
++ if st.Mode().Perm() != 0o600 {
++ t.Fatalf("mode %o", st.Mode().Perm())
++ }
++ }
++ out, err := Load(p, nil)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if out.Credential.Reveal() != secret || out.PublishKey.Reveal() != "pk" || out.DeviceID != "dev_1" {
++ t.Fatalf("%+v", out)
++ }
++}
++
++func TestLoadTightensMode(t *testing.T) {
++ if runtime.GOOS == "windows" {
++ t.Skip("no unix modes")
++ }
++ p := filepath.Join(t.TempDir(), "credential.json")
++ if err := Save(p, &Credentials{DeviceID: "d", Credential: NewSecret(secret)}); err != nil {
++ t.Fatal(err)
++ }
++ os.Chmod(p, 0o644)
++ var warned string
++ if _, err := Load(p, func(s string) { warned = s }); err != nil {
++ t.Fatal(err)
++ }
++ st, _ := os.Stat(p)
++ if st.Mode().Perm() != 0o600 || warned == "" || strings.Contains(warned, secret) {
++ t.Fatalf("mode %o warn %q", st.Mode().Perm(), warned)
++ }
++}
++
++func TestCorruptFileErrorHasNoContents(t *testing.T) {
++ p := filepath.Join(t.TempDir(), "credential.json")
++ os.WriteFile(p, []byte(`{"credential":"`+secret+`" broken`), 0o600)
++ _, err := Load(p, nil)
++ if err == nil || strings.Contains(err.Error(), secret) {
++ t.Fatalf("err %v", err)
++ }
++}
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
nnew file mode 100644
iindex 0000000..ddc4409
---- /dev/null
++++ b/internal/protocol/protocol.go
@@@ -0,0 +1,368 @@
++// Package protocol holds every wire detail of the link between the Node and OpenVibe.Bot: the pairing request, the
++// 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).
++package protocol
++
++import (
++ "bytes"
++ "encoding/json"
++ "errors"
++ "fmt"
++)
++
++// Version is the envelope version, the `v` field of every frame.
++const Version = 1
++
++// Endpoint paths, relative to the server origin (default https://openvibe.bot).
++const (
++ PairPath = "/api/v1/pair"
++ DevicePath = "/device"
++)
++
++// Server → device message types.
++const (
++ TypeHello = "hello"
++ TypeConfig = "config"
++ TypeCommand = "command"
++ TypeEstop = "estop"
++ TypeEstopClear = "estop_clear"
++ TypeHeartbeatAck = "heartbeat_ack"
++)
++
++// Device → server message types.
++const (
++ TypeStatus = "status"
++ TypeTelemetry = "telemetry"
++ TypeAck = "ack"
++ TypeNack = "nack"
++ TypeHeartbeat = "heartbeat"
++ TypeEstopState = "estop_state"
++)
++
++// Command kinds.
++const (
++ KindDrive = "drive"
++ KindActuator = "actuator"
++ KindPTZ = "ptz"
++ KindSay = "say"
++ KindDisplay = "display"
++ KindHalt = "halt"
++)
++
++// Kinds lists every command kind the Node understands.
++var Kinds = []string{KindDrive, KindActuator, KindPTZ, KindSay, KindDisplay, KindHalt}
++
++// GuardedKinds move something and are refused while an e-stop or the local kill switch is latched.
++var GuardedKinds = map[string]bool{KindDrive: true, KindActuator: true, KindPTZ: true}
++
++// Fault codes carried in nack.fault_code and status.faults[].code.
++const (
++ FaultBadFrame = "bad_frame" // the command could not be parsed
++ FaultBadValue = "bad_value" // a value is missing, not a number, or out of shape
++ FaultUnsupported = "unsupported" // no driver on this device handles the kind or the actuator
++ FaultNotAllowed = "not_allowed" // the kind is not in config.allowed
++ FaultEstopped = "estopped" // the remote e-stop is latched
++ FaultLocalStop = "local_stop" // the local kill switch (`openvibe-node stop`) is latched
++ FaultNoHeartbeat = "no_heartbeat" // a plugin lost the core's heartbeat
++ FaultExpired = "expired" // the deadline had already passed on arrival
++ FaultNotReady = "not_ready" // the driver is starting or restarting
++ FaultPluginDown = "plugin_down" // the driver process is not running
++ FaultPluginTimeout = "plugin_timeout" // the driver did not answer in time
++ FaultHardware = "hardware" // the driver reported a hardware error
++ FaultFirmware = "firmware_unsupported"
++ FaultNotConnected = "not_connected" // a bridge driver cannot reach its robot
++ FaultCliff = "cliff" // refused because the robot is at a cliff edge
++ FaultShuttingDown = "shutting_down"
++)
++
++// DefaultDeadlineMS is used when a motion command carries no deadline.
++const DefaultDeadlineMS = 300
++
++// DefaultMaxCommandMS caps deadlines when the profile sets no max_command_ms.
++const DefaultMaxCommandMS = 1000
++
++// DefaultHeartbeatMS is the device heartbeat interval; the link is declared lost after two intervals without a frame.
++const DefaultHeartbeatMS = 1000
++
++// Message is any typed payload.
++type Message interface{ MessageType() string }
++
++// Envelope is the part common to every frame.
++type Envelope struct {
++ V int `json:"v"`
++ Seq uint64 `json:"seq"`
++ TS int64 `json:"ts"`
++ Type string `json:"type"`
++}
++
++// Frame is a decoded frame: the envelope plus the typed message.
++type Frame struct {
++ Envelope
++ Msg Message
++}
++
++// ---- 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"`
++}
++
++type Limits struct {
++ // MaxSpeed and MaxTurn are fractions of full scale (0..1). Nil means no owner limit (1.0).
++ MaxSpeed *float64 `json:"max_speed,omitempty"`
++ MaxTurn *float64 `json:"max_turn,omitempty"`
++ MaxCommandMS int `json:"max_command_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
++}
++
++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
++}
++
++type Estop struct {
++ Reason string `json:"reason,omitempty"`
++ By string `json:"by,omitempty"`
++}
++
++type EstopClear struct {
++ By string `json:"by,omitempty"`
++}
++
++type HeartbeatAck struct {
++ T int64 `json:"t"`
++}
++
++// ---- device → server ----
++
++type DriverStatus struct {
++ Name string `json:"name"`
++ Driver string `json:"driver,omitempty"`
++ Version string `json:"version,omitempty"`
++ State string `json:"state"`
++ Capabilities map[string]any `json:"capabilities,omitempty"`
++}
++
++type Fault struct {
++ Code string `json:"code"`
++ Driver string `json:"driver,omitempty"`
++ Message string `json:"message,omitempty"`
++}
++
++type Status struct {
++ AgentVersion string `json:"agent_version"`
++ DeviceKind string `json:"device_kind,omitempty"`
++ OS string `json:"os,omitempty"`
++ Arch string `json:"arch,omitempty"`
++ Drivers []DriverStatus `json:"drivers"`
++ Faults []Fault `json:"faults"`
++ EstopLatched bool `json:"estop_latched"`
++ LocalStop bool `json:"local_stop"`
++ 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"`
++ TS int64 `json:"ts"`
++ Fields map[string]any `json:"fields,omitempty"`
++}
++
++type Telemetry struct {
++ Battery *Battery `json:"battery,omitempty"`
++ RSSI *int `json:"rssi,omitempty"`
++ Sensors map[string]any `json:"sensors,omitempty"`
++ Events []Event `json:"events,omitempty"`
++}
++
++type Ack struct {
++ ID string `json:"id"`
++ LatencyMS int64 `json:"latency_ms,omitempty"`
++}
++
++type Nack struct {
++ ID string `json:"id"`
++ FaultCode string `json:"fault_code"`
++ Message string `json:"message,omitempty"`
++}
++
++type Heartbeat struct {
++ T int64 `json:"t"`
++}
++
++type EstopState struct {
++ Latched bool `json:"latched"`
++ LocalStop bool `json:"local_stop"`
++ Reason string `json:"reason,omitempty"`
++}
++
++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 (Status) MessageType() string { return TypeStatus }
++func (Telemetry) MessageType() string { return TypeTelemetry }
++func (Ack) MessageType() string { return TypeAck }
++func (Nack) MessageType() string { return TypeNack }
++func (Heartbeat) MessageType() string { return TypeHeartbeat }
++func (EstopState) MessageType() string { return TypeEstopState }
++
++var registry = map[string]func() Message{
++ TypeHello: func() Message { return &Hello{} },
++ 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{} },
++ TypeStatus: func() Message { return &Status{} },
++ TypeTelemetry: func() Message { return &Telemetry{} },
++ TypeAck: func() Message { return &Ack{} },
++ TypeNack: func() Message { return &Nack{} },
++ TypeHeartbeat: func() Message { return &Heartbeat{} },
++ TypeEstopState: func() Message { return &EstopState{} },
++}
++
++// ErrUnknownType is returned (wrapped) by Decode for a frame whose type is not in this version's message set.
++var ErrUnknownType = errors.New("unknown frame type")
++
++// Encode builds one frame: the message's fields plus v, seq, ts and type at the top level.
++func Encode(seq uint64, ts int64, m Message) ([]byte, error) {
++ body, err := json.Marshal(m)
++ if err != nil {
++ return nil, err
++ }
++ env, err := json.Marshal(Envelope{V: Version, Seq: seq, TS: ts, Type: m.MessageType()})
++ if err != nil {
++ return nil, err
++ }
++ // Splice the two objects: {"v":..,"type":..} + {"id":..} → {"v":..,"type":..,"id":..}.
++ body = bytes.TrimSpace(body)
++ if len(body) < 2 || body[0] != '{' {
++ return nil, fmt.Errorf("message %s did not encode as an object", m.MessageType())
++ }
++ if len(body) == 2 { // {}
++ return env, nil
++ }
++ out := make([]byte, 0, len(env)+len(body))
++ out = append(out, env[:len(env)-1]...)
++ out = append(out, ',')
++ out = append(out, body[1:]...)
++ return out, nil
++}
++
++// Decode parses a frame. Unknown types return ErrUnknownType with the envelope filled, so the caller can ignore
++// frames from a newer server.
++func Decode(data []byte) (Frame, error) {
++ var f Frame
++ if err := json.Unmarshal(data, &f.Envelope); err != nil {
++ return f, fmt.Errorf("%s: %w", FaultBadFrame, err)
++ }
++ if f.V != Version {
++ return f, fmt.Errorf("%s: unsupported version %d", FaultBadFrame, f.V)
++ }
++ mk, ok := registry[f.Type]
++ if !ok {
++ return f, fmt.Errorf("%w %q", ErrUnknownType, f.Type)
++ }
++ m := mk()
++ if err := json.Unmarshal(data, m); err != nil {
++ return f, fmt.Errorf("%s: %s: %w", FaultBadFrame, f.Type, err)
++ }
++ f.Msg = derefMessage(m)
++ return f, nil
++}
++
++func derefMessage(m Message) Message {
++ switch v := m.(type) {
++ case *Hello:
++ return *v
++ case *Config:
++ return *v
++ case *Command:
++ return *v
++ case *Estop:
++ return *v
++ case *EstopClear:
++ return *v
++ case *HeartbeatAck:
++ return *v
++ case *Status:
++ return *v
++ case *Telemetry:
++ return *v
++ case *Ack:
++ return *v
++ case *Nack:
++ return *v
++ case *Heartbeat:
++ return *v
++ case *EstopState:
++ return *v
++ }
++ return m
++}
++
++// ---- pairing (HTTPS) ----
++
++// DeviceKind values for PairRequest.DeviceKind.
++const (
++ DeviceOnboard = "onboard"
++ DeviceBridge = "bridge"
++)
++
++type PairRequest struct {
++ 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"`
++}
++
++type ICEServer struct {
++ URLs []string `json:"urls"`
++ Username string `json:"username,omitempty"`
++ Credential string `json:"credential,omitempty"`
++}
++
++type PairResponse struct {
++ DeviceID string `json:"device_id"`
++ Credential string `json:"credential"`
++ PublishKey string `json:"publish_key"`
++ 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"`
++}
ddiff --git a/internal/protocol/protocol_test.go b/internal/protocol/protocol_test.go
nnew file mode 100644
iindex 0000000..acd8692
---- /dev/null
++++ b/internal/protocol/protocol_test.go
@@@ -0,0 +1,68 @@
++package protocol
++
++import (
++ "encoding/json"
++ "errors"
++ "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},
++ 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},
++ EstopState{Latched: true},
++ }
++ for i, m := range msgs {
++ b, err := Encode(uint64(i+1), 1234, m)
++ if err != nil {
++ t.Fatal(err)
++ }
++ var raw map[string]any
++ 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() {
++ t.Fatalf("envelope wrong: %s", b)
++ }
++ f, err := Decode(b)
++ if err != nil {
++ t.Fatalf("decode %s: %v", b, err)
++ }
++ if f.Msg.MessageType() != m.MessageType() {
++ t.Fatalf("type %s != %s", f.Msg.MessageType(), m.MessageType())
++ }
++ }
++}
++
++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"}`))
++ 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}` {
++ t.Fatalf("%+v", c)
++ }
++}
++
++func TestDecodeRejects(t *testing.T) {
++ if _, err := Decode([]byte(`{"v":2,"type":"hello"}`)); err == nil {
++ t.Fatal("version 2 accepted")
++ }
++ if _, err := Decode([]byte(`nope`)); err == nil {
++ t.Fatal("garbage accepted")
++ }
++ _, err := Decode([]byte(`{"v":1,"type":"future"}`))
++ if !errors.Is(err, ErrUnknownType) {
++ t.Fatalf("want ErrUnknownType, got %v", err)
++ }
++}