Core: safety latch, limits, command dedup, plugin supervisor

ddiff --git a/internal/plugins/manager.go b/internal/plugins/manager.go
nnew file mode 100644
iindex 0000000..0f516e2
---- /dev/null
++++ b/internal/plugins/manager.go
@@@ -0,0 +1,127 @@
++package plugins
++
++import (
++	"context"
++	"log/slog"
++	"sync"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// capabilityFor maps a command kind to the describe capability key that declares it.
++var capabilityFor = map[string]string{
++	protocol.KindDrive:    "drive",
++	protocol.KindPTZ:      "ptz",
++	protocol.KindActuator: "actuator",
++	protocol.KindSay:      "say",
++	protocol.KindDisplay:  "display",
++}
++
++// Manager supervises every configured plugin.
++type Manager struct {
++	log     *slog.Logger
++	plugins []*Plugin
++	out     chan Output
++	wg      sync.WaitGroup
++}
++
++func NewManager(specs []Spec, log *slog.Logger) *Manager {
++	m := &Manager{log: log, out: make(chan Output, 1024)}
++	for _, s := range specs {
++		m.plugins = append(m.plugins, newPlugin(s, log, m.out))
++	}
++	return m
++}
++
++// Output is the stream of telemetry, events, video frames, faults and state changes from every plugin.
++func (m *Manager) Output() <-chan Output { return m.out }
++
++// Start supervises every plugin until ctx ends. Wait returns once all have stopped their actuators and exited.
++func (m *Manager) Start(ctx context.Context) {
++	for _, p := range m.plugins {
++		m.wg.Add(1)
++		go func(p *Plugin) {
++			defer m.wg.Done()
++			p.run(ctx)
++		}(p)
++	}
++}
++
++func (m *Manager) Wait() { m.wg.Wait() }
++
++func (m *Manager) Plugins() []*Plugin { return m.plugins }
++
++func (m *Manager) Get(name string) *Plugin {
++	for _, p := range m.plugins {
++		if p.spec.Name == name {
++			return p
++		}
++	}
++	return nil
++}
++
++// StopAll tells every plugin to stop every actuator.
++func (m *Manager) StopAll() {
++	for _, p := range m.plugins {
++		p.Stop()
++	}
++}
++
++func (m *Manager) EstopAll() {
++	for _, p := range m.plugins {
++		p.Estop()
++	}
++}
++
++func (m *Manager) ResumeAll() {
++	for _, p := range m.plugins {
++		p.Resume()
++	}
++}
++
++// Infos snapshots every plugin.
++func (m *Manager) Infos() []Info {
++	out := make([]Info, 0, len(m.plugins))
++	for _, p := range m.plugins {
++		out = append(out, p.Info())
++	}
++	return out
++}
++
++// Route picks the plugin for a command: the named target, or the first plugin whose describe declares the kind's
++// capability (for actuators, the first that lists the actuator's name or "any"). Plugins that have not described
++// themselves yet are considered by their configured order only when targeted by name.
++func (m *Manager) Route(kind, target, actuator string) *Plugin {
++	if target != "" {
++		return m.Get(target)
++	}
++	key := capabilityFor[kind]
++	if key == "" {
++		return nil
++	}
++	var fallback *Plugin
++	for _, p := range m.plugins {
++		info := p.Info()
++		if info.Describe == nil {
++			continue
++		}
++		c, ok := info.Describe.Capabilities[key]
++		if !ok {
++			continue
++		}
++		if kind != protocol.KindActuator || actuator == "" {
++			return p
++		}
++		if names, ok := c.(map[string]any)["names"].([]any); ok {
++			for _, n := range names {
++				if n == actuator || n == "any" {
++					return p
++				}
++			}
++		}
++		if fallback == nil {
++			fallback = p
++		}
++	}
++	return fallback
++}
ddiff --git a/internal/plugins/plugin.go b/internal/plugins/plugin.go
nnew file mode 100644
iindex 0000000..54ecf42
---- /dev/null
++++ b/internal/plugins/plugin.go
@@@ -0,0 +1,556 @@
++// Package plugins runs driver plugins: separate executables speaking JSON lines over stdin/stdout (docs/plugins.md).
++//
++// The supervisor starts each plugin, sends hello, waits for describe, sends a heartbeat every HeartbeatInterval,
++// forwards commands and collects ack/nack, telemetry, events and video frames. A plugin that exits is restarted with
++// exponential backoff; it starts stopped, and it is sent `estop` again at once if the latch is set.
++package plugins
++
++import (
++	"bufio"
++	"context"
++	"encoding/json"
++	"errors"
++	"fmt"
++	"io"
++	"log/slog"
++	"os"
++	"os/exec"
++	"sync"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// Timing. Variables so tests can shorten them.
++var (
++	HeartbeatInterval = 250 * time.Millisecond
++	DescribeTimeout   = 15 * time.Second
++	StopGrace         = 2 * time.Second
++	BackoffMin        = 1 * time.Second
++	BackoffMax        = 30 * time.Second
++	StableAfter       = 60 * time.Second
++	MinReplyTimeout   = 1 * time.Second
++)
++
++// Plugin states.
++const (
++	StateStarting = "starting"
++	StateReady    = "ready"
++	StateFaulted  = "faulted" // running but reported a fault (e.g. firmware unsupported, robot unreachable)
++	StateDown     = "down"    // exited; waiting to restart
++	StateStopped  = "stopped" // supervisor finished
++)
++
++// Spec says how to start a plugin.
++type Spec struct {
++	Name   string
++	Argv   []string
++	Env    []string // extra KEY=VALUE entries
++	Dir    string
++	Config map[string]any
++}
++
++// Describe is the plugin's first message.
++type Describe struct {
++	Driver       string         `json:"driver"`
++	Version      string         `json:"version"`
++	Protocol     int            `json:"protocol"`
++	Capabilities map[string]any `json:"capabilities"`
++	MotionKinds  []string       `json:"motion_kinds"`
++}
++
++// Reply is a plugin's answer to one command.
++type Reply struct {
++	OK        bool
++	FaultCode string
++	Message   string
++}
++
++// Output kinds carried by Output.Kind.
++const (
++	OutTelemetry = "telemetry"
++	OutEvent     = "event"
++	OutVideo     = "video"
++	OutFault     = "fault"
++	OutState     = "state"
++)
++
++// Output is something a plugin said that the core forwards: telemetry, an event, a video frame, a fault or a state
++// change. Data is the plugin's message without "op".
++type Output struct {
++	Plugin string
++	Kind   string
++	Data   map[string]any
++	At     time.Time
++}
++
++// Plugin is one supervised driver process.
++type Plugin struct {
++	spec Spec
++	log  *slog.Logger
++	out  chan<- Output
++
++	mu       sync.Mutex
++	state    string
++	desc     *Describe
++	fault    *protocol.Fault
++	estopped bool
++	pending  map[string]chan Reply
++	queue    chan []byte // to the current process's stdin; nil while down
++	kill     func()
++	restarts int
++	pid      int
++}
++
++func newPlugin(spec Spec, log *slog.Logger, out chan<- Output) *Plugin {
++	return &Plugin{spec: spec, log: log.With("plugin", spec.Name), out: out, state: StateStarting,
++		pending: map[string]chan Reply{}}
++}
++
++// Name is the configured name.
++func (p *Plugin) Name() string { return p.spec.Name }
++
++// Info is a snapshot for status.
++type Info struct {
++	Name     string
++	State    string
++	Describe *Describe
++	Fault    *protocol.Fault
++	Restarts int
++	PID      int
++}
++
++func (p *Plugin) Info() Info {
++	p.mu.Lock()
++	defer p.mu.Unlock()
++	return Info{Name: p.spec.Name, State: p.state, Describe: p.desc, Fault: p.fault, Restarts: p.restarts, PID: p.pid}
++}
++
++func (p *Plugin) setState(s string) {
++	p.mu.Lock()
++	changed := p.state != s
++	p.state = s
++	p.mu.Unlock()
++	if changed {
++		p.emit(Output{Kind: OutState, Data: map[string]any{"state": s}}, true)
++	}
++}
++
++func (p *Plugin) emit(o Output, important bool) {
++	o.Plugin, o.At = p.spec.Name, time.Now()
++	if important {
++		select {
++		case p.out <- o:
++		case <-time.After(200 * time.Millisecond):
++			p.log.Warn("output channel full; dropped", "kind", o.Kind)
++		}
++		return
++	}
++	select {
++	case p.out <- o:
++	default:
++	}
++}
++
++// send queues one line for the plugin. A plugin that stops reading its stdin is killed: it can no longer be told to
++// stop, and its own heartbeat watchdog is the only thing left, so we make sure by ending it.
++func (p *Plugin) send(msg map[string]any) error {
++	b, err := json.Marshal(msg)
++	if err != nil {
++		return err
++	}
++	b = append(b, '\n')
++	p.mu.Lock()
++	q, kill := p.queue, p.kill
++	p.mu.Unlock()
++	if q == nil {
++		return errors.New(protocol.FaultPluginDown)
++	}
++	select {
++	case q <- b:
++		return nil
++	default:
++		p.log.Error("plugin is not reading its input; killing it")
++		if kill != nil {
++			kill()
++		}
++		return errors.New(protocol.FaultPluginDown)
++	}
++}
++
++// Stop tells the plugin to stop every actuator.
++func (p *Plugin) Stop() { _ = p.send(map[string]any{"op": "stop"}) }
++
++// Estop latches the plugin's e-stop (the core also refuses motion itself).
++func (p *Plugin) Estop() {
++	p.mu.Lock()
++	p.estopped = true
++	p.mu.Unlock()
++	_ = p.send(map[string]any{"op": "estop"})
++}
++
++// Resume clears the plugin's e-stop.
++func (p *Plugin) Resume() {
++	p.mu.Lock()
++	p.estopped = false
++	p.mu.Unlock()
++	_ = p.send(map[string]any{"op": "resume"})
++}
++
++// Command sends a command and waits for the plugin's ack or nack.
++func (p *Plugin) Command(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int) Reply {
++	p.mu.Lock()
++	if p.state != StateReady && kind != protocol.KindHalt {
++		st, fault := p.state, p.fault
++		p.mu.Unlock()
++		if st == StateFaulted && fault != nil {
++			return Reply{FaultCode: fault.Code, Message: fault.Message}
++		}
++		if st == StateDown || st == StateStopped {
++			return Reply{FaultCode: protocol.FaultPluginDown}
++		}
++		return Reply{FaultCode: protocol.FaultNotReady}
++	}
++	ch := make(chan Reply, 1)
++	p.pending[id] = ch
++	p.mu.Unlock()
++	defer func() {
++		p.mu.Lock()
++		delete(p.pending, id)
++		p.mu.Unlock()
++	}()
++	msg := map[string]any{"op": "command", "id": id, "kind": kind, "deadline_ms": deadlineMS}
++	if len(value) > 0 {
++		msg["value"] = value
++	}
++	if err := p.send(msg); err != nil {
++		return Reply{FaultCode: protocol.FaultPluginDown}
++	}
++	wait := time.Duration(deadlineMS) * time.Millisecond
++	if wait < MinReplyTimeout {
++		wait = MinReplyTimeout
++	}
++	t := time.NewTimer(wait)
++	defer t.Stop()
++	select {
++	case r := <-ch:
++		return r
++	case <-t.C:
++		p.Stop()
++		return Reply{FaultCode: protocol.FaultPluginTimeout}
++	case <-ctx.Done():
++		p.Stop()
++		return Reply{FaultCode: protocol.FaultShuttingDown}
++	}
++}
++
++func (p *Plugin) failPending(code string) {
++	p.mu.Lock()
++	defer p.mu.Unlock()
++	for id, ch := range p.pending {
++		select {
++		case ch <- Reply{FaultCode: code}:
++		default:
++		}
++		delete(p.pending, id)
++	}
++}
++
++// run supervises the plugin until ctx ends.
++func (p *Plugin) run(ctx context.Context) {
++	backoff := BackoffMin
++	for {
++		started := time.Now()
++		err := p.runOnce(ctx)
++		p.failPending(protocol.FaultPluginDown)
++		if ctx.Err() != nil {
++			p.setState(StateStopped)
++			return
++		}
++		p.mu.Lock()
++		p.restarts++
++		p.fault = &protocol.Fault{Code: protocol.FaultPluginDown, Driver: p.spec.Name, Message: errString(err)}
++		p.mu.Unlock()
++		p.setState(StateDown)
++		p.emit(Output{Kind: OutFault, Data: map[string]any{"fault_code": protocol.FaultPluginDown, "message": errString(err)}}, true)
++		if time.Since(started) > StableAfter {
++			backoff = BackoffMin
++		}
++		p.log.Warn("plugin exited; restarting", "err", errString(err), "in", backoff)
++		select {
++		case <-ctx.Done():
++			p.setState(StateStopped)
++			return
++		case <-time.After(backoff):
++		}
++		backoff *= 2
++		if backoff > BackoffMax {
++			backoff = BackoffMax
++		}
++	}
++}
++
++func errString(err error) string {
++	if err == nil {
++		return "exited"
++	}
++	return err.Error()
++}
++
++func (p *Plugin) runOnce(ctx context.Context) error {
++	if len(p.spec.Argv) == 0 {
++		return errors.New("no command")
++	}
++	p.setState(StateStarting)
++	cmd := exec.Command(p.spec.Argv[0], p.spec.Argv[1:]...)
++	cmd.Dir = p.spec.Dir
++	cmd.Env = append(os.Environ(), p.spec.Env...)
++	cmd.Env = append(cmd.Env, "PYTHONUNBUFFERED=1")
++	setProcAttr(cmd)
++	stdin, err := cmd.StdinPipe()
++	if err != nil {
++		return err
++	}
++	stdout, err := cmd.StdoutPipe()
++	if err != nil {
++		return err
++	}
++	stderr, err := cmd.StderrPipe()
++	if err != nil {
++		return err
++	}
++	if err := cmd.Start(); err != nil {
++		return fmt.Errorf("start %s: %w", p.spec.Argv[0], err)
++	}
++	queue := make(chan []byte, 256)
++	var killOnce sync.Once
++	kill := func() { killOnce.Do(func() { killProcess(cmd) }) }
++	p.mu.Lock()
++	p.queue, p.kill, p.pid, p.fault, p.desc = queue, kill, cmd.Process.Pid, nil, nil
++	estopped := p.estopped
++	p.mu.Unlock()
++
++	// Writer: the only goroutine touching stdin. Closing stdin is the plugin's signal to stop and exit.
++	writerDone := make(chan struct{})
++	stopWriter := make(chan struct{})
++	go func() {
++		defer close(writerDone)
++		defer stdin.Close()
++		for {
++			select {
++			case b := <-queue:
++				if _, err := stdin.Write(b); err != nil {
++					return
++				}
++			case <-stopWriter:
++				// Drain what is queued (a final stop) and close.
++				for {
++					select {
++					case b := <-queue:
++						if _, err := stdin.Write(b); err != nil {
++							return
++						}
++					default:
++						return
++					}
++				}
++			}
++		}
++	}()
++	go p.pipeStderr(stderr)
++
++	described := make(chan struct{})
++	readerDone := make(chan struct{})
++	go func() {
++		defer close(readerDone)
++		p.readStdout(stdout, described)
++	}()
++
++	_ = p.send(map[string]any{"op": "hello", "config": nonNil(p.spec.Config), "probe": false})
++	if estopped {
++		_ = p.send(map[string]any{"op": "estop"})
++	}
++
++	exited := make(chan error, 1)
++	go func() {
++		<-readerDone
++		exited <- cmd.Wait()
++	}()
++
++	hb := time.NewTicker(HeartbeatInterval)
++	defer hb.Stop()
++	describeTimer := time.NewTimer(DescribeTimeout)
++	defer describeTimer.Stop()
++	shutdown := func() error {
++		_ = p.send(map[string]any{"op": "stop"})
++		p.mu.Lock()
++		p.queue = nil
++		p.mu.Unlock()
++		close(stopWriter)
++		select {
++		case err := <-exited:
++			return err
++		case <-time.After(StopGrace):
++			kill()
++			return <-exited
++		}
++	}
++	for {
++		select {
++		case <-ctx.Done():
++			return shutdown()
++		case err := <-exited:
++			p.mu.Lock()
++			p.queue = nil
++			p.mu.Unlock()
++			close(stopWriter)
++			<-writerDone
++			if err == nil {
++				err = errors.New("plugin exited")
++			}
++			return err
++		case <-hb.C:
++			_ = p.send(map[string]any{"op": "heartbeat", "t": time.Now().UnixMilli()})
++		case <-described:
++			described = nil
++			describeTimer.Stop()
++		case <-describeTimer.C:
++			if described != nil {
++				p.log.Error("plugin did not describe itself in time; killing it")
++				kill()
++			}
++		}
++	}
++}
++
++func nonNil(m map[string]any) map[string]any {
++	if m == nil {
++		return map[string]any{}
++	}
++	return m
++}
++
++func (p *Plugin) pipeStderr(r io.Reader) {
++	sc := bufio.NewScanner(r)
++	sc.Buffer(make([]byte, 64*1024), 1024*1024)
++	for sc.Scan() {
++		p.log.Info(sc.Text(), "stream", "stderr")
++	}
++}
++
++func (p *Plugin) readStdout(r io.Reader, described chan struct{}) {
++	sc := bufio.NewScanner(r)
++	sc.Buffer(make([]byte, 64*1024), 4*1024*1024)
++	gotDescribe := false
++	for sc.Scan() {
++		line := sc.Bytes()
++		var m map[string]any
++		if err := json.Unmarshal(line, &m); err != nil {
++			p.log.Warn("plugin wrote a line that is not JSON", "line", truncate(string(line), 200))
++			continue
++		}
++		op, _ := m["op"].(string)
++		delete(m, "op")
++		switch op {
++		case "describe":
++			var d Describe
++			b, _ := json.Marshal(m)
++			_ = json.Unmarshal(b, &d)
++			p.mu.Lock()
++			p.desc = &d
++			p.mu.Unlock()
++			if !gotDescribe {
++				gotDescribe = true
++				close(described)
++			}
++			p.log.Info("plugin described", "driver", d.Driver, "version", d.Version)
++		case "ready":
++			p.setState(StateReady)
++		case "ack", "nack":
++			id, _ := m["id"].(string)
++			code, _ := m["fault_code"].(string)
++			msg, _ := m["message"].(string)
++			p.mu.Lock()
++			ch := p.pending[id]
++			p.mu.Unlock()
++			if ch != nil {
++				select {
++				case ch <- Reply{OK: op == "ack", FaultCode: code, Message: msg}:
++				default:
++				}
++			}
++		case "fault":
++			code, _ := m["fault_code"].(string)
++			msg, _ := m["message"].(string)
++			if code == "" {
++				code = protocol.FaultHardware
++			}
++			p.mu.Lock()
++			p.fault = &protocol.Fault{Code: code, Driver: p.spec.Name, Message: msg}
++			p.mu.Unlock()
++			p.setState(StateFaulted)
++			p.emit(Output{Kind: OutFault, Data: m}, true)
++			p.log.Error("plugin fault", "code", code, "message", msg)
++		case "telemetry":
++			if fl, ok := m["faults"].([]any); ok && len(fl) > 0 {
++				p.log.Debug("plugin reports faults", "faults", fl)
++			}
++			p.emit(Output{Kind: OutTelemetry, Data: m}, false)
++		case "event":
++			p.emit(Output{Kind: OutEvent, Data: m}, true)
++		case "video":
++			p.emit(Output{Kind: OutVideo, Data: m}, false)
++		default:
++			p.log.Warn("unknown op from plugin", "op", op)
++		}
++	}
++}
++
++func truncate(s string, n int) string {
++	if len(s) > n {
++		return s[:n] + "…"
++	}
++	return s
++}
++
++// Probe starts a plugin in probe mode and returns its describe without letting it touch hardware.
++func Probe(ctx context.Context, spec Spec) (*Describe, error) {
++	ctx, cancel := context.WithTimeout(ctx, DescribeTimeout)
++	defer cancel()
++	cmd := exec.CommandContext(ctx, spec.Argv[0], spec.Argv[1:]...)
++	cmd.Dir = spec.Dir
++	cmd.Env = append(append(os.Environ(), spec.Env...), "PYTHONUNBUFFERED=1")
++	stdin, err := cmd.StdinPipe()
++	if err != nil {
++		return nil, err
++	}
++	stdout, err := cmd.StdoutPipe()
++	if err != nil {
++		return nil, err
++	}
++	if err := cmd.Start(); err != nil {
++		return nil, err
++	}
++	defer func() { _ = cmd.Wait() }()
++	defer stdin.Close()
++	b, _ := json.Marshal(map[string]any{"op": "hello", "config": nonNil(spec.Config), "probe": true})
++	if _, err := stdin.Write(append(b, '\n')); err != nil {
++		return nil, err
++	}
++	sc := bufio.NewScanner(stdout)
++	sc.Buffer(make([]byte, 64*1024), 4*1024*1024)
++	for sc.Scan() {
++		var m struct {
++			Op string `json:"op"`
++			Describe
++		}
++		if json.Unmarshal(sc.Bytes(), &m) == nil && m.Op == "describe" {
++			d := m.Describe
++			return &d, nil
++		}
++	}
++	if ctx.Err() != nil {
++		return nil, errors.New("plugin did not describe itself in time")
++	}
++	return nil, errors.New("plugin exited without describing itself")
++}
ddiff --git a/internal/plugins/plugins_test.go b/internal/plugins/plugins_test.go
nnew file mode 100644
iindex 0000000..da05eca
---- /dev/null
++++ b/internal/plugins/plugins_test.go
@@@ -0,0 +1,303 @@
++package plugins
++
++import (
++	"bufio"
++	"context"
++	"encoding/json"
++	"fmt"
++	"io"
++	"log/slog"
++	"os"
++	"path/filepath"
++	"strings"
++	"sync"
++	"testing"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// The test binary doubles as a fake plugin: with OPENVIBE_FAKE_PLUGIN set it speaks the plugin protocol and appends
++// everything it receives to the file named by that variable.
++func TestMain(m *testing.M) {
++	if path := os.Getenv("OPENVIBE_FAKE_PLUGIN"); path != "" {
++		fakePlugin(path, os.Stdin, os.Stdout)
++		os.Exit(0)
++	}
++	HeartbeatInterval = 20 * time.Millisecond
++	BackoffMin = 20 * time.Millisecond
++	BackoffMax = 100 * time.Millisecond
++	StopGrace = 500 * time.Millisecond
++	os.Exit(m.Run())
++}
++
++func fakePlugin(logPath string, in io.Reader, out io.Writer) {
++	f, _ := os.OpenFile(logPath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
++	defer f.Close()
++	rec := func(s string) { fmt.Fprintln(f, s); f.Sync() }
++	w := json.NewEncoder(out)
++	sc := bufio.NewScanner(in)
++	mode := os.Getenv("OPENVIBE_FAKE_MODE")
++	for sc.Scan() {
++		var m map[string]any
++		json.Unmarshal(sc.Bytes(), &m)
++		op, _ := m["op"].(string)
++		if op != "heartbeat" {
++			rec(sc.Text())
++		}
++		switch op {
++		case "hello":
++			if mode == "mute" {
++				continue
++			}
++			w.Encode(map[string]any{"op": "describe", "driver": "fake", "version": "1",
++				"capabilities": map[string]any{"drive": map[string]any{}, "actuator": map[string]any{"names": []string{"lift"}}}})
++			if m["probe"] == true {
++				continue
++			}
++			if mode == "fault" {
++				w.Encode(map[string]any{"op": "fault", "fault_code": "firmware_unsupported", "message": "v1 not v2"})
++				continue
++			}
++			w.Encode(map[string]any{"op": "ready"})
++			w.Encode(map[string]any{"op": "event", "name": "booted"})
++		case "command":
++			switch m["kind"] {
++			case "crash":
++				rec("CRASH")
++				os.Exit(3)
++			case "hang":
++				time.Sleep(10 * time.Second)
++			case "slow":
++				continue
++			case "display":
++				w.Encode(map[string]any{"op": "nack", "id": m["id"], "fault_code": "unsupported"})
++			default:
++				w.Encode(map[string]any{"op": "ack", "id": m["id"]})
++				w.Encode(map[string]any{"op": "telemetry", "sensors": map[string]any{"last": m["kind"]}})
++			}
++		}
++	}
++	rec("EOF-STOP")
++}
++
++type fixture struct {
++	t    *testing.T
++	log  string
++	m    *Manager
++	stop context.CancelFunc
++	outs chan Output
++	seen []Output
++}
++
++func start(t *testing.T, mode string) *fixture {
++	t.Helper()
++	logPath := filepath.Join(t.TempDir(), "plugin.log")
++	exe, _ := os.Executable()
++	spec := Spec{Name: "fake", Argv: []string{exe}, Env: []string{"OPENVIBE_FAKE_PLUGIN=" + logPath, "OPENVIBE_FAKE_MODE=" + mode},
++		Config: map[string]any{"k": "v"}}
++	m := NewManager([]Spec{spec}, slog.New(slog.NewTextHandler(io.Discard, nil)))
++	ctx, cancel := context.WithCancel(context.Background())
++	m.Start(ctx)
++	fx := &fixture{t: t, log: logPath, m: m, stop: cancel, outs: make(chan Output, 1024)}
++	go func() {
++		for o := range m.Output() {
++			fx.outs <- o
++		}
++	}()
++	t.Cleanup(func() { cancel(); m.Wait() })
++	return fx
++}
++
++func (fx *fixture) waitState(s string) {
++	fx.t.Helper()
++	deadline := time.Now().Add(5 * time.Second)
++	for time.Now().Before(deadline) {
++		if fx.m.Get("fake").Info().State == s {
++			return
++		}
++		time.Sleep(5 * time.Millisecond)
++	}
++	fx.t.Fatalf("state %s never reached (now %s)", s, fx.m.Get("fake").Info().State)
++}
++
++func (fx *fixture) lines() []string {
++	b, _ := os.ReadFile(fx.log)
++	return strings.Split(strings.TrimSpace(string(b)), "\n")
++}
++
++func (fx *fixture) waitLog(substr string) {
++	fx.t.Helper()
++	deadline := time.Now().Add(5 * time.Second)
++	for time.Now().Before(deadline) {
++		b, _ := os.ReadFile(fx.log)
++		if strings.Contains(string(b), substr) {
++			return
++		}
++		time.Sleep(5 * time.Millisecond)
++	}
++	fx.t.Fatalf("plugin never logged %q; got:\n%s", substr, strings.Join(fx.lines(), "\n"))
++}
++
++func (fx *fixture) waitOutput(kind string) Output {
++	fx.t.Helper()
++	for i, o := range fx.seen {
++		if o.Kind == kind {
++			fx.seen = append(fx.seen[:i:i], fx.seen[i+1:]...)
++			return o
++		}
++	}
++	timeout := time.After(5 * time.Second)
++	for {
++		select {
++		case o := <-fx.outs:
++			if o.Kind == kind {
++				return o
++			}
++			fx.seen = append(fx.seen, o)
++		case <-timeout:
++			fx.t.Fatalf("no %s output", kind)
++		}
++	}
++}
++
++func TestHelloDescribeCommand(t *testing.T) {
++	fx := start(t, "")
++	fx.waitState(StateReady)
++	p := fx.m.Get("fake")
++	if d := p.Info().Describe; d == nil || d.Driver != "fake" {
++		t.Fatalf("describe %+v", d)
++	}
++	fx.waitLog(`"config":{"k":"v"}`)
++	r := p.Command(context.Background(), "c1", "drive", json.RawMessage(`{"throttle":0.5}`), 300)
++	if !r.OK {
++		t.Fatalf("%+v", r)
++	}
++	fx.waitLog(`"deadline_ms":300`)
++	fx.waitLog(`"value":{"throttle":0.5}`)
++	if r := p.Command(context.Background(), "c2", "display", nil, 300); r.OK || r.FaultCode != "unsupported" {
++		t.Fatalf("%+v", r)
++	}
++	if o := fx.waitOutput(OutTelemetry); o.Data["sensors"] == nil {
++		t.Fatalf("%+v", o)
++	}
++	if o := fx.waitOutput(OutEvent); o.Data["name"] != "booted" {
++		t.Fatalf("%+v", o)
++	}
++}
++
++func TestRouting(t *testing.T) {
++	fx := start(t, "")
++	fx.waitState(StateReady)
++	if fx.m.Route("drive", "", "") == nil || fx.m.Route("actuator", "", "lift") == nil {
++		t.Fatal("not routed")
++	}
++	if fx.m.Route("say", "", "") != nil {
++		t.Fatal("say routed to a plugin without it")
++	}
++	if fx.m.Route("drive", "nope", "") != nil {
++		t.Fatal("unknown target routed")
++	}
++}
++
++func TestShutdownStopsAndClosesStdin(t *testing.T) {
++	fx := start(t, "")
++	fx.waitState(StateReady)
++	fx.stop()
++	fx.m.Wait()
++	l := fx.lines()
++	if !strings.Contains(l[len(l)-2], `"op":"stop"`) || l[len(l)-1] != "EOF-STOP" {
++		t.Fatalf("want stop then EOF, got %v", l)
++	}
++}
++
++func TestCrashRestartsWithEstop(t *testing.T) {
++	fx := start(t, "")
++	fx.waitState(StateReady)
++	p := fx.m.Get("fake")
++	p.Estop()
++	fx.waitLog(`{"op":"estop"}`)
++	r := p.Command(context.Background(), "boom", "crash", nil, 300)
++	if r.OK || r.FaultCode != protocol.FaultPluginDown {
++		t.Fatalf("%+v", r)
++	}
++	fx.waitOutput(OutFault)
++	fx.waitState(StateReady)
++	if p.Info().Restarts < 1 {
++		t.Fatal("no restart counted")
++	}
++	// After the restart the latched e-stop is sent again right after hello.
++	l := fx.lines()
++	var afterCrash []string
++	for i, s := range l {
++		if s == "CRASH" {
++			afterCrash = l[i+1:]
++		}
++	}
++	if len(afterCrash) < 2 || !strings.Contains(afterCrash[0], `"op":"hello"`) || afterCrash[1] != `{"op":"estop"}` {
++		t.Fatalf("after restart: %v", afterCrash)
++	}
++}
++
++func TestReplyTimeout(t *testing.T) {
++	fx := start(t, "")
++	fx.waitState(StateReady)
++	old := MinReplyTimeout
++	MinReplyTimeout = 100 * time.Millisecond
++	defer func() { MinReplyTimeout = old }()
++	r := fx.m.Get("fake").Command(context.Background(), "s", "slow", nil, 50)
++	if r.FaultCode != protocol.FaultPluginTimeout {
++		t.Fatalf("%+v", r)
++	}
++	fx.waitLog(`{"op":"stop"}`)
++}
++
++func TestFaultState(t *testing.T) {
++	fx := start(t, "fault")
++	fx.waitState(StateFaulted)
++	r := fx.m.Get("fake").Command(context.Background(), "x", "drive", nil, 300)
++	if r.FaultCode != "firmware_unsupported" {
++		t.Fatalf("%+v", r)
++	}
++	if f := fx.m.Get("fake").Info().Fault; f == nil || f.Message != "v1 not v2" {
++		t.Fatalf("%+v", f)
++	}
++}
++
++func TestMuteKilledAfterDescribeTimeout(t *testing.T) {
++	old := DescribeTimeout
++	DescribeTimeout = 200 * time.Millisecond
++	defer func() { DescribeTimeout = old }()
++	fx := start(t, "mute")
++	fx.waitOutput(OutFault)
++}
++
++func TestProbe(t *testing.T) {
++	logPath := filepath.Join(t.TempDir(), "p.log")
++	exe, _ := os.Executable()
++	d, err := Probe(context.Background(), Spec{Name: "fake", Argv: []string{exe}, Env: []string{"OPENVIBE_FAKE_PLUGIN=" + logPath}})
++	if err != nil || d.Driver != "fake" {
++		t.Fatalf("%v %+v", err, d)
++	}
++	b, _ := os.ReadFile(logPath)
++	if !strings.Contains(string(b), `"probe":true`) {
++		t.Fatalf("%s", b)
++	}
++}
++
++func TestConcurrentCommands(t *testing.T) {
++	fx := start(t, "")
++	fx.waitState(StateReady)
++	p := fx.m.Get("fake")
++	var wg sync.WaitGroup
++	for i := 0; i < 50; i++ {
++		wg.Add(1)
++		go func(i int) {
++			defer wg.Done()
++			if r := p.Command(context.Background(), fmt.Sprint("id", i), "drive", nil, 300); !r.OK {
++				t.Errorf("%d: %+v", i, r)
++			}
++		}(i)
++	}
++	wg.Wait()
++}
ddiff --git a/internal/plugins/proc_unix.go b/internal/plugins/proc_unix.go
nnew file mode 100644
iindex 0000000..d1d3fc6
---- /dev/null
++++ b/internal/plugins/proc_unix.go
@@@ -0,0 +1,22 @@
++//go:build !windows
++
++package plugins
++
++import (
++	"os/exec"
++	"syscall"
++)
++
++// setProcAttr puts the plugin in its own process group so killing it also ends its children (speech synthesis,
++// camera helpers).
++func setProcAttr(cmd *exec.Cmd) {
++	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
++}
++
++func killProcess(cmd *exec.Cmd) {
++	if cmd.Process == nil {
++		return
++	}
++	_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
++	_ = cmd.Process.Kill()
++}
ddiff --git a/internal/plugins/proc_windows.go b/internal/plugins/proc_windows.go
nnew file mode 100644
iindex 0000000..834cdb6
---- /dev/null
++++ b/internal/plugins/proc_windows.go
@@@ -0,0 +1,13 @@
++//go:build windows
++
++package plugins
++
++import "os/exec"
++
++func setProcAttr(cmd *exec.Cmd) {}
++
++func killProcess(cmd *exec.Cmd) {
++	if cmd.Process != nil {
++		_ = cmd.Process.Kill()
++	}
++}
ddiff --git a/internal/safety/dedup.go b/internal/safety/dedup.go
nnew file mode 100644
iindex 0000000..a0b0585
---- /dev/null
++++ b/internal/safety/dedup.go
@@@ -0,0 +1,75 @@
++package safety
++
++import (
++	"container/list"
++	"sync"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// Entry is one command id's result, pending until Complete.
++type Entry struct {
++	id     string
++	done   chan struct{}
++	result protocol.Message
++	at     time.Time
++	elem   *list.Element
++}
++
++// Done is closed when the result is known.
++func (e *Entry) Done() <-chan struct{} { return e.done }
++
++// Result is the first result (an Ack or a Nack). Valid after Done.
++func (e *Entry) Result() protocol.Message { return e.result }
++
++// Dedup remembers command ids so a repeated id is answered with the first result and never executed twice. It lives
++// for the whole process, across reconnects, so a server that re-sends after a reconnect gets the old answer.
++type Dedup struct {
++	mu    sync.Mutex
++	max   int
++	ttl   time.Duration
++	items map[string]*Entry
++	order *list.List // oldest first
++	now   func() time.Time
++}
++
++func NewDedup(max int, ttl time.Duration) *Dedup {
++	return &Dedup{max: max, ttl: ttl, items: map[string]*Entry{}, order: list.New(), now: time.Now}
++}
++
++// Begin registers id. first is true when the caller must execute the command and later call Complete; otherwise the
++// caller waits on e.Done() and answers with e.Result().
++func (d *Dedup) Begin(id string) (e *Entry, first bool) {
++	d.mu.Lock()
++	defer d.mu.Unlock()
++	now := d.now()
++	for f := d.order.Front(); f != nil; f = d.order.Front() {
++		old := f.Value.(*Entry)
++		if len(d.items) <= d.max && now.Sub(old.at) < d.ttl {
++			break
++		}
++		d.order.Remove(f)
++		delete(d.items, old.id)
++	}
++	if e, ok := d.items[id]; ok {
++		return e, false
++	}
++	e = &Entry{id: id, done: make(chan struct{}), at: now}
++	e.elem = d.order.PushBack(e)
++	d.items[id] = e
++	return e, true
++}
++
++// Complete records the first result. Later calls are ignored.
++func (d *Dedup) Complete(e *Entry, result protocol.Message) {
++	d.mu.Lock()
++	defer d.mu.Unlock()
++	select {
++	case <-e.done:
++		return
++	default:
++	}
++	e.result = result
++	close(e.done)
++}
ddiff --git a/internal/safety/latch.go b/internal/safety/latch.go
nnew file mode 100644
iindex 0000000..3135abb
---- /dev/null
++++ b/internal/safety/latch.go
@@@ -0,0 +1,187 @@
++// Package safety holds the rules that keep motors from running when they should not: the persisted e-stop and local
++// kill-switch latch, the owner's limits applied to every value before it reaches a plugin, deadline capping, and the
++// command-id cache that makes a repeated command answer with its first result instead of running twice.
++package safety
++
++import (
++	"encoding/json"
++	"errors"
++	"io/fs"
++	"os"
++	"path/filepath"
++	"sync"
++	"time"
++)
++
++// LatchState is the persisted e-stop state. Two independent latches:
++//
++//   - Remote: set by a server `estop`; cleared by a server `estop_clear` (the owner) or a local `resume`.
++//   - Local: set by `openvibe-node stop` on the machine itself; cleared only by `openvibe-node resume`.
++//
++// Anything that moves is refused while either is set.
++type LatchState struct {
++	Remote       bool      `json:"remote"`
++	RemoteReason string    `json:"remote_reason,omitempty"`
++	RemoteAt     time.Time `json:"remote_at,omitempty"`
++	Local        bool      `json:"local"`
++	LocalAt      time.Time `json:"local_at,omitempty"`
++}
++
++// Stopped reports whether anything is latched.
++func (s LatchState) Stopped() bool { return s.Remote || s.Local }
++
++// Latch is the persisted latch. It survives restarts: a Node that was e-stopped comes back e-stopped.
++type Latch struct {
++	mu       sync.Mutex
++	path     string
++	state    LatchState
++	lastRaw  []byte
++	onChange []func(LatchState)
++}
++
++// OpenLatch reads the latch file (missing = clear). A corrupt file is treated as latched: failing safe.
++func OpenLatch(path string) (*Latch, error) {
++	l := &Latch{path: path}
++	st, raw, err := readLatch(path)
++	if err != nil {
++		l.state = LatchState{Local: true, LocalAt: time.Now()}
++		return l, err
++	}
++	l.state, l.lastRaw = st, raw
++	return l, nil
++}
++
++func readLatch(path string) (LatchState, []byte, error) {
++	var st LatchState
++	raw, err := os.ReadFile(path)
++	if errors.Is(err, fs.ErrNotExist) {
++		return st, nil, nil
++	}
++	if err != nil {
++		return st, nil, err
++	}
++	if err := json.Unmarshal(raw, &st); err != nil {
++		return st, raw, errors.New("latch file is corrupt; holding the local stop until `openvibe-node resume`")
++	}
++	return st, raw, nil
++}
++
++// State returns the current state.
++func (l *Latch) State() LatchState {
++	l.mu.Lock()
++	defer l.mu.Unlock()
++	return l.state
++}
++
++// OnChange registers a callback run (outside the lock) after every change.
++func (l *Latch) OnChange(f func(LatchState)) {
++	l.mu.Lock()
++	l.onChange = append(l.onChange, f)
++	l.mu.Unlock()
++}
++
++func (l *Latch) update(f func(*LatchState)) (LatchState, error) {
++	l.mu.Lock()
++	before := l.state
++	f(&l.state)
++	st := l.state
++	err := l.persistLocked()
++	cbs := append([]func(LatchState){}, l.onChange...)
++	l.mu.Unlock()
++	if before != st {
++		for _, cb := range cbs {
++			cb(st)
++		}
++	}
++	return st, err
++}
++
++func (l *Latch) persistLocked() error {
++	b, err := json.Marshal(l.state)
++	if err != nil {
++		return err
++	}
++	if err := os.MkdirAll(filepath.Dir(l.path), 0o700); err != nil {
++		return err
++	}
++	tmp := l.path + ".tmp"
++	if err := os.WriteFile(tmp, b, 0o600); err != nil {
++		return err
++	}
++	if err := os.Rename(tmp, l.path); err != nil {
++		return err
++	}
++	l.lastRaw = b
++	return nil
++}
++
++// SetRemote latches the server e-stop.
++func (l *Latch) SetRemote(reason string) (LatchState, error) {
++	return l.update(func(s *LatchState) {
++		if !s.Remote {
++			s.RemoteAt = time.Now().UTC()
++		}
++		s.Remote, s.RemoteReason = true, reason
++	})
++}
++
++// ClearRemote clears the server e-stop (owner's estop_clear). The local latch is untouched.
++func (l *Latch) ClearRemote() (LatchState, error) {
++	return l.update(func(s *LatchState) { s.Remote, s.RemoteReason, s.RemoteAt = false, "", time.Time{} })
++}
++
++// SetLocal latches the local kill switch.
++func (l *Latch) SetLocal() (LatchState, error) {
++	return l.update(func(s *LatchState) {
++		if !s.Local {
++			s.LocalAt = time.Now().UTC()
++		}
++		s.Local = true
++	})
++}
++
++// Resume clears both latches (`openvibe-node resume`).
++func (l *Latch) Resume() (LatchState, error) {
++	return l.update(func(s *LatchState) { *s = LatchState{} })
++}
++
++// Reload adopts the file's state if another process (the CLI, when the Node's socket was unreachable) changed it.
++// It reports whether anything changed.
++func (l *Latch) Reload() (bool, error) {
++	st, raw, err := readLatch(l.path)
++	l.mu.Lock()
++	if err != nil {
++		if raw == nil || string(raw) == string(l.lastRaw) {
++			l.mu.Unlock()
++			return false, err
++		}
++		st = l.state
++		st.Local, st.LocalAt = true, time.Now().UTC()
++	} else if string(raw) == string(l.lastRaw) {
++		l.mu.Unlock()
++		return false, nil
++	}
++	changed := st != l.state
++	l.state, l.lastRaw = st, raw
++	cbs := append([]func(LatchState){}, l.onChange...)
++	l.mu.Unlock()
++	if changed {
++		for _, cb := range cbs {
++			cb(st)
++		}
++	}
++	return changed, err
++}
++
++// WriteLocalFile latches the local stop (or clears everything, for resume) directly in the file. The CLI uses it
++// when no Node is running or its socket does not answer; a running Node picks the change up within its poll interval.
++func WriteLocalFile(path string, stop bool) error {
++	l, _ := OpenLatch(path)
++	var err error
++	if stop {
++		_, err = l.SetLocal()
++	} else {
++		_, err = l.Resume()
++	}
++	return err
++}
ddiff --git a/internal/safety/limits.go b/internal/safety/limits.go
nnew file mode 100644
iindex 0000000..5c14582
---- /dev/null
++++ b/internal/safety/limits.go
@@@ -0,0 +1,152 @@
++package safety
++
++import (
++	"bytes"
++	"encoding/json"
++	"fmt"
++	"math"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// Limits are the effective owner limits: the stricter of the server's config and the local config.
++type Limits struct {
++	MaxSpeed     float64 // 0..1, applied to throttle, x, y
++	MaxTurn      float64 // 0..1, applied to steer, rotation
++	MaxCommandMS int     // cap on deadline_ms
++}
++
++// DefaultLimits apply before the server sends config: full scale, deadlines capped at protocol.DefaultMaxCommandMS.
++func DefaultLimits() Limits {
++	return Limits{MaxSpeed: 1, MaxTurn: 1, MaxCommandMS: protocol.DefaultMaxCommandMS}
++}
++
++// Merge returns the stricter of the server limits and the local caps.
++func Merge(server protocol.Limits, localSpeed, localTurn *float64, localMaxMS int) Limits {
++	l := DefaultLimits()
++	pick := func(cur float64, vals ...*float64) float64 {
++		for _, v := range vals {
++			if v != nil && !math.IsNaN(*v) {
++				cur = math.Min(cur, math.Max(0, *v))
++			}
++		}
++		return cur
++	}
++	l.MaxSpeed = pick(l.MaxSpeed, server.MaxSpeed, localSpeed)
++	l.MaxTurn = pick(l.MaxTurn, server.MaxTurn, localTurn)
++	if server.MaxCommandMS > 0 {
++		l.MaxCommandMS = server.MaxCommandMS
++	}
++	if localMaxMS > 0 && localMaxMS < l.MaxCommandMS {
++		l.MaxCommandMS = localMaxMS
++	}
++	return l
++}
++
++// Deadline returns the deadline in ms for a command: the default when absent, never above MaxCommandMS.
++func (l Limits) Deadline(requested int) int {
++	d := requested
++	if d <= 0 {
++		d = protocol.DefaultDeadlineMS
++	}
++	max := l.MaxCommandMS
++	if max <= 0 {
++		max = protocol.DefaultMaxCommandMS
++	}
++	if d > max {
++		d = max
++	}
++	return d
++}
++
++var speedKeys = map[string]bool{"throttle": true, "x": true, "y": true}
++var turnKeys = map[string]bool{"steer": true, "rotation": true}
++var positionKeys = map[string]bool{"pan": true, "tilt": true, "zoom": true}
++
++// FaultError is an error carrying a protocol fault code.
++type FaultError struct{ Code, Message string }
++
++func (e *FaultError) Error() string { return e.Code + ": " + e.Message }
++
++func badValue(format string, a ...any) error {
++	return &FaultError{Code: protocol.FaultBadValue, Message: fmt.Sprintf(format, a...)}
++}
++
++// Clamp checks a command value and clamps every motion field to the limits. It runs in the core, before a plugin
++// sees the value, so no operator can exceed the owner's limits whatever the plugin does.
++//
++//	drive:    throttle, x, y → ±MaxSpeed; steer, rotation → ±MaxTurn (all must be numbers)
++//	ptz:      pan, tilt, zoom → -1..1
++//	actuator: a numeric "value" → -1..1; "name" must be a string
++func (l Limits) Clamp(kind string, value json.RawMessage) (json.RawMessage, error) {
++	v := map[string]any{}
++	if len(bytes.TrimSpace(value)) > 0 && string(bytes.TrimSpace(value)) != "null" {
++		d := json.NewDecoder(bytes.NewReader(value))
++		d.UseNumber()
++		if err := d.Decode(&v); err != nil {
++			return nil, badValue("value must be a JSON object")
++		}
++	}
++	num := func(k string, raw any) (float64, error) {
++		n, ok := raw.(json.Number)
++		if !ok {
++			return 0, badValue("%s must be a number", k)
++		}
++		f, err := n.Float64()
++		if err != nil || math.IsNaN(f) || math.IsInf(f, 0) {
++			return 0, badValue("%s must be a finite number", k)
++		}
++		return f, nil
++	}
++	clampTo := func(f, m float64) float64 { return math.Max(-m, math.Min(m, f)) }
++	switch kind {
++	case protocol.KindDrive:
++		for k, raw := range v {
++			switch {
++			case speedKeys[k]:
++				f, err := num(k, raw)
++				if err != nil {
++					return nil, err
++				}
++				v[k] = clampTo(clampTo(f, 1), l.MaxSpeed)
++			case turnKeys[k]:
++				f, err := num(k, raw)
++				if err != nil {
++					return nil, err
++				}
++				v[k] = clampTo(clampTo(f, 1), l.MaxTurn)
++			}
++		}
++	case protocol.KindPTZ:
++		for k, raw := range v {
++			if positionKeys[k] {
++				f, err := num(k, raw)
++				if err != nil {
++					return nil, err
++				}
++				v[k] = clampTo(f, 1)
++			}
++		}
++	case protocol.KindActuator:
++		name, ok := v["name"].(string)
++		if !ok || name == "" {
++			return nil, badValue("actuator needs a name")
++		}
++		if raw, ok := v["value"].(json.Number); ok {
++			f, err := num("value", raw)
++			if err != nil {
++				return nil, err
++			}
++			v["value"] = clampTo(f, 1)
++		}
++	case protocol.KindSay:
++		if s, ok := v["text"].(string); !ok || s == "" || len(s) > 1000 {
++			return nil, badValue("say needs text (1 to 1000 bytes)")
++		}
++	}
++	out, err := json.Marshal(v)
++	if err != nil {
++		return nil, badValue("value could not be re-encoded")
++	}
++	return out, nil
++}
ddiff --git a/internal/safety/safety_test.go b/internal/safety/safety_test.go
nnew file mode 100644
iindex 0000000..0500447
---- /dev/null
++++ b/internal/safety/safety_test.go
@@@ -0,0 +1,154 @@
++package safety
++
++import (
++	"encoding/json"
++	"os"
++	"path/filepath"
++	"testing"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++func f(v float64) *float64 { return &v }
++
++func TestLatchPersistsAcrossRestart(t *testing.T) {
++	p := filepath.Join(t.TempDir(), "latch.json")
++	l, err := OpenLatch(p)
++	if err != nil || l.State().Stopped() {
++		t.Fatalf("fresh latch: %v %+v", err, l.State())
++	}
++	var seen []LatchState
++	l.OnChange(func(s LatchState) { seen = append(seen, s) })
++	l.SetRemote("owner pressed stop")
++	l2, _ := OpenLatch(p)
++	if !l2.State().Remote || l2.State().RemoteReason != "owner pressed stop" {
++		t.Fatalf("not persisted: %+v", l2.State())
++	}
++	l.SetLocal()
++	l.ClearRemote()
++	if st := l.State(); st.Remote || !st.Local {
++		t.Fatalf("estop_clear must not clear the local stop: %+v", st)
++	}
++	l.Resume()
++	if l.State().Stopped() {
++		t.Fatal("resume did not clear")
++	}
++	if len(seen) != 4 {
++		t.Fatalf("callbacks %d", len(seen))
++	}
++}
++
++func TestCorruptLatchFailsSafe(t *testing.T) {
++	p := filepath.Join(t.TempDir(), "latch.json")
++	os.WriteFile(p, []byte("{nope"), 0o600)
++	l, err := OpenLatch(p)
++	if err == nil || !l.State().Local {
++		t.Fatalf("corrupt latch must hold the local stop: %v %+v", err, l.State())
++	}
++}
++
++func TestReloadSeesCLIWrite(t *testing.T) {
++	p := filepath.Join(t.TempDir(), "latch.json")
++	l, _ := OpenLatch(p)
++	if ch, _ := l.Reload(); ch {
++		t.Fatal("spurious change")
++	}
++	if err := WriteLocalFile(p, true); err != nil {
++		t.Fatal(err)
++	}
++	if ch, _ := l.Reload(); !ch || !l.State().Local {
++		t.Fatalf("reload missed the stop: %+v", l.State())
++	}
++	WriteLocalFile(p, false)
++	if ch, _ := l.Reload(); !ch || l.State().Stopped() {
++		t.Fatalf("reload missed the resume: %+v", l.State())
++	}
++}
++
++func TestMerge(t *testing.T) {
++	l := Merge(protocol.Limits{MaxSpeed: f(0.8), MaxTurn: f(0.5), MaxCommandMS: 600}, f(0.3), nil, 400)
++	if l.MaxSpeed != 0.3 || l.MaxTurn != 0.5 || l.MaxCommandMS != 400 {
++		t.Fatalf("%+v", l)
++	}
++	if d := DefaultLimits(); d.MaxSpeed != 1 || d.MaxCommandMS != protocol.DefaultMaxCommandMS {
++		t.Fatalf("%+v", d)
++	}
++}
++
++func TestDeadline(t *testing.T) {
++	l := Limits{MaxCommandMS: 500}
++	for in, want := range map[int]int{0: 300, -5: 300, 100: 100, 500: 500, 9000: 500} {
++		if got := l.Deadline(in); got != want {
++			t.Errorf("Deadline(%d)=%d want %d", in, got, want)
++		}
++	}
++}
++
++func TestClamp(t *testing.T) {
++	l := Limits{MaxSpeed: 0.5, MaxTurn: 0.25, MaxCommandMS: 1000}
++	out, err := l.Clamp("drive", json.RawMessage(`{"throttle":1,"steer":-1,"radius":3}`))
++	if err != nil {
++		t.Fatal(err)
++	}
++	var v map[string]float64
++	json.Unmarshal(out, &v)
++	if v["throttle"] != 0.5 || v["steer"] != -0.25 || v["radius"] != 3 {
++		t.Fatalf("%s", out)
++	}
++	out, _ = l.Clamp("drive", json.RawMessage(`{"x":-9,"y":0.1,"rotation":0.9}`))
++	json.Unmarshal(out, &v)
++	if v["x"] != -0.5 || v["y"] != 0.1 || v["rotation"] != 0.25 {
++		t.Fatalf("%s", out)
++	}
++	out, _ = l.Clamp("ptz", json.RawMessage(`{"pan":4,"tilt":-0.5}`))
++	json.Unmarshal(out, &v)
++	if v["pan"] != 1 || v["tilt"] != -0.5 {
++		t.Fatalf("%s", out)
++	}
++	for kind, bad := range map[string]string{
++		"drive":    `{"throttle":"fast"}`,
++		"ptz":      `{"pan":null}`,
++		"actuator": `{"value":1}`,
++		"say":      `{}`,
++	} {
++		if _, err := l.Clamp(kind, json.RawMessage(bad)); err == nil {
++			t.Errorf("%s %s accepted", kind, bad)
++		}
++	}
++	if _, err := l.Clamp("drive", json.RawMessage(`[1]`)); err == nil {
++		t.Error("array accepted")
++	}
++	if _, err := l.Clamp("halt", nil); err != nil {
++		t.Error(err)
++	}
++}
++
++func TestDedup(t *testing.T) {
++	d := NewDedup(2, time.Minute)
++	e, first := d.Begin("a")
++	if !first {
++		t.Fatal("first")
++	}
++	e2, first2 := d.Begin("a")
++	if first2 || e2 != e {
++		t.Fatal("duplicate treated as new")
++	}
++	d.Complete(e, protocol.Ack{ID: "a"})
++	d.Complete(e, protocol.Nack{ID: "a"})
++	<-e2.Done()
++	if _, ok := e2.Result().(protocol.Ack); !ok {
++		t.Fatal("first result not kept")
++	}
++	d.Begin("b")
++	d.Begin("c")
++	d.Begin("d")
++	if _, first := d.Begin("a"); !first {
++		t.Fatal("capacity not enforced")
++	}
++	now := time.Now()
++	d.now = func() time.Time { return now.Add(2 * time.Minute) }
++	if _, first := d.Begin("d"); !first {
++		t.Fatal("ttl not enforced")
++	}
++}