Safety review fixes

ddiff --git a/internal/plugins/plugin.go b/internal/plugins/plugin.go
iindex 0413e40..55368a6 100644
---- a/internal/plugins/plugin.go
++++ b/internal/plugins/plugin.go
@@@ -383,10 +383,24 @@ func (p *Plugin) runOnce(ctx context.Context) error {
  		_ = p.send(map[string]any{"op": "estop"})
  	}
  
++	// Watch the plugin process itself, not its stdout: a child it started (speech, a camera helper) can hold stdout
++	// open after the plugin died. When the plugin exits, its whole process group is killed, which also frees any
++	// motor bus or serial port a child still holds, and ends the stdout reader.
  	exited := make(chan error, 1)
  	go func() {
++		ps, err := cmd.Process.Wait()
++		kill()
++		select {
++		case <-readerDone:
++		case <-time.After(2 * time.Second):
++		}
++		stdout.Close()
++		stderr.Close()
  		<-readerDone
--		exited <- cmd.Wait()
++		if err == nil && ps != nil && !ps.Success() {
++			err = errors.New(ps.String())
++		}
++		exited <- err
  	}()
  
  	hb := time.NewTicker(HeartbeatInterval)
ddiff --git a/internal/plugins/plugins_test.go b/internal/plugins/plugins_test.go
iindex da05eca..9fa8f2a 100644
---- a/internal/plugins/plugins_test.go
++++ b/internal/plugins/plugins_test.go
@@@ -11,6 +11,7 @@ import (
  	"path/filepath"
  	"strings"
  	"sync"
++	"syscall"
  	"testing"
  	"time"
  
@@@ -301,3 +302,45 @@ func TestConcurrentCommands(t *testing.T) {
  	}
  	wg.Wait()
  }
++
++// A plugin that dies while a child still holds its stdout is noticed at once, and the child is killed.
++func TestCrashWithChildHoldingStdout(t *testing.T) {
++	if _, err := os.Stat("/bin/sh"); err != nil {
++		t.Skip("no /bin/sh")
++	}
++	dir := t.TempDir()
++	pidFile := filepath.Join(dir, "child.pid")
++	script := `sleep 30 & echo $! > ` + pidFile + `
++echo '{"op":"describe","driver":"sh","capabilities":{}}'
++echo '{"op":"ready"}'
++read line
++exit 1`
++	m := NewManager([]Spec{{Name: "sh", Argv: []string{"/bin/sh", "-c", script}}}, slog.New(slog.NewTextHandler(io.Discard, nil)))
++	ctx, cancel := context.WithCancel(context.Background())
++	defer func() { cancel(); m.Wait() }()
++	go func() {
++		for range m.Output() {
++		}
++	}()
++	m.Start(ctx)
++	deadline := time.Now().Add(3 * time.Second)
++	for m.Get("sh").Info().Restarts < 1 {
++		if time.Now().After(deadline) {
++			t.Fatalf("crash not noticed: %+v", m.Get("sh").Info())
++		}
++		time.Sleep(10 * time.Millisecond)
++	}
++	b, _ := os.ReadFile(pidFile)
++	var pid int
++	fmt.Sscan(string(b), &pid)
++	if pid > 0 {
++		time.Sleep(100 * time.Millisecond)
++		if p, err := os.FindProcess(pid); err == nil && p.Signal(syscall.Signal(0)) == nil {
++			// Zombie children of the killed shell are reaped by init; a live one is a failure.
++			stat, _ := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid))
++			if !strings.Contains(string(stat), ") Z ") && len(stat) > 0 {
++				t.Fatalf("child %d still running", pid)
++			}
++		}
++	}
++}
ddiff --git a/internal/safety/latch.go b/internal/safety/latch.go
iindex 3135abb..62e1f5a 100644
---- a/internal/safety/latch.go
++++ b/internal/safety/latch.go
@@@ -37,6 +37,9 @@ type Latch struct {
  	state    LatchState
  	lastRaw  []byte
  	onChange []func(LatchState)
++	// cbMu runs change callbacks one at a time, each with the state current when it runs, so a slow callback can
++	// never apply an older state after a newer one.
++	cbMu sync.Mutex
  }
  
  // OpenLatch reads the latch file (missing = clear). A corrupt file is treated as latched: failing safe.
@@@ -73,7 +76,8 @@ func (l *Latch) State() LatchState {
  	return l.state
  }
  
--// OnChange registers a callback run (outside the lock) after every change.
++// OnChange registers a callback run (outside the state lock, one at a time, with the current state) after every
++// change.
  func (l *Latch) OnChange(f func(LatchState)) {
  	l.mu.Lock()
  	l.onChange = append(l.onChange, f)
@@@ -86,16 +90,25 @@ func (l *Latch) update(f func(*LatchState)) (LatchState, error) {
  	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)
--		}
++		l.notify()
  	}
  	return st, err
  }
  
