openvibers/openvibe.link Commits

A command older than the robot's newest turn halt never reaches a driver, and plugins get turn_epoch (cmd.turn_epoch in Python) (#55)

OP OpenVibers committed 20 days ago · 575d26d9e8 · 9 files, +133 −20

Changes

ddiff --git a/docs/hardware-looi.md b/docs/hardware-looi.md
iindex 9477d38..f3c2361 100644
---- a/docs/hardware-looi.md
++++ b/docs/hardware-looi.md
@@@ -20,7 +20,9 @@ turns, cooldowns and per-role allowed commands, all enforced by Bot's command ga
The e-stop is latched in Bot and only the owner clears it. Each command tells the driver who sent it (`cmd.operator`,
`{"subject", "role"}`), and when a queue turn ends or its holder leaves Bot halts the robot before the next person
drives (the driver gets `stop`), so commands from the last driver never come after it
--([plugins.md](plugins.md#core--plugin)).
++([plugins.md](plugins.md#core--plugin)). The owner or an operator driving during someone's turn pauses that turn and
++halts the robot first, the same way. Each command also carries `cmd.turn_epoch`: the core already refuses one older
++than the newest halt's, and a driver may check it again against its own record.
To show the panel under a Live stream, bind the robot to the channel on the Live dashboard (the robot card). The
channel page then shows the robot's panel beside the phone's stream.
ddiff --git a/docs/plugins.md b/docs/plugins.md
iindex 62d0539..493b5dc 100644
---- a/docs/plugins.md
++++ b/docs/plugins.md
@@@ -48,7 +48,7 @@ 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) |
--| `command` | `id`, `kind`, `value`, `deadline_ms`, `operator`? | act, answer `ack` or `nack` with the same `id` |
++| `command` | `id`, `kind`, `value`, `deadline_ms`, `operator`?, `turn_epoch`? | act, answer `ack` or `nack` with the same `id` |
| `heartbeat` | `t` (core clock, ms) | note the time |
| `stop` | | stop every actuator |
| `estop` | | stop every actuator and refuse motion until `resume` |
@@@ -66,6 +66,13 @@ reach a plugin in the order Bot sent them, and when a public robot's turn ends o
before the next person's first command, which a plugin gets as `stop`: nothing from the last driver comes after it.
Python: `cmd.operator` (a dict, or `None`).
++`turn_epoch` is the robot's control epoch Bot decided the command under (milliseconds on Bot's clock; Bot#155,
++contracts 0.213.0). It rises when a queue turn ends, its holder leaves, or the owner takes the turn over, and Bot's halt
++for that change carries the new value. The core keeps the newest epoch a halt carried per robot and nacks (`expired`) a
++command older than it, so it never reaches a plugin; a driver may keep the newest it saw and refuse older ones too, as
++defence in depth (a desk robot's own controller, for one). Absent from older Bots and on robots without a queue.
++Python: `cmd.turn_epoch` (an int, or `None`).
++
### Plugin → core
| op | fields |
ddiff --git a/internal/node/botcontract_test.go b/internal/node/botcontract_test.go
iindex d54f93c..73fe813 100644
---- a/internal/node/botcontract_test.go
++++ b/internal/node/botcontract_test.go
@@@ -341,3 +341,40 @@ func TestLocalLatchSurvivesRemoteClear(t *testing.T) {
t.Fatal("the device's latched:false cleared Bot's latch")
}
}
++
++// turn_epoch (contracts 0.213.0, Bot#155): once a halt carried a robot's new control epoch, a command decided under an
++// older one is nacked expired and never reaches the plugin; the current epoch runs, and a command without one (an older
++// Bot) runs as before.
++func TestStaleTurnNeverReachesPlugin(t *testing.T) {
++ e := start(t)
++ send := func(c protocol.Command) protocol.Message {
++ t.Helper()
++ if err := e.conn.Command(c, 300); err != nil {
++ t.Fatal(err)
++ }
++ r, err := e.conn.Reply(c.ID, 5*time.Second)
++ if err != nil {
++ t.Fatal(err)
++ }
++ return r
++ }
++ drive := json.RawMessage(`{"throttle":0.3}`)
++ if r := send(protocol.Command{ID: "before", Kind: protocol.KindDrive, Value: drive, TurnEpoch: 100}); !isAck(r) {
++ t.Fatalf("a command before any halt: %+v", r)
++ }
++ if r := send(protocol.Command{ID: "halt-200", Kind: protocol.KindHalt, TurnEpoch: 200, Operator: &protocol.Operator{Role: "system"}}); !isAck(r) {
++ t.Fatalf("the turn's halt: %+v", r)
++ }
++ if r := send(protocol.Command{ID: "stale", Kind: protocol.KindDrive, Value: drive, TurnEpoch: 100}); nackCode(r) != protocol.FaultExpired {
++ t.Fatalf("a command from the ended turn: %+v", r)
++ }
++ if r := send(protocol.Command{ID: "current", Kind: protocol.KindDrive, Value: drive, TurnEpoch: 200}); !isAck(r) {
++ t.Fatalf("a command of the current turn: %+v", r)
++ }
++ if r := send(protocol.Command{ID: "no-epoch", Kind: protocol.KindDrive, Value: drive}); !isAck(r) {
++ t.Fatalf("a command without an epoch (an older Bot): %+v", r)
++ }
++ if len(e.commandRecords("stale")) != 0 || len(e.commandRecords("current")) != 1 {
++ t.Fatalf("records: %+v", e.records())
++ }
++}
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex 5908d53..a58a649 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -71,17 +71,19 @@ type Node struct {
ctx context.Context
-- mu sync.Mutex
-- limits safety.Limits
-- allowed map[string]bool
-- telemetry map[string]map[string]any // per plugin, latest
-- events []protocol.Event
-- telemMS int
-- jpeg *video.JPEGSource
-- jpegFrom string
-- videoSrc string
-- lastTelem time.Time
-- started time.Time
++ mu sync.Mutex
++ limits safety.Limits
++ // haltEpochs: per robot, the newest turn_epoch a halt carried (noteHaltEpoch); older commands are refused.
++ haltEpochs map[string]int64
++ allowed map[string]bool
++ telemetry map[string]map[string]any // per plugin, latest
++ events []protocol.Event
++ telemMS int
++ jpeg *video.JPEGSource
++ jpegFrom string
++ videoSrc string
++ lastTelem time.Time
++ started time.Time
// configured is set by the first config on a connection; until then only halt runs, so a latch the owner set
// while the link was down is applied before any motion.
@@@ -410,6 +412,7 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
limits, allowed, configured := n.limits, n.allowed, n.configured
n.mu.Unlock()
if c.Kind == protocol.KindHalt {
++ n.noteHaltEpoch(c.RobotID, c.TurnEpoch)
n.mgr.StopAll()
r := protocol.Ack{ID: c.ID, LatencyMS: time.Since(received).Milliseconds()}
n.dedup.Complete(entry, r)
@@@ -420,6 +423,12 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
nack(protocol.FaultNotReady, "the server's config has not arrived on this connection yet")
return
}
++ // A command decided under a turn that has since ended (its epoch is older than the newest halt's for this robot) never
++ // reaches a driver. Bot already guarantees it is not sent after that halt; this holds even if frames were delayed.
++ if last := n.haltEpoch(c.RobotID); c.TurnEpoch > 0 && c.TurnEpoch < last {
++ nack(protocol.FaultExpired, fmt.Sprintf("from a turn that ended: epoch %d is older than the newest halt's %d", c.TurnEpoch, last))
++ return
++ }
if allowed != nil && !allowed[c.Kind] {
nack(protocol.FaultNotAllowed, c.Kind+" is not in this robot's allowed_commands")
return
@@@ -471,7 +480,7 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
nack(protocol.FaultUnsupported, "no driver on this device handles "+c.Kind)
return
}
-- wait := p.BeginAs(n.ctx, c.ID, c.Kind, value, deadline, c.Operator)
++ wait := p.BeginTurn(n.ctx, c.ID, c.Kind, value, deadline, c.Operator, c.TurnEpoch)
go func() {
r := wait()
var msg protocol.Message
@@@ -562,6 +571,27 @@ func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, r
// serverNow is the server's clock when a frame stamped sentMS was received: the link's estimate, else the frame's
// own ts.
++// noteHaltEpoch records a halt's turn_epoch for its robot (the newest wins); haltEpoch reads it (0: none seen).
++func (n *Node) noteHaltEpoch(robotID string, epoch int64) {
++ if epoch <= 0 {
++ return
++ }
++ n.mu.Lock()
++ defer n.mu.Unlock()
++ if n.haltEpochs == nil {
++ n.haltEpochs = map[string]int64{}
++ }
++ if epoch > n.haltEpochs[robotID] {
++ n.haltEpochs[robotID] = epoch
++ }
++}
++
++func (n *Node) haltEpoch(robotID string) int64 {
++ n.mu.Lock()
++ defer n.mu.Unlock()
++ return n.haltEpochs[robotID]
++}
++
func (n *Node) serverNow(sentMS int64, received time.Time) int64 {
if n.link != nil {
if ms, ok := n.link.ServerNow(received); ok {
ddiff --git a/internal/plugins/plugin.go b/internal/plugins/plugin.go
iindex 46ec91e..18bc064 100644
---- a/internal/plugins/plugin.go
++++ b/internal/plugins/plugin.go
@@@ -226,6 +226,12 @@ func (p *Plugin) Begin(ctx context.Context, id, kind string, value json.RawMessa
// BeginAs is Begin with who sent the command, as Bot's gate saw them: the plugin gets it as `operator`
// {subject, role} (subject null for Bot's own commands, role "system"), and nothing when op is nil.
func (p *Plugin) BeginAs(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int, op *protocol.Operator) func() Reply {
++ return p.BeginTurn(ctx, id, kind, value, deadlineMS, op, 0)
++}
++
++// BeginTurn is BeginAs with the robot's control epoch the command was decided under (protocol.Command.TurnEpoch), sent
++// to the plugin as turn_epoch when it is set, so a driver can refuse a command older than the newest turn halt itself.
++func (p *Plugin) BeginTurn(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int, op *protocol.Operator, turnEpoch int64) func() Reply {
now := func(r Reply) func() Reply { return func() Reply { return r } }
p.mu.Lock()
if p.state != StateReady && kind != protocol.KindHalt {
@@@ -258,6 +264,9 @@ func (p *Plugin) BeginAs(ctx context.Context, id, kind string, value json.RawMes
}
msg["operator"] = map[string]any{"subject": subject, "role": op.Role}
}
++ if turnEpoch > 0 {
++ msg["turn_epoch"] = turnEpoch
++ }
if err := p.send(msg); err != nil {
done()
return now(Reply{FaultCode: protocol.FaultPluginDown})
ddiff --git a/internal/plugins/plugins_test.go b/internal/plugins/plugins_test.go
iindex f42f773..d1efa39 100644
---- a/internal/plugins/plugins_test.go
++++ b/internal/plugins/plugins_test.go
@@@ -192,6 +192,10 @@ func TestHelloDescribeCommand(t *testing.T) {
t.Fatalf("%+v", r)
}
fx.waitLog(`"operator":{"role":"system","subject":null}`)
++ if r := p.BeginTurn(context.Background(), "c1d", "drive", nil, 300, &protocol.Operator{Subject: "usr_1", Role: "queue"}, 1792900000123)(); !r.OK {
++ t.Fatalf("%+v", r)
++ }
++ fx.waitLog(`"turn_epoch":1792900000123`)
if r := p.Command(context.Background(), "c2", "display", nil, 300); r.OK || r.FaultCode != "unsupported" {
t.Fatalf("%+v", r)
}
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex 9cb7cb7..fb75b95 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -306,8 +306,12 @@ type Command struct {
// DeadlineMS is an absolute instant in epoch milliseconds on the server's clock, present on motion kinds only.
DeadlineMS int64 `json:"deadline_ms,omitempty"`
Operator *Operator `json:"operator,omitempty"`
-- RobotID string `json:"robot_id,omitempty"`
-- Target string `json:"target,omitempty"` // Node only: a driver name; default: the first driver with the capability
++ // TurnEpoch is the robot's control epoch when Bot decided the command (contracts 0.213.0, Bot#155): it rises when a
++ // queue turn ends, is left or is paused for an owner takeover, and Bot's halt for that change carries the new value.
++ // Zero when Bot sends none (older Bots, robots without a queue).
++ TurnEpoch int64 `json:"turn_epoch,omitempty"`
++ RobotID string `json:"robot_id,omitempty"`
++ Target string `json:"target,omitempty"` // Node only: a driver name; default: the first driver with the capability
}
// Estop is sent when the e-stop latches (Latched true) or the owner clears it (Latched false).
ddiff --git a/plugins/sdk/openvibe_plugin/__init__.py b/plugins/sdk/openvibe_plugin/__init__.py
iindex fbf54be..fdd0507 100644
---- a/plugins/sdk/openvibe_plugin/__init__.py
++++ b/plugins/sdk/openvibe_plugin/__init__.py
@@@ -59,10 +59,13 @@ def _ack(cid, data=None):
class Command:
"""operator is who sent it, as Bot's gate saw them: {"subject": "usr_…" or None, "role": "owner" | "operator" |
-- "queue" | …}, or None when the core passed none."""
-- __slots__ = ("id", "kind", "value", "deadline_ms", "deadline", "target", "operator")
++ "queue" | …}, or None when the core passed none. turn_epoch is the robot's control epoch Bot decided it under (an
++ int, or None from older Bots): it rises when a queue turn ends, is left or is paused for an owner takeover, and the
++ core already refuses a command older than the newest halt's, so a driver may also keep the newest it saw and refuse
++ anything older (defence in depth)."""
++ __slots__ = ("id", "kind", "value", "deadline_ms", "deadline", "target", "operator", "turn_epoch")
-- def __init__(self, id, kind, value, deadline_ms, now, target=None, operator=None):
++ def __init__(self, id, kind, value, deadline_ms, now, target=None, operator=None, turn_epoch=None):
self.id = id
self.kind = kind
self.value = value if value is not None else {}
@@@ -70,6 +73,7 @@ class Command:
self.deadline = now + deadline_ms / 1000.0
self.target = target
self.operator = operator if isinstance(operator, dict) else None
++ self.turn_epoch = turn_epoch if isinstance(turn_epoch, int) and not isinstance(turn_epoch, bool) and turn_epoch > 0 else None
def __repr__(self):
return "Command(id=%r, kind=%r, value=%r, deadline_ms=%r)" % (self.id, self.kind, self.value, self.deadline_ms)
@@@ -342,7 +346,7 @@ class Runtime:
dl = DEFAULT_DEADLINE_MS
moves = kind in GUARDED_KINDS or kind in self.plugin.motion_kinds
dl = max(1, min(dl, MAX_DEADLINE_MS if moves else MAX_LONG_DEADLINE_MS))
-- cmd = Command(cid, kind, msg.get("value"), dl, now, msg.get("target"), msg.get("operator"))
++ cmd = Command(cid, kind, msg.get("value"), dl, now, msg.get("target"), msg.get("operator"), msg.get("turn_epoch"))
if kind == "halt":
self.safe_stop("halt")
self.send({"op": "ack", "id": cid})
ddiff --git a/plugins/sdk/tests/test_runtime.py b/plugins/sdk/tests/test_runtime.py
iindex a5e375b..1e57c53 100644
---- a/plugins/sdk/tests/test_runtime.py
++++ b/plugins/sdk/tests/test_runtime.py
@@@ -109,6 +109,22 @@ class RuntimeTest(unittest.TestCase):
"operator": "nobody"}))
self.assertEqual(seen, [{"subject": "usr_1", "role": "queue"}, None, None])
++ def test_turn_epoch_reaches_handle(self):
++ seen = []
++
++ class Epoch(Recorder):
++ def handle(self, cmd):
++ seen.append(cmd.turn_epoch)
++
++ rt, p, out, clock = make(Epoch())
++ self.hello(rt)
++ rt.feed_line(json.dumps({"op": "command", "id": "c1", "kind": "drive", "value": {}, "deadline_ms": 300,
++ "turn_epoch": 1792900000123}))
++ rt.feed_line(json.dumps({"op": "command", "id": "c2", "kind": "drive", "value": {}, "deadline_ms": 300}))
++ rt.feed_line(json.dumps({"op": "command", "id": "c3", "kind": "drive", "value": {}, "deadline_ms": 300,
++ "turn_epoch": "soon"}))
++ self.assertEqual(seen, [1792900000123, None, None])
++
def test_newer_command_extends_deadline(self):
rt, p, out, clock = make()
self.hello(rt)