openvibers/openvibe.link Commits

the Node applies the machine's roles from the network's hub at its check-in (#90)

OP OpenVibers committed 11 days ago · 7d714d6582 · 6 files, +273 −22

* the Node applies the machine's roles from the network's hub at its check-in

The owner sets what a machine does for the network on openvibe.network/machines
(plan L34: codes.worker, desks.host, ai.host, bot.robot). Bot keeps the roles on
the device record and sends them in the config frame; the Node reads them at its
check-in, keeps them in its status, and applies the ones this machine was
installed for: the job worker starts when the local configuration has it, and
the machine says plainly which role it could not honour.

The local configuration always wins (ADR-045): a role never installs anything
and never turns on a feature the owner turned off here. A role the Node does not
know is ignored, never fatal.

* the roles test: unix only, and the files gofmt clean

The other end-to-end tests of the node package are unix only (the control socket is
a unix socket), so the roles test follows them and cannot use the windows build's
package. gofmt formats the new field and the new test.

* Merge branch 'main' into feat/machines-hub-node

* Merge branch 'main' into feat/machines-hub-node

Changes

ddiff --git a/README.md b/README.md
iindex ed13794..566d657 100644
---- a/README.md
++++ b/README.md
@@@ -71,6 +71,12 @@ this computer's models" on openvibe.bot/machines ([docs/ai.md](docs/ai.md)). The
job worker runs each job as an unprivileged user in a sandbox, and the OpenVibe.Codes job runner lets you start a coding
task on openvibe.codes/run from any browser and have this machine do it ([docs/worker.md](docs/worker.md)).
++**What the machine does for the network:** the owner sets the machine's roles on openvibe.network/machines
++(`codes.worker`, `desks.host`, `ai.host`, `bot.robot`). The Node reads them at its check-in, keeps them, and applies
++the ones this machine was installed for: the job worker starts if the local configuration has it, and the machine says
++plainly which role it could not honour. The local configuration always wins: a role never installs anything and never
++turns on a feature you turned off here.
++
Docs: [protocol](docs/protocol.md) (the wire format with OpenVibe.Bot), [plugins](docs/plugins.md) (the driver
protocol, the bundled drivers, and where Actor's computer-control plugin attaches), [install](docs/install.md),
[memory](docs/memory.md) (what the coding tools on a machine remember, how the Node reads it and how it is uploaded to
ddiff --git a/STATUS.json b/STATUS.json
iindex 8293a2c..2a85cac 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -19,6 +19,7 @@
"the computer plugin for OpenVibe.Actor: terminal, files, screen, input and its own browser, at the level the owner sets on the machine (observe, ask, trusted, full); approvals in the tray (Linux, Windows), a notification or openvibe-node approve",
"the live screen: the machine's screen on openvibe.bot for its owner only, up to 30 minutes",
"the job worker (--worker): sandboxed function, code and linux jobs, and the OpenVibe.Codes job runner",
++ "the machine's roles from the network's hub (codes.worker, desks.host, ai.host, bot.robot): read at check-in, kept, and applied to what the machine was installed for; the local configuration always wins",
"one-command install on a Raspberry Pi or other Linux robot (I2C, SPI, camera tools, kit detection)",
"system service on systemd, launchd and Windows (or a Windows sign-in start)",
"ESP32 library (OpenVibeNode): pairing, control link, drive and actuator commands, telemetry, heartbeat, deadman and e-stop"
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 88407aa..d7b8547 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -156,7 +156,7 @@ server's own counter, never the heartbeat's.
| type | fields |
|-----------------|--------------------------------------------------------------------------------------------------|
| `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` |
++| `config` | `heartbeat_ms`, `limits` {`max_speed` 0..1, `max_turn` 0..1, `max_command_ms`, `heartbeat_ms`}, `allowed_commands` [kinds], `estop_latched`, `roles` [ids] — right after `hello` |
| `command` | `id` (server-minted: the idempotency and ack key), `ref` (the operator's own id, logs only), `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` | `echo` (the heartbeat's `t`, or null), `t` (the same send time when the server has it), `server_time` (RFC 3339); the Node measures RTT from the echoed send time |
@@@ -166,13 +166,17 @@ server's own counter, never the heartbeat's.
latches the Node on reconnect, and a clear made while offline releases it. Until a connection's `config` has been
applied the Node runs nothing but `halt` (`nack not_ready`), so that latch is in force before any motion.
`config.allowed_commands` absent allows every kind; present (Bot always sends it, `halt` included) it is the list.
++`config.roles` are the machine's roles for the network, as its owner set them on the hub (`codes.worker`,
++`desks.host`, `ai.host`, `bot.robot`): what the machine does for the network. The Node keeps them and applies the ones
++this machine was installed for; the local configuration always wins (a role never installs anything and never turns on
++a feature the owner turned off here), and a role the Node does not know is ignored.
`estop` `latched:false` is the owner's clear: it clears the remote latch only, never the local kill switch.
### Device → server
| type | fields |
|---------------|-----------------------------------------------------------------------------------------------------|
--| `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` |
++| `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`, `roles` [ids] (the roles the machine applies) |
| `telemetry` | `battery` (0..1), `voltage` (V), `rssi`, `sensors` {…}; Node extra: `events` [{`name`, `driver`, `ts`, `fields`}] |
| `ack` | `id`; Node extra: `latency_ms` (receipt → plugin ack); `data` on a computer command (its answer, ≤ 512 KB) |
| `nack` | `id`, `fault_code`; Node extra: `message` |
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex 4741b47..e3e225d 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -96,12 +96,15 @@ type Node struct {
voiceOn bool
callWord string
voiceAskedBy map[string]string
-- telemMS int
-- jpeg *video.JPEGSource
-- jpegFrom string
-- videoSrc string
-- lastTelem time.Time
-- started time.Time
++ // roles are what the machine does for the network, as the owner set them on the hub (plan L34): the ids of the
++ // catalogue (codes.worker, desks.host, ai.host, bot.robot). They arrive in the config frame at check-in.
++ roles []string
++ telemMS int
++ jpeg *video.JPEGSource
++ jpegFrom string
++ videoSrc string
++ lastTelem time.Time
++ started time.Time
// configured is set by the first config on a connection; until then only halt runs, so a latch the owner set
// while the link was down is applied before any motion.
@@@ -230,20 +233,9 @@ func New(opt Options) (*Node, error) {
}
}
if opt.Config.Worker.Enabled {
-- wcfg := opt.Config.Worker
-- // No job may see the Node's config, credential, state or control socket (they all lie in these).
-- wcfg.NodeDirs = []string{opt.Paths.ConfigDir, opt.Paths.StateDir}
-- w := worker.New(wcfg, n.sendJobFrame, opt.Log)
-- if err := w.Probe(); err != nil {
-- opt.Log.Error("worker off: jobs cannot run isolated here, every job is refused", "err", err)
-- } else {
-- n.worker = w
-- w.SetStopped(stopFault(n.latch.State()))
-- n.jobs.admit = w.Admit
-- // Advertise a class only when an entry implements it, and only here, after the sandbox probe passed.
-- n.jobs.classes = w.Classes()
-- opt.Log.Info("worker on", "functions", len(opt.Config.Worker.Functions))
-- }
++ // The worker starts when the local configuration has it; the hub's role codes.worker asks for it again at
++ // check-in (startWorker), which never turns on what the local configuration turned off.
++ n.startWorker("the local configuration")
}
n.latch.OnChange(n.onLatch)
return n, nil
@@@ -422,8 +414,112 @@ func (n *Node) applyConfig(c protocol.Config) {
"allowed", c.AllowedCommands, "estop_latched", c.EstopLatched)
n.mu.Lock()
n.voiceOn, n.callWord = c.Voice, c.CallWord
++ n.roles = append([]string(nil), c.Roles...)
n.mu.Unlock()
n.tellVoice(nil)
++ n.applyRoles(c.Roles)
++}
++
++// The machine roles of the network's hub (plan L34; Contracts 0.257.0 network.machine-roles@1): what the machine does
++// for the network, as the owner set them. The catalogue is the products' own (each declares the roles it installs).
++const (
++ RoleCodesWorker = "codes.worker" // the Codes job runner: this machine runs its owner's coding jobs
++ RoleDesksHost = "desks.host" // the desk sandbox: this machine hosts desks for agents
++ RoleAIHost = "ai.host" // the local model server, as OpenVibe Node's ai plugin
++ RoleBotRobot = "bot.robot" // the robot drivers the profile asks for
++)
++
++// applyRoles turns on the features the roles name that this machine has.
++//
++// The local configuration always wins (ADR-045): a role never installs anything, never starts a plugin the machine
++// was not installed for, and never turns on a feature the owner turned off here. What it does is start what the
++// machine was installed for and has not started yet (the job worker), and say plainly which role it could not
++// honour, so the owner sees it on the machine as well as on the hub.
++func (n *Node) applyRoles(roles []string) {
++ if len(roles) == 0 {
++ return
++ }
++ for _, role := range roles {
++ switch role {
++ case RoleCodesWorker:
++ if n.startWorker("the role codes.worker") {
++ n.log.Info("role applied: the job worker runs", "role", role)
++ }
++ case RoleDesksHost:
++ if n.hasDeskHost() {
++ n.log.Info("role applied: this machine hosts desks", "role", role)
++ } else {
++ n.log.Info("role not applied: this machine has no desk host installed", "role", role)
++ }
++ case RoleAIHost:
++ if n.hasAIHost() {
++ n.log.Info("role applied: this machine serves its own models", "role", role)
++ } else {
++ n.log.Info("role not applied: this machine has no model server installed", "role", role)
++ }
++ case RoleBotRobot:
++ // The robot drivers are the machine's own job: the profile asked for them at install time.
++ n.log.Info("role applied: this machine drives its robot", "role", role)
++ default:
++ n.log.Info("role not known here: ignored", "role", role)
++ }
++ }
++}
++
++// startWorker starts the job worker when the local configuration has it and it is not running yet. It is the work the
++// boot does; the role codes.worker asks for it again at check-in, so a machine whose worker was configured but not
++// running starts it. → whether it is running afterwards.
++func (n *Node) startWorker(why string) bool {
++ n.mu.Lock()
++ running := n.worker != nil
++ n.mu.Unlock()
++ if running {
++ return true
++ }
++ if !n.opt.Config.Worker.Enabled {
++ n.log.Info("worker off: the local configuration has it off, and the local configuration wins", "why", why)
++ return false
++ }
++ wcfg := n.opt.Config.Worker
++ // No job may see the Node's config, credential, state or control socket (they all lie in these).
++ wcfg.NodeDirs = []string{n.opt.Paths.ConfigDir, n.opt.Paths.StateDir}
++ w := worker.New(wcfg, n.sendJobFrame, n.log)
++ if err := w.Probe(); err != nil {
++ n.log.Error("worker off: jobs cannot run isolated here, every job is refused", "err", err)
++ return false
++ }
++ n.mu.Lock()
++ n.worker = w
++ n.mu.Unlock()
++ w.SetStopped(stopFault(n.latch.State()))
++ n.jobs.admit = w.Admit
++ // Advertise a class only when an entry implements it, and only here, after the sandbox probe passed.
++ n.jobs.classes = w.Classes()
++ n.log.Info("worker on", "functions", len(wcfg.Functions), "why", why)
++ return true
++}
++
++// hasDeskHost says whether this machine was installed as a desk host: the computer plugin with a desk of its own.
++func (n *Node) hasDeskHost() bool {
++ for _, p := range n.opt.Config.Plugins {
++ if p.Name != "computer" && p.Name != "computer.py" {
++ continue
++ }
++ if d, ok := p.Config["desk"].(map[string]any); ok && len(d) > 0 {
++ return true
++ }
++ }
++ return false
++}
++
++// hasAIHost says whether this machine was installed with the ai plugin (the local model server).
++func (n *Node) hasAIHost() bool {
++ for _, p := range n.opt.Config.Plugins {
++ if p.Name == "ai" || p.Name == "ai.py" {
++ return true
++ }
++ }
++ return false
}
// tellVoice gives the voice plugins (or just p) the owner's voice setting: they listen for the call word only while on.
@@@ -1105,6 +1201,9 @@ type Status struct {
Telemetry protocol.Telemetry `json:"telemetry"`
// Jobs are the worker's jobs not ended yet; empty when the worker is off.
Jobs []worker.JobInfo `json:"jobs"`
++ // Roles are what the machine does for the network, as the owner set them on the hub (plan L34): the ids of the
++ // catalogue. Empty when the owner set none.
++ Roles []string `json:"roles,omitempty"`
}
type LinkStatus struct {
@@@ -1142,6 +1241,7 @@ func (n *Node) Status() Status {
s.Limits = n.limits
s.Video.Source = n.videoSrc
s.Telemetry = mergeTelemetry(n.telemetry, nil)
++ s.Roles = append([]string(nil), n.roles...)
n.mu.Unlock()
if n.link != nil {
st := n.link.Stats()
ddiff --git a/internal/node/roles_test.go b/internal/node/roles_test.go
nnew file mode 100644
iindex 0000000..882fec5
---- /dev/null
++++ b/internal/node/roles_test.go
@@@ -0,0 +1,135 @@
++//go:build !windows
++
++package node
++
++import (
++ "bytes"
++ "context"
++ "io"
++ "log/slog"
++ "os"
++ "strings"
++ "sync"
++ "testing"
++ "time"
++
++ "github.com/OpenVibers/OpenVibe.Node/internal/config"
++ "github.com/OpenVibers/OpenVibe.Node/internal/fakebot"
++ "github.com/OpenVibers/OpenVibe.Node/internal/link"
++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// The machine roles of the network's hub (plan L34): the Node reads them in the config frame at its check-in, keeps
++// them, and applies the ones this machine has. A role it does not know is ignored, never fatal, and the local
++// configuration always wins (ADR-045).
++func TestRolesAtCheckIn(t *testing.T) {
++ home, err := os.MkdirTemp("", "ovn")
++ if err != nil {
++ t.Fatal(err)
++ }
++ t.Cleanup(func() { os.RemoveAll(home) })
++ bot := fakebot.New()
++ t.Cleanup(bot.Close)
++ bot.Config = protocol.Config{HeartbeatMS: 100,
++ Limits: protocol.Limits{MaxSpeed: f64(0.5), MaxTurn: f64(0.4), MaxCommandMS: 1000},
++ AllowedCommands: append([]string(nil), protocol.Kinds...),
++ Roles: []string{"codes.worker", "desks.host", "nope.nope"}}
++ bot.AddCode("TEST2345")
++ creds, err := link.Pair(context.Background(), nil, bot.URL(), protocol.PairRequest{Code: "TEST2345", AgentVersion: "test",
++ DeviceKind: "onboard"})
++ if err != nil {
++ t.Fatal(err)
++ }
++ logs := &syncBuffer{}
++ cfg := config.Default()
++ cfg.Plugins = nil
++ cfg.Video.Source = "off"
++ n, err := New(Options{Config: cfg, Paths: config.HomePaths(home), Creds: creds, Version: "test", HeartbeatInterval: 100 * time.Millisecond,
++ LinkBackoffMin: 50 * time.Millisecond, LatchPoll: 50 * time.Millisecond,
++ Log: slog.New(slog.NewTextHandler(logs, nil))})
++ if err != nil {
++ t.Fatal(err)
++ }
++ ctx, cancel := context.WithCancel(context.Background())
++ done := make(chan error, 1)
++ go func() { done <- n.Run(ctx) }()
++ t.Cleanup(func() {
++ cancel()
++ select {
++ case <-done:
++ case <-time.After(10 * time.Second):
++ t.Error("the node did not stop")
++ }
++ })
++ if _, err := bot.NextConn(10 * time.Second); err != nil {
++ t.Fatal(err)
++ }
++ // The roles are in the machine's status as soon as the config was applied.
++ deadline := time.Now().Add(10 * time.Second)
++ for {
++ roles := n.Status().Roles
++ if len(roles) == 3 {
++ break
++ }
++ if time.Now().After(deadline) {
++ t.Fatalf("the roles never arrived: %v", roles)
++ }
++ time.Sleep(10 * time.Millisecond)
++ }
++ got := strings.Join(n.Status().Roles, ",")
++ if got != "codes.worker,desks.host,nope.nope" {
++ t.Fatalf("the roles are %q", got)
++ }
++ // The role the machine does not know is ignored, and the worker role does not turn on what the local
++ // configuration turned off.
++ text := logs.String()
++ if !strings.Contains(text, "role not known here") {
++ t.Errorf("an unknown role was not ignored: %s", text)
++ }
++ if !strings.Contains(text, "local configuration wins") {
++ t.Errorf("the local configuration did not win over the role: %s", text)
++ }
++ if !strings.Contains(text, "no desk host installed") {
++ t.Errorf("the desk host role was not said to be unapplied: %s", text)
++ }
++}
++
++// hasDeskHost and hasAIHost: the features a machine was installed for, which the roles name.
++func TestRoleFeatures(t *testing.T) {
++ cfg := config.Default()
++ cfg.Plugins = []config.PluginConfig{
++ {Name: "computer", Config: map[string]any{"desk": map[string]any{"max_desks": 2}}},
++ {Name: "ai"},
++ }
++ n := &Node{opt: Options{Config: cfg}}
++ if !n.hasDeskHost() {
++ t.Error("the computer plugin with a desk is a desk host")
++ }
++ if !n.hasAIHost() {
++ t.Error("the ai plugin serves the machine's own models")
++ }
++ plain := &Node{opt: Options{Config: config.Default()}}
++ if plain.hasDeskHost() || plain.hasAIHost() {
++ t.Error("a machine with neither plugin has neither feature")
++ }
++}
++
++// syncBuffer collects the log lines of a test.
++type syncBuffer struct {
++ mu sync.Mutex
++ buf bytes.Buffer
++}
++
++func (b *syncBuffer) Write(p []byte) (int, error) {
++ b.mu.Lock()
++ defer b.mu.Unlock()
++ return b.buf.Write(p)
++}
++
++func (b *syncBuffer) String() string {
++ b.mu.Lock()
++ defer b.mu.Unlock()
++ return b.buf.String()
++}
++
++var _ io.Writer = (*syncBuffer)(nil)
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex 722ce12..19fea9c 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -323,6 +323,11 @@ type Config struct {
// CallWord only while it is true.
Voice bool `json:"voice,omitempty"`
CallWord string `json:"call_word,omitempty"`
++ // Roles are what the machine does for the network, as the owner set them on the hub (plan L34; the catalogue is
++ // Contracts 0.257.0's network.machine-roles@1): codes.worker, desks.host, ai.host, bot.robot. The Node reads them
++ // at its check-in and applies the ones this machine has; a role it does not know, or a feature it was not
++ // installed for, is left alone. The local configuration always wins (ADR-045).
++ Roles []string `json:"roles,omitempty"`
}
// VoiceAsk is the owner's question after the machine's call word (device → server).