Core: node orchestrator, local control socket, end-to-end tests

ddiff --git a/go.mod b/go.mod
iindex 75168f3..309fb3d 100644
---- a/go.mod
++++ b/go.mod
@@@ -4,6 +4,7 @@ go 1.24.7
  
  require (
  	github.com/gorilla/websocket v1.5.3
++	github.com/pion/ice/v4 v4.4.4
  	github.com/pion/interceptor v0.1.49
  	github.com/pion/rtcp v1.2.18
  	github.com/pion/webrtc/v4 v4.2.22
@@@ -13,7 +14,6 @@ require (
  	github.com/google/uuid v1.6.0 // indirect
  	github.com/pion/datachannel v1.6.3 // indirect
  	github.com/pion/dtls/v3 v3.1.9 // indirect
--	github.com/pion/ice/v4 v4.4.4 // indirect
  	github.com/pion/logging v0.2.4 // indirect
  	github.com/pion/mdns/v2 v2.2.1 // indirect
  	github.com/pion/randutil v0.1.0 // indirect
ddiff --git a/internal/localctl/localctl.go b/internal/localctl/localctl.go
nnew file mode 100644
iindex 0000000..33d7ce6
---- /dev/null
++++ b/internal/localctl/localctl.go
@@@ -0,0 +1,100 @@
++// Package localctl is the running Node's local control socket: `openvibe-node stop`, `resume`, `status` and
++// `plugins` talk to it. It is a Unix domain socket (Windows 10+ supports them too) in the state directory with mode
++// 0600, so only the account running the Node and root can use it. One JSON request per connection, one JSON answer.
++package localctl
++
++import (
++	"context"
++	"encoding/json"
++	"errors"
++	"net"
++	"os"
++	"path/filepath"
++	"time"
++)
++
++// Commands.
++const (
++	CmdStatus  = "status"
++	CmdStop    = "stop"
++	CmdResume  = "resume"
++	CmdPlugins = "plugins"
++)
++
++type Request struct {
++	Cmd string `json:"cmd"`
++}
++
++type Response struct {
++	OK    bool            `json:"ok"`
++	Error string          `json:"error,omitempty"`
++	Data  json.RawMessage `json:"data,omitempty"`
++}
++
++// Handler answers one request.
++type Handler func(Request) Response
++
++// Serve listens on path until ctx ends.
++func Serve(ctx context.Context, path string, h Handler) error {
++	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
++		return err
++	}
++	// A socket left by a crashed Node: remove it if nobody answers.
++	if _, err := os.Stat(path); err == nil {
++		if c, err := net.DialTimeout("unix", path, 200*time.Millisecond); err == nil {
++			c.Close()
++			return errors.New("another openvibe-node is already running (its control socket answers)")
++		}
++		_ = os.Remove(path)
++	}
++	ln, err := net.Listen("unix", path)
++	if err != nil {
++		return err
++	}
++	_ = os.Chmod(path, 0o600)
++	go func() {
++		<-ctx.Done()
++		ln.Close()
++	}()
++	defer os.Remove(path)
++	for {
++		c, err := ln.Accept()
++		if err != nil {
++			if ctx.Err() != nil {
++				return nil
++			}
++			continue
++		}
++		go func(c net.Conn) {
++			defer c.Close()
++			_ = c.SetDeadline(time.Now().Add(5 * time.Second))
++			var req Request
++			if err := json.NewDecoder(c).Decode(&req); err != nil {
++				_ = json.NewEncoder(c).Encode(Response{Error: "bad request"})
++				return
++			}
++			_ = json.NewEncoder(c).Encode(h(req))
++		}(c)
++	}
++}
++
++// Call sends one request to a running Node.
++func Call(path string, req Request, timeout time.Duration) (Response, error) {
++	var resp Response
++	c, err := net.DialTimeout("unix", path, timeout)
++	if err != nil {
++		return resp, err
++	}
++	defer c.Close()
++	_ = c.SetDeadline(time.Now().Add(timeout))
++	if err := json.NewEncoder(c).Encode(req); err != nil {
++		return resp, err
++	}
++	if err := json.NewDecoder(c).Decode(&resp); err != nil {
++		return resp, err
++	}
++	if !resp.OK && resp.Error != "" {
++		return resp, errors.New(resp.Error)
++	}
++	return resp, nil
++}
ddiff --git a/internal/localctl/localctl_test.go b/internal/localctl/localctl_test.go
nnew file mode 100644
iindex 0000000..07c61d6
---- /dev/null
++++ b/internal/localctl/localctl_test.go
@@@ -0,0 +1,59 @@
++package localctl
++
++import (
++	"context"
++	"os"
++	"path/filepath"
++	"runtime"
++	"testing"
++	"time"
++)
++
++func TestServeCall(t *testing.T) {
++	dir, err := os.MkdirTemp("", "ovn") // short path: unix socket paths are limited to ~100 bytes
++	if err != nil {
++		t.Fatal(err)
++	}
++	defer os.RemoveAll(dir)
++	path := filepath.Join(dir, "n.sock")
++	ctx, cancel := context.WithCancel(context.Background())
++	done := make(chan error, 1)
++	go func() {
++		done <- Serve(ctx, path, func(r Request) Response {
++			if r.Cmd == CmdStop {
++				return Response{OK: true, Data: []byte(`"stopped"`)}
++			}
++			return Response{Error: "unknown"}
++		})
++	}()
++	var resp Response
++	deadline := time.Now().Add(2 * time.Second)
++	for {
++		resp, err = Call(path, Request{Cmd: CmdStop}, time.Second)
++		if err == nil || time.Now().After(deadline) {
++			break
++		}
++		time.Sleep(10 * time.Millisecond)
++	}
++	if err != nil || string(resp.Data) != `"stopped"` {
++		t.Fatalf("%v %+v", err, resp)
++	}
++	if runtime.GOOS != "windows" {
++		st, _ := os.Stat(path)
++		if st.Mode().Perm() != 0o600 {
++			t.Fatalf("socket mode %o", st.Mode().Perm())
++		}
++	}
++	if _, err := Call(path, Request{Cmd: "nope"}, time.Second); err == nil {
++		t.Fatal("unknown accepted")
++	}
++	// A second server refuses to start.
++	if err := Serve(context.Background(), path, nil); err == nil {
++		t.Fatal("second server started")
++	}
++	cancel()
++	<-done
++	if _, err := os.Stat(path); err == nil {
++		t.Fatal("socket left behind")
++	}
++}
ddiff --git a/internal/node/node.go b/internal/node/node.go
nnew file mode 100644
iindex 0000000..2f64c6d
---- /dev/null
++++ b/internal/node/node.go
@@@ -0,0 +1,693 @@
++// Package node ties the Node together: the control link, the plugins, the safety rules, video and the local control
++// socket. Every path that should stop the motors ends in Manager.StopAll or EstopAll:
++//
++//	link lost (disconnect, deadman)       → StopAll
++//	server estop                          → latch (persisted) → EstopAll
++//	`openvibe-node stop` (socket or file) → latch (persisted) → EstopAll
++//	core shutdown                         → StopAll, then each plugin's stdin is closed (plugins stop on EOF)
++//	plugin crash                          → restarted stopped, estop re-sent if latched
++//	core dies without cleanup             → plugins stop on EOF or after 1 s without a heartbeat
++package node
++
++import (
++	"context"
++	"encoding/json"
++	"errors"
++	"fmt"
++	"log/slog"
++	"os"
++	"os/exec"
++	"runtime"
++	"sort"
++	"strings"
++	"sync"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/config"
++	"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
++	"github.com/OpenVibers/OpenVibe.Node/internal/link"
++	"github.com/OpenVibers/OpenVibe.Node/internal/localctl"
++	"github.com/OpenVibers/OpenVibe.Node/internal/plugins"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++	"github.com/OpenVibers/OpenVibe.Node/internal/safety"
++	"github.com/OpenVibers/OpenVibe.Node/internal/video"
++)
++
++// Options configure a Node.
++type Options struct {
++	Config  *config.Config
++	Paths   config.Paths
++	Creds   *credentials.Credentials
++	Log     *slog.Logger
++	Version string
++
++	// For tests.
++	HeartbeatInterval time.Duration
++	LinkBackoffMin    time.Duration
++	VideoLoopback     bool
++	LatchPoll         time.Duration
++}
++
++// Node is a running device agent.
++type Node struct {
++	opt   Options
++	log   *slog.Logger
++	latch *safety.Latch
++	dedup *safety.Dedup
++	mgr   *plugins.Manager
++	link  *link.Link
++	pub   *video.Publisher
++
++	ctx context.Context
++
++	mu        sync.Mutex
++	limits    safety.Limits
++	allowed   map[string]bool
++	telemetry map[string]map[string]any // per plugin, latest
++	events    []protocol.Event
++	telemMS   int
++	jpeg      *video.JPEGSource
++	jpegFrom  string
++	videoSrc  string
++	lastTelem time.Time
++	started   time.Time
++}
++
++// New builds a Node. Run starts it.
++func New(opt Options) (*Node, error) {
++	if opt.Log == nil {
++		opt.Log = slog.Default()
++	}
++	if opt.LatchPoll == 0 {
++		opt.LatchPoll = 250 * time.Millisecond
++	}
++	latch, err := safety.OpenLatch(opt.Paths.LatchFile())
++	if err != nil {
++		opt.Log.Error("latch", "err", err)
++	}
++	n := &Node{opt: opt, log: opt.Log, latch: latch, dedup: safety.NewDedup(4096, 10*time.Minute),
++		telemetry: map[string]map[string]any{}, telemMS: 500}
++	n.limits = safety.Merge(protocol.Limits{}, opt.Config.Limits.MaxSpeed, opt.Config.Limits.MaxTurn, opt.Config.Limits.MaxCommandMS)
++
++	var specs []plugins.Spec
++	for _, p := range opt.Config.Plugins {
++		if p.Disabled {
++			continue
++		}
++		var env []string
++		for k, v := range p.Env {
++			env = append(env, k+"="+v)
++		}
++		sort.Strings(env)
++		env = append(env, "OPENVIBE_NODE_STATE="+opt.Paths.StateDir)
++		cfg := map[string]any{}
++		for k, v := range p.Config {
++			cfg[k] = v
++		}
++		if _, ok := cfg["frame_dir"]; !ok {
++			cfg["frame_dir"] = opt.Paths.FramesDir() + string(os.PathSeparator) + p.Name
++		}
++		specs = append(specs, plugins.Spec{Name: p.Name, Argv: opt.Config.PluginCommand(p, opt.Paths), Env: env, Dir: p.Dir, Config: cfg})
++	}
++	n.mgr = plugins.NewManager(specs, opt.Log)
++
++	if opt.Creds != nil {
++		u := opt.Creds.DeviceURL
++		if u == "" {
++			server := opt.Creds.Server
++			if server == "" {
++				server = opt.Config.Server
++			}
++			if u, err = link.DeviceURL(server); err != nil {
++				return nil, err
++			}
++		}
++		n.link = link.New(link.Options{URL: u, Credential: opt.Creds.Credential, UserAgent: "openvibe-node/" + opt.Version,
++			Log: opt.Log, HeartbeatInterval: opt.HeartbeatInterval, BackoffMin: opt.LinkBackoffMin}, n)
++		if opt.Creds.WHIPURL != "" && opt.Config.Video.Source != "off" {
++			n.pub = video.NewPublisher(video.Options{WHIPURL: opt.Creds.WHIPURL, PublishKey: opt.Creds.PublishKey,
++				ICEServers: opt.Creds.ICEServers, Log: opt.Log, IncludeLoopback: opt.VideoLoopback})
++		}
++	}
++	n.latch.OnChange(n.onLatch)
++	return n, nil
++}
++
++// Run runs until ctx ends, then stops every actuator and waits for the plugins to exit.
++func (n *Node) Run(ctx context.Context) error {
++	n.ctx = ctx
++	n.started = time.Now()
++	ctlErr := make(chan error, 1)
++	go func() { ctlErr <- localctl.Serve(ctx, n.opt.Paths.SocketPath(), n.control) }()
++	select {
++	case err := <-ctlErr:
++		if err != nil {
++			return err
++		}
++	case <-time.After(50 * time.Millisecond):
++	}
++
++	pctx, cancelPlugins := context.WithCancel(context.Background())
++	n.mgr.Start(pctx)
++	if n.latch.State().Stopped() {
++		n.log.Warn("starting with the stop latched", "remote", n.latch.State().Remote, "local", n.latch.State().Local)
++		n.mgr.EstopAll()
++	}
++
++	var wg sync.WaitGroup
++	wg.Add(3)
++	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.link != nil {
++		wg.Add(1)
++		go func() { defer wg.Done(); n.link.Run(ctx) }()
++	} else {
++		n.log.Warn("not paired: running local only (run `openvibe-node pair <CODE>`)")
++	}
++	<-ctx.Done()
++	n.log.Info("shutting down: stopping every actuator")
++	n.mgr.StopAll()
++	cancelPlugins()
++	n.mgr.Wait()
++	wg.Wait()
++	return nil
++}
++
++// ---- link.Handler ----
++
++func (n *Node) Connected() {
++	n.sendStatus()
++	n.sendEstopState()
++}
++
++func (n *Node) Disconnected(err error) {
++	n.log.Warn("link lost: stopping every actuator", "err", err)
++	n.mgr.StopAll()
++}
++
++func (n *Node) Frame(f protocol.Frame) {
++	switch m := f.Msg.(type) {
++	case protocol.Hello:
++		n.log.Info("server hello", "device", m.DeviceID)
++	case protocol.Config:
++		n.applyConfig(m)
++	case protocol.Command:
++		n.command(m, time.Now())
++	case protocol.Estop:
++		n.log.Warn("e-stop from the server", "reason", m.Reason, "by", m.By)
++		n.mgr.EstopAll() // at once, before the disk write
++		if _, err := n.latch.SetRemote(m.Reason); err != nil {
++			n.log.Error("could not persist the e-stop (it is held in memory)", "err", err)
++		}
++		n.sendEstopState()
++	case protocol.EstopClear:
++		n.log.Warn("e-stop cleared by the server", "by", m.By)
++		if _, err := n.latch.ClearRemote(); err != nil {
++			n.log.Error("could not persist the e-stop clear", "err", err)
++		}
++		n.sendEstopState()
++	}
++}
++
++func (n *Node) applyConfig(c protocol.Config) {
++	l := safety.Merge(c.Limits, n.opt.Config.Limits.MaxSpeed, n.opt.Config.Limits.MaxTurn, n.opt.Config.Limits.MaxCommandMS)
++	n.mu.Lock()
++	n.limits = l
++	n.allowed = nil
++	if len(c.Allowed) > 0 {
++		n.allowed = map[string]bool{protocol.KindHalt: true}
++		for _, k := range c.Allowed {
++			n.allowed[k] = true
++		}
++	}
++	if c.TelemetryMS >= 500 {
++		n.telemMS = c.TelemetryMS
++	}
++	n.mu.Unlock()
++	if n.link != nil {
++		n.link.SetTiming(c.HeartbeatMS, c.DeadmanMS)
++	}
++	n.log.Info("config", "max_speed", l.MaxSpeed, "max_turn", l.MaxTurn, "max_command_ms", l.MaxCommandMS)
++}
++
++// command validates and routes one command. The synchronous part (checks, clamping, queueing to the plugin) keeps
++// commands in arrival order; the reply is awaited on its own goroutine.
++func (n *Node) command(c protocol.Command, received time.Time) {
++	if c.ID == "" {
++		n.log.Warn("command without id ignored", "kind", c.Kind)
++		return
++	}
++	entry, first := n.dedup.Begin(c.ID)
++	if !first {
++		go func() {
++			select {
++			case <-entry.Done():
++				n.send(entry.Result())
++			case <-time.After(10 * time.Second):
++			}
++		}()
++		return
++	}
++	nack := func(code, msg string) {
++		r := protocol.Nack{ID: c.ID, FaultCode: code, Message: msg}
++		n.dedup.Complete(entry, r)
++		n.send(r)
++	}
++	known := false
++	for _, k := range protocol.Kinds {
++		known = known || k == c.Kind
++	}
++	if !known {
++		nack(protocol.FaultUnsupported, "unknown kind "+c.Kind)
++		return
++	}
++	n.mu.Lock()
++	limits, allowed := n.limits, n.allowed
++	n.mu.Unlock()
++	if c.Kind == protocol.KindHalt {
++		n.mgr.StopAll()
++		r := protocol.Ack{ID: c.ID, LatencyMS: time.Since(received).Milliseconds()}
++		n.dedup.Complete(entry, r)
++		n.send(r)
++		return
++	}
++	if allowed != nil && !allowed[c.Kind] {
++		nack(protocol.FaultNotAllowed, "")
++		return
++	}
++	if protocol.GuardedKinds[c.Kind] {
++		st := n.latch.State()
++		if st.Local {
++			nack(protocol.FaultLocalStop, "stopped on the device with `openvibe-node stop`")
++			return
++		}
++		if st.Remote {
++			nack(protocol.FaultEstopped, st.RemoteReason)
++			return
++		}
++	}
++	value, err := limits.Clamp(c.Kind, c.Value)
++	if err != nil {
++		var fe *safety.FaultError
++		if errors.As(err, &fe) {
++			nack(fe.Code, fe.Message)
++		} else {
++			nack(protocol.FaultBadValue, err.Error())
++		}
++		return
++	}
++	var actuator string
++	if c.Kind == protocol.KindActuator {
++		var v struct {
++			Name string `json:"name"`
++		}
++		_ = json.Unmarshal(value, &v)
++		actuator = v.Name
++	}
++	p := n.mgr.Route(c.Kind, c.Target, actuator)
++	if p == nil {
++		nack(protocol.FaultUnsupported, "no driver on this device handles "+c.Kind)
++		return
++	}
++	deadline := limits.Deadline(c.DeadlineMS)
++	wait := p.Begin(n.ctx, c.ID, c.Kind, value, deadline)
++	go func() {
++		r := wait()
++		var msg protocol.Message
++		if r.OK {
++			msg = protocol.Ack{ID: c.ID, LatencyMS: time.Since(received).Milliseconds()}
++		} else {
++			msg = protocol.Nack{ID: c.ID, FaultCode: r.FaultCode, Message: r.Message}
++		}
++		n.dedup.Complete(entry, msg)
++		n.send(msg)
++	}()
++}
++
++func (n *Node) send(m protocol.Message) {
++	if n.link == nil {
++		return
++	}
++	if err := n.link.Send(m); err != nil && !errors.Is(err, link.ErrOffline) {
++		n.log.Warn("send failed", "type", m.MessageType(), "err", err)
++	}
++}
++
++// ---- latch ----
++
++func (n *Node) onLatch(st safety.LatchState) {
++	if st.Stopped() {
++		n.log.Warn("stop latched", "remote", st.Remote, "local", st.Local)
++		n.mgr.EstopAll()
++	} else {
++		n.log.Warn("stop cleared; motion allowed again")
++		n.mgr.ResumeAll()
++	}
++	n.sendEstopState()
++	n.sendStatus()
++}
++
++func (n *Node) pollLatch(ctx context.Context) {
++	t := time.NewTicker(n.opt.LatchPoll)
++	defer t.Stop()
++	for {
++		select {
++		case <-ctx.Done():
++			return
++		case <-t.C:
++			if _, err := n.latch.Reload(); err != nil {
++				n.log.Error("latch file", "err", err)
++			}
++		}
++	}
++}
++
++func (n *Node) sendEstopState() {
++	st := n.latch.State()
++	n.send(protocol.EstopState{Latched: st.Remote, LocalStop: st.Local, Reason: st.RemoteReason})
++}
++
++// ---- plugin output ----
++
++func (n *Node) pump(ctx context.Context) {
++	t := time.NewTicker(100 * time.Millisecond)
++	defer t.Stop()
++	for {
++		select {
++		case <-ctx.Done():
++			return
++		case o := <-n.mgr.Output():
++			n.output(o)
++		case <-t.C:
++			n.flushTelemetry(false)
++		}
++	}
++}
++
++func (n *Node) output(o plugins.Output) {
++	switch o.Kind {
++	case plugins.OutTelemetry:
++		n.mu.Lock()
++		n.telemetry[o.Plugin] = o.Data
++		n.mu.Unlock()
++	case plugins.OutEvent:
++		name, _ := o.Data["name"].(string)
++		fields := map[string]any{}
++		for k, v := range o.Data {
++			if k != "name" {
++				fields[k] = v
++			}
++		}
++		if len(fields) == 0 {
++			fields = nil
++		}
++		n.log.Info("plugin event", "plugin", o.Plugin, "event", name)
++		n.mu.Lock()
++		n.events = append(n.events, protocol.Event{Name: name, Driver: o.Plugin, TS: o.At.UnixMilli(), Fields: fields})
++		n.mu.Unlock()
++		n.flushTelemetry(true)
++	case plugins.OutVideo:
++		n.mu.Lock()
++		j, from := n.jpeg, n.jpegFrom
++		n.mu.Unlock()
++		if j != nil && (from == "" || from == o.Plugin) {
++			if p, ok := o.Data["path"].(string); ok {
++				j.Push(p)
++			}
++		}
++	case plugins.OutFault, plugins.OutState:
++		n.sendStatus()
++	}
++}
++
++// flushTelemetry sends at most one telemetry frame per telemetry interval (≥ 500 ms: ≤ 2 Hz). Events go out at the
++// next 100 ms tick even inside the interval, so a cliff is not held back by the rate limit of sensor data.
++func (n *Node) flushTelemetry(eventsOnly bool) {
++	n.mu.Lock()
++	due := time.Since(n.lastTelem) >= time.Duration(n.telemMS)*time.Millisecond
++	if !due && len(n.events) == 0 {
++		n.mu.Unlock()
++		return
++	}
++	if eventsOnly && !due && time.Since(n.lastTelem) < 100*time.Millisecond {
++		n.mu.Unlock()
++		return
++	}
++	t := protocol.Telemetry{Events: n.events}
++	n.events = nil
++	if due {
++		t = mergeTelemetry(n.telemetry, t.Events)
++		n.lastTelem = time.Now()
++	}
++	n.mu.Unlock()
++	if t.Battery == nil && len(t.Sensors) == 0 && t.RSSI == nil && len(t.Events) == 0 {
++		return
++	}
++	n.send(t)
++}
++
++func mergeTelemetry(per map[string]map[string]any, events []protocol.Event) protocol.Telemetry {
++	t := protocol.Telemetry{Events: events}
++	names := make([]string, 0, len(per))
++	for k := range per {
++		names = append(names, k)
++	}
++	sort.Strings(names)
++	for _, name := range names {
++		d := per[name]
++		if b, ok := d["battery"].(map[string]any); ok && t.Battery == nil {
++			t.Battery = &protocol.Battery{Volts: num(b["volts"]), Percent: num(b["percent"])}
++		}
++		if r := num(d["rssi"]); r != nil && t.RSSI == nil {
++			v := int(*r)
++			t.RSSI = &v
++		}
++		if s, ok := d["sensors"].(map[string]any); ok {
++			if t.Sensors == nil {
++				t.Sensors = map[string]any{}
++			}
++			for k, v := range s {
++				if _, taken := t.Sensors[k]; taken {
++					k = name + "." + k
++				}
++				t.Sensors[k] = v
++			}
++		}
++		if f, ok := d["faults"].([]any); ok && len(f) > 0 {
++			if t.Sensors == nil {
++				t.Sensors = map[string]any{}
++			}
++			t.Sensors[name+".faults"] = f
++		}
++	}
++	return t
++}
++
++func num(v any) *float64 {
++	if f, ok := v.(float64); ok {
++		return &f
++	}
++	return nil
++}
++
++// ---- video ----
++
++func (n *Node) startVideo(ctx context.Context) {
++	if n.pub == nil {
++		return
++	}
++	src := n.chooseSource(ctx)
++	if src == nil {
++		n.log.Info("no camera to publish")
++		return
++	}
++	n.mu.Lock()
++	n.videoSrc = src.Name()
++	n.mu.Unlock()
++	n.log.Info("publishing video", "source", src.Name())
++	n.pub.Run(ctx, src)
++}
++
++// chooseSource applies video.source. For "auto" it waits (up to 20 s) for the plugins to describe their cameras.
++func (n *Node) chooseSource(ctx context.Context) video.Source {
++	vc := n.opt.Config.Video
++	fps := vc.FPS
++	switch vc.Source {
++	case "test":
++		return &video.TestPattern{W: vc.Width, H: vc.Height, FPS: fps}
++	case "command":
++		return &video.CommandSource{Argv: vc.Command, Log: n.log}
++	}
++	deadline := time.Now().Add(20 * time.Second)
++	for {
++		all := true
++		for _, info := range n.mgr.Infos() {
++			if info.Describe == nil {
++				all = false
++				continue
++			}
++			if vc.Source == "plugin" && vc.Plugin != "" && info.Name != vc.Plugin {
++				continue
++			}
++			cam, _ := info.Describe.Capabilities["camera"].(map[string]any)
++			if cam == nil {
++				continue
++			}
++			if argv := strSlice(cam["h264_command"]); len(argv) > 0 && vc.Source == "auto" {
++				if _, err := exec.LookPath(argv[0]); err == nil {
++					return &video.CommandSource{Argv: argv, Log: n.log}
++				}
++				n.log.Warn("camera program not found", "plugin", info.Name, "program", argv[0])
++			}
++			if j, _ := cam["jpeg"].(bool); j {
++				src := &video.JPEGSource{FFmpeg: vc.FFmpeg, FPS: fps, Log: n.log}
++				n.mu.Lock()
++				n.jpeg, n.jpegFrom = src, info.Name
++				n.mu.Unlock()
++				return src
++			}
++			if s, _ := cam["source"].(string); s == "test_pattern" {
++				return &video.TestPattern{W: vc.Width, H: vc.Height, FPS: fps}
++			}
++		}
++		if all || time.Now().After(deadline) {
++			return nil
++		}
++		select {
++		case <-ctx.Done():
++			return nil
++		case <-time.After(100 * time.Millisecond):
++		}
++	}
++}
++
++func strSlice(v any) []string {
++	a, ok := v.([]any)
++	if !ok {
++		return nil
++	}
++	var out []string
++	for _, x := range a {
++		if s, ok := x.(string); ok {
++			out = append(out, s)
++		}
++	}
++	return out
++}
++
++// ---- status and local control ----
++
++// Status is what `openvibe-node status` prints. It never contains the credential or the publish key.
++type Status struct {
++	Version   string             `json:"version"`
++	DeviceID  string             `json:"device_id,omitempty"`
++	Server    string             `json:"server,omitempty"`
++	Paired    bool               `json:"paired"`
++	Uptime    string             `json:"uptime"`
++	Link      LinkStatus         `json:"link"`
++	Latch     safety.LatchState  `json:"latch"`
++	Limits    safety.Limits      `json:"limits"`
++	Plugins   []PluginStatus     `json:"plugins"`
++	Video     VideoStatus        `json:"video"`
++	ConfigDir string             `json:"config_dir"`
++	StateDir  string             `json:"state_dir"`
++	Telemetry protocol.Telemetry `json:"telemetry"`
++}
++
++type LinkStatus struct {
++	Connected  bool   `json:"connected"`
++	RTTMS      int64  `json:"rtt_ms,omitempty"`
++	Reconnects int64  `json:"reconnects"`
++	LastError  string `json:"last_error,omitempty"`
++}
++
++type PluginStatus struct {
++	Name         string          `json:"name"`
++	State        string          `json:"state"`
++	Driver       string          `json:"driver,omitempty"`
++	Version      string          `json:"version,omitempty"`
++	Restarts     int             `json:"restarts"`
++	PID          int             `json:"pid,omitempty"`
++	Fault        *protocol.Fault `json:"fault,omitempty"`
++	Capabilities map[string]any  `json:"capabilities,omitempty"`
++}
++
++type VideoStatus struct {
++	State     string `json:"state"`
++	Source    string `json:"source,omitempty"`
++	Frames    int64  `json:"frames"`
++	LastError string `json:"last_error,omitempty"`
++}
++
++func (n *Node) Status() Status {
++	s := Status{Version: n.opt.Version, Latch: n.latch.State(), ConfigDir: n.opt.Paths.ConfigDir, StateDir: n.opt.Paths.StateDir,
++		Uptime: time.Since(n.started).Round(time.Second).String()}
++	if n.opt.Creds != nil {
++		s.Paired, s.DeviceID, s.Server = true, n.opt.Creds.DeviceID, n.opt.Creds.Server
++	}
++	n.mu.Lock()
++	s.Limits = n.limits
++	s.Video.Source = n.videoSrc
++	s.Telemetry = mergeTelemetry(n.telemetry, nil)
++	n.mu.Unlock()
++	if n.link != nil {
++		st := n.link.Stats()
++		s.Link = LinkStatus{Connected: st.Connected, RTTMS: st.RTT.Milliseconds(), Reconnects: st.Reconnects, LastError: st.LastError}
++	}
++	for _, i := range n.mgr.Infos() {
++		ps := PluginStatus{Name: i.Name, State: i.State, Restarts: i.Restarts, PID: i.PID, Fault: i.Fault}
++		if i.Describe != nil {
++			ps.Driver, ps.Version, ps.Capabilities = i.Describe.Driver, i.Describe.Version, i.Describe.Capabilities
++		}
++		s.Plugins = append(s.Plugins, ps)
++	}
++	s.Video.State = video.StateOff
++	if n.pub != nil {
++		s.Video.State, s.Video.Frames, s.Video.LastError = n.pub.State(), n.pub.Frames(), n.pub.LastError()
++	}
++	return s
++}
++
++func (n *Node) sendStatus() {
++	st := n.latch.State()
++	s := protocol.Status{AgentVersion: n.opt.Version, DeviceKind: n.opt.Config.DeviceKind, OS: runtime.GOOS, Arch: runtime.GOARCH,
++		EstopLatched: st.Remote, LocalStop: st.Local, Drivers: []protocol.DriverStatus{}, Faults: []protocol.Fault{}}
++	for _, i := range n.mgr.Infos() {
++		d := protocol.DriverStatus{Name: i.Name, State: i.State}
++		if i.Describe != nil {
++			d.Driver, d.Version, d.Capabilities = i.Describe.Driver, i.Describe.Version, i.Describe.Capabilities
++		}
++		s.Drivers = append(s.Drivers, d)
++		if i.Fault != nil {
++			s.Faults = append(s.Faults, *i.Fault)
++		}
++	}
++	if n.pub != nil {
++		s.Video = n.pub.State()
++	}
++	n.send(s)
++}
++
++func (n *Node) control(r localctl.Request) localctl.Response {
++	switch r.Cmd {
++	case localctl.CmdStop:
++		n.mgr.EstopAll() // at once, then persist
++		if _, err := n.latch.SetLocal(); err != nil {
++			return localctl.Response{OK: true, Error: "stopped, but the latch could not be saved: " + err.Error()}
++		}
++		return localctl.Response{OK: true}
++	case localctl.CmdResume:
++		if _, err := n.latch.Resume(); err != nil {
++			return localctl.Response{Error: err.Error()}
++		}
++		return localctl.Response{OK: true}
++	case localctl.CmdStatus, localctl.CmdPlugins:
++		b, err := json.Marshal(n.Status())
++		if err != nil {
++			return localctl.Response{Error: err.Error()}
++		}
++		return localctl.Response{OK: true, Data: b}
++	}
++	return localctl.Response{Error: fmt.Sprintf("unknown command %q", strings.TrimSpace(r.Cmd))}
++}
ddiff --git a/internal/node/node_test.go b/internal/node/node_test.go
nnew file mode 100644
iindex 0000000..a2aa9e1
---- /dev/null
++++ b/internal/node/node_test.go
@@@ -0,0 +1,458 @@
++package node
++
++import (
++	"bufio"
++	"context"
++	"encoding/json"
++	"io"
++	"log/slog"
++	"os"
++	"os/exec"
++	"path/filepath"
++	"runtime"
++	"strings"
++	"syscall"
++	"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/localctl"
++	"github.com/OpenVibers/OpenVibe.Node/internal/plugins"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++	"github.com/OpenVibers/OpenVibe.Node/internal/safety"
++	"github.com/OpenVibers/OpenVibe.Node/internal/video"
++)
++
++func init() {
++	plugins.HeartbeatInterval = 100 * time.Millisecond
++	plugins.BackoffMin = 50 * time.Millisecond
++}
++
++func f64(v float64) *float64 { return &v }
++
++type env struct {
++	t      *testing.T
++	srv    *fakebot.Server
++	whip   *video.WHIPReceiver
++	node   *Node
++	conn   *fakebot.Conn
++	record string
++	paths  config.Paths
++	cancel context.CancelFunc
++	done   chan struct{}
++}
++
++// start pairs a Node running the real Python dry-run plugin with the fake Bot server and waits until the plugin is
++// ready and the link is up.
++func start(t *testing.T) *env {
++	t.Helper()
++	if runtime.GOOS == "windows" {
++		t.Skip("end-to-end tests run on unix")
++	}
++	py, err := exec.LookPath("python3")
++	if err != nil {
++		t.Skip("python3 not installed")
++	}
++	root, _ := filepath.Abs("../..")
++	home, err := os.MkdirTemp("", "ovn") // short: the control socket path must stay under ~100 bytes
++	if err != nil {
++		t.Fatal(err)
++	}
++	t.Cleanup(func() { os.RemoveAll(home) })
++	paths := config.HomePaths(home)
++
++	srv := fakebot.New()
++	t.Cleanup(srv.Close)
++	srv.Config = protocol.Config{HeartbeatMS: 100, Limits: protocol.Limits{MaxSpeed: f64(0.5), MaxTurn: f64(0.4), MaxCommandMS: 1000}}
++	srv.AddCode("TEST2345")
++	creds, err := link.Pair(context.Background(), nil, srv.URL(), protocol.PairRequest{Code: "TEST2345", AgentVersion: "test",
++		DeviceKind: "onboard", Drivers: []string{"dryrun"}})
++	if err != nil {
++		t.Fatal(err)
++	}
++	whip := &video.WHIPReceiver{PublishKey: creds.PublishKey.Reveal()}
++	srv.SetWHIP(whip)
++	t.Cleanup(whip.Close)
++
++	record := filepath.Join(home, "record.jsonl")
++	cfg := config.Default()
++	cfg.Python = py
++	cfg.Plugins = []config.PluginConfig{{Name: "dryrun", Config: map[string]any{"record": record, "log_commands": false},
++		Env: map[string]string{"PYTHONPATH": filepath.Join(root, "plugins/sdk") + string(os.PathListSeparator) + filepath.Join(root, "plugins/dryrun")}}}
++	cfg.Video = config.VideoConfig{Source: "auto", Width: 64, Height: 48, FPS: 10}
++
++	n, err := New(Options{Config: cfg, Paths: paths, Creds: creds, Version: "test", HeartbeatInterval: 100 * time.Millisecond,
++		LinkBackoffMin: 50 * time.Millisecond, VideoLoopback: true, LatchPoll: 50 * time.Millisecond,
++		Log: slog.New(slog.NewTextHandler(io.Discard, nil))})
++	if err != nil {
++		t.Fatal(err)
++	}
++	ctx, cancel := context.WithCancel(context.Background())
++	done := make(chan struct{})
++	go func() { n.Run(ctx); close(done) }()
++	e := &env{t: t, srv: srv, whip: whip, node: n, record: record, paths: paths, cancel: cancel, done: done}
++	t.Cleanup(e.stop)
++	e.conn = e.nextConn()
++	e.waitReady()
++	return e
++}
++
++func (e *env) stop() {
++	e.cancel()
++	select {
++	case <-e.done:
++	case <-time.After(10 * time.Second):
++		e.t.Error("node did not stop")
++	}
++}
++
++func (e *env) nextConn() *fakebot.Conn {
++	e.t.Helper()
++	c, err := e.srv.NextConn(10 * time.Second)
++	if err != nil {
++		e.t.Fatal(err)
++	}
++	return c
++}
++
++func (e *env) waitReady() {
++	e.t.Helper()
++	_, err := e.conn.Expect(protocol.TypeStatus, 20*time.Second, func(m protocol.Message) bool {
++		s := m.(protocol.Status)
++		return len(s.Drivers) == 1 && s.Drivers[0].State == plugins.StateReady
++	})
++	if err != nil {
++		e.t.Fatal(err)
++	}
++}
++
++var seq int
++
++func (e *env) cmd(kind, value string, deadline int) protocol.Message {
++	e.t.Helper()
++	seq++
++	id := "cmd-" + strings.ReplaceAll(e.t.Name(), "/", "-") + "-" + string(rune('a'+seq%26)) + time.Now().Format("150405.000000")
++	return e.cmdID(id, kind, value, deadline)
++}
++
++func (e *env) cmdID(id, kind, value string, deadline int) protocol.Message {
++	e.t.Helper()
++	if err := e.conn.Send(protocol.Command{ID: id, Kind: kind, Value: json.RawMessage(value), DeadlineMS: deadline, Operator: "usr_1", Role: "operator"}); err != nil {
++		e.t.Fatal(err)
++	}
++	r, err := e.conn.Reply(id, 5*time.Second)
++	if err != nil {
++		e.t.Fatal(err)
++	}
++	return r
++}
++
++type rec struct {
++	T          float64         `json:"t"`
++	What       string          `json:"what"`
++	ID         string          `json:"id"`
++	Kind       string          `json:"kind"`
++	Value      json.RawMessage `json:"value"`
++	DeadlineMS int             `json:"deadline_ms"`
++	WasMoving  bool            `json:"was_moving"`
++}
++
++func (e *env) records() []rec {
++	f, err := os.Open(e.record)
++	if err != nil {
++		return nil
++	}
++	defer f.Close()
++	var out []rec
++	sc := bufio.NewScanner(f)
++	for sc.Scan() {
++		var r rec
++		if json.Unmarshal(sc.Bytes(), &r) == nil {
++			out = append(out, r)
++		}
++	}
++	return out
++}
++
++// waitStop waits until the plugin records a stop after the given number of records, and returns how long it took.
++func (e *env) waitStop(after int, within time.Duration) time.Duration {
++	e.t.Helper()
++	start := time.Now()
++	for time.Since(start) < within {
++		r := e.records()
++		for i := after; i < len(r); i++ {
++			if r[i].What == "stop" && r[i].WasMoving {
++				return time.Since(start)
++			}
++		}
++		time.Sleep(10 * time.Millisecond)
++	}
++	e.t.Fatalf("no stop within %s; records: %+v", within, e.records()[after:])
++	return 0
++}
++
++func isAck(m protocol.Message) bool { _, ok := m.(protocol.Ack); return ok }
++
++func nackCode(m protocol.Message) string {
++	if n, ok := m.(protocol.Nack); ok {
++		return n.FaultCode
++	}
++	return ""
++}
++
++func TestPairConnectAndStatus(t *testing.T) {
++	e := start(t)
++	reqs := e.srv.PairRequests()
++	if len(reqs) != 1 || reqs[0].Code != "TEST2345" {
++		t.Fatalf("%+v", reqs)
++	}
++	var sawEstopState bool
++	for _, f := range e.conn.Frames() {
++		if f.Type == protocol.TypeEstopState {
++			sawEstopState = true
++			if f.Msg.(protocol.EstopState).Latched {
++				t.Fatal("latched on a fresh device")
++			}
++		}
++	}
++	if !sawEstopState {
++		t.Fatal("no estop_state after connect")
++	}
++	if e.srv.CredentialSeenInURL() {
++		t.Fatal("credential in URL")
++	}
++	resp, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdStatus}, 2*time.Second)
++	if err != nil {
++		t.Fatal(err)
++	}
++	var st Status
++	json.Unmarshal(resp.Data, &st)
++	if !st.Paired || !st.Link.Connected || len(st.Plugins) != 1 || st.Plugins[0].Driver != "dryrun" || st.Limits.MaxSpeed != 0.5 {
++		t.Fatalf("%+v", st)
++	}
++	// The status (and so `openvibe-node status`) never shows the credential or publish key.
++	if strings.Contains(string(resp.Data), e.node.opt.Creds.Credential.Reveal()) || strings.Contains(string(resp.Data), e.node.opt.Creds.PublishKey.Reveal()) {
++		t.Fatal("secret in status")
++	}
++}
++
++func TestCommandsClampedAndIdempotent(t *testing.T) {
++	e := start(t)
++	id := "dup-1"
++	if r := e.cmdID(id, "drive", `{"throttle":2,"steer":-1}`, 300); !isAck(r) {
++		t.Fatalf("%+v", r)
++	}
++	if r := e.cmdID(id, "drive", `{"throttle":2,"steer":-1}`, 300); !isAck(r) {
++		t.Fatalf("duplicate: %+v", r)
++	}
++	var drives []rec
++	for _, r := range e.records() {
++		if r.What == "command" && r.ID == id {
++			drives = append(drives, r)
++		}
++	}
++	if len(drives) != 1 {
++		t.Fatalf("executed %d times", len(drives))
++	}
++	var v map[string]float64
++	json.Unmarshal(drives[0].Value, &v)
++	if v["throttle"] != 0.5 || v["steer"] != -0.4 {
++		t.Fatalf("not clamped to the owner limits: %s", drives[0].Value)
++	}
++	if drives[0].DeadlineMS != 300 {
++		t.Fatalf("deadline %d", drives[0].DeadlineMS)
++	}
++	// Deadlines are capped at max_command_ms.
++	e.cmd("drive", `{"throttle":0.1}`, 60000)
++	r := e.records()
++	if last := r[len(r)-1]; last.DeadlineMS != 1000 {
++		t.Fatalf("deadline not capped: %d", last.DeadlineMS)
++	}
++	if c := nackCode(e.cmd("drive", `{"throttle":"fast"}`, 300)); c != protocol.FaultBadValue {
++		t.Fatalf("%q", c)
++	}
++	if c := nackCode(e.cmd("teleport", `{}`, 300)); c != protocol.FaultUnsupported {
++		t.Fatalf("%q", c)
++	}
++	if r := e.cmd("say", `{"text":"hello"}`, 300); !isAck(r) {
++		t.Fatalf("%+v", r)
++	}
++}
++
++func TestDeadlineStops(t *testing.T) {
++	e := start(t)
++	n := len(e.records())
++	e.cmd("drive", `{"throttle":0.3}`, 300)
++	d := e.waitStop(n, 2*time.Second)
++	if d < 200*time.Millisecond {
++		t.Fatalf("stopped after %s, before the deadline", d)
++	}
++}
++
++func TestLinkLossStops(t *testing.T) {
++	e := start(t)
++	n := len(e.records())
++	e.cmd("drive", `{"throttle":0.3}`, 1000)
++	e.conn.Close()
++	if d := e.waitStop(n, 2*time.Second); d > 500*time.Millisecond {
++		t.Fatalf("link loss stop took %s", d)
++	}
++	// It reconnects, and nothing is replayed.
++	e.conn = e.nextConn()
++	e.waitReady()
++}
++
++func TestMissedHeartbeatsStop(t *testing.T) {
++	e := start(t)
++	n := len(e.records())
++	e.cmd("drive", `{"throttle":0.3}`, 1000)
++	e.conn.Mute()
++	if d := e.waitStop(n, 2*time.Second); d > 600*time.Millisecond {
++		t.Fatalf("deadman stop took %s", d)
++	}
++}
++
++func TestEstopLatchedUntilOwnerClears(t *testing.T) {
++	e := start(t)
++	n := len(e.records())
++	e.cmd("drive", `{"throttle":0.3}`, 1000)
++	e.conn.Send(protocol.Estop{Reason: "owner", By: "usr_owner"})
++	e.waitStop(n, time.Second)
++	if _, err := e.conn.Expect(protocol.TypeEstopState, 2*time.Second, func(m protocol.Message) bool { return m.(protocol.EstopState).Latched }); err != nil {
++		t.Fatal(err)
++	}
++	if c := nackCode(e.cmd("drive", `{"throttle":0.3}`, 300)); c != protocol.FaultEstopped {
++		t.Fatalf("%q", c)
++	}
++	if !isAck(e.cmd("halt", `{}`, 300)) {
++		t.Fatal("halt refused")
++	}
++	// Persisted: a restarted Node comes back latched.
++	l, _ := safety.OpenLatch(e.paths.LatchFile())
++	if !l.State().Remote {
++		t.Fatal("e-stop not persisted")
++	}
++	// Survives a reconnect.
++	e.conn.Close()
++	e.conn = e.nextConn()
++	if _, err := e.conn.Expect(protocol.TypeEstopState, 5*time.Second, func(m protocol.Message) bool { return m.(protocol.EstopState).Latched }); err != nil {
++		t.Fatal(err)
++	}
++	e.conn.Send(protocol.EstopClear{By: "usr_owner"})
++	if _, err := e.conn.Expect(protocol.TypeEstopState, 2*time.Second, func(m protocol.Message) bool { return !m.(protocol.EstopState).Latched }); err != nil {
++		t.Fatal(err)
++	}
++	if r := e.cmd("drive", `{"throttle":0.3}`, 300); !isAck(r) {
++		t.Fatalf("after clear: %+v", r)
++	}
++}
++
++func TestLocalKillSwitch(t *testing.T) {
++	e := start(t)
++	n := len(e.records())
++	e.cmd("drive", `{"throttle":0.3}`, 1000)
++	if _, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdStop}, 2*time.Second); err != nil {
++		t.Fatal(err)
++	}
++	e.waitStop(n, time.Second)
++	if c := nackCode(e.cmd("drive", `{"throttle":0.3}`, 300)); c != protocol.FaultLocalStop {
++		t.Fatalf("%q", c)
++	}
++	// The owner's estop_clear does not clear the local kill switch.
++	e.conn.Send(protocol.EstopClear{By: "usr_owner"})
++	time.Sleep(100 * time.Millisecond)
++	if c := nackCode(e.cmd("drive", `{"throttle":0.3}`, 300)); c != protocol.FaultLocalStop {
++		t.Fatalf("after server clear: %q", c)
++	}
++	if _, err := localctl.Call(e.paths.SocketPath(), localctl.Request{Cmd: localctl.CmdResume}, 2*time.Second); err != nil {
++		t.Fatal(err)
++	}
++	if r := e.cmd("drive", `{"throttle":0.3}`, 300); !isAck(r) {
++		t.Fatalf("after resume: %+v", r)
++	}
++}
++
++func TestKillSwitchThroughFileWhenSocketUnused(t *testing.T) {
++	e := start(t)
++	n := len(e.records())
++	e.cmd("drive", `{"throttle":0.3}`, 1000)
++	if err := safety.WriteLocalFile(e.paths.LatchFile(), true); err != nil {
++		t.Fatal(err)
++	}
++	e.waitStop(n, time.Second)
++	if c := nackCode(e.cmd("drive", `{"throttle":0.3}`, 300)); c != protocol.FaultLocalStop {
++		t.Fatalf("%q", c)
++	}
++}
++
++func TestPluginCrashRestartsStopped(t *testing.T) {
++	e := start(t)
++	pid := e.node.mgr.Get("dryrun").Info().PID
++	syscall.Kill(pid, syscall.SIGKILL)
++	deadline := time.Now().Add(10 * time.Second)
++	for {
++		i := e.node.mgr.Get("dryrun").Info()
++		if i.Restarts >= 1 && i.State == plugins.StateReady && i.PID != pid {
++			break
++		}
++		if time.Now().After(deadline) {
++			t.Fatalf("not restarted: %+v", i)
++		}
++		time.Sleep(20 * time.Millisecond)
++	}
++	r := e.records()
++	// The restarted plugin ran setup and stopped before taking commands.
++	if len(r) < 2 || r[len(r)-2].What != "setup" || r[len(r)-1].What != "stop" {
++		t.Fatalf("restart records: %+v", r)
++	}
++	if !isAck(e.cmd("drive", `{"throttle":0.1}`, 300)) {
++		t.Fatal("not driving after restart")
++	}
++}
++
++func TestShutdownStops(t *testing.T) {
++	e := start(t)
++	n := len(e.records())
++	e.cmd("drive", `{"throttle":0.3}`, 1000)
++	e.cancel()
++	<-e.done
++	found := false
++	for _, r := range e.records()[n:] {
++		found = found || (r.What == "stop" && r.WasMoving)
++	}
++	if !found {
++		t.Fatalf("no stop on shutdown: %+v", e.records()[n:])
++	}
++}
++
++func TestVideoAndTelemetry(t *testing.T) {
++	e := start(t)
++	deadline := time.Now().Add(20 * time.Second)
++	for e.whip.Packets.Load() < 10 {
++		if time.Now().After(deadline) {
++			t.Fatalf("no video: %+v", e.node.Status().Video)
++		}
++		time.Sleep(50 * time.Millisecond)
++	}
++	f, err := e.conn.Expect(protocol.TypeTelemetry, 5*time.Second, func(m protocol.Message) bool { return m.(protocol.Telemetry).Battery != nil })
++	if err != nil {
++		t.Fatal(err)
++	}
++	if tl := f.Msg.(protocol.Telemetry); tl.Battery.Percent == nil || tl.Sensors["distance_cm"] == nil {
++		t.Fatalf("%+v", tl)
++	}
++	// ≤ 2 Hz.
++	var times []int64
++	for _, fr := range e.conn.Frames() {
++		if fr.Type == protocol.TypeTelemetry && fr.Msg.(protocol.Telemetry).Battery != nil {
++			times = append(times, fr.TS)
++		}
++	}
++	for i := 1; i < len(times); i++ {
++		if times[i]-times[i-1] < 450 {
++			t.Fatalf("telemetry faster than 2 Hz: %v", times)
++		}
++	}
++}
ddiff --git a/internal/video/whip.go b/internal/video/whip.go
iindex 3e8913f..df69467 100644
---- a/internal/video/whip.go
++++ b/internal/video/whip.go
@@@ -17,6 +17,7 @@ import (
  	"sync/atomic"
  	"time"
  
++	"github.com/pion/ice/v4"
  	"github.com/pion/interceptor"
  	"github.com/pion/rtcp"
  	"github.com/pion/webrtc/v4"
@@@ -166,6 +167,7 @@ func (p *Publisher) newAPI() (*webrtc.API, error) {
  		return nil, err
  	}
  	se := webrtc.SettingEngine{}
++	se.SetICEMulticastDNSMode(ice.MulticastDNSModeDisabled)
  	if p.opt.IncludeLoopback {
  		se.SetIncludeLoopbackCandidate(true)
  	}
ddiff --git a/internal/video/whiptest.go b/internal/video/whiptest.go
iindex a437aa9..8f9299e 100644
---- a/internal/video/whiptest.go
++++ b/internal/video/whiptest.go
@@@ -8,6 +8,7 @@ import (
  	"sync/atomic"
  	"time"
  
++	"github.com/pion/ice/v4"
  	"github.com/pion/webrtc/v4"
  )
  
@@@ -57,6 +58,7 @@ func (w *WHIPReceiver) post(rw http.ResponseWriter, r *http.Request) {
  	m := &webrtc.MediaEngine{}
  	_ = m.RegisterDefaultCodecs()
  	se := webrtc.SettingEngine{}
++	se.SetICEMulticastDNSMode(ice.MulticastDNSModeDisabled)
  	se.SetIncludeLoopbackCandidate(true)
  	api := webrtc.NewAPI(webrtc.WithMediaEngine(m), webrtc.WithSettingEngine(se))
  	pc, err := api.NewPeerConnection(webrtc.Configuration{})
ddiff --git a/plugins/dryrun/openvibe_dryrun/__init__.py b/plugins/dryrun/openvibe_dryrun/__init__.py
iindex 4197bc0..1fa018a 100644
---- a/plugins/dryrun/openvibe_dryrun/__init__.py
++++ b/plugins/dryrun/openvibe_dryrun/__init__.py
@@@ -23,6 +23,7 @@ class DryRun(Plugin):
          self.battery = 100.0
          self.started = time.monotonic()
          self.log_commands = True
++        self.record_path = None
  
      def describe(self, config):
          return {
@@@ -36,8 +37,17 @@ class DryRun(Plugin):
              "camera": {"source": "test_pattern"},
          }
  
++    def record(self, what, **fields):
++        """Append one JSON line to config["record"] (if set): what the pretend robot was told, for end-to-end tests."""
++        if not self.record_path:
++            return
++        with open(self.record_path, "a") as f:
++            f.write(json.dumps(dict(t=round(time.time(), 3), what=what, **fields), sort_keys=True) + "\n")
++
      def setup(self, config):
          self.log_commands = bool(config.get("log_commands", True))
++        self.record_path = config.get("record")
++        self.record("setup")
          self.log("dryrun: ready (no hardware)")
  
      def handle(self, cmd):
@@@ -65,10 +75,12 @@ class DryRun(Plugin):
              raise Fault("unsupported", "kind %s" % cmd.kind)
          if self.log_commands:
              self.log("dryrun: %s %s (deadline %d ms)", cmd.kind, json.dumps(v, sort_keys=True), cmd.deadline_ms)
++        self.record("command", id=cmd.id, kind=cmd.kind, value=v, deadline_ms=cmd.deadline_ms)
  
      def stop(self):
          if self.state["throttle"] or self.state["steer"]:
              self.log("dryrun: stop")
++        self.record("stop", was_moving=bool(self.state["throttle"] or self.state["steer"]))
          self.state["throttle"] = 0.0
          self.state["steer"] = 0.0
  
ddiff --git a/plugins/dryrun/tests/test_dryrun.py b/plugins/dryrun/tests/test_dryrun.py
iindex 7448719..3dc41ef 100644
---- a/plugins/dryrun/tests/test_dryrun.py
++++ b/plugins/dryrun/tests/test_dryrun.py
@@@ -72,6 +72,18 @@ class DryRunTest(unittest.TestCase):
          self.assertEqual(d["driver"], "dryrun")
          self.assertEqual(d["capabilities"]["camera"]["source"], "test_pattern")
  
++    def test_record(self):
++        import tempfile
++        path = os.path.join(tempfile.mkdtemp(), "rec.jsonl")
++        p = DryRun()
++        p.log = lambda *a: None
++        rt = Runtime(p, infile=io.StringIO(""), outfile=io.StringIO(), clock=self.clock)
++        rt.feed_line(json.dumps({"op": "hello", "config": {"record": path}}))
++        rt.feed_line(json.dumps({"op": "command", "id": "r", "kind": "drive", "value": {"throttle": 0.5}}))
++        rt.feed_line(json.dumps({"op": "estop"}))
++        whats = [json.loads(x)["what"] for x in open(path)]
++        self.assertEqual(whats, ["setup", "stop", "command", "stop"])
++
  
  if __name__ == "__main__":
      unittest.main()