openvibers/openvibe.link Commits
A plugin's command carries operator {subject, role} as Bot's gate saw it (cmd.operator in the Python SDK) (#54)
Changes
@@@ -17,7 +17,10 @@ them per robot.Who may drive is Bot's job, not the driver's: the owner, invited operators, an optional public queue with timedturns, cooldowns and per-role allowed commands, all enforced by Bot's command gate before a command reaches the Node.--The e-stop is latched in Bot and only the owner clears it.++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)).To show the panel under a Live stream, bind the robot to the channel on the Live dashboard (the robot card). Thechannel page then shows the robot's panel beside the phone's stream.@@@ -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` | act, answer `ack` or `nack` with the same `id` |++| `command` | `id`, `kind`, `value`, `deadline_ms`, `operator`? | 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` |@@@ -59,6 +59,13 @@ describes itself and waits for EOF without opening any hardware.`value` has already been validated and clamped to the owner's limits by the core ([protocol.md](protocol.md#commands)).`deadline_ms` is relative to receipt and already capped.++`operator` is who sent the command, as Bot's gate saw them: `{"subject": "usr_…", "role": "owner" | "operator" |++"queue"}` (`queue` is the holder of a public robot's turn); it is absent when the core got none. Bot has already++decided the command may run, so a driver uses it to log or to apply a rule of its own, never to grant more. Commands++reach a plugin in the order Bot sent them, and when a public robot's turn ends or its holder leaves Bot halts the robot++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`).++### Plugin → core| op | fields |@@@ -471,7 +471,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.Begin(n.ctx, c.ID, c.Kind, value, deadline)++ wait := p.BeginAs(n.ctx, c.ID, c.Kind, value, deadline, c.Operator)go func() {r := wait()var msg protocol.Message@@@ -532,7 +532,7 @@ func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, rn.inputs[c.ID] = truen.mu.Unlock()}-- wait := p.Begin(n.ctx, c.ID, c.Kind, c.Value, int(window))++ wait := p.BeginAs(n.ctx, c.ID, c.Kind, c.Value, int(window), c.Operator)go func() {r := wait()if c.Kind == protocol.KindInput {@@@ -220,6 +220,12 @@ func (p *Plugin) Command(ctx context.Context, id, kind string, value json.RawMes// waits for its ack or nack. A plugin that does not answer in max(deadline, MinReplyTimeout) is told to stop and the// command fails with plugin_timeout.func (p *Plugin) Begin(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int) func() Reply {++ return p.BeginAs(ctx, id, kind, value, deadlineMS, nil)++}++++// 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 {now := func(r Reply) func() Reply { return func() Reply { return r } }p.mu.Lock()if p.state != StateReady && kind != protocol.KindHalt {@@@ -245,6 +251,13 @@ func (p *Plugin) Begin(ctx context.Context, id, kind string, value json.RawMessaif len(value) > 0 {msg["value"] = value}++ if op != nil {++ var subject any++ if op.Subject != "" {++ subject = op.Subject++ }++ msg["operator"] = map[string]any{"subject": subject, "role": op.Role}++ }if err := p.send(msg); err != nil {done()return now(Reply{FaultCode: protocol.FaultPluginDown})@@@ -184,6 +184,14 @@ func TestHelloDescribeCommand(t *testing.T) {}fx.waitLog(`"deadline_ms":300`)fx.waitLog(`"value":{"throttle":0.5}`)++ if r := p.BeginAs(context.Background(), "c1b", "drive", nil, 300, &protocol.Operator{Subject: "usr_1", Role: "queue"})(); !r.OK {++ t.Fatalf("%+v", r)++ }++ fx.waitLog(`"operator":{"role":"queue","subject":"usr_1"}`)++ if r := p.BeginAs(context.Background(), "c1c", "drive", nil, 300, &protocol.Operator{Role: "system"})(); !r.OK {++ t.Fatalf("%+v", r)++ }++ fx.waitLog(`"operator":{"role":"system","subject":null}`)if r := p.Command(context.Background(), "c2", "display", nil, 300); r.OK || r.FaultCode != "unsupported" {t.Fatalf("%+v", r)}@@@ -58,15 +58,18 @@ def _ack(cid, data=None):class Command:-- __slots__ = ("id", "kind", "value", "deadline_ms", "deadline", "target")++ """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")-- def __init__(self, id, kind, value, deadline_ms, now, target=None):++ def __init__(self, id, kind, value, deadline_ms, now, target=None, operator=None):self.id = idself.kind = kindself.value = value if value is not None else {}self.deadline_ms = deadline_msself.deadline = now + deadline_ms / 1000.0self.target = target++ self.operator = operator if isinstance(operator, dict) else Nonedef __repr__(self):return "Command(id=%r, kind=%r, value=%r, deadline_ms=%r)" % (self.id, self.kind, self.value, self.deadline_ms)@@@ -339,7 +342,7 @@ class Runtime:dl = DEFAULT_DEADLINE_MSmoves = kind in GUARDED_KINDS or kind in self.plugin.motion_kindsdl = 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"))++ cmd = Command(cid, kind, msg.get("value"), dl, now, msg.get("target"), msg.get("operator"))if kind == "halt":self.safe_stop("halt")self.send({"op": "ack", "id": cid})@@@ -93,6 +93,22 @@ class RuntimeTest(unittest.TestCase):self.assertFalse(p.moving)self.assertIn("deadline", rt.stop_reasons)++ def test_operator_reaches_handle(self):++ seen = []++++ class Who(Recorder):++ def handle(self, cmd):++ seen.append(cmd.operator)++++ rt, p, out, clock = make(Who())++ self.hello(rt)++ rt.feed_line(json.dumps({"op": "command", "id": "c1", "kind": "drive", "value": {}, "deadline_ms": 300,++ "operator": {"subject": "usr_1", "role": "queue"}}))++ 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,++ "operator": "nobody"}))++ self.assertEqual(seen, [{"subject": "usr_1", "role": "queue"}, None, None])++def test_newer_command_extends_deadline(self):rt, p, out, clock = make()self.hello(rt)
@@ -17,7 +17,10 @@ them per robot.
@@ -17,7 +17,10 @@ them per robot.
Who may drive is Bot's job, not the driver's: the owner, invited operators, an optional public queue with timed
Who may drive is Bot's job, not the driver's: the owner, invited operators, an optional public queue with timed
turns, cooldowns and per-role allowed commands, all enforced by Bot's command gate before a command reaches the Node.
turns, cooldowns and per-role allowed commands, all enforced by Bot's command gate before a command reaches the Node.
-The e-stop is latched in Bot and only the owner clears it.
+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)).
To show the panel under a Live stream, bind the robot to the channel on the Live dashboard (the robot card). The
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.
channel page then shows the robot's panel beside the phone's stream.
@@ -48,7 +48,7 @@ describes itself and waits for EOF without opening any hardware.
@@ -48,7 +48,7 @@ describes itself and waits for EOF without opening any hardware.
| op | fields | the plugin must |
| op | fields | the plugin must |
|-------------|---------------------------------------------------------------|--------------------------------------------------------|
|-------------|---------------------------------------------------------------|--------------------------------------------------------|
| `hello` | `config` {…}, `probe` bool | answer `describe`; then set up (unless probe) |
| `hello` | `config` {…}, `probe` bool | answer `describe`; then set up (unless probe) |
-| `command` | `id`, `kind`, `value`, `deadline_ms` | act, answer `ack` or `nack` with the same `id` |
+| `command` | `id`, `kind`, `value`, `deadline_ms`, `operator`? | act, answer `ack` or `nack` with the same `id` |
| `heartbeat` | `t` (core clock, ms) | note the time |
| `heartbeat` | `t` (core clock, ms) | note the time |
| `stop` | | stop every actuator |
| `stop` | | stop every actuator |
| `estop` | | stop every actuator and refuse motion until `resume` |
| `estop` | | stop every actuator and refuse motion until `resume` |
@@ -59,6 +59,13 @@ describes itself and waits for EOF without opening any hardware.
@@ -59,6 +59,13 @@ describes itself and waits for EOF without opening any hardware.
`value` has already been validated and clamped to the owner's limits by the core ([protocol.md](protocol.md#commands)).
`value` has already been validated and clamped to the owner's limits by the core ([protocol.md](protocol.md#commands)).
`deadline_ms` is relative to receipt and already capped.
`deadline_ms` is relative to receipt and already capped.
+`operator` is who sent the command, as Bot's gate saw them: `{"subject": "usr_…", "role": "owner" | "operator" |
+"queue"}` (`queue` is the holder of a public robot's turn); it is absent when the core got none. Bot has already
+decided the command may run, so a driver uses it to log or to apply a rule of its own, never to grant more. Commands
+reach a plugin in the order Bot sent them, and when a public robot's turn ends or its holder leaves Bot halts the robot
+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`).
+
### Plugin → core
### Plugin → core
| op | fields |
| op | fields |
@@ -471,7 +471,7 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
@@ -471,7 +471,7 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
nack(protocol.FaultUnsupported, "no driver on this device handles "+c.Kind)
nack(protocol.FaultUnsupported, "no driver on this device handles "+c.Kind)
return
return
}
}
- wait := p.Begin(n.ctx, c.ID, c.Kind, value, deadline)
+ wait := p.BeginAs(n.ctx, c.ID, c.Kind, value, deadline, c.Operator)
go func() {
go func() {
r := wait()
r := wait()
var msg protocol.Message
var msg protocol.Message
@@ -532,7 +532,7 @@ func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, r
@@ -532,7 +532,7 @@ func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, r
n.inputs[c.ID] = true
n.inputs[c.ID] = true
n.mu.Unlock()
n.mu.Unlock()
}
}
- wait := p.Begin(n.ctx, c.ID, c.Kind, c.Value, int(window))
+ wait := p.BeginAs(n.ctx, c.ID, c.Kind, c.Value, int(window), c.Operator)
go func() {
go func() {
r := wait()
r := wait()
if c.Kind == protocol.KindInput {
if c.Kind == protocol.KindInput {
@@ -220,6 +220,12 @@ func (p *Plugin) Command(ctx context.Context, id, kind string, value json.RawMes
@@ -220,6 +220,12 @@ func (p *Plugin) Command(ctx context.Context, id, kind string, value json.RawMes
// waits for its ack or nack. A plugin that does not answer in max(deadline, MinReplyTimeout) is told to stop and the
// waits for its ack or nack. A plugin that does not answer in max(deadline, MinReplyTimeout) is told to stop and the
// command fails with plugin_timeout.
// command fails with plugin_timeout.
func (p *Plugin) Begin(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int) func() Reply {
func (p *Plugin) Begin(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int) func() Reply {
+ return p.BeginAs(ctx, id, kind, value, deadlineMS, nil)
+}
+
+// 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 {
now := func(r Reply) func() Reply { return func() Reply { return r } }
now := func(r Reply) func() Reply { return func() Reply { return r } }
p.mu.Lock()
p.mu.Lock()
if p.state != StateReady && kind != protocol.KindHalt {
if p.state != StateReady && kind != protocol.KindHalt {
@@ -245,6 +251,13 @@ func (p *Plugin) Begin(ctx context.Context, id, kind string, value json.RawMessa
@@ -245,6 +251,13 @@ func (p *Plugin) Begin(ctx context.Context, id, kind string, value json.RawMessa
if len(value) > 0 {
if len(value) > 0 {
msg["value"] = value
msg["value"] = value
}
}
+ if op != nil {
+ var subject any
+ if op.Subject != "" {
+ subject = op.Subject
+ }
+ msg["operator"] = map[string]any{"subject": subject, "role": op.Role}
+ }
if err := p.send(msg); err != nil {
if err := p.send(msg); err != nil {
done()
done()
return now(Reply{FaultCode: protocol.FaultPluginDown})
return now(Reply{FaultCode: protocol.FaultPluginDown})
@@ -184,6 +184,14 @@ func TestHelloDescribeCommand(t *testing.T) {
@@ -184,6 +184,14 @@ func TestHelloDescribeCommand(t *testing.T) {
}
}
fx.waitLog(`"deadline_ms":300`)
fx.waitLog(`"deadline_ms":300`)
fx.waitLog(`"value":{"throttle":0.5}`)
fx.waitLog(`"value":{"throttle":0.5}`)
+ if r := p.BeginAs(context.Background(), "c1b", "drive", nil, 300, &protocol.Operator{Subject: "usr_1", Role: "queue"})(); !r.OK {
+ t.Fatalf("%+v", r)
+ }
+ fx.waitLog(`"operator":{"role":"queue","subject":"usr_1"}`)
+ if r := p.BeginAs(context.Background(), "c1c", "drive", nil, 300, &protocol.Operator{Role: "system"})(); !r.OK {
+ t.Fatalf("%+v", r)
+ }
+ fx.waitLog(`"operator":{"role":"system","subject":null}`)
if r := p.Command(context.Background(), "c2", "display", nil, 300); r.OK || r.FaultCode != "unsupported" {
if r := p.Command(context.Background(), "c2", "display", nil, 300); r.OK || r.FaultCode != "unsupported" {
t.Fatalf("%+v", r)
t.Fatalf("%+v", r)
}
}
@@ -58,15 +58,18 @@ def _ack(cid, data=None):
@@ -58,15 +58,18 @@ def _ack(cid, data=None):
class Command:
class Command:
- __slots__ = ("id", "kind", "value", "deadline_ms", "deadline", "target")
+ """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")
- def __init__(self, id, kind, value, deadline_ms, now, target=None):
+ def __init__(self, id, kind, value, deadline_ms, now, target=None, operator=None):
self.id = id
self.id = id
self.kind = kind
self.kind = kind
self.value = value if value is not None else {}
self.value = value if value is not None else {}
self.deadline_ms = deadline_ms
self.deadline_ms = deadline_ms
self.deadline = now + deadline_ms / 1000.0
self.deadline = now + deadline_ms / 1000.0
self.target = target
self.target = target
+ self.operator = operator if isinstance(operator, dict) else None
def __repr__(self):
def __repr__(self):
return "Command(id=%r, kind=%r, value=%r, deadline_ms=%r)" % (self.id, self.kind, self.value, self.deadline_ms)
return "Command(id=%r, kind=%r, value=%r, deadline_ms=%r)" % (self.id, self.kind, self.value, self.deadline_ms)
@@ -339,7 +342,7 @@ class Runtime:
@@ -339,7 +342,7 @@ class Runtime:
dl = DEFAULT_DEADLINE_MS
dl = DEFAULT_DEADLINE_MS
moves = kind in GUARDED_KINDS or kind in self.plugin.motion_kinds
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))
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"))
+ cmd = Command(cid, kind, msg.get("value"), dl, now, msg.get("target"), msg.get("operator"))
if kind == "halt":
if kind == "halt":
self.safe_stop("halt")
self.safe_stop("halt")
self.send({"op": "ack", "id": cid})
self.send({"op": "ack", "id": cid})
@@ -93,6 +93,22 @@ class RuntimeTest(unittest.TestCase):
@@ -93,6 +93,22 @@ class RuntimeTest(unittest.TestCase):
self.assertFalse(p.moving)
self.assertFalse(p.moving)
self.assertIn("deadline", rt.stop_reasons)
self.assertIn("deadline", rt.stop_reasons)
+ def test_operator_reaches_handle(self):
+ seen = []
+
+ class Who(Recorder):
+ def handle(self, cmd):
+ seen.append(cmd.operator)
+
+ rt, p, out, clock = make(Who())
+ self.hello(rt)
+ rt.feed_line(json.dumps({"op": "command", "id": "c1", "kind": "drive", "value": {}, "deadline_ms": 300,
+ "operator": {"subject": "usr_1", "role": "queue"}}))
+ 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,
+ "operator": "nobody"}))
+ self.assertEqual(seen, [{"subject": "usr_1", "role": "queue"}, None, None])
+
def test_newer_command_extends_deadline(self):
def test_newer_command_extends_deadline(self):
rt, p, out, clock = make()
rt, p, out, clock = make()
self.hello(rt)
self.hello(rt)