++func (l *Latch) notify() {
++	l.cbMu.Lock()
++	defer l.cbMu.Unlock()
++	l.mu.Lock()
++	st := l.state
++	cbs := append([]func(LatchState){}, l.onChange...)
++	l.mu.Unlock()
++	for _, cb := range cbs {
++		cb(st)
++	}
++}
++
  func (l *Latch) persistLocked() error {
  	b, err := json.Marshal(l.state)
  	if err != nil {
@@@ -148,8 +161,9 @@ func (l *Latch) Resume() (LatchState, error) {
  // 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)
++	// Read under the lock: an update between our read and our adoption would otherwise be overwritten by stale bytes.
  	l.mu.Lock()
++	st, raw, err := readLatch(l.path)
  	if err != nil {
  		if raw == nil || string(raw) == string(l.lastRaw) {
  			l.mu.Unlock()
@@@ -163,12 +177,9 @@ func (l *Latch) Reload() (bool, error) {
  	}
  	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)
--		}
++		l.notify()
  	}
  	return changed, err
  }
ddiff --git a/internal/safety/safety_test.go b/internal/safety/safety_test.go
iindex 0500447..c8d92d6 100644
---- a/internal/safety/safety_test.go
++++ b/internal/safety/safety_test.go
@@@ -152,3 +152,25 @@ func TestDedup(t *testing.T) {
  		t.Fatal("ttl not enforced")
  	}
  }
