openvibers/openvibe.link Commits
Node: the heartbeat carries the machine's fleet of coding agents (#86)
Changes
@@@ -146,7 +146,8 @@ Every frame is one JSON object in one text message with the envelope fields at t| `type` | string | message type |Unknown `type`s are ignored (a newer server may send more). Frames are at most 1 MiB. A `heartbeat` carries the--device's send time in `t` (Unix ms) and the last measured `rtt_ms`; the server echoes `t` back in `heartbeat_ack` as++device's send time in `t` (Unix ms), the last measured `rtt_ms` and, on a machine that runs coding agents, the `fleet`++its logins probe last read (below); the server echoes `t` back in `heartbeat_ack` as`echo` (and `t`), so the Node measures the round trip against its own clock. A `heartbeat_ack`'s envelope `seq` is theserver's own counter, never the heartbeat's.@@@ -175,7 +176,7 @@ applied the Node runs nothing but `halt` (`nack not_ready`), so that latch is in| `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` |--| `heartbeat` | `t` (send time, Unix ms), `rtt_ms` (the last measured round trip, once there is one; 0 is a real measurement) |++| `heartbeat` | `t` (send time, Unix ms), `rtt_ms` (the last measured round trip, once there is one; 0 is a real measurement); Node extra: `fleet` [{`agent`, `installed`, `signed_in`, `windows` [{`name`, `used`, `limit`, `resets_at`}]}] — what the machine's coding agents have left (Codes C9c) || `reauth` | `token`: a fresh node token for the same principal (a Network-paired machine only; refused → `error` `bot.reauth_refused`) || `estop_state` | `latched` (the device is stopped: remote e-stop or local kill switch), `by` (`device`), `at`; Node extras: `robot_id` (when `hello` named exactly one robot), `local_stop` (local kill switch), `reason` |@@@ -186,6 +187,13 @@ clears that); `latched:false` is a report only. Plugin events (`cliff`, `picked_`low_battery`, `heartbeat_lost`, …) travel in `telemetry.events` and are sent within 100 ms; sensor telemetry is sentat most every 500 ms (≤ 2 Hz, Bot's cap). There is no separate `event` message type.++`fleet` is the OpenVibe.Codes runner's own probe (`openvibe-codes logins --json`) read as the Node's own user, once a++minute and once per heartbeat when read: each agent's id, whether it is installed and signed in, and each limit++window's label, used and limit numbers and when it frees. Nothing else leaves the machine — no key, token or account++name — and a reading is bounded (16 agents, 10 windows each, short labels). Absent means the machine has no reading++(no Codes runner, or its probe has not answered yet), never "nothing installed"; a probe that fails keeps the last++reading until the next one.++### Commands| kind | value | motion |nnew file mode 100644@@@ -0,0 +1,215 @@++// Package fleet reports what this machine's coding agents have left (Codes C9c).++//++// The reading is the OpenVibe.Codes runner's own probe — `openvibe-codes logins --json` — the same one a person sees++// on openvibe.codes/agents: each CLI, whether it is installed, whether it is signed in, and the limit windows its++// login file carries with when they free. Nothing else leaves the machine: no key, no token and no account name; the++// probe reads the owner's own files read-only and reports labels, numbers and times.++//++// The prober runs the CLI as the Node's own user (worker.plan_logins are files under that user's home, so the same home++// is where the logins are), caches the last answer, and the link sends it with the heartbeat. A probe that fails keeps++// the previous reading until the next one, and says so in the log; it never invents a state.++package fleet++++import (++ "bytes"++ "context"++ "encoding/json"++ "log/slog"++ "os"++ "os/exec"++ "regexp"++ "strings"++ "sync"++ "time"++++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"++)++++// Defaults for the prober: a minute between readings (a plan window moves in hours, not seconds) and a probe that++// may take this long before it is killed.++const (++ DefaultInterval = 60 * time.Second++ DefaultTimeout = 20 * time.Second++ MaxAgents = 16++ MaxWindows = 10++)++++var (++ agentID = regexp.MustCompile(`^[a-z0-9-]{1,40}$`)++ windowMax = 40++)++++// Options configure a Prober.++type Options struct {++ Command []string // the probe: [node, /opt/openvibe-codes/…/openvibe-codes.js, logins, --json]++ Home string // HOME for the probe; empty means the Node user's home++ Interval time.Duration // default DefaultInterval++ Timeout time.Duration // default DefaultTimeout++ Log *slog.Logger++}++++// Prober runs the Codes logins probe on a timer and keeps its last answer.++type Prober struct {++ opt Options++ log *slog.Logger++++ mu sync.Mutex++ last []protocol.FleetAgent++ at time.Time++ // probed is false until the first probe finished, so a machine that has never read its logins sends no fleet++ // rather than an empty one (Bot keeps what it has; Codes does not conclude "nothing installed" from silence).++ probed bool++}++++// New builds a Prober. A nil Command makes Latest return nothing: the machine has no Codes runner, so it has no++// coding agents to report.++func New(opt Options) *Prober {++ if opt.Log == nil {++ opt.Log = slog.Default()++ }++ if opt.Interval <= 0 {++ opt.Interval = DefaultInterval++ }++ if opt.Timeout <= 0 {++ opt.Timeout = DefaultTimeout++ }++ return &Prober{opt: opt, log: opt.Log.With("component", "fleet")}++}++++// Latest is the last reading, for the link's heartbeat. It never blocks on a probe.++func (p *Prober) Latest() []protocol.FleetAgent {++ if p == nil || len(p.opt.Command) == 0 {++ return nil++ }++ p.mu.Lock()++ defer p.mu.Unlock()++ if !p.probed {++ return nil++ }++ out := make([]protocol.FleetAgent, len(p.last))++ for i, a := range p.last {++ out[i] = a++ if len(a.Windows) > 0 {++ out[i].Windows = append([]protocol.FleetWindow(nil), a.Windows...)++ }++ }++ return out++}++++// At is when the last reading was taken (the zero time before the first one).++func (p *Prober) At() time.Time {++ if p == nil {++ return time.Time{}++ }++ p.mu.Lock()++ defer p.mu.Unlock()++ return p.at++}++++// Run probes now and then every Interval until ctx ends.++func (p *Prober) Run(ctx context.Context) {++ if p == nil || len(p.opt.Command) == 0 {++ return++ }++ p.probe(ctx)++ t := time.NewTicker(p.opt.Interval)++ defer t.Stop()++ for {++ select {++ case <-ctx.Done():++ return++ case <-t.C:++ p.probe(ctx)++ }++ }++}++++// probe runs the CLI once and replaces the reading on success.++func (p *Prober) probe(parent context.Context) {++ ctx, cancel := context.WithTimeout(parent, p.opt.Timeout)++ defer cancel()++ cmd := exec.CommandContext(ctx, p.opt.Command[0], p.opt.Command[1:]...)++ home := p.opt.Home++ if home == "" {++ home, _ = os.UserHomeDir()++ }++ cmd.Env = append(os.Environ(), "HOME="+home)++ cmd.Stderr = nil++ var out bytes.Buffer++ cmd.Stdout = &out++ if err := cmd.Run(); err != nil {++ p.log.Warn("logins probe failed; the last reading stands", "err", err)++ return++ }++ agents, err := Parse(out.Bytes())++ if err != nil {++ p.log.Warn("logins probe answered something that is not a fleet; the last reading stands", "err", err)++ return++ }++ p.mu.Lock()++ p.last, p.at, p.probed = agents, time.Now(), true++ p.mu.Unlock()++}++++// Parse reads the OpenVibe.Codes runner's `logins --json` answer into the fleet the heartbeat carries. Only the four++// things a person may see are kept, and everything is bounded: an agent id shaped like an id, at most MaxAgents++// agents, at most MaxWindows windows each, a short label and reset time, and two finite numbers.++func Parse(raw []byte) ([]protocol.FleetAgent, error) {++ var rows []struct {++ ID string `json:"id"`++ Installed bool `json:"installed"`++ SignedIn bool `json:"signed_in"`++ Windows []struct {++ Name string `json:"name"`++ Used *float64 `json:"used"`++ Limit *float64 `json:"limit"`++ ResetsAt string `json:"resets_at"`++ } `json:"windows"`++ }++ if err := json.Unmarshal(raw, &rows); err != nil {++ return nil, err++ }++ out := make([]protocol.FleetAgent, 0, len(rows))++ seen := map[string]bool{}++ for _, r := range rows {++ if !agentID.MatchString(r.ID) || seen[r.ID] {++ continue++ }++ seen[r.ID] = true++ a := protocol.FleetAgent{Agent: r.ID, Installed: r.Installed, SignedIn: r.SignedIn}++ for _, w := range r.Windows {++ if len(a.Windows) >= MaxWindows {++ break++ }++ name := strings.TrimSpace(w.Name)++ if name == "" {++ name = "window"++ }++ if len(name) > windowMax {++ name = name[:windowMax]++ }++ resets := strings.TrimSpace(w.ResetsAt)++ if len(resets) > windowMax {++ resets = resets[:windowMax]++ }++ a.Windows = append(a.Windows, protocol.FleetWindow{Name: name, Used: finite(w.Used), Limit: finite(w.Limit), ResetsAt: resets})++ }++ out = append(out, a)++ if len(out) >= MaxAgents {++ break++ }++ }++ return out, nil++}++++// finite drops a number that would not survive JSON (NaN, ±Inf), so the frame is always encodable.++func finite(v *float64) *float64 {++ if v == nil {++ return nil++ }++ f := *v++ if f != f || f > 1e18 || f < -1e18 {++ return nil++ }++ return &f++}nnew file mode 100644@@@ -0,0 +1,152 @@++package fleet++++import (++ "context"++ "fmt"++ "io"++ "log/slog"++ "os"++ "path/filepath"++ "testing"++ "time"++)++++func quiet() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }++++func TestParseKeepsOnlyWhatAPersonMaySee(t *testing.T) {++ raw := []byte(`[++ {"id":"codex","installed":true,"signed_in":true,"version":"1.2.3","name":"Codex","api_key":"sk-secret",++ "windows":[{"name":"week","used":100,"limit":100,"resets_at":"2026-11-04T00:00:00Z","account":"[email protected]"}]},++ {"id":"opencode","installed":true,"signed_in":false,"windows":[]},++ {"id":"Bad ID!","installed":true,"signed_in":true},++ {"id":"deepseek","installed":false,"signed_in":false}++ ]`)++ got, err := Parse(raw)++ if err != nil {++ t.Fatal(err)++ }++ if len(got) != 3 {++ t.Fatalf("agents: %+v", got)++ }++ if got[0].Agent != "codex" || !got[0].Installed || !got[0].SignedIn {++ t.Fatalf("codex: %+v", got[0])++ }++ if len(got[0].Windows) != 1 || got[0].Windows[0].Name != "week" || got[0].Windows[0].Used == nil || *got[0].Windows[0].Used != 100 ||++ got[0].Windows[0].ResetsAt != "2026-11-04T00:00:00Z" {++ t.Fatalf("window: %+v", got[0].Windows)++ }++ if got[1].Agent != "opencode" || got[1].SignedIn {++ t.Fatalf("opencode: %+v", got[1])++ }++ if got[2].Agent != "deepseek" || got[2].Installed {++ t.Fatalf("deepseek: %+v", got[2])++ }++}++++func TestParseIsBounded(t *testing.T) {++ var rows []byte++ rows = append(rows, '[')++ for i := 0; i < 40; i++ {++ if i > 0 {++ rows = append(rows, ',')++ }++ rows = append(rows, []byte(`{"id":"agent-`+"0123456789"[i%10:i%10+1]+`","windows":[{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"}]}`)...)++ }++ rows = append(rows, ']')++ got, err := Parse(rows)++ if err != nil {++ t.Fatal(err)++ }++ if len(got) > MaxAgents {++ t.Fatalf("agents: %d", len(got))++ }++ for _, a := range got {++ if len(a.Windows) > MaxWindows {++ t.Fatalf("%s windows: %d", a.Agent, len(a.Windows))++ }++ }++}++++// The probe runs a command; the tests run the test binary itself with a mode in the environment, the way the Go++// standard library does it, so the test says the same thing on every platform (a shell script is no Windows).++const helperEnv = "CODES_FLEET_HELPER"++const helperCountEnv = "CODES_FLEET_HELPER_COUNT"++++// TestFleetHelperProcess is the probe the tests below run: it is not a test of its own.++func TestFleetHelperProcess(t *testing.T) {++ mode := os.Getenv(helperEnv)++ if mode == "" {++ return++ }++ switch mode {++ case "ok":++ fmt.Fprint(os.Stdout, `[{"id":"codex","installed":true,"signed_in":true,"windows":[{"name":"week","used":10,"limit":100,"resets_at":"2026-11-04T00:00:00Z"}]}]`)++ case "empty":++ fmt.Fprint(os.Stdout, `[]`)++ case "bad":++ fmt.Fprint(os.Stdout, "not json")++ case "count":++ if path := os.Getenv(helperCountEnv); path != "" {++ if f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644); err == nil {++ fmt.Fprintln(f, "x")++ f.Close()++ }++ }++ fmt.Fprint(os.Stdout, `[]`)++ }++ os.Exit(0)++}++++// The probe command: this test binary, asked for the helper.++func helperCommand() []string { return []string{os.Args[0], "-test.run=TestFleetHelperProcess"} }++++// withHelper sets the mode the helper answers with, for the time of one test.++func withHelper(t *testing.T, mode string) {++ t.Helper()++ t.Setenv(helperEnv, mode)++}++++func TestProberReadsTheRunnerAndKeepsItsLastAnswer(t *testing.T) {++ withHelper(t, "ok")++ p := New(Options{Command: helperCommand(), Log: quiet()})++ if got := p.Latest(); got != nil {++ t.Fatalf("nothing is reported before the first probe: %+v", got)++ }++ p.probe(context.Background())++ got := p.Latest()++ if len(got) != 1 || got[0].Agent != "codex" || len(got[0].Windows) != 1 {++ t.Fatalf("probe: %+v", got)++ }++ if p.At().IsZero() {++ t.Fatal("the reading has a time")++ }++++ // A probe that answers something else keeps the last reading rather than reporting an empty machine.++ withHelper(t, "bad")++ p.probe(context.Background())++ if got := p.Latest(); len(got) != 1 {++ t.Fatalf("the last reading stands: %+v", got)++ }++++ // A machine that never answered reports nothing.++ withHelper(t, "bad")++ broken := New(Options{Command: helperCommand(), Log: quiet()})++ broken.probe(context.Background())++ if got := broken.Latest(); got != nil {++ t.Fatalf("a machine that never answered reports nothing: %+v", got)++ }++}++++func TestRunProbesAgainOnItsInterval(t *testing.T) {++ count := filepath.Join(t.TempDir(), "count")++ t.Setenv(helperCountEnv, count)++ withHelper(t, "count")++ p := New(Options{Command: helperCommand(), Interval: 20 * time.Millisecond, Log: quiet()})++ ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)++ defer cancel()++ p.Run(ctx)++ data, _ := os.ReadFile(count)++ if len(data) < 2 {++ t.Fatalf("the prober ran once: %q", data)++ }++}@@@ -67,6 +67,9 @@ type Options struct {Deadman time.Duration // default 2 × heartbeatBackoffMin time.Duration // default 500 msBackoffMax time.Duration // default 30 s++ // Fleet, when set, is read once per heartbeat and sent with it: what this machine's coding agents have left++ // (Codes C9c). It returns the last reading the prober took, never blocks, and is bounded by its caller.++ Fleet func() []protocol.FleetAgent}// Tokens supplies node tokens; *TokenSource is the real one.@@@ -467,7 +470,11 @@ func (l *Link) session(ctx context.Context) error {}if !now.Before(next) {next = now.Add(hb)-- _ = l.Send(protocol.Heartbeat{})++ beat := protocol.Heartbeat{}++ if l.opt.Fleet != nil {++ beat.Fleet = l.opt.Fleet()++ }++ _ = l.Send(beat)}if now.Sub(time.Unix(0, lastRx.Load())) > deadman {end(fmt.Errorf("heartbeat lost: nothing from the server for %s", deadman))@@@ -55,7 +55,7 @@ func (r *recorder) Disconnected(err error) {r.gotDown <- err}--func setup(t *testing.T, hb time.Duration) (*fakebot.Server, *Link, *recorder, *lockedWriter, context.CancelFunc) {++func setup(t *testing.T, hb time.Duration, tweak ...func(*Options)) (*fakebot.Server, *Link, *recorder, *lockedWriter, context.CancelFunc) {t.Helper()srv := fakebot.New()t.Cleanup(srv.Close)@@@ -67,8 +67,12 @@ func setup(t *testing.T, hb time.Duration) (*fakebot.Server, *Link, *recorder, *}logs := &lockedWriter{&bytes.Buffer{}, &sync.Mutex{}}rec := newRecorder()-- l := New(Options{URL: u, Credential: credentials.NewSecret(cred), HeartbeatInterval: hb, BackoffMin: 20 * time.Millisecond,-- BackoffMax: 100 * time.Millisecond, Log: slog.New(slog.NewTextHandler(logs, nil))}, rec)++ opt := Options{URL: u, Credential: credentials.NewSecret(cred), HeartbeatInterval: hb, BackoffMin: 20 * time.Millisecond,++ BackoffMax: 100 * time.Millisecond, Log: slog.New(slog.NewTextHandler(logs, nil))}++ for _, f := range tweak {++ f(&opt)++ }++ l := New(opt, rec)ctx, cancel := context.WithCancel(context.Background())done := make(chan struct{})go func() { l.Run(ctx); close(done) }()@@@ -146,6 +150,65 @@ func TestConnectHelloConfigHeartbeat(t *testing.T) {}}++// The heartbeat carries what this machine's coding agents have left (Codes C9c). Options.Fleet is read once per beat,++// so a new reading is on the next one; a machine with no reading sends no fleet rather than an empty one.++func TestHeartbeatCarriesTheFleet(t *testing.T) {++ var mu sync.Mutex++ var fleet []protocol.FleetAgent++ set := func(f []protocol.FleetAgent) {++ mu.Lock()++ fleet = f++ mu.Unlock()++ }++ used, limit := 100.0, 100.0++ set([]protocol.FleetAgent{{Agent: "codex", Installed: true, SignedIn: true, Windows: []protocol.FleetWindow{++ {Name: "week", Used: &used, Limit: &limit, ResetsAt: "2026-11-04T00:00:00Z"},++ }}})++ // 100 ms heartbeats (a 200 ms deadman), as above: this test is about the beat's contents, not the deadman.++ srv, _, rec, _, _ := setup(t, 100*time.Millisecond, func(o *Options) {++ o.Fleet = func() []protocol.FleetAgent {++ mu.Lock()++ defer mu.Unlock()++ return fleet++ }++ })++ c, err := srv.NextConn(3 * time.Second)++ if err != nil {++ t.Fatal(err)++ }++ waitFrame(t, rec, protocol.TypeHello)++ waitFrame(t, rec, protocol.TypeConfig)++ beat := func(match func([]protocol.FleetAgent) bool) protocol.Heartbeat {++ t.Helper()++ f, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, func(m protocol.Message) bool {++ hb, ok := m.(protocol.Heartbeat)++ return ok && match(hb.Fleet)++ })++ if err != nil {++ t.Fatal(err)++ }++ return f.Msg.(protocol.Heartbeat)++ }++ hb := beat(func(f []protocol.FleetAgent) bool { return len(f) > 0 })++ if len(hb.Fleet) != 1 || hb.Fleet[0].Agent != "codex" || !hb.Fleet[0].Installed || !hb.Fleet[0].SignedIn ||++ len(hb.Fleet[0].Windows) != 1 || hb.Fleet[0].Windows[0].Name != "week" ||++ hb.Fleet[0].Windows[0].Used == nil || *hb.Fleet[0].Windows[0].Used != 100 ||++ hb.Fleet[0].Windows[0].ResetsAt != "2026-11-04T00:00:00Z" {++ t.Fatalf("fleet on the beat: %+v", hb.Fleet)++ }++++ // The next beat carries the next reading.++ set([]protocol.FleetAgent{{Agent: "opencode", Installed: true}})++ hb = beat(func(f []protocol.FleetAgent) bool { return len(f) == 1 && f[0].Agent == "opencode" })++ if hb.Fleet[0].SignedIn || len(hb.Fleet[0].Windows) != 0 {++ t.Fatalf("the reading that replaced it: %+v", hb.Fleet)++ }++++ // No reading (never probed, or the CLI gone) is silence, not "nothing installed".++ set(nil)++ beat(func(f []protocol.FleetAgent) bool { return f == nil })++}++// Bot's heartbeat_ack echoes the device's send time (echo/t) over its own envelope seq, so the Node measures the// round trip from that; a fresh connection starts with no measurement.func TestHeartbeatRTTReportedAfterAck(t *testing.T) {nnew file mode 100644@@@ -0,0 +1,21 @@++package node++++import "github.com/OpenVibers/OpenVibe.Node/internal/config"++++// codesProbeCommand builds the fleet probe (Codes C9c) from the worker's declared OpenVibe.Codes entry: the same++// interpreter and script the jobs run, asked for `logins --json` instead of `job`. Nil when the machine has no such++// entry — a Node that never installed the Codes runner has no coding agents to report.++func codesProbeCommand(fns []config.FunctionConfig) []string {++ for _, f := range fns {++ if f.Name != "openvibe-codes" || f.EffectiveClass() != config.ClassCode || len(f.Command) == 0 {++ continue++ }++ argv := append([]string(nil), f.Command...)++ // The installer declares the entry as `node …/openvibe-codes.js job`: the same command with another verb.++ if len(argv) > 1 && argv[len(argv)-1] == "job" {++ argv = argv[:len(argv)-1]++ }++ return append(argv, "logins", "--json")++ }++ return nil++}nnew file mode 100644@@@ -0,0 +1,30 @@++package node++++import (++ "reflect"++ "testing"++++ "github.com/OpenVibers/OpenVibe.Node/internal/config"++)++++func TestCodesProbeCommand(t *testing.T) {++ entry := config.FunctionConfig{Name: "openvibe-codes", Version: "1.0.0", Class: "code",++ Command: []string{"/usr/bin/node", "/opt/openvibe-codes/node_modules/openvibe-codes/bin/openvibe-codes.js", "job"}}++ cases := []struct {++ name string++ fns []config.FunctionConfig++ want []string++ }{++ {"the installer's entry, asked for logins", []config.FunctionConfig{entry},++ []string{"/usr/bin/node", "/opt/openvibe-codes/node_modules/openvibe-codes/bin/openvibe-codes.js", "logins", "--json"}},++ {"an entry with no verb keeps its command and gets one", []config.FunctionConfig{{Name: "openvibe-codes", Class: "code", Command: []string{"/opt/codes/bin/run"}}},++ []string{"/opt/codes/bin/run", "logins", "--json"}},++ {"no Codes runner", []config.FunctionConfig{{Name: "thumbnail", Class: "code", Command: []string{"/opt/fn/run"}}}, nil},++ {"another class is not the runner", []config.FunctionConfig{{Name: "openvibe-codes", Class: "function", Command: []string{"/opt/codes/run", "job"}}}, nil},++ }++ for _, c := range cases {++ if got := codesProbeCommand(c.fns); !reflect.DeepEqual(got, c.want) {++ t.Errorf("%s: %v, want %v", c.name, got, c.want)++ }++ }++}@@@ -27,6 +27,7 @@ import ("github.com/OpenVibers/OpenVibe.Node/internal/config""github.com/OpenVibers/OpenVibe.Node/internal/credentials"++ "github.com/OpenVibers/OpenVibe.Node/internal/fleet""github.com/OpenVibers/OpenVibe.Node/internal/link""github.com/OpenVibers/OpenVibe.Node/internal/localctl""github.com/OpenVibers/OpenVibe.Node/internal/mono"@@@ -69,9 +70,12 @@ type Node struct {jobs *jobs// worker runs function jobs; nil unless config.worker is enabled on Linux and its boot-time probe passed.worker *worker.Worker-- mgr *plugins.Manager-- link *link.Link-- pub *video.Publisher++ // fleet is what this machine's coding agents have left (Codes C9c): the Codes runner's own logins probe, sent++ // with every heartbeat. Nil when the machine has no OpenVibe.Codes runner declared.++ fleet *fleet.Prober++ mgr *plugins.Manager++ link *link.Link++ pub *video.Publisher// screen is the owner's live view of the machine's screen (screen.go), apart from a robot's camera.screen screenState@@@ -181,6 +185,11 @@ func New(opt Options) (*Node, error) {}n.mgr = plugins.NewManager(specs, opt.Log)++ // Codes C9c: the same Codes runner the worker runs, asked for its logins instead of a job. The probe runs as the++ // Node's user, the home worker.plan_logins names are under, and only its answer (labels, numbers, times) leaves.++ if cmd := codesProbeCommand(opt.Config.Worker.Functions); cmd != nil {++ n.fleet = fleet.New(fleet.Options{Command: cmd, Log: opt.Log})++ }if opt.Creds != nil {u := opt.Creds.DeviceURLif u == "" {@@@ -194,6 +203,9 @@ func New(opt Options) (*Node, error) {}lo := link.Options{URL: u, Credential: opt.Creds.Credential, UserAgent: "openvibe-node/" + opt.Version,Log: opt.Log, HeartbeatInterval: opt.HeartbeatInterval, BackoffMin: opt.LinkBackoffMin, ReauthInterval: opt.ReauthInterval}++ if n.fleet != nil {++ lo.Fleet = n.fleet.Latest++ }if opt.Creds.NetworkPaired() {// A Network-paired machine authenticates with node tokens, never a Bot credential.lo.Credential = credentials.Secret{}@@@ -274,6 +286,10 @@ func (n *Node) Run(parent context.Context) error {go func() { defer wg.Done(); n.pump(ctx) }()go func() { defer wg.Done(); n.pollLatch(ctx) }()go func() { defer wg.Done(); n.startVideo(ctx) }()++ if n.fleet != nil {++ wg.Add(1)++ go func() { defer wg.Done(); n.fleet.Run(ctx) }()++ }if n.link != nil {wg.Add(1)go func() {@@@ -654,10 +654,32 @@ type Nack struct {// Heartbeat is sent every config.heartbeat_ms. Seq mirrors the envelope seq for older servers that echo it; T is the// device's send time (unix ms), which Bot echoes back as heartbeat_ack.t/echo and the Node measures RTT from. RTTMS is// the last measured round trip, absent until one has been measured (a measured 0 ms is still sent, hence the pointer).++//++// Fleet is what this machine's coding agents report (Codes C9c): installed, signed in, and their plan windows. It is++// sent with the heartbeat — refreshed every fleet.Interval, not every second — so Bot keeps the last reading per machine++// and Codes can route a task away from an agent whose window is full. Labels, numbers and times only: never a key, a++// token or an account name (the Codes runner's own `logins` probe).type Heartbeat struct {-- Seq uint64 `json:"seq"`-- T int64 `json:"t,omitempty"`-- RTTMS *int64 `json:"rtt_ms,omitempty"`++ Seq uint64 `json:"seq"`++ T int64 `json:"t,omitempty"`++ RTTMS *int64 `json:"rtt_ms,omitempty"`++ Fleet []FleetAgent `json:"fleet,omitempty"`++}++++// FleetAgent is one coding agent on the machine, as its CLI's own login file reports it.++type FleetAgent struct {++ Agent string `json:"agent"` // the OpenVibe.Codes harness id (codex, opencode, command-code, …)++ Installed bool `json:"installed"`++ SignedIn bool `json:"signed_in"`++ Windows []FleetWindow `json:"windows,omitempty"`++}++++// FleetWindow is one limit window: a label, what it has used of its limit, and when it frees.++type FleetWindow struct {++ Name string `json:"name"`++ Used *float64 `json:"used,omitempty"`++ Limit *float64 `json:"limit,omitempty"`++ ResetsAt string `json:"resets_at,omitempty"` // RFC 3339, as the provider's login file carries it}// Reauth carries a fresh node token for the same principal; Bot closes the socket with 4002 when none arrives within@@@ -419,3 +419,36 @@ func TestComputerActsBrowser(t *testing.T) {t.Error("browser is a computer kind")}}++++// A heartbeat carries the machine's fleet (Codes C9c) without changing any existing field, and a heartbeat with no++// fleet encodes exactly as before (the field is absent).++func TestHeartbeatCarriesTheFleet(t *testing.T) {++ used, limit := 96.0, 100.0++ h := Heartbeat{Seq: 42, Fleet: []FleetAgent{++ {Agent: "codex", Installed: true, SignedIn: true, Windows: []FleetWindow{++ {Name: "week", Used: &used, Limit: &limit, ResetsAt: "2026-11-04T00:00:00Z"},++ }},++ {Agent: "opencode", Installed: true, SignedIn: false},++ }}++ b, err := Encode(42, 1738065600500, h)++ if err != nil {++ t.Fatal(err)++ }++ f, err := Decode(b)++ if err != nil {++ t.Fatal(err)++ }++ got := f.Msg.(Heartbeat)++ if got.Seq != 42 || len(got.Fleet) != 2 || got.Fleet[0].Agent != "codex" || !got.Fleet[0].SignedIn ||++ len(got.Fleet[0].Windows) != 1 || *got.Fleet[0].Windows[0].Used != 96 || got.Fleet[0].Windows[0].ResetsAt != "2026-11-04T00:00:00Z" ||++ got.Fleet[1].Agent != "opencode" || got.Fleet[1].SignedIn {++ t.Fatalf("heartbeat with fleet: %+v", got)++ }++ plain, err := Encode(7, 1, Heartbeat{Seq: 7})++ if err != nil {++ t.Fatal(err)++ }++ if strings.Contains(string(plain), `"fleet"`) {++ t.Fatalf("an empty fleet is absent, not []: %s", plain)++ }++}
@@ -146,7 +146,8 @@ Every frame is one JSON object in one text message with the envelope fields at t
@@ -146,7 +146,8 @@ Every frame is one JSON object in one text message with the envelope fields at t
| `type` | string | message type |
| `type` | string | message type |
Unknown `type`s are ignored (a newer server may send more). Frames are at most 1 MiB. A `heartbeat` carries the
Unknown `type`s are ignored (a newer server may send more). Frames are at most 1 MiB. A `heartbeat` carries the
-device's send time in `t` (Unix ms) and the last measured `rtt_ms`; the server echoes `t` back in `heartbeat_ack` as
+device's send time in `t` (Unix ms), the last measured `rtt_ms` and, on a machine that runs coding agents, the `fleet`
+its logins probe last read (below); the server echoes `t` back in `heartbeat_ack` as
`echo` (and `t`), so the Node measures the round trip against its own clock. A `heartbeat_ack`'s envelope `seq` is the
`echo` (and `t`), so the Node measures the round trip against its own clock. A `heartbeat_ack`'s envelope `seq` is the
server's own counter, never the heartbeat's.
server's own counter, never the heartbeat's.
@@ -175,7 +176,7 @@ applied the Node runs nothing but `halt` (`nack not_ready`), so that latch is in
@@ -175,7 +176,7 @@ applied the Node runs nothing but `halt` (`nack not_ready`), so that latch is in
| `telemetry` | `battery` (0..1), `voltage` (V), `rssi`, `sensors` {…}; Node extra: `events` [{`name`, `driver`, `ts`, `fields`}] |
| `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) |
| `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` |
| `nack` | `id`, `fault_code`; Node extra: `message` |
-| `heartbeat` | `t` (send time, Unix ms), `rtt_ms` (the last measured round trip, once there is one; 0 is a real measurement) |
+| `heartbeat` | `t` (send time, Unix ms), `rtt_ms` (the last measured round trip, once there is one; 0 is a real measurement); Node extra: `fleet` [{`agent`, `installed`, `signed_in`, `windows` [{`name`, `used`, `limit`, `resets_at`}]}] — what the machine's coding agents have left (Codes C9c) |
| `reauth` | `token`: a fresh node token for the same principal (a Network-paired machine only; refused → `error` `bot.reauth_refused`) |
| `reauth` | `token`: a fresh node token for the same principal (a Network-paired machine only; refused → `error` `bot.reauth_refused`) |
| `estop_state` | `latched` (the device is stopped: remote e-stop or local kill switch), `by` (`device`), `at`; Node extras: `robot_id` (when `hello` named exactly one robot), `local_stop` (local kill switch), `reason` |
| `estop_state` | `latched` (the device is stopped: remote e-stop or local kill switch), `by` (`device`), `at`; Node extras: `robot_id` (when `hello` named exactly one robot), `local_stop` (local kill switch), `reason` |
@@ -186,6 +187,13 @@ clears that); `latched:false` is a report only. Plugin events (`cliff`, `picked_
@@ -186,6 +187,13 @@ clears that); `latched:false` is a report only. Plugin events (`cliff`, `picked_
`low_battery`, `heartbeat_lost`, …) travel in `telemetry.events` and are sent within 100 ms; sensor telemetry is sent
`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.
at most every 500 ms (≤ 2 Hz, Bot's cap). There is no separate `event` message type.
+`fleet` is the OpenVibe.Codes runner's own probe (`openvibe-codes logins --json`) read as the Node's own user, once a
+minute and once per heartbeat when read: each agent's id, whether it is installed and signed in, and each limit
+window's label, used and limit numbers and when it frees. Nothing else leaves the machine — no key, token or account
+name — and a reading is bounded (16 agents, 10 windows each, short labels). Absent means the machine has no reading
+(no Codes runner, or its probe has not answered yet), never "nothing installed"; a probe that fails keeps the last
+reading until the next one.
+
### Commands
### Commands
| kind | value | motion |
| kind | value | motion |
new file mode 100644
new file mode 100644
@@ -0,0 +1,215 @@
@@ -0,0 +1,215 @@
+// Package fleet reports what this machine's coding agents have left (Codes C9c).
+//
+// The reading is the OpenVibe.Codes runner's own probe — `openvibe-codes logins --json` — the same one a person sees
+// on openvibe.codes/agents: each CLI, whether it is installed, whether it is signed in, and the limit windows its
+// login file carries with when they free. Nothing else leaves the machine: no key, no token and no account name; the
+// probe reads the owner's own files read-only and reports labels, numbers and times.
+//
+// The prober runs the CLI as the Node's own user (worker.plan_logins are files under that user's home, so the same home
+// is where the logins are), caches the last answer, and the link sends it with the heartbeat. A probe that fails keeps
+// the previous reading until the next one, and says so in the log; it never invents a state.
+package fleet
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "log/slog"
+ "os"
+ "os/exec"
+ "regexp"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
+)
+
+// Defaults for the prober: a minute between readings (a plan window moves in hours, not seconds) and a probe that
+// may take this long before it is killed.
+const (
+ DefaultInterval = 60 * time.Second
+ DefaultTimeout = 20 * time.Second
+ MaxAgents = 16
+ MaxWindows = 10
+)
+
+var (
+ agentID = regexp.MustCompile(`^[a-z0-9-]{1,40}$`)
+ windowMax = 40
+)
+
+// Options configure a Prober.
+type Options struct {
+ Command []string // the probe: [node, /opt/openvibe-codes/…/openvibe-codes.js, logins, --json]
+ Home string // HOME for the probe; empty means the Node user's home
+ Interval time.Duration // default DefaultInterval
+ Timeout time.Duration // default DefaultTimeout
+ Log *slog.Logger
+}
+
+// Prober runs the Codes logins probe on a timer and keeps its last answer.
+type Prober struct {
+ opt Options
+ log *slog.Logger
+
+ mu sync.Mutex
+ last []protocol.FleetAgent
+ at time.Time
+ // probed is false until the first probe finished, so a machine that has never read its logins sends no fleet
+ // rather than an empty one (Bot keeps what it has; Codes does not conclude "nothing installed" from silence).
+ probed bool
+}
+
+// New builds a Prober. A nil Command makes Latest return nothing: the machine has no Codes runner, so it has no
+// coding agents to report.
+func New(opt Options) *Prober {
+ if opt.Log == nil {
+ opt.Log = slog.Default()
+ }
+ if opt.Interval <= 0 {
+ opt.Interval = DefaultInterval
+ }
+ if opt.Timeout <= 0 {
+ opt.Timeout = DefaultTimeout
+ }
+ return &Prober{opt: opt, log: opt.Log.With("component", "fleet")}
+}
+
+// Latest is the last reading, for the link's heartbeat. It never blocks on a probe.
+func (p *Prober) Latest() []protocol.FleetAgent {
+ if p == nil || len(p.opt.Command) == 0 {
+ return nil
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if !p.probed {
+ return nil
+ }
+ out := make([]protocol.FleetAgent, len(p.last))
+ for i, a := range p.last {
+ out[i] = a
+ if len(a.Windows) > 0 {
+ out[i].Windows = append([]protocol.FleetWindow(nil), a.Windows...)
+ }
+ }
+ return out
+}
+
+// At is when the last reading was taken (the zero time before the first one).
+func (p *Prober) At() time.Time {
+ if p == nil {
+ return time.Time{}
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.at
+}
+
+// Run probes now and then every Interval until ctx ends.
+func (p *Prober) Run(ctx context.Context) {
+ if p == nil || len(p.opt.Command) == 0 {
+ return
+ }
+ p.probe(ctx)
+ t := time.NewTicker(p.opt.Interval)
+ defer t.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-t.C:
+ p.probe(ctx)
+ }
+ }
+}
+
+// probe runs the CLI once and replaces the reading on success.
+func (p *Prober) probe(parent context.Context) {
+ ctx, cancel := context.WithTimeout(parent, p.opt.Timeout)
+ defer cancel()
+ cmd := exec.CommandContext(ctx, p.opt.Command[0], p.opt.Command[1:]...)
+ home := p.opt.Home
+ if home == "" {
+ home, _ = os.UserHomeDir()
+ }
+ cmd.Env = append(os.Environ(), "HOME="+home)
+ cmd.Stderr = nil
+ var out bytes.Buffer
+ cmd.Stdout = &out
+ if err := cmd.Run(); err != nil {
+ p.log.Warn("logins probe failed; the last reading stands", "err", err)
+ return
+ }
+ agents, err := Parse(out.Bytes())
+ if err != nil {
+ p.log.Warn("logins probe answered something that is not a fleet; the last reading stands", "err", err)
+ return
+ }
+ p.mu.Lock()
+ p.last, p.at, p.probed = agents, time.Now(), true
+ p.mu.Unlock()
+}
+
+// Parse reads the OpenVibe.Codes runner's `logins --json` answer into the fleet the heartbeat carries. Only the four
+// things a person may see are kept, and everything is bounded: an agent id shaped like an id, at most MaxAgents
+// agents, at most MaxWindows windows each, a short label and reset time, and two finite numbers.
+func Parse(raw []byte) ([]protocol.FleetAgent, error) {
+ var rows []struct {
+ ID string `json:"id"`
+ Installed bool `json:"installed"`
+ SignedIn bool `json:"signed_in"`
+ Windows []struct {
+ Name string `json:"name"`
+ Used *float64 `json:"used"`
+ Limit *float64 `json:"limit"`
+ ResetsAt string `json:"resets_at"`
+ } `json:"windows"`
+ }
+ if err := json.Unmarshal(raw, &rows); err != nil {
+ return nil, err
+ }
+ out := make([]protocol.FleetAgent, 0, len(rows))
+ seen := map[string]bool{}
+ for _, r := range rows {
+ if !agentID.MatchString(r.ID) || seen[r.ID] {
+ continue
+ }
+ seen[r.ID] = true
+ a := protocol.FleetAgent{Agent: r.ID, Installed: r.Installed, SignedIn: r.SignedIn}
+ for _, w := range r.Windows {
+ if len(a.Windows) >= MaxWindows {
+ break
+ }
+ name := strings.TrimSpace(w.Name)
+ if name == "" {
+ name = "window"
+ }
+ if len(name) > windowMax {
+ name = name[:windowMax]
+ }
+ resets := strings.TrimSpace(w.ResetsAt)
+ if len(resets) > windowMax {
+ resets = resets[:windowMax]
+ }
+ a.Windows = append(a.Windows, protocol.FleetWindow{Name: name, Used: finite(w.Used), Limit: finite(w.Limit), ResetsAt: resets})
+ }
+ out = append(out, a)
+ if len(out) >= MaxAgents {
+ break
+ }
+ }
+ return out, nil
+}
+
+// finite drops a number that would not survive JSON (NaN, ±Inf), so the frame is always encodable.
+func finite(v *float64) *float64 {
+ if v == nil {
+ return nil
+ }
+ f := *v
+ if f != f || f > 1e18 || f < -1e18 {
+ return nil
+ }
+ return &f
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,152 @@
@@ -0,0 +1,152 @@
+package fleet
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "log/slog"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+)
+
+func quiet() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) }
+
+func TestParseKeepsOnlyWhatAPersonMaySee(t *testing.T) {
+ raw := []byte(`[
+ {"id":"codex","installed":true,"signed_in":true,"version":"1.2.3","name":"Codex","api_key":"sk-secret",
+ "windows":[{"name":"week","used":100,"limit":100,"resets_at":"2026-11-04T00:00:00Z","account":"[email protected]"}]},
+ {"id":"opencode","installed":true,"signed_in":false,"windows":[]},
+ {"id":"Bad ID!","installed":true,"signed_in":true},
+ {"id":"deepseek","installed":false,"signed_in":false}
+ ]`)
+ got, err := Parse(raw)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 3 {
+ t.Fatalf("agents: %+v", got)
+ }
+ if got[0].Agent != "codex" || !got[0].Installed || !got[0].SignedIn {
+ t.Fatalf("codex: %+v", got[0])
+ }
+ if len(got[0].Windows) != 1 || got[0].Windows[0].Name != "week" || got[0].Windows[0].Used == nil || *got[0].Windows[0].Used != 100 ||
+ got[0].Windows[0].ResetsAt != "2026-11-04T00:00:00Z" {
+ t.Fatalf("window: %+v", got[0].Windows)
+ }
+ if got[1].Agent != "opencode" || got[1].SignedIn {
+ t.Fatalf("opencode: %+v", got[1])
+ }
+ if got[2].Agent != "deepseek" || got[2].Installed {
+ t.Fatalf("deepseek: %+v", got[2])
+ }
+}
+
+func TestParseIsBounded(t *testing.T) {
+ var rows []byte
+ rows = append(rows, '[')
+ for i := 0; i < 40; i++ {
+ if i > 0 {
+ rows = append(rows, ',')
+ }
+ rows = append(rows, []byte(`{"id":"agent-`+"0123456789"[i%10:i%10+1]+`","windows":[{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"},{"name":"w"}]}`)...)
+ }
+ rows = append(rows, ']')
+ got, err := Parse(rows)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(got) > MaxAgents {
+ t.Fatalf("agents: %d", len(got))
+ }
+ for _, a := range got {
+ if len(a.Windows) > MaxWindows {
+ t.Fatalf("%s windows: %d", a.Agent, len(a.Windows))
+ }
+ }
+}
+
+// The probe runs a command; the tests run the test binary itself with a mode in the environment, the way the Go
+// standard library does it, so the test says the same thing on every platform (a shell script is no Windows).
+const helperEnv = "CODES_FLEET_HELPER"
+const helperCountEnv = "CODES_FLEET_HELPER_COUNT"
+
+// TestFleetHelperProcess is the probe the tests below run: it is not a test of its own.
+func TestFleetHelperProcess(t *testing.T) {
+ mode := os.Getenv(helperEnv)
+ if mode == "" {
+ return
+ }
+ switch mode {
+ case "ok":
+ fmt.Fprint(os.Stdout, `[{"id":"codex","installed":true,"signed_in":true,"windows":[{"name":"week","used":10,"limit":100,"resets_at":"2026-11-04T00:00:00Z"}]}]`)
+ case "empty":
+ fmt.Fprint(os.Stdout, `[]`)
+ case "bad":
+ fmt.Fprint(os.Stdout, "not json")
+ case "count":
+ if path := os.Getenv(helperCountEnv); path != "" {
+ if f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644); err == nil {
+ fmt.Fprintln(f, "x")
+ f.Close()
+ }
+ }
+ fmt.Fprint(os.Stdout, `[]`)
+ }
+ os.Exit(0)
+}
+
+// The probe command: this test binary, asked for the helper.
+func helperCommand() []string { return []string{os.Args[0], "-test.run=TestFleetHelperProcess"} }
+
+// withHelper sets the mode the helper answers with, for the time of one test.
+func withHelper(t *testing.T, mode string) {
+ t.Helper()
+ t.Setenv(helperEnv, mode)
+}
+
+func TestProberReadsTheRunnerAndKeepsItsLastAnswer(t *testing.T) {
+ withHelper(t, "ok")
+ p := New(Options{Command: helperCommand(), Log: quiet()})
+ if got := p.Latest(); got != nil {
+ t.Fatalf("nothing is reported before the first probe: %+v", got)
+ }
+ p.probe(context.Background())
+ got := p.Latest()
+ if len(got) != 1 || got[0].Agent != "codex" || len(got[0].Windows) != 1 {
+ t.Fatalf("probe: %+v", got)
+ }
+ if p.At().IsZero() {
+ t.Fatal("the reading has a time")
+ }
+
+ // A probe that answers something else keeps the last reading rather than reporting an empty machine.
+ withHelper(t, "bad")
+ p.probe(context.Background())
+ if got := p.Latest(); len(got) != 1 {
+ t.Fatalf("the last reading stands: %+v", got)
+ }
+
+ // A machine that never answered reports nothing.
+ withHelper(t, "bad")
+ broken := New(Options{Command: helperCommand(), Log: quiet()})
+ broken.probe(context.Background())
+ if got := broken.Latest(); got != nil {
+ t.Fatalf("a machine that never answered reports nothing: %+v", got)
+ }
+}
+
+func TestRunProbesAgainOnItsInterval(t *testing.T) {
+ count := filepath.Join(t.TempDir(), "count")
+ t.Setenv(helperCountEnv, count)
+ withHelper(t, "count")
+ p := New(Options{Command: helperCommand(), Interval: 20 * time.Millisecond, Log: quiet()})
+ ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+ defer cancel()
+ p.Run(ctx)
+ data, _ := os.ReadFile(count)
+ if len(data) < 2 {
+ t.Fatalf("the prober ran once: %q", data)
+ }
+}
@@ -67,6 +67,9 @@ type Options struct {
@@ -67,6 +67,9 @@ type Options struct {
Deadman time.Duration // default 2 × heartbeat
Deadman time.Duration // default 2 × heartbeat
BackoffMin time.Duration // default 500 ms
BackoffMin time.Duration // default 500 ms
BackoffMax time.Duration // default 30 s
BackoffMax time.Duration // default 30 s
+ // Fleet, when set, is read once per heartbeat and sent with it: what this machine's coding agents have left
+ // (Codes C9c). It returns the last reading the prober took, never blocks, and is bounded by its caller.
+ Fleet func() []protocol.FleetAgent
}
}
// Tokens supplies node tokens; *TokenSource is the real one.
// Tokens supplies node tokens; *TokenSource is the real one.
@@ -467,7 +470,11 @@ func (l *Link) session(ctx context.Context) error {
@@ -467,7 +470,11 @@ func (l *Link) session(ctx context.Context) error {
}
}
if !now.Before(next) {
if !now.Before(next) {
next = now.Add(hb)
next = now.Add(hb)
- _ = l.Send(protocol.Heartbeat{})
+ beat := protocol.Heartbeat{}
+ if l.opt.Fleet != nil {
+ beat.Fleet = l.opt.Fleet()
+ }
+ _ = l.Send(beat)
}
}
if now.Sub(time.Unix(0, lastRx.Load())) > deadman {
if now.Sub(time.Unix(0, lastRx.Load())) > deadman {
end(fmt.Errorf("heartbeat lost: nothing from the server for %s", deadman))
end(fmt.Errorf("heartbeat lost: nothing from the server for %s", deadman))
@@ -55,7 +55,7 @@ func (r *recorder) Disconnected(err error) {
@@ -55,7 +55,7 @@ func (r *recorder) Disconnected(err error) {
r.gotDown <- err
r.gotDown <- err
}
}
-func setup(t *testing.T, hb time.Duration) (*fakebot.Server, *Link, *recorder, *lockedWriter, context.CancelFunc) {
+func setup(t *testing.T, hb time.Duration, tweak ...func(*Options)) (*fakebot.Server, *Link, *recorder, *lockedWriter, context.CancelFunc) {
t.Helper()
t.Helper()
srv := fakebot.New()
srv := fakebot.New()
t.Cleanup(srv.Close)
t.Cleanup(srv.Close)
@@ -67,8 +67,12 @@ func setup(t *testing.T, hb time.Duration) (*fakebot.Server, *Link, *recorder, *
@@ -67,8 +67,12 @@ func setup(t *testing.T, hb time.Duration) (*fakebot.Server, *Link, *recorder, *
}
}
logs := &lockedWriter{&bytes.Buffer{}, &sync.Mutex{}}
logs := &lockedWriter{&bytes.Buffer{}, &sync.Mutex{}}
rec := newRecorder()
rec := newRecorder()
- l := New(Options{URL: u, Credential: credentials.NewSecret(cred), HeartbeatInterval: hb, BackoffMin: 20 * time.Millisecond,
- BackoffMax: 100 * time.Millisecond, Log: slog.New(slog.NewTextHandler(logs, nil))}, rec)
+ opt := Options{URL: u, Credential: credentials.NewSecret(cred), HeartbeatInterval: hb, BackoffMin: 20 * time.Millisecond,
+ BackoffMax: 100 * time.Millisecond, Log: slog.New(slog.NewTextHandler(logs, nil))}
+ for _, f := range tweak {
+ f(&opt)
+ }
+ l := New(opt, rec)
ctx, cancel := context.WithCancel(context.Background())
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
done := make(chan struct{})
go func() { l.Run(ctx); close(done) }()
go func() { l.Run(ctx); close(done) }()
@@ -146,6 +150,65 @@ func TestConnectHelloConfigHeartbeat(t *testing.T) {
@@ -146,6 +150,65 @@ func TestConnectHelloConfigHeartbeat(t *testing.T) {
}
}
}
}
+// The heartbeat carries what this machine's coding agents have left (Codes C9c). Options.Fleet is read once per beat,
+// so a new reading is on the next one; a machine with no reading sends no fleet rather than an empty one.
+func TestHeartbeatCarriesTheFleet(t *testing.T) {
+ var mu sync.Mutex
+ var fleet []protocol.FleetAgent
+ set := func(f []protocol.FleetAgent) {
+ mu.Lock()
+ fleet = f
+ mu.Unlock()
+ }
+ used, limit := 100.0, 100.0
+ set([]protocol.FleetAgent{{Agent: "codex", Installed: true, SignedIn: true, Windows: []protocol.FleetWindow{
+ {Name: "week", Used: &used, Limit: &limit, ResetsAt: "2026-11-04T00:00:00Z"},
+ }}})
+ // 100 ms heartbeats (a 200 ms deadman), as above: this test is about the beat's contents, not the deadman.
+ srv, _, rec, _, _ := setup(t, 100*time.Millisecond, func(o *Options) {
+ o.Fleet = func() []protocol.FleetAgent {
+ mu.Lock()
+ defer mu.Unlock()
+ return fleet
+ }
+ })
+ c, err := srv.NextConn(3 * time.Second)
+ if err != nil {
+ t.Fatal(err)
+ }
+ waitFrame(t, rec, protocol.TypeHello)
+ waitFrame(t, rec, protocol.TypeConfig)
+ beat := func(match func([]protocol.FleetAgent) bool) protocol.Heartbeat {
+ t.Helper()
+ f, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, func(m protocol.Message) bool {
+ hb, ok := m.(protocol.Heartbeat)
+ return ok && match(hb.Fleet)
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ return f.Msg.(protocol.Heartbeat)
+ }
+ hb := beat(func(f []protocol.FleetAgent) bool { return len(f) > 0 })
+ if len(hb.Fleet) != 1 || hb.Fleet[0].Agent != "codex" || !hb.Fleet[0].Installed || !hb.Fleet[0].SignedIn ||
+ len(hb.Fleet[0].Windows) != 1 || hb.Fleet[0].Windows[0].Name != "week" ||
+ hb.Fleet[0].Windows[0].Used == nil || *hb.Fleet[0].Windows[0].Used != 100 ||
+ hb.Fleet[0].Windows[0].ResetsAt != "2026-11-04T00:00:00Z" {
+ t.Fatalf("fleet on the beat: %+v", hb.Fleet)
+ }
+
+ // The next beat carries the next reading.
+ set([]protocol.FleetAgent{{Agent: "opencode", Installed: true}})
+ hb = beat(func(f []protocol.FleetAgent) bool { return len(f) == 1 && f[0].Agent == "opencode" })
+ if hb.Fleet[0].SignedIn || len(hb.Fleet[0].Windows) != 0 {
+ t.Fatalf("the reading that replaced it: %+v", hb.Fleet)
+ }
+
+ // No reading (never probed, or the CLI gone) is silence, not "nothing installed".
+ set(nil)
+ beat(func(f []protocol.FleetAgent) bool { return f == nil })
+}
+
// Bot's heartbeat_ack echoes the device's send time (echo/t) over its own envelope seq, so the Node measures the
// Bot's heartbeat_ack echoes the device's send time (echo/t) over its own envelope seq, so the Node measures the
// round trip from that; a fresh connection starts with no measurement.
// round trip from that; a fresh connection starts with no measurement.
func TestHeartbeatRTTReportedAfterAck(t *testing.T) {
func TestHeartbeatRTTReportedAfterAck(t *testing.T) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,21 @@
@@ -0,0 +1,21 @@
+package node
+
+import "github.com/OpenVibers/OpenVibe.Node/internal/config"
+
+// codesProbeCommand builds the fleet probe (Codes C9c) from the worker's declared OpenVibe.Codes entry: the same
+// interpreter and script the jobs run, asked for `logins --json` instead of `job`. Nil when the machine has no such
+// entry — a Node that never installed the Codes runner has no coding agents to report.
+func codesProbeCommand(fns []config.FunctionConfig) []string {
+ for _, f := range fns {
+ if f.Name != "openvibe-codes" || f.EffectiveClass() != config.ClassCode || len(f.Command) == 0 {
+ continue
+ }
+ argv := append([]string(nil), f.Command...)
+ // The installer declares the entry as `node …/openvibe-codes.js job`: the same command with another verb.
+ if len(argv) > 1 && argv[len(argv)-1] == "job" {
+ argv = argv[:len(argv)-1]
+ }
+ return append(argv, "logins", "--json")
+ }
+ return nil
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,30 @@
@@ -0,0 +1,30 @@
+package node
+
+import (
+ "reflect"
+ "testing"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/config"
+)
+
+func TestCodesProbeCommand(t *testing.T) {
+ entry := config.FunctionConfig{Name: "openvibe-codes", Version: "1.0.0", Class: "code",
+ Command: []string{"/usr/bin/node", "/opt/openvibe-codes/node_modules/openvibe-codes/bin/openvibe-codes.js", "job"}}
+ cases := []struct {
+ name string
+ fns []config.FunctionConfig
+ want []string
+ }{
+ {"the installer's entry, asked for logins", []config.FunctionConfig{entry},
+ []string{"/usr/bin/node", "/opt/openvibe-codes/node_modules/openvibe-codes/bin/openvibe-codes.js", "logins", "--json"}},
+ {"an entry with no verb keeps its command and gets one", []config.FunctionConfig{{Name: "openvibe-codes", Class: "code", Command: []string{"/opt/codes/bin/run"}}},
+ []string{"/opt/codes/bin/run", "logins", "--json"}},
+ {"no Codes runner", []config.FunctionConfig{{Name: "thumbnail", Class: "code", Command: []string{"/opt/fn/run"}}}, nil},
+ {"another class is not the runner", []config.FunctionConfig{{Name: "openvibe-codes", Class: "function", Command: []string{"/opt/codes/run", "job"}}}, nil},
+ }
+ for _, c := range cases {
+ if got := codesProbeCommand(c.fns); !reflect.DeepEqual(got, c.want) {
+ t.Errorf("%s: %v, want %v", c.name, got, c.want)
+ }
+ }
+}
@@ -27,6 +27,7 @@ import (
@@ -27,6 +27,7 @@ import (
"github.com/OpenVibers/OpenVibe.Node/internal/config"
"github.com/OpenVibers/OpenVibe.Node/internal/config"
"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
+ "github.com/OpenVibers/OpenVibe.Node/internal/fleet"
"github.com/OpenVibers/OpenVibe.Node/internal/link"
"github.com/OpenVibers/OpenVibe.Node/internal/link"
"github.com/OpenVibers/OpenVibe.Node/internal/localctl"
"github.com/OpenVibers/OpenVibe.Node/internal/localctl"
"github.com/OpenVibers/OpenVibe.Node/internal/mono"
"github.com/OpenVibers/OpenVibe.Node/internal/mono"
@@ -69,9 +70,12 @@ type Node struct {
@@ -69,9 +70,12 @@ type Node struct {
jobs *jobs
jobs *jobs
// worker runs function jobs; nil unless config.worker is enabled on Linux and its boot-time probe passed.
// worker runs function jobs; nil unless config.worker is enabled on Linux and its boot-time probe passed.
worker *worker.Worker
worker *worker.Worker
- mgr *plugins.Manager
- link *link.Link
- pub *video.Publisher
+ // fleet is what this machine's coding agents have left (Codes C9c): the Codes runner's own logins probe, sent
+ // with every heartbeat. Nil when the machine has no OpenVibe.Codes runner declared.
+ fleet *fleet.Prober
+ mgr *plugins.Manager
+ link *link.Link
+ pub *video.Publisher
// screen is the owner's live view of the machine's screen (screen.go), apart from a robot's camera.
// screen is the owner's live view of the machine's screen (screen.go), apart from a robot's camera.
screen screenState
screen screenState
@@ -181,6 +185,11 @@ func New(opt Options) (*Node, error) {
@@ -181,6 +185,11 @@ func New(opt Options) (*Node, error) {
}
}
n.mgr = plugins.NewManager(specs, opt.Log)
n.mgr = plugins.NewManager(specs, opt.Log)
+ // Codes C9c: the same Codes runner the worker runs, asked for its logins instead of a job. The probe runs as the
+ // Node's user, the home worker.plan_logins names are under, and only its answer (labels, numbers, times) leaves.
+ if cmd := codesProbeCommand(opt.Config.Worker.Functions); cmd != nil {
+ n.fleet = fleet.New(fleet.Options{Command: cmd, Log: opt.Log})
+ }
if opt.Creds != nil {
if opt.Creds != nil {
u := opt.Creds.DeviceURL
u := opt.Creds.DeviceURL
if u == "" {
if u == "" {
@@ -194,6 +203,9 @@ func New(opt Options) (*Node, error) {
@@ -194,6 +203,9 @@ func New(opt Options) (*Node, error) {
}
}
lo := link.Options{URL: u, Credential: opt.Creds.Credential, UserAgent: "openvibe-node/" + opt.Version,
lo := link.Options{URL: u, Credential: opt.Creds.Credential, UserAgent: "openvibe-node/" + opt.Version,
Log: opt.Log, HeartbeatInterval: opt.HeartbeatInterval, BackoffMin: opt.LinkBackoffMin, ReauthInterval: opt.ReauthInterval}
Log: opt.Log, HeartbeatInterval: opt.HeartbeatInterval, BackoffMin: opt.LinkBackoffMin, ReauthInterval: opt.ReauthInterval}
+ if n.fleet != nil {
+ lo.Fleet = n.fleet.Latest
+ }
if opt.Creds.NetworkPaired() {
if opt.Creds.NetworkPaired() {
// A Network-paired machine authenticates with node tokens, never a Bot credential.
// A Network-paired machine authenticates with node tokens, never a Bot credential.
lo.Credential = credentials.Secret{}
lo.Credential = credentials.Secret{}
@@ -274,6 +286,10 @@ func (n *Node) Run(parent context.Context) error {
@@ -274,6 +286,10 @@ func (n *Node) Run(parent context.Context) error {
go func() { defer wg.Done(); n.pump(ctx) }()
go func() { defer wg.Done(); n.pump(ctx) }()
go func() { defer wg.Done(); n.pollLatch(ctx) }()
go func() { defer wg.Done(); n.pollLatch(ctx) }()
go func() { defer wg.Done(); n.startVideo(ctx) }()
go func() { defer wg.Done(); n.startVideo(ctx) }()
+ if n.fleet != nil {
+ wg.Add(1)
+ go func() { defer wg.Done(); n.fleet.Run(ctx) }()
+ }
if n.link != nil {
if n.link != nil {
wg.Add(1)
wg.Add(1)
go func() {
go func() {
@@ -654,10 +654,32 @@ type Nack struct {
@@ -654,10 +654,32 @@ type Nack struct {
// Heartbeat is sent every config.heartbeat_ms. Seq mirrors the envelope seq for older servers that echo it; T is the
// Heartbeat is sent every config.heartbeat_ms. Seq mirrors the envelope seq for older servers that echo it; T is the
// device's send time (unix ms), which Bot echoes back as heartbeat_ack.t/echo and the Node measures RTT from. RTTMS is
// device's send time (unix ms), which Bot echoes back as heartbeat_ack.t/echo and the Node measures RTT from. RTTMS is
// the last measured round trip, absent until one has been measured (a measured 0 ms is still sent, hence the pointer).
// the last measured round trip, absent until one has been measured (a measured 0 ms is still sent, hence the pointer).
+//
+// Fleet is what this machine's coding agents report (Codes C9c): installed, signed in, and their plan windows. It is
+// sent with the heartbeat — refreshed every fleet.Interval, not every second — so Bot keeps the last reading per machine
+// and Codes can route a task away from an agent whose window is full. Labels, numbers and times only: never a key, a
+// token or an account name (the Codes runner's own `logins` probe).
type Heartbeat struct {
type Heartbeat struct {
- Seq uint64 `json:"seq"`
- T int64 `json:"t,omitempty"`
- RTTMS *int64 `json:"rtt_ms,omitempty"`
+ Seq uint64 `json:"seq"`
+ T int64 `json:"t,omitempty"`
+ RTTMS *int64 `json:"rtt_ms,omitempty"`
+ Fleet []FleetAgent `json:"fleet,omitempty"`
+}
+
+// FleetAgent is one coding agent on the machine, as its CLI's own login file reports it.
+type FleetAgent struct {
+ Agent string `json:"agent"` // the OpenVibe.Codes harness id (codex, opencode, command-code, …)
+ Installed bool `json:"installed"`
+ SignedIn bool `json:"signed_in"`
+ Windows []FleetWindow `json:"windows,omitempty"`
+}
+
+// FleetWindow is one limit window: a label, what it has used of its limit, and when it frees.
+type FleetWindow struct {
+ Name string `json:"name"`
+ Used *float64 `json:"used,omitempty"`
+ Limit *float64 `json:"limit,omitempty"`
+ ResetsAt string `json:"resets_at,omitempty"` // RFC 3339, as the provider's login file carries it
}
}
// Reauth carries a fresh node token for the same principal; Bot closes the socket with 4002 when none arrives within
// Reauth carries a fresh node token for the same principal; Bot closes the socket with 4002 when none arrives within
@@ -419,3 +419,36 @@ func TestComputerActsBrowser(t *testing.T) {
@@ -419,3 +419,36 @@ func TestComputerActsBrowser(t *testing.T) {
t.Error("browser is a computer kind")
t.Error("browser is a computer kind")
}
}
}
}
+
+// A heartbeat carries the machine's fleet (Codes C9c) without changing any existing field, and a heartbeat with no
+// fleet encodes exactly as before (the field is absent).
+func TestHeartbeatCarriesTheFleet(t *testing.T) {
+ used, limit := 96.0, 100.0
+ h := Heartbeat{Seq: 42, Fleet: []FleetAgent{
+ {Agent: "codex", Installed: true, SignedIn: true, Windows: []FleetWindow{
+ {Name: "week", Used: &used, Limit: &limit, ResetsAt: "2026-11-04T00:00:00Z"},
+ }},
+ {Agent: "opencode", Installed: true, SignedIn: false},
+ }}
+ b, err := Encode(42, 1738065600500, h)
+ if err != nil {
+ t.Fatal(err)
+ }
+ f, err := Decode(b)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got := f.Msg.(Heartbeat)
+ if got.Seq != 42 || len(got.Fleet) != 2 || got.Fleet[0].Agent != "codex" || !got.Fleet[0].SignedIn ||
+ len(got.Fleet[0].Windows) != 1 || *got.Fleet[0].Windows[0].Used != 96 || got.Fleet[0].Windows[0].ResetsAt != "2026-11-04T00:00:00Z" ||
+ got.Fleet[1].Agent != "opencode" || got.Fleet[1].SignedIn {
+ t.Fatalf("heartbeat with fleet: %+v", got)
+ }
+ plain, err := Encode(7, 1, Heartbeat{Seq: 7})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if strings.Contains(string(plain), `"fleet"`) {
+ t.Fatalf("an empty fleet is absent, not []: %s", plain)
+ }
+}