openvibers/openvibe.link Commits

v0.17.0: deadlines against a bounded clock; every stop names its cause (on_stop / StopInfo) (#63)

OP OpenVibers committed 17 days ago · 632207fa5d · 15 files, +603 −86

* v0.17.0: deadlines count from the latest the server's clock can be (estimate + the connection's fastest round trip) and motion waits for that bound; every stop and estop names its cause, hello carries the newest halt epoch, and drivers get on_stop(StopInfo) and on_resume

* TestRunDryRun: send the first drive again while it is refused not_ready (motion waits for the connection's first round trip)

Changes

ddiff --git a/STATUS.json b/STATUS.json
iindex 901bf83..db5683f 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -5,8 +5,8 @@
"wave": "T15",
"plan": "Plan track T15 (Bot complete) and ADR-043 (Bot: devices, pairing, control and safety); the computer plugin and the job worker serve OpenVibe.Actor and the Codes job runner (tracks T14/T17)",
"kind": "client",
-- "version": "0.16.1",
-- "release": "v0.16.1, published 2026-10-27 as a GitHub release (the Python runtime refuses a motion command whose deadline passed while it waited behind a slow handler, so a driver never moves on a stale command; plugins.md answers where deadlines come from, re-stamping, order across a change of driver, stop versus estop, and that nothing is held past its validity; v0.16.0: openvibe-node quit ends a running node cleanly with nothing latched, for launchers; each driver command carries its provenance (link session, seq, Bot send time and deadline) and a halt's stop its turn_epoch, and the Python runtime refuses a replay on a session and a command from an ended turn; the no-redelivery contract written down in plugins.md): binaries for linux amd64/arm64/armv7, darwin amd64/arm64 and windows amd64, the plugins tarball, install.sh and SHA256SUMS",
++ "version": "0.17.0",
++ "release": "v0.17.0, published 2026-10-27 as a GitHub release (deadlines are checked against the latest the server's clock can be: the estimate plus its bound, the connection's fastest round trip, and motion waits for that bound on every new connection; every stop and estop names its cause, hello carries the newest halt epoch, and the Python SDK gives drivers on_stop(StopInfo) and on_resume to tell an ordinary stop from an emergency without re-arming anything; v0.16.1: the Python runtime refuses a motion command whose deadline passed while it waited behind a slow handler, so a driver never moves on a stale command; plugins.md answers where deadlines come from, re-stamping, order across a change of driver, stop versus estop, and that nothing is held past its validity; v0.16.0: openvibe-node quit ends a running node cleanly with nothing latched, for launchers; each driver command carries its provenance (link session, seq, Bot send time and deadline) and a halt's stop its turn_epoch, and the Python runtime refuses a replay on a session and a command from an ended turn; the no-redelivery contract written down in plugins.md): binaries for linux amd64/arm64/armv7, darwin amd64/arm64 and windows amd64, the plugins tarball, install.sh and SHA256SUMS",
"runtime": "go 1.24 (one static binary per platform); plugins in Python 3; an ESP32 Arduino library",
"database": "none",
"deployed": false,
ddiff --git a/cmd/openvibe-node/main_test.go b/cmd/openvibe-node/main_test.go
iindex cc96ece..34f964a 100644
---- a/cmd/openvibe-node/main_test.go
++++ b/cmd/openvibe-node/main_test.go
@@@ -7,6 +7,7 @@ import (
"os/exec"
"path/filepath"
"runtime"
++ "strconv"
"strings"
"syscall"
"testing"
@@@ -251,8 +252,18 @@ func TestRunDryRun(t *testing.T) {
}
return r
}
-- if _, ok := send("c1", "drive", `{"throttle":0.4}`).(protocol.Ack); !ok {
-- t.Fatal("drive not acked")
++ // Timed motion waits for the connection's first measured round trip (v0.17.0): a drive sent the moment the driver is
++ // ready can still be refused not_ready, as Bot would see it, so send again (a new id each time) until it runs.
++ var first protocol.Message
++ for i := 0; i < 40; i++ {
++ first = send("c1-"+strconv.Itoa(i), "drive", `{"throttle":0.4}`)
++ if n, ok := first.(protocol.Nack); !ok || n.FaultCode != protocol.FaultNotReady {
++ break
++ }
++ time.Sleep(50 * time.Millisecond)
++ }
++ if _, ok := first.(protocol.Ack); !ok {
++ t.Fatalf("drive not acked: %+v", first)
}
out, errOut, code := cli(t, "status", "--home", home)
if code != 0 || !strings.Contains(out, "Link: up") || !strings.Contains(out, "dryrun") || strings.Contains(out, c.Credential.Reveal()) {
ddiff --git a/docs/plugins.md b/docs/plugins.md
iindex 0bcc317..6b4b09c 100644
---- a/docs/plugins.md
++++ b/docs/plugins.md
@@@ -47,12 +47,12 @@ describes itself and waits for EOF without opening any hardware.
| op | fields | the plugin must |
|-------------|---------------------------------------------------------------|--------------------------------------------------------|
--| `hello` | `config` {…}, `probe` bool | answer `describe`; then set up (unless probe) |
++| `hello` | `config` {…}, `probe` bool, `halt_epoch`? (the newest a halt carried, 0.17) | answer `describe`; then set up (unless probe) |
| `command` | `id`, `kind`, `value`, `deadline_ms`, `operator`?, `turn_epoch`?, `provenance`? | act, answer `ack` or `nack` with the same `id` |
| `heartbeat` | `t` (core clock, ms) | note the time |
--| `stop` | `turn_epoch`? (a halt's) | stop every actuator |
--| `estop` | | stop every actuator and refuse motion until `resume` |
--| `resume` | | accept motion again |
++| `stop` | `cause` (0.17), `turn_epoch`? (a halt's) | stop every actuator; latch nothing |
++| `estop` | `cause`, `reason`?, `replayed`?, `first_cause`? (0.17) | stop every actuator and refuse motion until `resume` |
++| `resume` | `cause` (`cleared`, 0.17) | accept motion again; never re-arm anything of its own |
| `approval` | `id`, `approved` bool | the owner's answer to an `approval` event it raised |
| `tray` | `attached` bool, `ttl_s` | (computer plugin only) the tray shows its approvals for `ttl_s` s unless told again: its own desktop prompts may stay quiet |
| `voice` | `on` bool, `call_word` | (plugins that describe a `voice` capability) the owner's Vibe by voice setting from Bot's config: listen for `call_word` only while `on` ([voice.md](voice.md)) |
@@@ -85,6 +85,7 @@ and refuses an older command itself (`expired`).
| `seq` | its frame's number on that session: strictly increasing, with gaps (other frames share the count) |
| `issued_at_ms` | Bot's clock when it sent the command, which is when it minted the id (`cmd_<ULID>`, whose time part is the same) |
| `expires_at_ms` | the absolute deadline on Bot's clock (motion kinds); `deadline_ms` is what is left of it at receipt |
++| `clock_uncertainty_ms` | (0.17) how far the core's estimate of Bot's clock may trail it: the connection's fastest round trip. `deadline_ms` already counts from estimate + this |
Python: `cmd.provenance` (a dict, or `None` from an older Node). The runtime refuses (`expired`) a command whose `seq`
is not above the last one it handled on the same `session`.
@@@ -131,9 +132,10 @@ Asked by the Generic LOOI driver's authors (2026-10-27). Each answer is what Bot
- Only `drive`, `actuator`, `ptz` and a held button's `down` get one.
- The command is sent in the same step. Its id (`cmd_<ULID>`) and `issued_at_ms` are minted then, after the gate's
last await.
-- - The core turns the absolute instant into what is left of it. It compares the instant with its estimate of Bot's
-- clock, which is the local monotonic clock plus the largest `ts − arrival` seen on the connection. So time a frame
-- spent queued on the way counts as lateness, and a delayed run of frames can never make a deadline look open again.
++ - The core turns the absolute instant into what is left of it. It compares the instant with the latest Bot's clock
++ can be: its estimate (the local monotonic clock plus the largest `ts − arrival` seen on the connection) plus that
++ estimate's bound (answer 6). So time a frame spent queued on the way counts as lateness, and a delayed run of
++ frames can never make a deadline look open again.
- It caps what is left at the robot's `max_command_ms`.
- The plugin gets that remainder as `deadline_ms`, and `expires_at_ms` in `provenance` unchanged.
- Neither a viewer nor Live can choose a later deadline.
@@@ -189,6 +191,59 @@ Asked by the Generic LOOI driver's authors (2026-10-27). Each answer is what Bot
- A driver without the runtime should do the same: note the time each line is read (on a reader thread) and refuse
motion that is already past its `deadline_ms` when its turn comes.
++Asked next by the same authors (2026-10-27, P7 LOOI shadow integration), and changed in Node 0.17.0 to answer them:
++
++6. **The clock's error, its bound, and when motion waits for it.**
++ - **The error.** Each frame from Bot gives one sample, `ts − arrival` = the clock offset less that frame's one-way
++ transit. The estimate keeps the largest sample on the connection, so it trails Bot's real clock by the fastest
++ one-way transit among them. That is a late reading: on its own it would grant a command up to that transit more
++ than its deadline allows.
++ - **The bound.** Every `heartbeat_ack` is one of those samples, and its one-way transit is at most its own round
++ trip. So the connection's fastest heartbeat round trip bounds the error, whatever the asymmetry of the path.
++ The core checks every deadline against estimate + bound (the latest Bot's clock can be), rounded up to the
++ millisecond. This is fail-closed: a command can only get less time than its deadline allows, never more.
++ - **What else holds.** Clock drift over a connection is tracked by the newest samples or reads early (fail-closed).
++ The local clock is monotonic, so a wall-clock step on the device changes nothing. Bot's clock stepping back reads
++ as early until the next connection; stepping forward moves the estimate with the next frame. Latency and jitter
++ only make commands arrive later, so they read as expired, never as open. The window a plugin gets is also capped
++ at the robot's `max_command_ms`, and two missed heartbeats end the link (`stop`, `link_lost`).
++ - **Before a bound exists.** The first heartbeat goes out as soon as Bot's hello arrives, and its ack sets the bound
++ (one round trip, usually tens of milliseconds). Until then the core refuses `drive`, `actuator`, `ptz`, `button` and
++ `point` that carry a deadline (`not_ready`, "the link's clock is not synchronized yet"). Every new connection starts
++ unsynchronized. A link whose fastest round trip is over 1 s (`MaxClockUncertaintyMS`) times no motion either
++ (`not_ready`). Speech and other non-motion kinds are not held.
++ - **A stricter cap of your own.** The bound reaches the plugin as `provenance.clock_uncertainty_ms`. A driver may refuse
++ motion above a cap it chooses (say 150 ms for P7) without inventing another clock or replay mechanism. The deadline,
++ epoch and seq contract stays as it is.
++7. **Why a plugin is stopped (`cause`), and how a driver tells an ordinary stop from an emergency.**
++ - Every `stop` and `estop` names its cause, and `hello` carries the newest halt epoch, so the meaning holds across a
++ plugin restart and a Node reconnect.
++
++ | op | `cause` | when | latches |
++ |---|---|---|---|
++ | `stop` | `halt` | Bot's halt: a turn ended, a viewer handoff, a release, a timeout, an unbind (with `turn_epoch`) | no |
++ | `stop` | `link_lost` | the link to Bot dropped (two missed heartbeats, a close) | no |
++ | `stop` | `shutdown` | the Node is stopping (service stop, `openvibe-node quit`), or the core is stopping this process | no |
++ | `stop` | `timeout` | a command got no answer from the driver within its window | no |
++ | `estop` | `remote` | the platform's e-stop (a robot owner or staff), with its `reason` | yes, until the owner clears it |
++ | `estop` | `local` | `openvibe-node stop` on the device | yes, until `openvibe-node resume` |
++ | `estop` | `driver` | a driver's own `local_stop` (this plugin's or another's) | yes, until `openvibe-node resume` |
++ | `estop` | `latched` | the latch was already set when the Node or this process started (`replayed: true`, `first_cause` when known) | yes |
++ | `resume` | `cleared` | the owner cleared the platform's latch | clears the core's and the runtime's refusal only |
++
++ - **Python (`openvibe_plugin` 0.17).** Every stop the runtime makes calls `on_stop(info)`, and `info` is a `StopInfo`:
++ `cause`, `emergency` (true for `estop`), `latched`, `reason`, `turn_epoch`, `halt_epoch`, `replayed`, `first_cause`,
++ `motion_only`. The runtime's own causes are `setup` (the stop every setup ends with), `deadline` (motion only: stop
++ what moves on its own), `heartbeat` (the core went quiet), `handler_error` and `eof`. An older core sends no cause:
++ `stop` or `estop`. The default `on_stop` calls `stop_motion()` for a deadline and `stop()` otherwise, so existing
++ drivers behave as before. `on_resume(info)` does nothing by default.
++ - **For P7.** Override `on_stop`: stop every motor in every case. Latch P7's own STOP only when `info.emergency` is
++ true; an ordinary halt, a disconnect, a timeout or a handoff only revokes remote motion. Treat `on_resume` as
++ information. Never ARM P7, clear its physical latch or restore a remote grant there; the next command still passes
++ every check. Across a plugin restart the latch arrives again as `estop` `latched` with `replayed: true`, and the halt
++ epoch comes in `hello`. Across a Node reconnect the link's loss is a `stop` (`link_lost`); a platform e-stop set
++ while offline is applied from Bot's config before any command.
++
### Plugin → core
| op | fields |
@@@ -368,6 +423,19 @@ if __name__ == "__main__":
raise SystemExit(run(Arm()))
```
++A driver with a safety latch of its own tells an ordinary stop from an emergency with `on_stop` (Node 0.17; "Why a
++plugin is stopped" above):
++
++```python
++ def on_stop(self, info): # every stop: a halt, a lost link, a deadline, an e-stop…
++ self.stop() # always stop every actuator first
++ if info.emergency: # the platform's e-stop (remote, local, driver, or replayed as latched)
++ self.hardware_latch.set() # only then latch the hardware's own STOP
++
++ def on_resume(self, info): # the owner cleared the platform's latch: information, never a re-arm
++ self.log("platform latch cleared; the hardware latch stays until the person on site clears it")
++```
++
Test it with `openvibe_plugin.Runtime(plugin, infile, outfile, clock)` and a fake clock, as the bundled plugins do.
## Where OpenVibe.Actor attaches (later)
ddiff --git a/internal/fakebot/fakebot.go b/internal/fakebot/fakebot.go
iindex 2849b31..1786b33 100644
---- a/internal/fakebot/fakebot.go
++++ b/internal/fakebot/fakebot.go
@@@ -42,6 +42,9 @@ type Server struct {
PairBody []byte
// WHIP, when set, serves POST/DELETE under /whip/.
WHIP http.Handler
++ // HoldHeartbeatAcks makes new connections leave heartbeats unanswered until Conn.SetAutoAck(true): the device has
++ // no measured round trip, so no bound on its clock estimate.
++ HoldHeartbeatAcks bool
mu sync.Mutex
codes map[string]string // code → robot id
@@@ -329,7 +332,7 @@ func (s *Server) device(w http.ResponseWriter, r *http.Request) {
old.CloseWith(protocol.CloseReplaced, "replaced by a newer connection")
}
s.count.Add(1)
-- c.autoAck.Store(true)
++ c.autoAck.Store(!s.HoldHeartbeatAcks)
if greeting != nil {
for _, b := range greeting {
_ = c.SendRaw(b)
@@@ -469,6 +472,9 @@ func (c *Conn) SendRaw(b []byte) error {
return c.ws.WriteMessage(websocket.TextMessage, b)
}
++// SetAutoAck turns the automatic heartbeat_ack answers on or off for this connection.
++func (c *Conn) SetAutoAck(on bool) { c.autoAck.Store(on) }
++
// Mute stops answering heartbeats and sending anything, as a hung server would.
func (c *Conn) Mute() { c.muted.Store(true); c.autoAck.Store(false) }
ddiff --git a/internal/link/link.go b/internal/link/link.go
iindex 8afe5c7..2cedad7 100644
---- a/internal/link/link.go
++++ b/internal/link/link.go
@@@ -110,6 +110,7 @@ type Link struct {
connected atomic.Bool
lastRTT atomic.Int64
++ minRTT atomic.Int64 // the fastest heartbeat round trip on this connection (ns), 0 until one is measured
reconnects atomic.Int64
lastErr atomic.Value // string
connectedAt atomic.Int64
@@@ -181,7 +182,8 @@ func (l *Link) localMS(at time.Time) int64 {
// ServerNow estimates the server's clock (epoch ms) at local time at, from the frames of the current connection. The
// offset is the largest ts − receive time seen on the connection: the clock skew less the fastest transit, so a frame
--// that sat in a queue on the way reads as late by however long it sat there. The best sample is kept for the whole
++// that sat in a queue on the way reads as late by however long it sat there. The estimate itself trails the server by
++// that fastest transit; ClockUncertainty bounds it, and deadlines are checked against estimate + bound. The best sample is kept for the whole
// connection: a run of delayed frames can never make an expired deadline look open again. A server clock that steps
// back only makes deadlines read as expired (fail closed) until the next connection. ok is false before any frame.
func (l *Link) ServerNow(at time.Time) (ms int64, ok bool) {
@@@ -193,6 +195,18 @@ func (l *Link) ServerNow(at time.Time) (ms int64, ok bool) {
return l.localMS(at) + l.skew, true
}
++// ClockUncertainty bounds how far ServerNow can trail the server's real clock on this connection: the fastest
++// heartbeat round trip measured on it. ServerNow trails the server by the fastest one-way transit among the frames it
++// sampled; every heartbeat_ack is one of those frames, and its one-way transit is at most its own round trip, so the
++// fastest round trip is an upper bound whatever the asymmetry of the path. ok is false until a round trip is measured
++// (the first heartbeat goes out as soon as the server's hello arrives).
++func (l *Link) ClockUncertainty() (time.Duration, bool) {
++ if v := l.minRTT.Load(); v > 0 {
++ return time.Duration(v), true
++ }
++ return 0, false
++}
++
// Connected reports whether a connection is up.
func (l *Link) Connected() bool { return l.connected.Load() }
@@@ -376,8 +390,10 @@ func (l *Link) session(ctx context.Context) error {
l.mu.Lock()
l.out, l.seq, l.hbSent, l.hasSkew = nil, 0, map[uint64]hbSend{}, false
l.mu.Unlock()
-- // A fresh connection has no measured round trip: the first heartbeat must not claim one.
++ // A fresh connection has no measured round trip: the first heartbeat must not claim one, and its clock estimate
++ // has no bound until one is measured (ClockUncertainty).
l.lastRTT.Store(0)
++ l.minRTT.Store(0)
sessCtx, cancel := context.WithCancel(ctx)
var endErr error
@@@ -515,7 +531,14 @@ func (l *Link) session(ctx context.Context) error {
l.mu.Unlock()
if !sent.IsZero() {
// A coarse clock (Windows) can read 0 on a fast loopback; 0 means "not measured yet", so floor at 1 ns.
-- l.lastRTT.Store(int64(max(time.Since(sent), 1)))
++ rtt := int64(max(time.Since(sent), 1))
++ l.lastRTT.Store(rtt)
++ for {
++ cur := l.minRTT.Load()
++ if (cur != 0 && cur <= rtt) || l.minRTT.CompareAndSwap(cur, rtt) {
++ break
++ }
++ }
}
continue
case protocol.Error:
ddiff --git a/internal/link/link_test.go b/internal/link/link_test.go
iindex cc3e265..d0f9bbb 100644
---- a/internal/link/link_test.go
++++ b/internal/link/link_test.go
@@@ -201,6 +201,71 @@ func TestHeartbeatRTTReportedAfterAck(t *testing.T) {
}
}
++// The clock estimate's bound is the connection's fastest round trip: none before the first heartbeat_ack, the
++// minimum after several, and none again on a new connection until its own first ack.
++func TestClockUncertaintyIsTheFastestRoundTrip(t *testing.T) {
++ // 400 ms heartbeats: the muted server answers only these acks, well inside the 800 ms deadman.
++ srv, l, rec, _, _ := setup(t, 400*time.Millisecond)
++ c, err := srv.NextConn(3 * time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ waitFrame(t, rec, protocol.TypeConfig)
++ c.Mute()
++ if _, ok := l.ClockUncertainty(); ok {
++ t.Fatal("a bound before any round trip")
++ }
++ ack := func(delay time.Duration) time.Duration {
++ f, err := c.Expect(protocol.TypeHeartbeat, 2*time.Second, nil)
++ if err != nil {
++ t.Fatal(err)
++ }
++ echo := f.Msg.(protocol.Heartbeat).T
++ time.Sleep(delay)
++ before := l.Stats().RTT
++ if err := c.Send(protocol.HeartbeatAck{Echo: &echo}); err != nil {
++ t.Fatal(err)
++ }
++ deadline := time.Now().Add(2 * time.Second)
++ for l.Stats().RTT == before {
++ if time.Now().After(deadline) {
++ t.Fatal("the ack set no RTT")
++ }
++ time.Sleep(2 * time.Millisecond)
++ }
++ return l.Stats().RTT
++ }
++ slow := ack(80 * time.Millisecond)
++ u, ok := l.ClockUncertainty()
++ if !ok || u != slow || u < 80*time.Millisecond {
++ t.Fatalf("after one slow round trip: %v %v (rtt %v)", u, ok, slow)
++ }
++ fast := ack(0)
++ if u, _ := l.ClockUncertainty(); u != fast || fast >= slow {
++ t.Fatalf("the faster round trip did not tighten the bound: %v (fast %v, slow %v)", u, fast, slow)
++ }
++ ack(60 * time.Millisecond)
++ if u, _ := l.ClockUncertainty(); u != fast {
++ t.Fatalf("a slower round trip loosened the bound: %v (fast %v)", u, fast)
++ }
++ c.Close()
++ c2, err := srv.NextConn(3 * time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ c2.Mute()
++ deadline := time.Now().Add(2 * time.Second)
++ for {
++ if _, ok := l.ClockUncertainty(); !ok {
++ break
++ }
++ if time.Now().After(deadline) {
++ t.Fatal("a new connection kept the old connection's bound")
++ }
++ time.Sleep(2 * time.Millisecond)
++ }
++}
++
// An ack that carries only echo (Bot's newer field, no t) still sets the RTT.
func TestHeartbeatAckEchoOnlySetsRTT(t *testing.T) {
srv, l, rec, _, _ := setup(t, 200*time.Millisecond)
ddiff --git a/internal/node/botcontract_test.go b/internal/node/botcontract_test.go
iindex 3b5d2d4..4f6ee5c 100644
---- a/internal/node/botcontract_test.go
++++ b/internal/node/botcontract_test.go
@@@ -405,8 +405,12 @@ func TestProvenanceReachesPlugin(t *testing.T) {
if pa.IssuedAtMS <= 0 || pa.ExpiresAtMS != pa.IssuedAtMS+300 {
t.Fatalf("times: %+v", pa)
}
++ if pa.ClockUncertaintyMS == nil || *pa.ClockUncertaintyMS < 0 || *pa.ClockUncertaintyMS > protocol.MaxClockUncertaintyMS {
++ t.Fatalf("clock uncertainty: %+v", pa)
++ }
e.conn.Close()
e.conn = e.nextConn()
++ e.waitSynced()
if err := e.conn.Command(protocol.Command{ID: "p3", Kind: protocol.KindDrive, Value: drive}, 300); err != nil {
t.Fatal(err)
}
@@@ -417,3 +421,43 @@ func TestProvenanceReachesPlugin(t *testing.T) {
t.Fatalf("a new connection, a new session: %+v", c)
}
}
++
++// Motion is timed only on a bounded clock (Maticus, 2026-10-27): before the connection's first measured round trip a
++// drive with a deadline is refused as not_ready and never reaches the plugin; once one is measured, the same drive runs,
++// with the bound in its provenance, and its window is the deadline less that bound.
++func TestMotionWaitsForABoundedClock(t *testing.T) {
++ e := startWith(t, func(s *fakebot.Server) { s.HoldHeartbeatAcks = true }, true)
++ drive := json.RawMessage(`{"throttle":0.3}`)
++ if err := e.conn.Command(protocol.Command{ID: "unsynced", Kind: protocol.KindDrive, Value: drive}, 300); err != nil {
++ t.Fatal(err)
++ }
++ r, err := e.conn.Reply("unsynced", 5*time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if n, ok := r.(protocol.Nack); !ok || n.FaultCode != protocol.FaultNotReady || !strings.Contains(n.Message, "not synchronized") {
++ t.Fatalf("before a round trip: %+v", r)
++ }
++ if got := e.commandRecords("unsynced"); len(got) != 0 {
++ t.Fatalf("an unsynchronized drive reached the plugin: %+v", got)
++ }
++ // A say is not motion: it is not held for the clock.
++ if r := e.cmdID("say-unsynced", protocol.KindSay, `{"text":"hi"}`, 300); !isAck(r) {
++ t.Fatalf("say before a round trip: %+v", r)
++ }
++ e.conn.SetAutoAck(true)
++ e.waitSynced()
++ if err := e.conn.Command(protocol.Command{ID: "synced", Kind: protocol.KindDrive, Value: drive}, 300); err != nil {
++ t.Fatal(err)
++ }
++ if r, err := e.conn.Reply("synced", 5*time.Second); err != nil || !isAck(r) {
++ t.Fatalf("after a round trip: %+v %v", r, err)
++ }
++ got := e.commandRecords("synced")
++ if len(got) != 1 || got[0].Provenance == nil || got[0].Provenance.ClockUncertaintyMS == nil {
++ t.Fatalf("records: %+v", got)
++ }
++ if unc := *got[0].Provenance.ClockUncertaintyMS; got[0].DeadlineMS > 300-int(unc) {
++ t.Fatalf("the window %d ms ignores the clock bound %d ms", got[0].DeadlineMS, unc)
++ }
++}
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex ef76c36..b05e9f8 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -253,9 +253,9 @@ func (n *Node) Run(parent context.Context) error {
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()
++ if st := n.latch.State(); st.Stopped() {
++ n.log.Warn("starting with the stop latched", "remote", st.Remote, "local", st.Local)
++ n.mgr.EstopAll(plugins.EstopLatched, latchReason(st))
}
var wg sync.WaitGroup
@@@ -277,7 +277,7 @@ func (n *Node) Run(parent context.Context) error {
}
<-ctx.Done()
n.log.Info("shutting down: stopping every actuator")
-- n.mgr.StopAll()
++ n.mgr.StopAll(plugins.StopShutdown)
if n.worker != nil {
n.worker.Close()
}
@@@ -302,7 +302,7 @@ func (n *Node) Connected() {
func (n *Node) Disconnected(err error) {
n.log.Warn("link lost: stopping every actuator", "err", err)
-- n.mgr.StopAll()
++ n.mgr.StopAll(plugins.StopLinkLost)
n.mu.Lock()
n.configured = false
n.mu.Unlock()
@@@ -344,7 +344,7 @@ func (n *Node) Frame(f protocol.Frame) {
func (n *Node) remoteEstop(reason, by string) {
n.log.Warn("e-stop from the server", "by", by)
-- n.mgr.EstopAll() // at once, before the disk write
++ n.mgr.EstopAll(plugins.EstopRemote, safety.CleanReason(reason)) // at once, before the disk write
if _, err := n.latch.SetRemote(reason); err != nil {
n.log.Error("could not persist the e-stop (it is held in memory)", "err", err)
}
@@@ -531,13 +531,29 @@ func (n *Node) command(c protocol.Command, sentMS int64, seq uint64, received ti
}
}
deadline := limits.Deadline(0)
++ var uncertainty *int64
if c.DeadlineMS > 0 {
-- left := c.DeadlineMS - n.serverNow(sentMS, received)
++ left, unc, synced := n.timeLeft(c.DeadlineMS, sentMS, received)
++ if protocol.GuardedKinds[c.Kind] {
++ // Motion is timed only on a clock Node can bound (Maticus, 2026-10-27): not before the connection's first
++ // round trip, and not on a link whose fastest round trip is slower than MaxClockUncertaintyMS.
++ if !synced {
++ nack(protocol.FaultNotReady, "the link's clock is not synchronized yet: no round trip has been measured on this connection")
++ return
++ }
++ if unc > protocol.MaxClockUncertaintyMS {
++ nack(protocol.FaultNotReady, fmt.Sprintf("the link is too slow to time motion: its fastest round trip is %d ms (at most %d)", unc, protocol.MaxClockUncertaintyMS))
++ return
++ }
++ }
if left <= 0 {
-- nack(protocol.FaultExpired, fmt.Sprintf("the deadline passed %d ms before the command arrived", -left))
++ nack(protocol.FaultExpired, fmt.Sprintf("the deadline passed %d ms before the command arrived (counting %d ms of clock uncertainty)", -left, unc))
return
}
deadline = limits.Deadline(int(min(left, int64(limits.MaxCommandMS)+1)))
++ if synced && n.link != nil {
++ uncertainty = &unc
++ }
}
value, err := limits.Clamp(c.Kind, c.Value)
if err != nil {
@@@ -563,7 +579,7 @@ func (n *Node) command(c protocol.Command, sentMS int64, seq uint64, received ti
return
}
// Where it came from, unchanged, so a driver can check it without keeping old ids (Maticus, 2026-10-26, LOOI).
-- prov := &plugins.Provenance{Session: session, Seq: seq, IssuedAtMS: sentMS, ExpiresAtMS: c.DeadlineMS}
++ prov := &plugins.Provenance{Session: session, Seq: seq, IssuedAtMS: sentMS, ExpiresAtMS: c.DeadlineMS, ClockUncertaintyMS: uncertainty}
wait := p.BeginWith(n.ctx, c.ID, c.Kind, value, deadline, c.Operator, c.TurnEpoch, prov)
go func() {
r := wait()
@@@ -608,7 +624,7 @@ func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, r
}
window := int64(protocol.DefaultComputerMS)
if c.DeadlineMS > 0 {
-- window = c.DeadlineMS - n.serverNow(sentMS, received)
++ window, _, _ = n.timeLeft(c.DeadlineMS, sentMS, received)
if window <= 0 {
nack(protocol.FaultExpired, fmt.Sprintf("the deadline passed %d ms before the command arrived", -window))
return
@@@ -674,6 +690,33 @@ func (n *Node) haltEpoch(robotID string) int64 {
return n.haltEpochs[robotID]
}
++// timeLeft is how long a command stamped sentMS and received at received has before its deadline_ms, an instant on the
++// server's clock, counted from the LATEST the server's clock can be at receipt: the link's estimate plus its bound
++// (Link.ClockUncertainty, the connection's fastest round trip). unc is that bound in ms; synced is false while no round
++// trip has been measured on the connection (left then counts from the bare estimate). Without a link (local commands)
++// it is synced with no uncertainty.
++func (n *Node) timeLeft(deadlineMS, sentMS int64, received time.Time) (left, unc int64, synced bool) {
++ now := n.serverNow(sentMS, received)
++ if n.link == nil {
++ return deadlineMS - now, 0, true
++ }
++ d, ok := n.link.ClockUncertainty()
++ if !ok {
++ return deadlineMS - now, 0, false
++ }
++ // Round up: a 0.4 ms bound is still a bound.
++ unc = (d.Microseconds() + 999) / 1000
++ return deadlineMS - now - unc, unc, true
++}
++
++// latchReason is the reason a latched stop gives a driver: the remote one when the platform set it, else the local one.
++func latchReason(st safety.LatchState) string {
++ if st.Remote {
++ return st.RemoteReason
++ }
++ return st.LocalReason
++}
++
// serverNow is the server's clock when a frame stamped sentMS was received: the link's estimate, else the frame's
// own ts.
func (n *Node) serverNow(sentMS int64, received time.Time) int64 {
@@@ -705,7 +748,11 @@ func (n *Node) onLatch(st safety.LatchState) {
}
if st.Stopped() {
n.log.Warn("stop latched", "remote", st.Remote, "local", st.Local)
-- n.mgr.EstopAll()
++ cause := plugins.EstopRemote
++ if st.Local {
++ cause = plugins.EstopLocal
++ }
++ n.mgr.EstopAll(cause, latchReason(st))
} else {
n.log.Warn("stop cleared; motion allowed again")
n.mgr.ResumeAll()
@@@ -825,11 +872,11 @@ func (n *Node) output(o plugins.Output) {
case plugins.OutLocalStop:
// A driver's own stop: everything stops at once, then the local latch is persisted with its reason. It is
// never a clear: nothing a plugin says releases a latch.
-- n.mgr.EstopAll()
reason, _ := o.Data["reason"].(string)
if reason = safety.CleanReason(reason); reason == "" {
reason = o.Plugin + " stop"
}
++ n.mgr.EstopAll(plugins.EstopDriver, reason)
if _, err := n.latch.SetLocalWith(reason); err != nil {
n.log.Error("a plugin latched the local stop, but the latch could not be saved", "plugin", o.Plugin, "err", err)
}
@@@ -1224,7 +1271,7 @@ func (n *Node) control(r localctl.Request) localctl.Response {
n.tellTray(r.On)
return localctl.Response{OK: true}
case localctl.CmdStop:
-- n.mgr.EstopAll() // at once, then persist
++ n.mgr.EstopAll(plugins.EstopLocal, "openvibe-node stop") // at once, then persist
if _, err := n.latch.SetLocalWith("openvibe-node stop"); err != nil {
return localctl.Response{OK: true, Error: "stopped, but the latch could not be saved: " + err.Error()}
}
@@@ -1237,7 +1284,7 @@ func (n *Node) control(r localctl.Request) localctl.Response {
"(sudo systemctl stop openvibe-node; on Windows, Services), or remove it with `openvibe-node uninstall`"}
}
n.log.Info("quit by its owner (openvibe-node quit)")
-- n.mgr.StopAll() // at once; Run stops everything again on the way out
++ n.mgr.StopAll(plugins.StopShutdown) // at once; Run stops everything again on the way out
n.quitting.Store(true)
go func() { time.Sleep(100 * time.Millisecond); n.quit() }() // after this answer is written
return localctl.Response{OK: true}
ddiff --git a/internal/node/node_test.go b/internal/node/node_test.go
iindex cbba2f5..90195b4 100644
---- a/internal/node/node_test.go
++++ b/internal/node/node_test.go
@@@ -44,6 +44,10 @@ type env struct {
paths config.Paths
cancel context.CancelFunc
done chan struct{}
++ // holdAcks: the fake Bot leaves heartbeats unanswered (no measured round trip); connAt: when the current
++ // connection began (nextConn), so waitSynced only counts a round trip measured on it.
++ holdAcks bool
++ connAt time.Time
}
// start pairs a Node running the real Python dry-run plugin with the fake Bot server and waits until the plugin is
@@@ -105,11 +109,15 @@ func startWith(t *testing.T, prep func(*fakebot.Server), ready bool) *env {
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}
++ e := &env{t: t, srv: srv, whip: whip, node: n, record: record, paths: paths, cancel: cancel, done: done,
++ holdAcks: srv.HoldHeartbeatAcks}
t.Cleanup(e.stop)
e.conn = e.nextConn()
if ready {
e.waitReady()
++ if !e.holdAcks {
++ e.waitSynced()
++ }
}
return e
}
@@@ -125,6 +133,7 @@ func (e *env) stop() {
func (e *env) nextConn() *fakebot.Conn {
e.t.Helper()
++ e.connAt = time.Now()
c, err := e.srv.NextConn(10 * time.Second)
if err != nil {
e.t.Fatal(err)
@@@ -132,6 +141,24 @@ func (e *env) nextConn() *fakebot.Conn {
return c
}
++// waitSynced waits until the current connection (the one nextConn last returned) has a measured round trip, so timed
++// motion is accepted on it: the link is up since after that connection began, and its clock has a bound.
++func (e *env) waitSynced() {
++ e.t.Helper()
++ deadline := time.Now().Add(10 * time.Second)
++ for {
++ st := e.node.link.Stats()
++ _, ok := e.node.link.ClockUncertainty()
++ if ok && st.Connected && st.Since.UnixMilli() >= e.connAt.UnixMilli()-1 {
++ return
++ }
++ if time.Now().After(deadline) {
++ e.t.Fatal("no round trip measured on the connection")
++ }
++ time.Sleep(5 * time.Millisecond)
++ }
++}
++
func (e *env) waitReady() {
e.t.Helper()
_, err := e.conn.Expect(protocol.TypeStatus, 20*time.Second, func(m protocol.Message) bool {
@@@ -154,6 +181,9 @@ func (e *env) cmd(kind, value string, deadline int) protocol.Message {
func (e *env) cmdID(id, kind, value string, deadline int) protocol.Message {
e.t.Helper()
++ if protocol.GuardedKinds[kind] && !e.holdAcks {
++ e.waitSynced() // motion is timed only on a bounded clock
++ }
if err := e.conn.Command(protocol.Command{ID: id, Kind: kind, Value: json.RawMessage(value)}, deadline); err != nil {
e.t.Fatal(err)
}
@@@ -173,10 +203,11 @@ type rec struct {
DeadlineMS int `json:"deadline_ms"`
WasMoving bool `json:"was_moving"`
Provenance *struct {
-- Session string `json:"session"`
-- Seq uint64 `json:"seq"`
-- IssuedAtMS int64 `json:"issued_at_ms"`
-- ExpiresAtMS int64 `json:"expires_at_ms"`
++ Session string `json:"session"`
++ Seq uint64 `json:"seq"`
++ IssuedAtMS int64 `json:"issued_at_ms"`
++ ExpiresAtMS int64 `json:"expires_at_ms"`
++ ClockUncertaintyMS *int64 `json:"clock_uncertainty_ms"`
} `json:"provenance"`
}
ddiff --git a/internal/plugins/manager.go b/internal/plugins/manager.go
iindex a27dc1b..0e1237e 100644
---- a/internal/plugins/manager.go
++++ b/internal/plugins/manager.go
@@@ -70,10 +70,10 @@ func (m *Manager) Get(name string) *Plugin {
return nil
}
--// StopAll tells every plugin to stop every actuator.
--func (m *Manager) StopAll() {
++// StopAll tells every plugin to stop every actuator, saying why (StopLinkLost, StopShutdown).
++func (m *Manager) StopAll(cause string) {
for _, p := range m.plugins {
-- p.Stop()
++ p.Stop(cause)
}
}
@@@ -84,9 +84,10 @@ func (m *Manager) HaltAll(turnEpoch int64) {
}
}
--func (m *Manager) EstopAll() {
++// EstopAll latches every plugin's e-stop, saying why (EstopRemote, EstopLocal, EstopDriver, EstopLatched).
++func (m *Manager) EstopAll(cause, reason string) {
for _, p := range m.plugins {
-- p.Estop()
++ p.Estop(cause, reason)
}
}
ddiff --git a/internal/plugins/plugin.go b/internal/plugins/plugin.go
iindex baaf670..854e88b 100644
---- a/internal/plugins/plugin.go
++++ b/internal/plugins/plugin.go
@@@ -99,11 +99,15 @@ type Plugin struct {
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
++ // The e-stop's cause and reason, sent again (replayed) to a restarted process while the stop is latched.
++ estopCause, estopReason string
++ // The newest turn_epoch a halt carried: hello tells a restarted process, so a stale turn stays refused there too.
++ haltEpoch int64
++ 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 {
@@@ -182,33 +186,51 @@ func (p *Plugin) send(msg map[string]any) error {
}
}
--// Stop tells the plugin to stop every actuator.
--func (p *Plugin) Stop() { _ = p.send(map[string]any{"op": "stop"}) }
++// Why a plugin is told to stop (the `cause` of its stop and estop ops; docs/plugins.md "Why a plugin is stopped"). A
++// `stop` is ordinary: revoke remote motion, latch nothing. An `estop` is the emergency: the platform's latch is set and
++// stays set until the owner clears it, which a driver only hears as `resume` and must never treat as a re-arm.
++const (
++ StopHalt = "halt" // Bot's halt: the controlling turn ended (release, handoff, timeout, unbind)
++ StopLinkLost = "link_lost" // the link to Bot was lost
++ StopShutdown = "shutdown" // the Node is stopping (service stop, `openvibe-node quit`)
++ StopTimeout = "timeout" // a command got no answer from the driver within its window
++ EstopRemote = "remote" // an e-stop from the platform (a robot owner or staff)
++ EstopLocal = "local" // `openvibe-node stop` on the device
++ EstopDriver = "driver" // a driver's own local stop (a plugin reported one)
++ EstopLatched = "latched" // the stop was already latched when the Node or this process started
++)
++
++// Stop tells the plugin to stop every actuator, saying why.
++func (p *Plugin) Stop(cause string) { _ = p.send(map[string]any{"op": "stop", "cause": cause}) }
// Halt is Stop for a halt command: it carries the halt's turn_epoch (when Bot sent one), the newest epoch a driver
// should take commands from.
func (p *Plugin) Halt(turnEpoch int64) {
-- if turnEpoch <= 0 {
-- p.Stop()
-- return
++ msg := map[string]any{"op": "stop", "cause": StopHalt}
++ if turnEpoch > 0 {
++ msg["turn_epoch"] = turnEpoch
++ p.mu.Lock()
++ p.haltEpoch = max(p.haltEpoch, turnEpoch)
++ p.mu.Unlock()
}
-- _ = p.send(map[string]any{"op": "stop", "turn_epoch": turnEpoch})
++ _ = p.send(msg)
}
--// Estop latches the plugin's e-stop (the core also refuses motion itself).
--func (p *Plugin) Estop() {
++// Estop latches the plugin's e-stop (the core also refuses motion itself), saying why.
++func (p *Plugin) Estop(cause, reason string) {
p.mu.Lock()
-- p.estopped = true
++ p.estopped, p.estopCause, p.estopReason = true, cause, reason
p.mu.Unlock()
-- _ = p.send(map[string]any{"op": "estop"})
++ _ = p.send(map[string]any{"op": "estop", "cause": cause, "reason": reason})
}
--// Resume clears the plugin's e-stop.
++// Resume tells the plugin the platform's e-stop latch was cleared by its owner. It is information, never a re-arm: the
++// driver keeps its own hardware latch and grants as they are.
func (p *Plugin) Resume() {
p.mu.Lock()
-- p.estopped = false
++ p.estopped, p.estopCause, p.estopReason = false, "", ""
p.mu.Unlock()
-- _ = p.send(map[string]any{"op": "resume"})
++ _ = p.send(map[string]any{"op": "resume", "cause": "cleared"})
}
// Answer gives the plugin its owner's answer to an approval it asked for (`openvibe-node approve|deny <id>`).
@@@ -276,6 +298,10 @@ type Provenance struct {
Seq uint64 `json:"seq,omitempty"`
IssuedAtMS int64 `json:"issued_at_ms,omitempty"`
ExpiresAtMS int64 `json:"expires_at_ms,omitempty"`
++ // How far Node's estimate of the server's clock may trail it (the connection's fastest round trip): the deadline
++ // was already checked against estimate + this, and a driver may apply a stricter cap of its own. Absent when the
++ // Node has no link clock (local commands).
++ ClockUncertaintyMS *int64 `json:"clock_uncertainty_ms,omitempty"`
}
// BeginWith is BeginTurn with the command's provenance (nil: none).
@@@ -334,10 +360,10 @@ func (p *Plugin) BeginWith(ctx context.Context, id, kind string, value json.RawM
case r := <-ch:
return r
case <-t.C:
-- p.Stop()
++ p.Stop(StopTimeout)
return Reply{FaultCode: protocol.FaultPluginTimeout}
case <-ctx.Done():
-- p.Stop()
++ p.Stop(StopShutdown)
return Reply{FaultCode: protocol.FaultShuttingDown}
}
}
@@@ -426,7 +452,7 @@ func (p *Plugin) runOnce(ctx context.Context) error {
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
++ estopped, estopCause, estopReason, haltEpoch := p.estopped, p.estopCause, p.estopReason, p.haltEpoch
p.mu.Unlock()
// Writer: the only goroutine touching stdin. Closing stdin is the plugin's signal to stop and exit.
@@@ -465,9 +491,14 @@ func (p *Plugin) runOnce(ctx context.Context) error {
p.readStdout(stdout, described)
}()
-- _ = p.send(map[string]any{"op": "hello", "config": nonNil(p.spec.Config), "probe": false})
++ hello := map[string]any{"op": "hello", "config": nonNil(p.spec.Config), "probe": false}
++ if haltEpoch > 0 {
++ hello["halt_epoch"] = haltEpoch
++ }
++ _ = p.send(hello)
if estopped {
-- _ = p.send(map[string]any{"op": "estop"})
++ // Still latched: the new process hears the same emergency again, marked as a replay, never a fresh one.
++ _ = p.send(map[string]any{"op": "estop", "cause": EstopLatched, "reason": estopReason, "first_cause": estopCause, "replayed": true})
}
// Watch the plugin process itself, not its stdout: a child it started (speech, a camera helper) can hold stdout
@@@ -495,7 +526,7 @@ func (p *Plugin) runOnce(ctx context.Context) error {
describeTimer := time.NewTimer(DescribeTimeout)
defer describeTimer.Stop()
shutdown := func() error {
-- _ = p.send(map[string]any{"op": "stop"})
++ _ = p.send(map[string]any{"op": "stop", "cause": StopShutdown})
p.mu.Lock()
p.queue = nil
p.mu.Unlock()
ddiff --git a/internal/plugins/plugins_test.go b/internal/plugins/plugins_test.go
iindex d1efa39..986ac92 100644
---- a/internal/plugins/plugins_test.go
++++ b/internal/plugins/plugins_test.go
@@@ -253,7 +253,7 @@ func TestShutdownStopsAndClosesStdin(t *testing.T) {
fx.stop()
fx.m.Wait()
l := fx.lines()
-- if !strings.Contains(l[len(l)-2], `"op":"stop"`) || l[len(l)-1] != "EOF-STOP" {
++ if l[len(l)-2] != `{"cause":"shutdown","op":"stop"}` || l[len(l)-1] != "EOF-STOP" {
t.Fatalf("want stop then EOF, got %v", l)
}
}
@@@ -262,8 +262,8 @@ func TestCrashRestartsWithEstop(t *testing.T) {
fx := start(t, "")
fx.waitState(StateReady)
p := fx.m.Get("fake")
-- p.Estop()
-- fx.waitLog(`{"op":"estop"}`)
++ p.Estop(EstopRemote, "staff: test")
++ fx.waitLog(`{"cause":"remote","op":"estop","reason":"staff: test"}`)
r := p.Command(context.Background(), "boom", "crash", nil, 300)
if r.OK || r.FaultCode != protocol.FaultPluginDown {
t.Fatalf("%+v", r)
@@@ -273,7 +273,7 @@ func TestCrashRestartsWithEstop(t *testing.T) {
if p.Info().Restarts < 1 {
t.Fatal("no restart counted")
}
-- // After the restart the latched e-stop is sent again right after hello.
++ // After the restart the latched e-stop is sent again right after hello, as a replay of the same emergency.
l := fx.lines()
var afterCrash []string
for i, s := range l {
@@@ -281,7 +281,8 @@ func TestCrashRestartsWithEstop(t *testing.T) {
afterCrash = l[i+1:]
}
}
-- if len(afterCrash) < 2 || !strings.Contains(afterCrash[0], `"op":"hello"`) || afterCrash[1] != `{"op":"estop"}` {
++ replay := `{"cause":"latched","first_cause":"remote","op":"estop","reason":"staff: test","replayed":true}`
++ if len(afterCrash) < 2 || !strings.Contains(afterCrash[0], `"op":"hello"`) || afterCrash[1] != replay {
t.Fatalf("after restart: %v", afterCrash)
}
}
@@@ -296,7 +297,7 @@ func TestReplyTimeout(t *testing.T) {
if r.FaultCode != protocol.FaultPluginTimeout {
t.Fatalf("%+v", r)
}
-- fx.waitLog(`{"op":"stop"}`)
++ fx.waitLog(`{"cause":"timeout","op":"stop"}`)
}
func TestFaultState(t *testing.T) {
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex e7dcf8b..7662ac5 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -259,6 +259,10 @@ const DefaultDeadlineMS = 300
// DefaultMaxCommandMS caps deadlines when the profile sets no max_command_ms.
const DefaultMaxCommandMS = 1000
++// MaxClockUncertaintyMS is the most clock uncertainty (the connection's fastest round trip) Node times motion with: past
++// it, a motion command with a deadline is refused as not_ready rather than run on a clock it cannot bound.
++const MaxClockUncertaintyMS = 1000
++
// DefaultHeartbeatMS is the device heartbeat interval; the link is declared lost after two intervals without a frame.
const DefaultHeartbeatMS = 1000
ddiff --git a/plugins/sdk/openvibe_plugin/__init__.py b/plugins/sdk/openvibe_plugin/__init__.py
iindex 8d10d6f..911a234 100644
---- a/plugins/sdk/openvibe_plugin/__init__.py
++++ b/plugins/sdk/openvibe_plugin/__init__.py
@@@ -19,8 +19,9 @@ import sys
import threading
import time
--__all__ = ["Plugin", "Fault", "Deferred", "Command", "Runtime", "run", "clamp", "MOTION_KINDS", "HEARTBEAT_TIMEOUT_S",
-- "DEFAULT_DEADLINE_MS", "PROTOCOL_VERSION"]
++__all__ = ["Plugin", "Fault", "Deferred", "Command", "StopInfo", "Runtime", "run", "clamp", "MOTION_KINDS",
++ "HEARTBEAT_TIMEOUT_S", "DEFAULT_DEADLINE_MS", "PROTOCOL_VERSION", "STOP_CAUSES", "EMERGENCY_CAUSES",
++ "RUNTIME_CAUSES"]
PROTOCOL_VERSION = 1
HEARTBEAT_TIMEOUT_S = 1.0
@@@ -67,8 +68,10 @@ class Command:
provenance is where it came from, unchanged from Bot (a dict, or None from older Nodes): "session" (the link
session it arrived on; a new one on every connection), "seq" (the frame's number on that session, strictly
increasing with gaps), "issued_at_ms" (Bot's clock when it sent the command and minted its id) and "expires_at_ms"
-- (the absolute deadline on Bot's clock, motion kinds only). Bot sends a command once and never again; the runtime
-- already refuses one whose seq is not above the last on its session (plugins.md, "Provenance")."""
++ (the absolute deadline on Bot's clock, motion kinds only), and "clock_uncertainty_ms" (Node v0.17.0: how far the
++ core's estimate of Bot's clock may trail it, the connection's fastest round trip; the deadline was already checked
++ against estimate + this, and a driver may refuse motion above a cap of its own). Bot sends a command once and never
++ again; the runtime already refuses one whose seq is not above the last on its session (plugins.md, "Provenance")."""
__slots__ = ("id", "kind", "value", "deadline_ms", "deadline", "target", "operator", "turn_epoch", "provenance")
def __init__(self, id, kind, value, deadline_ms, now, target=None, operator=None, turn_epoch=None, provenance=None):
@@@ -97,9 +100,65 @@ def _provenance(p):
v = p.get(k)
if isinstance(v, int) and not isinstance(v, bool) and v > 0:
out[k] = v
++ u = p.get("clock_uncertainty_ms")
++ if isinstance(u, int) and not isinstance(u, bool) and u >= 0:
++ out["clock_uncertainty_ms"] = u
return out or None
++# Why a driver is stopped (StopInfo.cause). From the core: an ordinary stop revokes remote motion and latches nothing;
++# an emergency latches the platform's stop until its owner clears it.
++STOP_CAUSES = ("halt", "link_lost", "shutdown", "timeout")
++EMERGENCY_CAUSES = ("remote", "local", "driver", "latched")
++# From the runtime itself: a motion command's own deadline, the core going quiet, the handler failing, stdin closing.
++RUNTIME_CAUSES = ("setup", "deadline", "heartbeat", "handler_error", "eof")
++
++
++class StopInfo:
++ """Why the plugin is told to stop: what on_stop() and on_resume() receive (Node v0.17.0; plugins.md, "Why a plugin
++ is stopped").
++
++ cause, from the core: "halt" (Bot ended the controlling turn: a release, a handoff to the next viewer, a timeout or
++ an unbind; turn_epoch is that halt's epoch), "link_lost" (the link to Bot dropped), "shutdown" (the Node is
++ stopping), "timeout" (a command got no answer within its window). From the runtime: "setup" (the stop every setup
++ ends with, so all actuators start stopped), "deadline" (a motion command's
++ own deadline passed; motion_only is True: stop what moves on its own), "heartbeat" (the core went quiet),
++ "handler_error", "eof". An older core sends no cause: "stop".
++ emergency is True for the platform's e-stop, with cause "remote" (a robot owner or staff), "local"
++ (`openvibe-node stop` on the device), "driver" (a driver's own local stop) or "latched" (the stop was already
++ latched when this process or the Node started: replayed is True and first_cause is the original cause when the core
++ knows it); an older core sends "estop". reason is the latch's reason text, when there is one.
++ latched is whether the platform's stop latch is set as this is delivered. It is cleared only by its owner, which
++ on_resume() reports with cause "cleared": information, never a re-arm.
++ halt_epoch is the newest halt epoch the runtime knows; it survives a restart of this process (the core's hello)."""
++ __slots__ = ("cause", "emergency", "latched", "reason", "turn_epoch", "halt_epoch", "replayed", "first_cause",
++ "motion_only")
++
++ def __init__(self, cause, emergency=False, latched=False, reason=None, turn_epoch=None, halt_epoch=0,
++ replayed=False, first_cause=None, motion_only=False):
++ self.cause = cause
++ self.emergency = emergency
++ self.latched = latched
++ self.reason = reason
++ self.turn_epoch = turn_epoch
++ self.halt_epoch = halt_epoch
++ self.replayed = replayed
++ self.first_cause = first_cause
++ self.motion_only = motion_only
++
++ def __repr__(self):
++ return "StopInfo(cause=%r, emergency=%r, latched=%r, turn_epoch=%r, halt_epoch=%r, replayed=%r)" % (
++ self.cause, self.emergency, self.latched, self.turn_epoch, self.halt_epoch, self.replayed)
++
++
++def _text(v, limit=200):
++ return str(v)[:limit] if isinstance(v, str) and v else None
++
++
++def _epoch(v):
++ return v if isinstance(v, int) and not isinstance(v, bool) and v > 0 else None
++
++
def clamp(v, lo, hi):
try:
v = float(v)
@@@ -147,6 +206,22 @@ class Plugin:
def stop(self):
"""Stop every actuator. Must be idempotent and safe to call at any time, including before setup."""
++ def on_stop(self, info):
++ """Stop, knowing why: info is a StopInfo (its cause, whether it is an emergency, the halt epoch, the latch).
++ Every stop the runtime makes calls this. The default calls stop_motion() for a motion deadline and stop() for
++ everything else, so a plugin that overrides only those behaves as before. Override it to tell an ordinary stop
++ (revoke remote motion; latch nothing of your own) from an emergency (info.emergency). Like stop() it must stop
++ every actuator it is asked to, be idempotent, and is reported, not retried, if it raises."""
++ if info.motion_only:
++ self.stop_motion()
++ else:
++ self.stop()
++
++ def on_resume(self, info):
++ """The platform's stop latch was cleared by its owner (info.cause "cleared"). Information only: never re-arm
++ hardware, clear a physical latch or restore a remote grant here. Commands are accepted again by the runtime and
++ each still passes every check. The default does nothing."""
++
def poll(self):
"""Return a telemetry dict (battery, rssi, sensors) or None. Called every telemetry_interval_s."""
return None
@@@ -237,6 +312,7 @@ class Runtime:
self.setup_retry_at = None
self.closed = False
self.stop_reasons = [] # for tests: why stop() was called, in order
++ self.stop_infos = [] # and the StopInfo each on_stop() got
# The newest turn_epoch a halt carried ("stop" with turn_epoch); a command from an older turn is refused.
self.halt_epoch = 0
# The link session of the last command handled, and its seq: a command on that session whose seq is not above
@@@ -255,15 +331,17 @@ class Runtime:
pass
# Safety.
-- def safe_stop(self, reason, motion_only=False):
++ def safe_stop(self, reason, motion_only=False, info=None):
++ """Stop through the plugin's on_stop with why (info; built from reason when not given). reason is what
++ stop_reasons records: the op or the runtime's own cause."""
++ if info is None:
++ info = StopInfo(reason, latched=self.estopped, halt_epoch=self.halt_epoch, motion_only=motion_only)
with self.lock:
self.stop_reasons.append(reason)
++ self.stop_infos.append(info)
self.motion_deadline = None
try:
-- if motion_only:
-- self.plugin.stop_motion()
-- else:
-- self.plugin.stop()
++ self.plugin.on_stop(info)
except Exception as e: # a failing stop is the worst case; report it loudly
self.plugin.log("stop failed (%s): %r", reason, e)
self.send({"op": "event", "name": "stop_failed", "reason": reason, "error": str(e)})
@@@ -272,7 +350,8 @@ class Runtime:
with self.lock:
self.estopped = True
self.local_stop_pending = True
-- self.safe_stop("local_stop")
++ self.safe_stop("local_stop", info=StopInfo("driver", emergency=True, latched=True, reason=_text(reason, 120),
++ halt_epoch=self.halt_epoch))
self.send({"op": "local_stop", "reason": str(reason or "")[:120]})
def tick(self):
@@@ -313,14 +392,19 @@ class Runtime:
elif op == "command":
self.on_command(msg, at)
elif op == "stop":
-- epoch = msg.get("turn_epoch")
-- if isinstance(epoch, int) and not isinstance(epoch, bool) and epoch > self.halt_epoch:
++ epoch = _epoch(msg.get("turn_epoch"))
++ if epoch and epoch > self.halt_epoch:
self.halt_epoch = epoch
-- self.safe_stop("stop")
++ cause = _text(msg.get("cause"), 40) or "stop"
++ self.safe_stop("stop", info=StopInfo(cause, latched=self.estopped, turn_epoch=epoch,
++ halt_epoch=self.halt_epoch))
elif op == "estop":
self.estopped = True
self.local_stop_pending = False
-- self.safe_stop("estop")
++ self.safe_stop("estop", info=StopInfo(_text(msg.get("cause"), 40) or "estop", emergency=True, latched=True,
++ reason=_text(msg.get("reason")), halt_epoch=self.halt_epoch,
++ replayed=msg.get("replayed") is True,
++ first_cause=_text(msg.get("first_cause"), 40)))
elif op == "approval":
try:
with self.lock:
@@@ -347,12 +431,22 @@ class Runtime:
self.plugin.log("resume ignored: the local stop has not reached the core yet")
else:
self.estopped = False
++ try:
++ with self.lock:
++ self.plugin.on_resume(StopInfo(_text(msg.get("cause"), 40) or "cleared",
++ halt_epoch=self.halt_epoch))
++ except Exception as e:
++ self.plugin.log("resume handler failed: %r", e)
else:
self.plugin.log("unknown op %r", op)
def on_hello(self, msg):
self.config = msg.get("config") or {}
self.probe = bool(msg.get("probe"))
++ # The newest halt epoch the core has seen (Node v0.17.0): a restarted process keeps refusing older turns.
++ epoch = _epoch(msg.get("halt_epoch"))
++ if epoch and epoch > self.halt_epoch:
++ self.halt_epoch = epoch
try:
caps = self.plugin.describe(self.config)
except Exception as e:
@@@ -371,7 +465,7 @@ class Runtime:
try:
with self.lock:
self.plugin.setup(self.config)
-- self.plugin.stop()
++ self.plugin.on_stop(StopInfo("setup", latched=self.estopped, halt_epoch=self.halt_epoch))
self.ready = True
self.last_heartbeat = self.clock()
self.send({"op": "ready"})
@@@ -400,7 +494,8 @@ class Runtime:
if kind == "halt":
if cmd.turn_epoch and cmd.turn_epoch > self.halt_epoch:
self.halt_epoch = cmd.turn_epoch
-- self.safe_stop("halt")
++ self.safe_stop("halt", info=StopInfo("halt", latched=self.estopped, turn_epoch=cmd.turn_epoch,
++ halt_epoch=self.halt_epoch))
self.send({"op": "ack", "id": cid})
return
if cmd.turn_epoch and cmd.turn_epoch < self.halt_epoch:
ddiff --git a/plugins/sdk/tests/test_runtime.py b/plugins/sdk/tests/test_runtime.py
iindex 0a15efd..afa46d0 100644
---- a/plugins/sdk/tests/test_runtime.py
++++ b/plugins/sdk/tests/test_runtime.py
@@@ -9,7 +9,7 @@ import unittest
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.dirname(HERE))
--from openvibe_plugin import Deferred, Fault, Plugin, Runtime # noqa: E402
++from openvibe_plugin import Deferred, Fault, Plugin, Runtime, StopInfo # noqa: E402
class FakeClock:
@@@ -180,6 +180,96 @@ class RuntimeTest(unittest.TestCase):
rt.feed_line(json.dumps({"op": "stop"}))
self.assertEqual(drive("c5", 300)["op"], "ack", "a stop without an epoch changes nothing")
++ def test_stop_causes_reach_on_stop(self):
++ """A driver tells an ordinary stop from an emergency (Maticus, 2026-10-27, P7): every stop still stops; only an
++ emergency latches the driver's own hardware stop; resume reports the platform's clear and re-arms nothing."""
++
++ class P7Like(Recorder):
++ def __init__(self):
++ super().__init__()
++ self.infos, self.resumes, self.physical_latch = [], [], False
++
++ def on_stop(self, info):
++ self.infos.append(info)
++ if info.emergency:
++ self.physical_latch = True
++ self.stop()
++
++ def on_resume(self, info):
++ self.resumes.append(info)
++
++ rt, p, out, clock = make(P7Like())
++ self.hello(rt)
++ feed = lambda m: rt.feed_line(json.dumps(m)) # noqa: E731
++ feed({"op": "stop", "cause": "halt", "turn_epoch": 7})
++ feed({"op": "stop", "cause": "link_lost"})
++ self.assertEqual([(i.cause, i.emergency, i.latched) for i in p.infos],
++ [("setup", False, False), ("halt", False, False), ("link_lost", False, False)])
++ self.assertEqual((p.infos[1].turn_epoch, p.infos[1].halt_epoch, p.infos[2].halt_epoch), (7, 7, 7))
++ self.assertFalse(p.physical_latch, "an ordinary stop latches nothing")
++ feed({"op": "estop", "cause": "remote", "reason": "staff: it wobbled"})
++ e = p.infos[-1]
++ self.assertEqual((e.cause, e.emergency, e.latched, e.reason, e.replayed), ("remote", True, True, "staff: it wobbled", False))
++ self.assertTrue(p.physical_latch)
++ feed({"op": "resume", "cause": "cleared"})
++ self.assertEqual([r.cause for r in p.resumes], ["cleared"])
++ self.assertTrue(p.physical_latch, "resume never clears the driver's own latch")
++ self.assertFalse(rt.estopped)
++ # The same emergency replayed to a restarted process, and an older core's bare ops.
++ feed({"op": "estop", "cause": "latched", "first_cause": "remote", "reason": "staff: it wobbled", "replayed": True})
++ self.assertEqual((p.infos[-1].cause, p.infos[-1].replayed, p.infos[-1].first_cause), ("latched", True, "remote"))
++ feed({"op": "stop"})
++ feed({"op": "estop"})
++ self.assertEqual([(i.cause, i.emergency) for i in p.infos[-2:]], [("stop", False), ("estop", True)])
++ self.assertEqual(p.calls.count("stop"), len(p.infos), "every stop stopped")
++ # The runtime's own stops say why too: the driver's own local stop is an emergency.
++ rt.local_stop("bumper")
++ self.assertEqual((p.infos[-1].cause, p.infos[-1].emergency, p.infos[-1].reason), ("driver", True, "bumper"))
++
++ def test_on_stop_defaults_to_stop_and_stop_motion(self):
++ """A plugin that overrides only stop() and stop_motion() behaves as before; the runtime still records why."""
++
++ class Wheels(Recorder):
++ def stop_motion(self):
++ self.calls.append("stop_motion")
++
++ rt, p, out, clock = make(Wheels())
++ self.hello(rt)
++ rt.feed_line(json.dumps({"op": "command", "id": "d", "kind": "drive", "value": {}, "deadline_ms": 300}))
++ clock.t += 1.0
++ rt.tick()
++ rt.feed_line(json.dumps({"op": "estop", "cause": "remote"}))
++ self.assertEqual(p.calls[-2:], ["stop_motion", "stop"])
++ self.assertEqual([(i.cause, i.motion_only, i.emergency) for i in rt.stop_infos[-2:]],
++ [("deadline", True, False), ("remote", False, True)])
++ self.assertIsInstance(rt.stop_infos[-1], StopInfo)
++
++ def test_halt_epoch_from_hello_survives_a_restart(self):
++ rt, p, out, clock = make()
++ rt.feed_line(json.dumps({"op": "hello", "config": {}, "halt_epoch": 50}))
++
++ def drive(cid, epoch):
++ rt.feed_line(json.dumps({"op": "command", "id": cid, "kind": "drive", "value": {}, "deadline_ms": 300,
++ "turn_epoch": epoch}))
++ return lines(out)[-1]
++
++ self.assertEqual(drive("old", 40)["fault_code"], "expired")
++ self.assertEqual(drive("now", 50)["op"], "ack")
++
++ def test_clock_uncertainty_reaches_handle(self):
++ seen = []
++
++ class Keep(Recorder):
++ def handle(self, cmd):
++ seen.append(cmd.provenance)
++
++ rt, p, out, clock = make(Keep())
++ self.hello(rt)
++ for i, u in enumerate((0, 42)):
++ rt.feed_line(json.dumps({"op": "command", "id": "c%d" % i, "kind": "say", "value": {}, "deadline_ms": 300,
++ "provenance": {"session": "s", "seq": i + 1, "clock_uncertainty_ms": u}}))
++ self.assertEqual([s.get("clock_uncertainty_ms") for s in seen], [0, 42])
++
def test_newer_command_extends_deadline(self):
rt, p, out, clock = make()
self.hello(rt)