openvibers/openvibe.link Commits

v0.16.0: each driver command carries its provenance (session, seq, Bot's send time and deadline) and a halt's stop its turn_epoch; the runtime refuses replays and ended turns; the no-redelivery contract in plugins.md (#61)

OP OpenVibers committed 17 days ago · 09fdc67031 · 11 files, +255 −21

Changes

ddiff --git a/STATUS.json b/STATUS.json
iindex ce803fd..b3dbff5 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.15.0",
--  "release": "v0.15.0, published 2026-10-26 as a GitHub release (a private desk for agents: desk start/stop/status, and screen, live screen, input and the browser on it with desk: private, never the owner's screen; Vibe by voice: the voice plugin hears the call word on the machine and sends only what follows, and Bot's answers are spoken there): binaries for linux amd64/arm64/armv7, darwin amd64/arm64 and windows amd64, the plugins tarball, install.sh and SHA256SUMS",
++  "version": "0.16.0",
++  "release": "v0.16.0, published 2026-10-27 as a GitHub release (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/docs/hardware-looi.md b/docs/hardware-looi.md
iindex f3c2361..676a4bb 100644
---- a/docs/hardware-looi.md
++++ b/docs/hardware-looi.md
@@@ -22,10 +22,15 @@ The e-stop is latched in Bot and only the owner clears it. Each command tells th
  drives (the driver gets `stop`), so commands from the last driver never come after it
  ([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.
++than the newest halt's, and a driver may check it again against its own record. Commands are never redelivered: Bot
++sends each once under a new id, the Node passes it on at most once and never after its deadline, and each carries
++`cmd.provenance` (the link session, the frame's seq, Bot's send time and deadline). The Python runtime refuses a
++replay on a session and a command from an ended turn by itself
++([plugins.md](plugins.md#what-a-driver-can-rely-on-commands-are-never-redelivered)).
++
++To show the panel under a Live stream, press **Put the buttons under my OpenVibe.Live stream** on the robot's page on
++openvibe.bot: it shows the panel where it is embedded and opens your channel to confirm. The channel page then shows
++the robot's panel beside the phone's stream, and viewers signed in on Live press Drive there.
  
  ## What the driver receives
  
ddiff --git a/docs/plugins.md b/docs/plugins.md
iindex eb2301a..f73c52a 100644
---- a/docs/plugins.md
++++ b/docs/plugins.md
@@@ -48,9 +48,9 @@ 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`?, `turn_epoch`? | act, answer `ack` or `nack` with the same `id`      |
++| `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`      |                                                               | stop every actuator                                    |
++| `stop`      | `turn_epoch`? (a halt's)                                      | stop every actuator                                    |
  | `estop`     |                                                               | stop every actuator and refuse motion until `resume`   |
  | `resume`    |                                                               | accept motion again                                    |
  | `approval`  | `id`, `approved` bool                                         | the owner's answer to an `approval` event it raised    |
@@@ -73,7 +73,52 @@ contracts 0.213.0). It rises when a queue turn ends, its holder leaves, or the o
  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`).
++A halt's epoch also comes with the `stop` it causes (`{"op": "stop", "turn_epoch": …}`), so a driver knows the newest
++epoch even before the next command. Python: `cmd.turn_epoch` (an int, or `None`); the runtime keeps the newest halt's
++and refuses an older command itself (`expired`).
++
++`provenance` is where the command came from, passed unchanged from Bot (Node 0.16):
++
++| field | meaning |
++|---|---|
++| `session` | the link session it arrived on (Bot's `hello`; a new id on every connection, and ids sort by time) |
++| `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 |
++
++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`.
++
++#### What a driver can rely on: commands are never redelivered
++
++Asked by the Generic LOOI driver's authors (2026-10-26). From the code of Bot and this Node:
++
++- **Bot sends a command once.** Each send gets a new id (`cmd_<ULID>`, minted at that moment). A command is sent only
++  while the device is online and is never queued: offline, it is refused, not held. One that is not acknowledged by
++  its deadline plus 1.5 s is recorded as `expired` and never sent again. An operator who repeats their own id gets the
++  first result back and causes no second device command. Nothing is replayed after a reconnect, a Bot restart or a turn
++  change. Jobs (`job_…`) do resend after a reconnect, but they are a different frame and never reach a driver as a
++  `command`.
++- **The Node delivers a command at most once.** It remembers command ids for 10 minutes (up to 4096) across
++  reconnects and answers a repeated id with the first result, without passing it on. It refuses a command whose absolute
++  deadline has passed on Bot's clock (`expired`), so a frame delayed or replayed with its original deadline never
++  reaches a driver. Within a connection, commands reach a plugin in the order Bot sent them.
++- **The boundaries.**
++  - A turn change is a halt carrying a new `turn_epoch`. Epochs are milliseconds on Bot's clock, so they rise across
++    Bot restarts. The core and the Python runtime both refuse older commands after it.
++  - A connection is a `session`. The link is authenticated with the device's credential, and a plugin's stdin comes
++    only from the Node that started it, so whatever arrives there is from that Node.
++- **What is not kept.** The Node's id memory and newest-halt epochs start empty when the Node restarts. So does a
++  plugin's state when the plugin restarts. Commands sent before a restart can no longer arrive afterwards: their
++  connection is gone and their deadlines (at most `max_command_ms`, 300 ms on LOOI) have long passed.
++
++So a driver needs no growing replay cache. Refuse a command if any of these is true:
++- `deadline_ms` has run out;
++- `turn_epoch` is older than the newest halt's;
++- `seq` is not above the last on the same `session`.
++
++The runtime does all three. A driver that wants more can also refuse an `issued_at_ms` older than the newest it has
++handled by more than `max_command_ms`.
  
  ### Plugin → core
  
ddiff --git a/internal/node/botcontract_test.go b/internal/node/botcontract_test.go
iindex 73fe813..3b5d2d4 100644
---- a/internal/node/botcontract_test.go
++++ b/internal/node/botcontract_test.go
@@@ -7,6 +7,7 @@ import (
  	"encoding/json"
  	"os"
  	"path/filepath"
++	"strings"
  	"testing"
  	"time"
  
@@@ -378,3 +379,41 @@ func TestStaleTurnNeverReachesPlugin(t *testing.T) {
  		t.Fatalf("records: %+v", e.records())
  	}
  }
++
++// Provenance (Maticus, 2026-10-26, for the LOOI driver): each command reaches the driver with the link session it came
++// on (Bot's hello), its frame's seq (strictly increasing on that session), Bot's send time and its absolute deadline,
++// so a driver can refuse a replay with constant state. A new connection has a new session.
++func TestProvenanceReachesPlugin(t *testing.T) {
++	e := start(t)
++	drive := json.RawMessage(`{"throttle":0.3}`)
++	for _, id := range []string{"p1", "p2"} {
++		if err := e.conn.Command(protocol.Command{ID: id, Kind: protocol.KindDrive, Value: drive}, 300); err != nil {
++			t.Fatal(err)
++		}
++		if r, err := e.conn.Reply(id, 5*time.Second); err != nil || !isAck(r) {
++			t.Fatalf("%s: %+v %v", id, r, err)
++		}
++	}
++	a, b := e.commandRecords("p1"), e.commandRecords("p2")
++	if len(a) != 1 || len(b) != 1 || a[0].Provenance == nil || b[0].Provenance == nil {
++		t.Fatalf("records: %+v", e.records())
++	}
++	pa, pb := a[0].Provenance, b[0].Provenance
++	if !strings.HasPrefix(pa.Session, "sess_") || pa.Session != pb.Session || pb.Seq <= pa.Seq {
++		t.Fatalf("session and seq: %+v %+v", pa, pb)
++	}
++	if pa.IssuedAtMS <= 0 || pa.ExpiresAtMS != pa.IssuedAtMS+300 {
++		t.Fatalf("times: %+v", pa)
++	}
++	e.conn.Close()
++	e.conn = e.nextConn()
++	if err := e.conn.Command(protocol.Command{ID: "p3", Kind: protocol.KindDrive, Value: drive}, 300); err != nil {
++		t.Fatal(err)
++	}
++	if r, err := e.conn.Reply("p3", 5*time.Second); err != nil || !isAck(r) {
++		t.Fatalf("p3: %+v %v", r, err)
++	}
++	if c := e.commandRecords("p3"); len(c) != 1 || c[0].Provenance == nil || c[0].Provenance.Session == pa.Session {
++		t.Fatalf("a new connection, a new session: %+v", c)
++	}
++}
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex 7a32203..ccb5206 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -103,6 +103,7 @@ type Node struct {
  	configured bool
  
  	robotIDs []string // the robots this device serves, from the connection's hello
++	session  string   // the connection's session id, from its hello: each command's provenance
  
  	// approvals waiting on the machine (the computer plugin at level "ask"), by command id: what
  	// `openvibe-node approvals` lists and `approve`/`deny` answer.
@@@ -295,7 +296,7 @@ func (n *Node) Run(parent context.Context) error {
  // has been applied (applyConfig), so they report the latch the owner may have set while the link was down.
  func (n *Node) Connected() {
  	n.mu.Lock()
--	n.configured, n.robotIDs = false, nil
++	n.configured, n.robotIDs, n.session = false, nil, ""
  	n.mu.Unlock()
  }
  
@@@ -311,13 +312,13 @@ func (n *Node) Frame(f protocol.Frame) {
  	switch m := f.Msg.(type) {
  	case protocol.Hello:
  		n.mu.Lock()
--		n.robotIDs = m.RobotIDs
++		n.robotIDs, n.session = m.RobotIDs, m.SessionID
  		n.mu.Unlock()
  		n.log.Info("server hello", "device", m.DeviceID, "robots", m.RobotIDs, "session", m.SessionID)
  	case protocol.Config:
  		n.applyConfig(m)
  	case protocol.Command:
--		n.command(m, f.TS, time.Now())
++		n.command(m, f.TS, f.Seq, time.Now())
  	case protocol.Estop:
  		// latched:false is the owner's clear. It clears only the remote latch: a local (CLI or file) stop stays.
  		if m.Latched {
@@@ -457,7 +458,7 @@ func (n *Node) voiceAnswer(a protocol.VoiceAnswer) {
  // commands in arrival order; the reply is awaited on its own goroutine. deadline_ms is an absolute instant on the
  // server's clock: it is compared with the link's estimate of that clock at receipt (serverNow), so a command that
  // arrives after its deadline is nacked and never reaches a plugin, and a plugin gets only the time that is left.
--func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
++func (n *Node) command(c protocol.Command, sentMS int64, seq uint64, received time.Time) {
  	if c.ID == "" {
  		n.log.Warn("command without id ignored", "kind", c.Kind)
  		return
@@@ -490,11 +491,11 @@ func (n *Node) command(c protocol.Command, sentMS int64, received time.Time) {
  		return
  	}
  	n.mu.Lock()
--	limits, allowed, configured := n.limits, n.allowed, n.configured
++	limits, allowed, configured, session := n.limits, n.allowed, n.configured, n.session
  	n.mu.Unlock()
  	if c.Kind == protocol.KindHalt {
  		n.noteHaltEpoch(c.RobotID, c.TurnEpoch)
--		n.mgr.StopAll()
++		n.mgr.HaltAll(c.TurnEpoch)
  		r := protocol.Ack{ID: c.ID, LatencyMS: time.Since(received).Milliseconds()}
  		n.dedup.Complete(entry, r)
  		n.send(r)
@@@ -561,7 +562,9 @@ 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.BeginTurn(n.ctx, c.ID, c.Kind, value, deadline, c.Operator, c.TurnEpoch)
++	// 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}
++	wait := p.BeginWith(n.ctx, c.ID, c.Kind, value, deadline, c.Operator, c.TurnEpoch, prov)
  	go func() {
  		r := wait()
  		var msg protocol.Message
ddiff --git a/internal/node/node_test.go b/internal/node/node_test.go
iindex 20256e2..cbba2f5 100644
---- a/internal/node/node_test.go
++++ b/internal/node/node_test.go
@@@ -172,6 +172,12 @@ type rec struct {
  	Value      json.RawMessage `json:"value"`
  	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"`
++	} `json:"provenance"`
  }
  
  func (e *env) records() []rec {
ddiff --git a/internal/plugins/manager.go b/internal/plugins/manager.go
iindex 9d6a302..a27dc1b 100644
---- a/internal/plugins/manager.go
++++ b/internal/plugins/manager.go
@@@ -77,6 +77,13 @@ func (m *Manager) StopAll() {
  	}
  }
  
++// HaltAll stops every plugin for a halt command, telling each the halt's turn_epoch (Plugin.Halt).
++func (m *Manager) HaltAll(turnEpoch int64) {
++	for _, p := range m.plugins {
++		p.Halt(turnEpoch)
++	}
++}
++
  func (m *Manager) EstopAll() {
  	for _, p := range m.plugins {
  		p.Estop()
ddiff --git a/internal/plugins/plugin.go b/internal/plugins/plugin.go
iindex 7360705..baaf670 100644
---- a/internal/plugins/plugin.go
++++ b/internal/plugins/plugin.go
@@@ -185,6 +185,16 @@ 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"}) }
  
++// 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
++	}
++	_ = p.send(map[string]any{"op": "stop", "turn_epoch": turnEpoch})
++}
++
  // Estop latches the plugin's e-stop (the core also refuses motion itself).
  func (p *Plugin) Estop() {
  	p.mu.Lock()
@@@ -253,6 +263,23 @@ func (p *Plugin) BeginAs(ctx context.Context, id, kind string, value json.RawMes
  // 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 {
++	return p.BeginWith(ctx, id, kind, value, deadlineMS, op, turnEpoch, nil)
++}
++
++// Provenance is where a command came from, passed unchanged from Bot to the driver as `provenance` (plugins.md): the
++// link session it arrived on (Bot's hello; a new one on every connection, and session ids sort by time), the frame's
++// seq on that session (strictly increasing, with gaps), Bot's clock when it sent the command (and minted its id), and
++// the absolute deadline on that clock for a motion kind. Bot sends a command once and never again under any id, so
++// a driver can refuse a replay with constant state: the same session with a seq not above the last one it handled.
++type Provenance struct {
++	Session     string `json:"session,omitempty"`
++	Seq         uint64 `json:"seq,omitempty"`
++	IssuedAtMS  int64  `json:"issued_at_ms,omitempty"`
++	ExpiresAtMS int64  `json:"expires_at_ms,omitempty"`
++}
++
++// BeginWith is BeginTurn with the command's provenance (nil: none).
++func (p *Plugin) BeginWith(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int, op *protocol.Operator, turnEpoch int64, prov *Provenance) func() Reply {
  	now := func(r Reply) func() Reply { return func() Reply { return r } }
  	p.mu.Lock()
  	if p.state != StateReady && kind != protocol.KindHalt {
@@@ -288,6 +315,9 @@ func (p *Plugin) BeginTurn(ctx context.Context, id, kind string, value json.RawM
  	if turnEpoch > 0 {
  		msg["turn_epoch"] = turnEpoch
  	}
++	if prov != nil {
++		msg["provenance"] = prov
++	}
  	if err := p.send(msg); err != nil {
  		done()
  		return now(Reply{FaultCode: protocol.FaultPluginDown})
ddiff --git a/plugins/dryrun/openvibe_dryrun/__init__.py b/plugins/dryrun/openvibe_dryrun/__init__.py
iindex 4fd4597..ffe1466 100644
---- a/plugins/dryrun/openvibe_dryrun/__init__.py
++++ b/plugins/dryrun/openvibe_dryrun/__init__.py
@@@ -76,7 +76,7 @@ class DryRun(Plugin):
              raise Fault("unsupported", "kind %s" % cmd.kind)
          if self.log_commands:
              self.log("dryrun: %s %s (deadline %d ms)", cmd.kind, json.dumps(v, sort_keys=True), cmd.deadline_ms)
--        self.record("command", id=cmd.id, kind=cmd.kind, value=v, deadline_ms=cmd.deadline_ms)
++        self.record("command", id=cmd.id, kind=cmd.kind, value=v, deadline_ms=cmd.deadline_ms, provenance=cmd.provenance)
  
      def stop(self):
          if self.state["throttle"] or self.state["steer"]:
ddiff --git a/plugins/sdk/openvibe_plugin/__init__.py b/plugins/sdk/openvibe_plugin/__init__.py
iindex aaacb87..01eaf7e 100644
---- a/plugins/sdk/openvibe_plugin/__init__.py
++++ b/plugins/sdk/openvibe_plugin/__init__.py
@@@ -62,10 +62,16 @@ class Command:
      "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")
++    anything older (defence in depth).
  
--    def __init__(self, id, kind, value, deadline_ms, now, target=None, operator=None, turn_epoch=None):
++    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")."""
++    __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):
          self.id = id
          self.kind = kind
          self.value = value if value is not None else {}
@@@ -74,11 +80,26 @@ class Command:
          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
++        self.provenance = _provenance(provenance)
  
      def __repr__(self):
          return "Command(id=%r, kind=%r, value=%r, deadline_ms=%r)" % (self.id, self.kind, self.value, self.deadline_ms)
  
  
++def _provenance(p):
++    """The core's provenance dict with only well-typed fields, or None."""
++    if not isinstance(p, dict):
++        return None
++    out = {}
++    if isinstance(p.get("session"), str) and p["session"]:
++        out["session"] = p["session"]
++    for k in ("seq", "issued_at_ms", "expires_at_ms"):
++        v = p.get(k)
++        if isinstance(v, int) and not isinstance(v, bool) and v > 0:
++            out[k] = v
++    return out or None
++
++
  def clamp(v, lo, hi):
      try:
          v = float(v)
@@@ -216,6 +237,12 @@ class Runtime:
          self.setup_retry_at = None
          self.closed = False
          self.stop_reasons = []  # for tests: why stop() was called, in order
++        # 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
++        # it is a replay (Bot never sends one; this holds even if something did).
++        self.last_session = None
++        self.last_seq = 0
  
      # Output.
      def send(self, obj):
@@@ -286,6 +313,9 @@ 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:
++                self.halt_epoch = epoch
              self.safe_stop("stop")
          elif op == "estop":
              self.estopped = True
@@@ -365,11 +395,25 @@ 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"), msg.get("turn_epoch"))
++        cmd = Command(cid, kind, msg.get("value"), dl, now, msg.get("target"), msg.get("operator"), msg.get("turn_epoch"),
++                      msg.get("provenance"))
          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.send({"op": "ack", "id": cid})
              return
++        if cmd.turn_epoch and cmd.turn_epoch < self.halt_epoch:
++            self.send({"op": "nack", "id": cid, "fault_code": "expired",
++                       "message": "from a turn that ended: epoch %d is older than the newest halt's %d" % (cmd.turn_epoch, self.halt_epoch)})
++            return
++        prov = cmd.provenance or {}
++        if prov.get("session") and prov.get("seq"):
++            if prov["session"] == self.last_session and prov["seq"] <= self.last_seq:
++                self.send({"op": "nack", "id": cid, "fault_code": "expired",
++                           "message": "a replay: seq %d is not above %d on this session" % (prov["seq"], self.last_seq)})
++                return
++            self.last_session, self.last_seq = prov["session"], prov["seq"]
          if not self.ready:
              self.send({"op": "nack", "id": cid, "fault_code": "not_ready"})
              return
ddiff --git a/plugins/sdk/tests/test_runtime.py b/plugins/sdk/tests/test_runtime.py
iindex 1e57c53..30c05a3 100644
---- a/plugins/sdk/tests/test_runtime.py
++++ b/plugins/sdk/tests/test_runtime.py
@@@ -125,6 +125,61 @@ class RuntimeTest(unittest.TestCase):
                                   "turn_epoch": "soon"}))
          self.assertEqual(seen, [1792900000123, None, None])
  
++    def test_provenance_reaches_handle(self):
++        seen = []
++
++        class Prov(Recorder):
++            def handle(self, cmd):
++                seen.append(cmd.provenance)
++
++        rt, p, out, clock = make(Prov())
++        self.hello(rt)
++        prov = {"session": "sess_01J9", "seq": 7, "issued_at_ms": 1792900000000, "expires_at_ms": 1792900000300}
++        rt.feed_line(json.dumps({"op": "command", "id": "c1", "kind": "drive", "value": {}, "deadline_ms": 300,
++                                 "provenance": prov}))
++        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,
++                                 "provenance": {"session": 5, "seq": "8", "issued_at_ms": True}}))
++        self.assertEqual(seen, [prov, None, None])
++
++    def test_replay_on_the_same_session_is_refused(self):
++        # Bot sends a command once and never again; if a frame came back anyway, its seq is not above the last one
++        # handled on its session, and the runtime refuses it with no state but that pair.
++        rt, p, out, clock = make()
++        self.hello(rt)
++
++        def cmd(cid, session, seq):
++            rt.feed_line(json.dumps({"op": "command", "id": cid, "kind": "drive", "value": {}, "deadline_ms": 300,
++                                     "provenance": {"session": session, "seq": seq, "issued_at_ms": 1792900000000}}))
++            return lines(out)[-1]
++
++        self.assertEqual(cmd("c1", "sess_A", 5)["op"], "ack")
++        replay = cmd("c1-again", "sess_A", 5)
++        self.assertEqual((replay["op"], replay["fault_code"]), ("nack", "expired"))
++        self.assertIn("replay", replay["message"])
++        self.assertEqual(cmd("c0", "sess_A", 3)["op"], "nack", "an older frame on the same session")
++        self.assertEqual(cmd("c2", "sess_A", 9)["op"], "ack", "seq has gaps")
++        self.assertEqual(cmd("c3", "sess_B", 1)["op"], "ack", "a new connection starts its own count")
++
++    def test_a_halts_epoch_refuses_older_turns(self):
++        rt, p, out, clock = make()
++        self.hello(rt)
++
++        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("c1", 100)["op"], "ack")
++        rt.feed_line(json.dumps({"op": "stop", "turn_epoch": 200}))
++        stale = drive("c2", 100)
++        self.assertEqual((stale["op"], stale["fault_code"]), ("nack", "expired"))
++        self.assertEqual(drive("c3", 200)["op"], "ack")
++        rt.feed_line(json.dumps({"op": "command", "id": "h", "kind": "halt", "turn_epoch": 300}))
++        self.assertEqual(drive("c4", 200)["op"], "nack", "a halt command's epoch counts too")
++        rt.feed_line(json.dumps({"op": "stop"}))
++        self.assertEqual(drive("c5", 300)["op"], "ack", "a stop without an epoch changes nothing")
++
      def test_newer_command_extends_deadline(self):
          rt, p, out, clock = make()
          self.hello(rt)