++
++// A stop must never be undone by a concurrent reload of the (older) file.
++func TestReloadNeverUndoesConcurrentStop(t *testing.T) {
++	for i := 0; i < 300; i++ {
++		p := filepath.Join(t.TempDir(), "latch.json")
++		l, _ := OpenLatch(p)
++		l.Resume()
++		done := make(chan struct{})
++		go func() { l.Reload(); close(done) }()
++		l.SetLocal()
++		<-done
++		l.Reload()
++		if !l.State().Local {
++			t.Fatalf("iteration %d: local stop undone by a reload", i)
++		}
++		l.SetRemote("x")
++		l2, _ := OpenLatch(p)
++		if !l2.State().Local {
++			t.Fatalf("iteration %d: local stop lost on disk", i)
++		}
++	}
++}
ddiff --git a/internal/video/whip.go b/internal/video/whip.go
iindex df69467..72dc9cd 100644
---- a/internal/video/whip.go
++++ b/internal/video/whip.go
@@@ -298,7 +298,7 @@ func (p *Publisher) post(ctx context.Context, sdp string) (answer, resource stri
  	defer resp.Body.Close()
  	body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
  	if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusOK {
--		return "", "", fmt.Errorf("whip: HTTP %d %s", resp.StatusCode, bytes.TrimSpace(body[:min(len(body), 200)]))
++		return "", "", fmt.Errorf("whip: HTTP %d %s", resp.StatusCode, scrub(string(bytes.TrimSpace(body[:min(len(body), 200)])), p.opt.PublishKey))
  	}
  	loc := resp.Header.Get("Location")
  	if loc != "" {
@@@ -321,7 +321,8 @@ func (p *Publisher) delete(resource string) {
  	if err != nil {
  		return
  	}
--	if !p.opt.PublishKey.IsZero() {
++	// The publish key goes only to the WHIP endpoint's own origin, whatever Location the server returned.
++	if !p.opt.PublishKey.IsZero() && sameOrigin(resource, p.opt.WHIPURL) {
  		req.Header.Set("Authorization", "Bearer "+p.opt.PublishKey.Reveal())
  	}
  	if resp, err := p.opt.HTTPClient.Do(req); err == nil {
@@@ -329,6 +330,12 @@ func (p *Publisher) delete(resource string) {
  	}
  }
  
++func sameOrigin(a, b string) bool {
++	ua, err1 := url.Parse(a)
++	ub, err2 := url.Parse(b)
++	return err1 == nil && err2 == nil && ua.Scheme == ub.Scheme && strings.EqualFold(ua.Host, ub.Host)
++}
++
  func redactURL(s string) string {
  	u, err := url.Parse(s)
  	if err != nil {
ddiff --git a/internal/video/whip_test.go b/internal/video/whip_test.go
iindex d7396b0..4131299 100644
---- a/internal/video/whip_test.go
++++ b/internal/video/whip_test.go
@@@ -72,3 +72,11 @@ func contains(s, sub string) bool {
  	}
  	return false
  }
++
++func TestSameOrigin(t *testing.T) {
++	if !sameOrigin("https://re.example/whip/a/s1", "https://re.example/whip/a") ||
++		sameOrigin("https://evil.example/x", "https://re.example/whip/a") ||
++		sameOrigin("http://re.example/x", "https://re.example/whip/a") {
++		t.Fatal("sameOrigin wrong")
++	}
++}
ddiff --git a/plugins/sdk/openvibe_plugin/__init__.py b/plugins/sdk/openvibe_plugin/__init__.py
iindex 73d726c..4755f08 100644
---- a/plugins/sdk/openvibe_plugin/__init__.py
++++ b/plugins/sdk/openvibe_plugin/__init__.py
@@@ -195,16 +195,19 @@ class Runtime:
                  self.send(dict(op="telemetry", **t))
  
      # Input.
--    def dispatch(self, msg):
++    def dispatch(self, msg, at=None):
++        """Handle one message. at is when the line arrived (defaults to now): deadlines and heartbeats count from
++        arrival, so a backlog after a slow handler cannot make stale commands or heartbeats look fresh."""
++        at = self.clock() if at is None else at
          op = msg.get("op")
          if op == "hello":
              self.on_hello(msg)
          elif op == "heartbeat":
--            self.last_heartbeat = self.clock()
++            self.last_heartbeat = max(self.last_heartbeat or at, at)
              if self.heartbeat_lost:
                  self.heartbeat_lost = False
          elif op == "command":
--            self.on_command(msg)
++            self.on_command(msg, at)
          elif op == "stop":
              self.safe_stop("stop")
          elif op == "estop":
@@@ -240,10 +243,10 @@ class Runtime:
          except Exception as e:
              self.send({"op": "fault", "fault_code": "hardware", "message": str(e)})
  
--    def on_command(self, msg):
++    def on_command(self, msg, now=None):
          cid = msg.get("id")
          kind = msg.get("kind")
--        now = self.clock()
++        now = self.clock() if now is None else now
          try:
              dl = int(msg.get("deadline_ms") or DEFAULT_DEADLINE_MS)
          except (TypeError, ValueError):
@@@ -279,7 +282,7 @@ class Runtime:
              self.safe_stop("handler_error")
              self.send({"op": "nack", "id": cid, "fault_code": "hardware", "message": str(e)})
  
--    def feed_line(self, line):
++    def feed_line(self, line, at=None):
          line = line.strip()
          if not line:
              return
@@@ -289,7 +292,7 @@ class Runtime:
              self.plugin.log("bad json from core: %r", line[:200])
              return
          if isinstance(msg, dict):
--            self.dispatch(msg)
++            self.dispatch(msg, at)
  
      def shutdown(self, reason="eof"):
          if self.closed:
@@@ -305,7 +308,7 @@ class Runtime:
      def _reader(self):
          try:
              for line in self.infile:
--                self.inbox.put(line)
++                self.inbox.put((line, self.clock()))
          except Exception:
              pass
          self.inbox.put(None)
@@@ -316,13 +319,14 @@ class Runtime:
          try:
              while True:
                  try:
--                    line = self.inbox.get(timeout=0.02)
++                    item = self.inbox.get(timeout=0.02)
                  except queue.Empty:
--                    line = ""
--                if line is None:
++                    item = ("", None)
++                if item is None:
                      break
++                line, at = item
                  if line:
--                    self.feed_line(line)
++                    self.feed_line(line, at)
                  self.tick()
          finally:
              self.shutdown("eof")
ddiff --git a/plugins/sdk/tests/test_runtime.py b/plugins/sdk/tests/test_runtime.py
iindex 1b5cc72..f701cf2 100644
---- a/plugins/sdk/tests/test_runtime.py
++++ b/plugins/sdk/tests/test_runtime.py
@@@ -107,6 +107,18 @@ class RuntimeTest(unittest.TestCase):
          rt.tick()
          self.assertFalse(p.moving)
  
++    def test_backlog_counts_from_arrival(self):
++        # A command that waited in the inbox while a handler blocked keeps its arrival-based deadline.
++        rt, p, out, clock = make()
++        self.hello(rt)
++        arrived = clock.t
++        clock.t += 0.5  # the handler was stuck for 500 ms
++        rt.feed_line(json.dumps({"op": "heartbeat", "t": 1}), arrived)
++        rt.feed_line(json.dumps({"op": "command", "id": "old", "kind": "drive", "deadline_ms": 300}), arrived)
++        rt.tick()
++        self.assertFalse(p.moving)
++        self.assertIn("deadline", rt.stop_reasons)
++
      def test_heartbeat_loss_stops_and_refuses(self):
          rt, p, out, clock = make()
          self.hello(rt)