v0.18.0: queue and pipe time count against a motion command's deadline (the writer stamps what is left; expires_at_mono_ns on the shared clock); the clock bound is exact to the nanosecond and counts Bot's millisecond stamps; a Node booting latched sets the e-stop before starting its plugins (#64)

ddiff --git a/STATUS.json b/STATUS.json
iindex db5683f..4a8b80c 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.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",
++  "version": "0.18.0",
++  "release": "v0.18.0, published 2026-10-27 as a GitHub release (a motion command's validity also counts the time it waits in the queue to its plugin and in the pipe: the core fixes the instant it expires at receipt, writes what is left at the moment it writes the line with expires_at_mono_ns on the clock it shares with the plugin, and never writes one that has run out, and the Python runtime refuses one that ran out in the pipe before handle(); the clock bound is kept in nanoseconds and counts Bot's whole-millisecond stamps, rounded up from nanoseconds; a Node booting latched sets the e-stop on its plugins before starting them; the Python SDK package reports the Node release; v0.17.0: 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,
@@@ -29,5 +29,5 @@
      "a tray on macOS (approvals there are a dialog with Deny as its default)",
      "the computer plugin's browser on Windows"
    ],
--  "updated": "2026-10-22"
++  "updated": "2026-10-27"
  }
ddiff --git a/docs/plugins.md b/docs/plugins.md
iindex 6b4b09c..b9e19d8 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, `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` |
++| `command`   | `id`, `kind`, `value`, `deadline_ms`, `expires_at_mono_ns`? (0.18), `operator`?, `turn_epoch`?, `provenance`? | act, answer `ack` or `nack` with the same `id` |
  | `heartbeat` | `t` (core clock, ms)                                          | note the time                                          |
  | `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`   |
@@@ -59,7 +59,10 @@ describes itself and waits for EOF without opening any hardware.
  | `voice_answer` | `id`, `state` (working, answered, refused), `text`, `url`  | (voice plugins) Vibe's answer to a `voice_ask` event the plugin raised: speak and show it |
  
  `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. Since Node 0.18 the core counts it down to the moment it writes
++the line, and adds `expires_at_mono_ns`: the instant the command expires on `CLOCK_MONOTONIC`, the clock every process on
++the machine shares (absent on Windows). A plugin that reads that clock when the line arrives takes off the time the line
++sat in the pipe too (answer 8); the Python runtime does.
  
  `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
@@@ -84,8 +87,8 @@ and refuses an older command itself (`expired`).
  | `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 |
--| `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 |
++| `expires_at_ms` | the absolute deadline on Bot's clock (motion kinds); `deadline_ms` is what is left of it when the core writes the line |
++| `clock_uncertainty_ms` | (0.17) how far the core's estimate of Bot's clock may trail it: the connection's fastest round trip, plus 1 ms for Bot's whole-millisecond stamps since 0.18, rounded up from nanoseconds. `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`.
@@@ -136,8 +139,10 @@ Asked by the Generic LOOI driver's authors (2026-10-27). Each answer is what Bot
       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.
++   - It caps what is left at the robot's `max_command_ms`, and fixes the instant it runs out on the machine's own clock:
++     receipt plus that remainder.
++   - The plugin gets what is left of it when the core writes the line, as `deadline_ms`, with that instant as
++     `expires_at_mono_ns` (0.18, answer 8), and `expires_at_ms` in `provenance` unchanged.
     - Neither a viewer nor Live can choose a later deadline.
  2. **Nothing re-stamps a command.**
     - Bot sends each device command once and never again, under any id. One with no answer by its deadline plus 1.5 s
@@@ -182,12 +187,15 @@ Asked by the Generic LOOI driver's authors (2026-10-27). Each answer is what Bot
     - The core decides at arrival:
       - it refuses a command whose deadline has passed;
       - it refuses a plugin that is down or not ready at once (`plugin_down`, `not_ready`) rather than holding the command;
--     - it hands the rest straight to the plugin's stdin. A plugin that stops reading is killed, so no line waits there.
++     - it queues the rest for the plugin's stdin. A plugin that stops reading until 256 lines are waiting is killed.
++       Until then lines can wait in that queue behind a full pipe; since 0.18 that time counts against a command, and one
++       whose time runs out there is answered `expired` and never written (answer 8).
     - A plugin that does not answer within max(deadline, 1 s) is told to `stop`, and the command fails with
       `plugin_timeout`.
     - The Python runtime counts each deadline from the moment its line arrived. It refuses (`expired`) a motion command
       whose deadline passed while it waited behind a slow handler, so the driver never sees it. Before Node 0.16.1 such a
--     command still reached `handle` and was stopped on the next tick.
++     command still reached `handle` and was stopped on the next tick. Since 0.18 it also takes off the time the line sat
++     in the pipe before it was read (`expires_at_mono_ns`).
     - 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.
  
@@@ -200,8 +208,8 @@ Asked next by the same authors (2026-10-27, P7 LOOI shadow integration), and cha
       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.
++     The core checks every deadline against estimate + bound (the latest Bot's clock can be). This is fail-closed: a
++     command can only get less time than its deadline allows, never more. (0.18 made the arithmetic strict: answer 8.)
     - **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
@@@ -244,6 +252,93 @@ Asked next by the same authors (2026-10-27, P7 LOOI shadow integration), and cha
       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.
  
++Reviewed again by the same authors against v0.17.0 (2026-10-27, `632207f`), and changed in Node 0.18.0:
++
++8. **Node-to-plugin transit, the bound's rounding, and what a driver may assume.**
++   - **Transit extended a command (confirmed, fixed).** In 0.17 the core turned the deadline into `deadline_ms` when the
++     frame arrived, then queued the line for the plugin's stdin. Lines wait in that queue while the pipe is full (a
++     plugin slow to read), and the Python runtime counted `deadline_ms` from when its reader got the line. So queue and
++     pipe time were not counted: a command with 100 ms left that waited 200 ms could still move for 100 ms. Nothing in
++     0.17 prevented it. In 0.18:
++     - The core fixes the instant a command expires when it arrives: receipt plus what was left (capped at
++       `max_command_ms`).
++     - The writer stamps the line when it writes it: `deadline_ms` is the whole milliseconds left then, and
++       `expires_at_mono_ns` the instant on `CLOCK_MONOTONIC`. A command with less than 1 ms left is never written. The
++       core answers it `expired`, saying how long it waited.
++     - The Python runtime reads both clocks when a line arrives (its own clock first, then the shared one, so the time
++       left can only come out short). It refuses (`expired`) a motion command whose `expires_at_mono_ns` has passed, so
++       `handle()` never runs. Otherwise the command gets the smaller of `deadline_ms` and what that instant leaves.
++     - Time the line then waits behind a slow handler counts as before (answer 5).
++     - The tests:
++       - `TestWriterQueueDelayCountsAgainstTheDeadline` (Go) fills the pipe while the plugin sleeps 400 ms. A command
++         with 100 ms is answered `expired` and never reaches the plugin. One with 1.5 s reaches it with at most what is
++         left: its `deadline_ms`, and on the plugin's own clock.
++       - `test_a_real_pipe_delay_never_extends_a_deadline` (Python) leaves lines in an OS pipe for 200 ms. The 100 ms
++         command never reaches `handle()`, and the 1 s one gets at most 800 ms.
++       - `test_pipe_delay_counts_against_the_deadline` covers the edges with fake clocks: 0.999999 ms left is expired,
++         and pipe plus inbox delay add up.
++       - Each test fails with the correction removed.
++   - **Clock domains.** There are three:
++     - Bot's clock: the absolute deadline, read through the link's estimate and its bound (answer 6).
++     - The core's monotonic clock: Go's, so a wall-clock step changes nothing.
++     - `CLOCK_MONOTONIC`, between the core and the plugin. The core reads it with `clock_gettime`; Python reads it with
++       `time.clock_gettime_ns(time.CLOCK_MONOTONIC)`. It is one clock for every process on the machine, and the core's
++       monotonic clock runs at the same rate.
++
++     The correction assumes the plugin runs on the same machine as the core (the core starts it, over a pipe). It also
++     assumes both read the same clock id. On Windows there is no shared clock: `expires_at_mono_ns` is absent, and only
++     the core's queue time is counted, up to the write.
++   - **The bound's rounding (confirmed, fixed).** 0.17 cut a round trip to whole microseconds before rounding up, so
++     1,000,001 ns came out as 1 ms. Two more roundings were not counted at all:
++     - Bot stamps `ts` in whole milliseconds, rounded down. A frame may have left up to 1 ms after its stamp, so the
++       estimate could trail by that much more than the round trip.
++     - The core kept receipt times in whole milliseconds.
++
++     In 0.18:
++     - The estimate is kept in nanoseconds.
++     - The bound is the fastest round trip plus 1 ms for the stamps.
++     - `timeLeft` works in nanoseconds and rounds the time left down.
++     - `clock_uncertainty_ms` and the `MaxClockUncertaintyMS` check round the bound up from nanoseconds. A bound of
++       exactly 1 s is still admitted, and 1 s + 1 ns is refused.
++     - So `clock_uncertainty_ms` reads about 1 ms higher than in 0.17.
++     - Admission is otherwise unchanged: no motion before a connection's first round trip, none over 1 s.
++     - The tests: `TestCeilMSRoundsUpFromNanoseconds` (nanosecond and microsecond boundaries),
++       `TestServerNowKeepsNanoseconds`, and the bound in `TestClockUncertaintyIsTheFastestRoundTrip`.
++   - **Stops and local authority (confirmed, no change).**
++     - Every stop the runtime makes calls `on_stop(StopInfo)`:
++       - `info.emergency` is true only for `estop`.
++       - `info.cause` says which (answer 7).
++       - `info.halt_epoch` is the newest halt epoch the runtime knows, and comes from `hello` after a restart.
++       - `info.turn_epoch` is the epoch of the halt that caused this stop.
++     - An ordinary stop (`halt`, `link_lost`, `shutdown`, `timeout`, `deadline`, `heartbeat`) only says "stop now". It
++       latches nothing in the core or the runtime, so an adapter may revoke remote motion without touching a newer local
++       owner's command.
++     - `on_resume` is information. Its default does nothing, and the runtime re-arms nothing, so P7 is not re-armed and
++       no local grant comes back.
++   - **A replayed e-stop comes before any motion.** The core keeps the latch on disk and sets it before it starts any
++     plugin.
++     - **Plugin restart.** The process gets `hello` with the latest halt epoch, then the e-stop again (`latched`,
++       `replayed: true`, `first_cause`). No command is routed to it before its `ready`, and lines reach it in the order
++       they were queued. So the replay arrives before any command, and the runtime refuses motion (`estopped`) from
++       then on.
++     - **Node reconnect.** No command is accepted until Bot's config has been applied on the new connection, which sets
++       the latch first.
++     - **Defence in depth.** The core itself refuses `drive`, `actuator`, `ptz`, `button` and `point` while latched,
++       whatever the plugin's state. `test_a_replayed_estop_is_applied_before_any_motion`, `TestCrashRestartsWithEstop` and
++       `TestLatchSetBeforeStartIsReplayedFirst` hold the order. (Before 0.18 a Node booting latched told its plugins just
++       after starting them. That was also safe, by the core's own refusal, but is now in the order written here.)
++   - **What a strict guarantee still assumes.** Each of these makes a command read as expired, never as open:
++     - Bot's clock does not step back during a connection (that reads as expired until the next one).
++     - The host is not suspended mid-command: `CLOCK_MONOTONIC` stops while suspended, and the link is lost on resume
++       anyway.
++     - The plugin is on the same machine.
++
++     Bounded but not removed:
++     - The gap between the runtime's check and the driver's own write to the hardware (the driver's code).
++     - Time inside `handle()`.
++   - **Versions.** The Python SDK package now carries the Node release (`openvibe_plugin.__version__`, `pyproject.toml`:
++     0.18.0; it said 0.1.0 before). For an exact source, pin the commit, not the version.
++
  ### Plugin → core
  
  | op          | fields                                                                                      |
ddiff --git a/go.mod b/go.mod
iindex 3a189c7..607c683 100644
---- a/go.mod
++++ b/go.mod
@@@ -11,6 +11,7 @@ require (
  	github.com/pion/interceptor v0.1.49
  	github.com/pion/rtcp v1.2.18
  	github.com/pion/webrtc/v4 v4.2.22
++	golang.org/x/sys v0.41.0
  )
  
  require (
@@@ -30,6 +31,5 @@ require (
  	github.com/wlynxg/anet v0.0.5 // indirect
  	golang.org/x/crypto v0.48.0 // indirect
  	golang.org/x/net v0.50.0 // indirect
--	golang.org/x/sys v0.41.0 // indirect
  	golang.org/x/time v0.14.0 // indirect
  )
ddiff --git a/internal/link/link.go b/internal/link/link.go
iindex 2cedad7..9e3c6e2 100644
---- a/internal/link/link.go
++++ b/internal/link/link.go
@@@ -102,8 +102,9 @@ type Link struct {
  	heartbeat time.Duration
  	deadman   time.Duration
  
--	// The server clock estimate of this connection (see ServerNow): skew is the largest ts − local receive time (ms)
--	// seen since clockBase, the local time of its first frame; hasSkew is false until a frame arrives.
++	// The server clock estimate of this connection (see ServerNow): skew is the largest ts − local receive time seen
++	// (ns; receive times are measured from clockBase, the local time of its first frame, on the monotonic clock);
++	// hasSkew is false until a frame arrives.
  	skew      int64
  	hasSkew   bool
  	clockBase time.Time
@@@ -168,41 +169,42 @@ func (l *Link) observe(ts int64, at time.Time) {
  	l.mu.Lock()
  	defer l.mu.Unlock()
  	if !l.hasSkew {
--		l.clockBase, l.skew, l.hasSkew = at, ts-at.UnixMilli(), true
++		l.clockBase, l.skew, l.hasSkew = at, ts*int64(time.Millisecond), true
  		return
  	}
--	l.skew = max(l.skew, ts-l.localMS(at))
++	l.skew = max(l.skew, ts*int64(time.Millisecond)-int64(at.Sub(l.clockBase)))
  }
  
--// localMS is local time at in epoch ms, measured from clockBase on the monotonic clock so a step of the wall clock
--// cannot move the estimate.
--func (l *Link) localMS(at time.Time) int64 {
--	return l.clockBase.UnixMilli() + at.Sub(l.clockBase).Milliseconds()
--}
++// TSResolution is the resolution of the server's ts: Bot stamps whole milliseconds (Date.now(), rounded down), so a
++// frame may have left up to this much after its ts says.
++const TSResolution = time.Millisecond
  
--// ServerNow estimates the server's clock (epoch ms) at local time at, from the frames of the current connection. The
++// ServerNow estimates the server's clock (epoch ns) 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 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) {
++// that fastest transit (and up to TSResolution more); ClockUncertainty bounds it, and deadlines are checked against
++// estimate + bound. Local times are measured in nanoseconds on the monotonic clock, so neither a step of the wall clock
++// nor rounding moves the estimate later. 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) (ns int64, ok bool) {
  	l.mu.Lock()
  	defer l.mu.Unlock()
  	if !l.hasSkew {
  		return 0, false
  	}
--	return l.localMS(at) + l.skew, true
++	return int64(at.Sub(l.clockBase)) + 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).
++// heartbeat round trip measured on it, plus TSResolution. ServerNow trails the server by the fastest one-way transit
++// among the frames it sampled, plus how far that frame's whole-millisecond ts was rounded down; 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 plus the ts
++// resolution is an upper bound whatever the asymmetry of the path. Round trips are kept to the nanosecond. 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 time.Duration(v) + TSResolution, true
  	}
  	return 0, false
  }
ddiff --git a/internal/link/link_test.go b/internal/link/link_test.go
iindex d0f9bbb..7c78105 100644
---- a/internal/link/link_test.go
++++ b/internal/link/link_test.go
@@@ -235,17 +235,18 @@ func TestClockUncertaintyIsTheFastestRoundTrip(t *testing.T) {
  		}
  		return l.Stats().RTT
  	}
++	// The bound is the fastest round trip plus the server's ts resolution (it stamps whole milliseconds).
  	slow := ack(80 * time.Millisecond)
  	u, ok := l.ClockUncertainty()
--	if !ok || u != slow || u < 80*time.Millisecond {
++	if !ok || u != slow+TSResolution || 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 {
++	if u, _ := l.ClockUncertainty(); u != fast+TSResolution || 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 {
++	if u, _ := l.ClockUncertainty(); u != fast+TSResolution {
  		t.Fatalf("a slower round trip loosened the bound: %v (fast %v)", u, fast)
  	}
  	c.Close()
@@@ -495,26 +496,44 @@ func TestServerNow(t *testing.T) {
  	if _, ok := l.ServerNow(time.Now()); ok {
  		t.Fatal("an estimate before any frame")
  	}
++	ms := func(at time.Time) int64 {
++		ns, _ := l.ServerNow(at)
++		return ns / int64(time.Millisecond)
++	}
  	l.observe(10_000, time.UnixMilli(9_900))  // server 100 ms ahead, fast frame
  	l.observe(10_000, time.UnixMilli(10_500)) // a frame that sat 600 ms in a queue
--	if ms, ok := l.ServerNow(time.UnixMilli(20_000)); !ok || ms != 20_100 {
--		t.Fatalf("server now %d %v", ms, ok)
++	if _, ok := l.ServerNow(time.UnixMilli(20_000)); !ok || ms(time.UnixMilli(20_000)) != 20_100 {
++		t.Fatalf("server now %d %v", ms(time.UnixMilli(20_000)), ok)
  	}
  	// A long run of frames that each sat 500 ms on the way never displaces the timely one: an expired deadline must
  	// not read as open again.
  	for i := 0; i < 100; i++ {
  		l.observe(int64(30_000+i*10), time.UnixMilli(int64(30_000+i*10)+400))
  	}
--	if ms, _ := l.ServerNow(time.UnixMilli(40_000)); ms != 40_100 {
--		t.Fatalf("delayed frames moved the estimate: %d", ms)
++	if got := ms(time.UnixMilli(40_000)); got != 40_100 {
++		t.Fatalf("delayed frames moved the estimate: %d", got)
  	}
  	// A new connection starts a new estimate.
  	l.mu.Lock()
  	l.hasSkew = false
  	l.mu.Unlock()
  	l.observe(50_000, time.UnixMilli(50_000))
--	if ms, _ := l.ServerNow(time.UnixMilli(60_000)); ms != 60_000 {
--		t.Fatalf("estimate across connections: %d", ms)
++	if got := ms(time.UnixMilli(60_000)); got != 60_000 {
++		t.Fatalf("estimate across connections: %d", got)
++	}
++}
++
++// The estimate keeps nanoseconds (Maticus, 2026-10-27): a frame received 0.6 ms into a millisecond moves the estimate
++// by those 0.6 ms, not by a whole millisecond either way.
++func TestServerNowKeepsNanoseconds(t *testing.T) {
++	l := New(Options{URL: "ws://127.0.0.1:1/device"}, newRecorder())
++	base := time.Unix(1_000, 0)
++	l.observe(5_000_000, base)
++	// A later frame that left 0.6 ms closer: the server is 0.6 ms further ahead than the first sample said.
++	l.observe(5_000_001, base.Add(400*time.Microsecond))
++	ns, _ := l.ServerNow(base.Add(10 * time.Millisecond))
++	if want := int64(5_000_000)*int64(time.Millisecond) + int64(10*time.Millisecond) + int64(600*time.Microsecond); ns != want {
++		t.Fatalf("server now %d, want %d (off by %v)", ns, want, time.Duration(ns-want))
  	}
  }
  
ddiff --git a/internal/node/clock_test.go b/internal/node/clock_test.go
nnew file mode 100644
iindex 0000000..d217a45
---- /dev/null
++++ b/internal/node/clock_test.go
@@@ -0,0 +1,48 @@
++package node
++
++import (
++	"testing"
++	"time"
++)
++
++// The clock bound is rounded up from nanoseconds (Maticus, 2026-10-27): v0.17.0 truncated a round trip to whole
++// microseconds first, so 1,000,001 ns came out as 1 ms.
++func TestCeilMSRoundsUpFromNanoseconds(t *testing.T) {
++	for _, c := range []struct {
++		d    time.Duration
++		want int64
++	}{
++		{0, 0},
++		{1, 1},                        // one nanosecond is still a bound
++		{999, 1},                      // under a microsecond
++		{time.Microsecond, 1},         // exactly one microsecond
++		{999_999, 1},                  // a nanosecond under a millisecond
++		{time.Millisecond, 1},         // exactly one millisecond
++		{1_000_001, 2},                // the reported case
++		{1_000_999, 2},                // under a microsecond over (v0.17.0: 1)
++		{1_001_000, 2},                // a whole microsecond over
++		{2*time.Millisecond - 1, 2},   // a nanosecond under two
++		{2 * time.Millisecond, 2},     // exactly two
++		{2*time.Millisecond + 1, 3},   // a nanosecond over two
++		{25*time.Millisecond - 1, 25}, // a typical round trip less a nanosecond
++		{25*time.Millisecond + 1, 26}, // and plus one
++		{999_999_999, 1000},           // a nanosecond under MaxClockUncertaintyMS
++		{time.Second, 1000},           // exactly MaxClockUncertaintyMS: still admitted
++		{time.Second + 1, 1001},       // a nanosecond over: refused
++		{time.Hour, 3_600_000},        // no overflow on a long bound
++	} {
++		if got := ceilMS(c.d); got != c.want {
++			t.Errorf("ceilMS(%d ns) = %d, want %d", int64(c.d), got, c.want)
++		}
++	}
++}
++
++// Without a link (local commands) the time left is exact to the nanosecond: nothing rounds a deadline later.
++func TestTimeLeftIsExact(t *testing.T) {
++	n := &Node{}
++	received := time.UnixMilli(1_000_000).Add(400 * time.Microsecond)
++	left, unc, synced := n.timeLeft(1_000_100, 0, received)
++	if left != 100*time.Millisecond-400*time.Microsecond || unc != 0 || !synced {
++		t.Fatalf("left %v unc %d synced %v", left, unc, synced)
++	}
++}
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex b05e9f8..57b0240 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -252,11 +252,17 @@ func (n *Node) Run(parent context.Context) error {
  	}
  
  	pctx, cancelPlugins := context.WithCancel(context.Background())
--	n.mgr.Start(pctx)
++	// A stop latched on disk is set on every plugin before any starts: each process's first lines are then hello and
++	// the e-stop replayed (cause latched, first_cause what set it), before it can be routed any command.
  	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))
++		first := plugins.EstopLocal
++		if st.Remote {
++			first = plugins.EstopRemote
++		}
++		n.mgr.EstopAll(first, latchReason(st))
  	}
++	n.mgr.Start(pctx)
  
  	var wg sync.WaitGroup
  	wg.Add(3)
@@@ -457,7 +463,9 @@ func (n *Node) voiceAnswer(a protocol.VoiceAnswer) {
  // command validates and routes one command. The synchronous part (checks, clamping, queueing to the plugin) keeps
  // 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.
++// arrives after its deadline is nacked and never reaches a plugin, and a plugin gets only the time that is left. What is
++// left becomes an instant on this machine (receipt + left): time the command then spends in the Node and in the queue
++// to the plugin counts against it, and the plugin is told the instant on the clock it shares with the Node (plugins.md).
  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)
@@@ -532,8 +540,10 @@ func (n *Node) command(c protocol.Command, sentMS int64, seq uint64, received ti
  	}
  	deadline := limits.Deadline(0)
  	var uncertainty *int64
++	var unc int64
  	if c.DeadlineMS > 0 {
--		left, unc, synced := n.timeLeft(c.DeadlineMS, sentMS, received)
++		left, u, synced := n.timeLeft(c.DeadlineMS, sentMS, received)
++		unc = u
  		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.
@@@ -546,11 +556,12 @@ func (n *Node) command(c protocol.Command, sentMS int64, seq uint64, received ti
  				return
  			}
  		}
--		if left <= 0 {
--			nack(protocol.FaultExpired, fmt.Sprintf("the deadline passed %d ms before the command arrived (counting %d ms of clock uncertainty)", -left, unc))
++		// deadline_ms is whole milliseconds, at least 1: a command with less than that left has none.
++		if left < time.Millisecond {
++			nack(protocol.FaultExpired, fmt.Sprintf("the deadline passed %d ms before the command arrived (counting %d ms of clock uncertainty)", max(0, ceilMS(-left)), unc))
  			return
  		}
--		deadline = limits.Deadline(int(min(left, int64(limits.MaxCommandMS)+1)))
++		deadline = limits.Deadline(int(min(int64(left/time.Millisecond), int64(limits.MaxCommandMS)+1)))
  		if synced && n.link != nil {
  			uncertainty = &unc
  		}
@@@ -580,7 +591,9 @@ func (n *Node) command(c protocol.Command, sentMS int64, seq uint64, received ti
  	}
  	// 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, ClockUncertaintyMS: uncertainty}
--	wait := p.BeginWith(n.ctx, c.ID, c.Kind, value, deadline, c.Operator, c.TurnEpoch, prov)
++	// The window counts from receipt, not from now: nothing the Node does after the frame arrived extends it.
++	expires := received.Add(time.Duration(deadline) * time.Millisecond)
++	wait := p.BeginWith(n.ctx, c.ID, c.Kind, value, expires, c.Operator, c.TurnEpoch, prov)
  	go func() {
  		r := wait()
  		var msg protocol.Message
@@@ -624,12 +637,12 @@ func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, r
  	}
  	window := int64(protocol.DefaultComputerMS)
  	if c.DeadlineMS > 0 {
--		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))
++		left, _, _ := n.timeLeft(c.DeadlineMS, sentMS, received)
++		if left < time.Millisecond {
++			nack(protocol.FaultExpired, fmt.Sprintf("the deadline passed %d ms before the command arrived", max(0, ceilMS(-left))))
  			return
  		}
--		window = min(window, int64(protocol.MaxComputerMS))
++		window = min(int64(left/time.Millisecond), int64(protocol.MaxComputerMS))
  	}
  	p := n.mgr.Route(c.Kind, c.Target, "")
  	if p == nil {
@@@ -692,21 +705,26 @@ func (n *Node) haltEpoch(robotID string) int64 {
  
  // 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) {
++// (Link.ClockUncertainty: the connection's fastest round trip and the server's ts resolution). Everything is kept in
++// nanoseconds, so no rounding can add time: left is exact, and unc, the bound reported in ms, is rounded up from
++// nanoseconds (Maticus, 2026-10-27: 1,000,001 ns is 2 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 time.Duration, unc int64, synced bool) {
++	end := deadlineMS * int64(time.Millisecond)
  	now := n.serverNow(sentMS, received)
  	if n.link == nil {
--		return deadlineMS - now, 0, true
++		return time.Duration(end - now), 0, true
  	}
  	d, ok := n.link.ClockUncertainty()
  	if !ok {
--		return deadlineMS - now, 0, false
++		return time.Duration(end - 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
++	return time.Duration(end-now) - d, ceilMS(d), true
++}
++
++// ceilMS is d in whole milliseconds, rounded up (d ≥ 0): a bound of 1,000,001 ns is 2 ms.
++func ceilMS(d time.Duration) int64 {
++	return int64((d + time.Millisecond - 1) / time.Millisecond)
  }
  
  // latchReason is the reason a latched stop gives a driver: the remote one when the platform set it, else the local one.
@@@ -717,18 +735,18 @@ func latchReason(st safety.LatchState) string {
  	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.
++// serverNow is the server's clock (epoch ns) 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 {
  	if n.link != nil {
--		if ms, ok := n.link.ServerNow(received); ok {
--			return ms
++		if ns, ok := n.link.ServerNow(received); ok {
++			return ns
  		}
  	}
  	if sentMS > 0 {
--		return sentMS
++		return sentMS * int64(time.Millisecond)
  	}
--	return received.UnixMilli()
++	return received.UnixNano()
  }
  
  func (n *Node) send(m protocol.Message) {
ddiff --git a/internal/node/node_test.go b/internal/node/node_test.go
iindex 90195b4..4a5e4b3 100644
---- a/internal/node/node_test.go
++++ b/internal/node/node_test.go
@@@ -327,10 +327,11 @@ func TestCommandsClampedAndIdempotent(t *testing.T) {
  	if d := drives[0].DeadlineMS; d < 150 || d > 305 {
  		t.Fatalf("deadline %d", d)
  	}
--	// Deadlines are capped at max_command_ms.
++	// Deadlines are capped at max_command_ms, counted from the Node's receipt: the plugin gets the cap less the time
++	// the command then spent in the Node and its queue (Node 0.18).
  	e.cmd("drive", `{"throttle":0.1}`, 60000)
  	r := e.records()
--	if last := r[len(r)-1]; last.DeadlineMS != 1000 {
++	if last := r[len(r)-1]; last.DeadlineMS > 1000 || last.DeadlineMS < 900 {
  		t.Fatalf("deadline not capped: %d", last.DeadlineMS)
  	}
  	if c := nackCode(e.cmd("drive", `{"throttle":"fast"}`, 300)); c != protocol.FaultBadValue {
ddiff --git a/internal/plugins/mono_unix.go b/internal/plugins/mono_unix.go
nnew file mode 100644
iindex 0000000..057a325
---- /dev/null
++++ b/internal/plugins/mono_unix.go
@@@ -0,0 +1,15 @@
++//go:build !windows
++
++package plugins
++
++import "golang.org/x/sys/unix"
++
++// monotonicNS reads CLOCK_MONOTONIC: one clock for every process on this machine (a plugin reads the same one, in
++// Python time.clock_gettime_ns(time.CLOCK_MONOTONIC)), which a step of the wall clock does not move.
++func monotonicNS() (int64, bool) {
++	var ts unix.Timespec
++	if unix.ClockGettime(unix.CLOCK_MONOTONIC, &ts) != nil {
++		return 0, false
++	}
++	return ts.Nano(), true
++}
ddiff --git a/internal/plugins/mono_windows.go b/internal/plugins/mono_windows.go
nnew file mode 100644
iindex 0000000..33dfc2f
---- /dev/null
++++ b/internal/plugins/mono_windows.go
@@@ -0,0 +1,7 @@
++//go:build windows
++
++package plugins
++
++// monotonicNS has no clock to share with a plugin on Windows: commands carry deadline_ms alone (counted down to the
++// moment the line is written), and expires_at_mono_ns is left out.
++func monotonicNS() (int64, bool) { return 0, false }
ddiff --git a/internal/plugins/plugin.go b/internal/plugins/plugin.go
iindex 854e88b..0851090 100644
---- a/internal/plugins/plugin.go
++++ b/internal/plugins/plugin.go
@@@ -104,7 +104,7 @@ type Plugin struct {
  	// 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
++	queue     chan line // to the current process's stdin; nil while down
  	kill      func()
  	restarts  int
  	pid       int
@@@ -160,6 +160,57 @@ func (p *Plugin) emit(o Output, important bool) {
  	}
  }
  
++// line is one write to the plugin's stdin: bytes, or a timed command the writer renders at the moment it writes it.
++type line struct {
++	b   []byte
++	cmd *timed
++}
++
++// timed is a command that expires at an instant on this machine. Lines can wait in the queue to the plugin while the
++// writer is blocked on a full pipe (a plugin slow to read), so the writer, not the sender, stamps what is left: a timed
++// command is written with deadline_ms = the whole milliseconds left at that moment and expires_at_mono_ns = the instant
++// on the clock the plugin shares (monotonicNS), and is never written once less than a millisecond is left (Maticus,
++// 2026-10-27: queue and pipe time must not extend a command's validity).
++type timed struct {
++	id      string
++	msg     map[string]any
++	expires time.Time // on this process's monotonic clock
++	queued  time.Time
++}
++
++// render is the command's line as it is written now, or ok false when less than a millisecond is left.
++func (c *timed) render() (b []byte, ok bool) {
++	// The shared clock is read before the time left, so expires_at_mono_ns can only come out early.
++	mono, hasMono := monotonicNS()
++	left := time.Until(c.expires)
++	if left < time.Millisecond {
++		return nil, false
++	}
++	c.msg["deadline_ms"] = int64(left / time.Millisecond)
++	if hasMono {
++		c.msg["expires_at_mono_ns"] = mono + int64(left)
++	}
++	b, err := json.Marshal(c.msg)
++	if err != nil {
++		return nil, false
++	}
++	return append(b, '\n'), true
++}
++
++// expired answers a timed command that was never written: its time ran out in the queue to the plugin.
++func (p *Plugin) expired(c *timed) {
++	p.mu.Lock()
++	ch := p.pending[c.id]
++	p.mu.Unlock()
++	if ch == nil {
++		return
++	}
++	select {
++	case ch <- Reply{FaultCode: protocol.FaultExpired, Message: fmt.Sprintf("its deadline passed while it waited %d ms in the queue to the driver; it was never sent", time.Since(c.queued).Milliseconds())}:
++	default:
++	}
++}
++
  // send queues one line for the plugin. A plugin that stops reading its stdin is killed: it can no longer be told to
  // stop, and its own heartbeat watchdog is the only thing left, so we make sure by ending it.
  func (p *Plugin) send(msg map[string]any) error {
@@@ -167,7 +218,10 @@ func (p *Plugin) send(msg map[string]any) error {
  	if err != nil {
  		return err
  	}
--	b = append(b, '\n')
++	return p.enqueue(line{b: append(b, '\n')})
++}
++
++func (p *Plugin) enqueue(l line) error {
  	p.mu.Lock()
  	q, kill := p.queue, p.kill
  	p.mu.Unlock()
@@@ -175,7 +229,7 @@ func (p *Plugin) send(msg map[string]any) error {
  		return errors.New(protocol.FaultPluginDown)
  	}
  	select {
--	case q <- b:
++	case q <- l:
  		return nil
  	default:
  		p.log.Error("plugin is not reading its input; killing it")
@@@ -285,7 +339,7 @@ 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)
++	return p.begin(ctx, id, kind, value, deadlineMS, time.Time{}, op, turnEpoch, nil)
  }
  
  // Provenance is where a command came from, passed unchanged from Bot to the driver as `provenance` (plugins.md): the
@@@ -304,8 +358,15 @@ type Provenance struct {
  	ClockUncertaintyMS *int64 `json:"clock_uncertainty_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 {
++// BeginWith is BeginTurn for a command that expires at an instant on this machine (the Node's receipt of it plus the
++// time its deadline left), with its provenance (nil: none). The plugin gets what is left when the line is written to
++// it, never more, and a command whose time runs out before then is answered expired without reaching the plugin.
++func (p *Plugin) BeginWith(ctx context.Context, id, kind string, value json.RawMessage, expires time.Time, op *protocol.Operator, turnEpoch int64, prov *Provenance) func() Reply {
++	return p.begin(ctx, id, kind, value, 0, expires, op, turnEpoch, prov)
++}
++
++// begin queues a command: with deadline_ms as given when expires is zero, else timed (BeginWith).
++func (p *Plugin) begin(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int, expires time.Time, 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 {
@@@ -344,13 +405,22 @@ func (p *Plugin) BeginWith(ctx context.Context, id, kind string, value json.RawM
  	if prov != nil {
  		msg["provenance"] = prov
  	}
--	if err := p.send(msg); err != nil {
++	var err error
++	if expires.IsZero() {
++		err = p.send(msg)
++	} else {
++		err = p.enqueue(line{cmd: &timed{id: id, msg: msg, expires: expires, queued: time.Now()}})
++	}
++	if err != nil {
  		done()
  		return now(Reply{FaultCode: protocol.FaultPluginDown})
  	}
  	return func() Reply {
  		defer done()
  		wait := time.Duration(deadlineMS) * time.Millisecond
++		if !expires.IsZero() {
++			wait = time.Until(expires)
++		}
  		if wait < MinReplyTimeout {
  			wait = MinReplyTimeout
  		}
@@@ -447,7 +517,7 @@ func (p *Plugin) runOnce(ctx context.Context) error {
  	if err := cmd.Start(); err != nil {
  		return fmt.Errorf("start %s: %w", p.spec.Argv[0], err)
  	}
--	queue := make(chan []byte, 256)
++	queue := make(chan line, 256)
  	var killOnce sync.Once
  	kill := func() { killOnce.Do(func() { killProcess(cmd) }) }
  	p.mu.Lock()
@@@ -458,21 +528,33 @@ func (p *Plugin) runOnce(ctx context.Context) error {
  	// Writer: the only goroutine touching stdin. Closing stdin is the plugin's signal to stop and exit.
  	writerDone := make(chan struct{})
  	stopWriter := make(chan struct{})
++	write := func(l line) error {
++		b := l.b
++		if l.cmd != nil {
++			var ok bool
++			if b, ok = l.cmd.render(); !ok {
++				p.expired(l.cmd)
++				return nil
++			}
++		}
++		_, err := stdin.Write(b)
++		return err
++	}
  	go func() {
  		defer close(writerDone)
  		defer stdin.Close()
  		for {
  			select {
--			case b := <-queue:
--				if _, err := stdin.Write(b); err != nil {
++			case l := <-queue:
++				if err := write(l); err != nil {
  					return
  				}
  			case <-stopWriter:
  				// Drain what is queued (a final stop) and close.
  				for {
  					select {
--					case b := <-queue:
--						if _, err := stdin.Write(b); err != nil {
++					case l := <-queue:
++						if err := write(l); err != nil {
  							return
  						}
  					default:
ddiff --git a/internal/plugins/plugins_test.go b/internal/plugins/plugins_test.go
iindex 986ac92..15e319e 100644
---- a/internal/plugins/plugins_test.go
++++ b/internal/plugins/plugins_test.go
@@@ -40,13 +40,26 @@ func fakePlugin(logPath string, in io.Reader, out io.Writer) {
  	rec := func(s string) { fmt.Fprintln(f, s) }
  	w := json.NewEncoder(out)
  	sc := bufio.NewScanner(in)
++	sc.Buffer(make([]byte, 0, 64*1024), 4<<20) // a filler line can be larger than a pipe
  	mode := os.Getenv("OPENVIBE_FAKE_MODE")
  	for sc.Scan() {
++		// What a command had left when its line was read, on the clock shared with the core.
++		mono, hasMono := monotonicNS()
  		var m map[string]any
--		json.Unmarshal(sc.Bytes(), &m)
++		d := json.NewDecoder(strings.NewReader(sc.Text()))
++		d.UseNumber()
++		d.Decode(&m)
  		op, _ := m["op"].(string)
  		if op != "heartbeat" {
--			rec(sc.Text())
++			if len(sc.Text()) > 4096 {
++				rec(fmt.Sprintf("LONG %v", m["id"]))
++			} else {
++				rec(sc.Text())
++			}
++		}
++		if exp, ok := m["expires_at_mono_ns"].(json.Number); ok && hasMono {
++			e, _ := exp.Int64()
++			rec(fmt.Sprintf("LEFT %v %d", m["id"], e-mono))
  		}
  		switch op {
  		case "hello":
@@@ -77,6 +90,11 @@ func fakePlugin(logPath string, in io.Reader, out io.Writer) {
  				os.Exit(3)
  			case "hang":
  				time.Sleep(10 * time.Second)
++			case "pause":
++				// Stop reading stdin for a while, as a plugin stuck in a handler would.
++				ms, _ := m["value"].(map[string]any)["ms"].(json.Number).Int64()
++				time.Sleep(time.Duration(ms) * time.Millisecond)
++				w.Encode(map[string]any{"op": "ack", "id": m["id"]})
  			case "slow":
  				continue
  			case "display":
@@@ -287,6 +305,29 @@ func TestCrashRestartsWithEstop(t *testing.T) {
  	}
  }
  
++// A latch set before the manager starts (the Node booting with its stop latched on disk) is the first thing each
++// process hears after hello: the e-stop replayed, naming what set it, before the plugin can be routed any command.
++func TestLatchSetBeforeStartIsReplayedFirst(t *testing.T) {
++	logPath := filepath.Join(t.TempDir(), "plugin.log")
++	exe, _ := os.Executable()
++	spec := Spec{Name: "fake", Argv: []string{exe}, Env: []string{"OPENVIBE_FAKE_PLUGIN=" + logPath, "OPENVIBE_FAKE_MODE="}}
++	m := NewManager([]Spec{spec}, slog.New(slog.NewTextHandler(io.Discard, nil)))
++	m.EstopAll(EstopRemote, "staff: it wobbled")
++	ctx, cancel := context.WithCancel(context.Background())
++	m.Start(ctx)
++	t.Cleanup(func() { cancel(); m.Wait() })
++	fx := &fixture{t: t, log: logPath, m: m}
++	fx.waitState(StateReady)
++	if r := m.Get("fake").Command(context.Background(), "after", "say", nil, 300); !r.OK {
++		t.Fatalf("%+v", r)
++	}
++	l := fx.lines()
++	replay := `{"cause":"latched","first_cause":"remote","op":"estop","reason":"staff: it wobbled","replayed":true}`
++	if len(l) < 3 || !strings.Contains(l[0], `"op":"hello"`) || l[1] != replay || !strings.Contains(l[2], `"id":"after"`) {
++		t.Fatalf("lines: %v", l)
++	}
++}
++
  func TestReplyTimeout(t *testing.T) {
  	fx := start(t, "")
  	fx.waitState(StateReady)
@@@ -300,6 +341,67 @@ func TestReplyTimeout(t *testing.T) {
  	fx.waitLog(`{"cause":"timeout","op":"stop"}`)
  }
  
++// Time a command waits in the queue to the plugin counts against it (Maticus, 2026-10-27). The plugin stops reading
++// for 400 ms and a line larger than the pipe fills it, so the writer is blocked and two motion commands wait in the
++// queue: the one with 100 ms is answered expired and never written; the one with 1.5 s is written with only what is
++// left (deadline_ms and expires_at_mono_ns, counted down to the moment it was written).
++func TestWriterQueueDelayCountsAgainstTheDeadline(t *testing.T) {
++	fx := start(t, "")
++	fx.waitState(StateReady)
++	p := fx.m.Get("fake")
++	ctx := context.Background()
++	held := time.Now()
++	pause := p.Begin(ctx, "hold", "pause", json.RawMessage(`{"ms":400}`), 3000)
++	fx.waitLog(`"id":"hold"`)
++	filler, _ := json.Marshal(map[string]any{"text": strings.Repeat("x", 1<<20)})
++	fill := p.Begin(ctx, "fill", "say", filler, 3000)
++	queued := time.Now()
++	late := p.BeginWith(ctx, "late", "drive", nil, queued.Add(100*time.Millisecond), nil, 0, nil)
++	ontime := p.BeginWith(ctx, "ontime", "drive", nil, queued.Add(1500*time.Millisecond), nil, 0, nil)
++	if r := late(); r.FaultCode != protocol.FaultExpired || !strings.Contains(r.Message, "never sent") {
++		t.Fatalf("late: %+v", r)
++	}
++	if r := ontime(); !r.OK {
++		t.Fatalf("ontime: %+v", r)
++	}
++	if r := pause(); !r.OK {
++		t.Fatalf("pause: %+v", r)
++	}
++	if r := fill(); !r.OK {
++		t.Fatalf("fill: %+v", r)
++	}
++	// Nothing could be written before the plugin read again, 400 ms after it read "hold" (which was after held).
++	maxLeft := queued.Add(1500 * time.Millisecond).Sub(held.Add(400 * time.Millisecond))
++	var deadlineMS int64 = -1
++	sawLeft := false
++	for _, l := range fx.lines() {
++		if strings.Contains(l, `"id":"late"`) || strings.HasPrefix(l, "LEFT late") {
++			t.Fatalf("the expired command reached the plugin: %s", l)
++		}
++		if strings.Contains(l, `"id":"ontime"`) {
++			var m struct {
++				DeadlineMS int64 `json:"deadline_ms"`
++			}
++			json.Unmarshal([]byte(l), &m)
++			deadlineMS = m.DeadlineMS
++		}
++		if strings.HasPrefix(l, "LEFT ontime ") {
++			var ns int64
++			fmt.Sscanf(strings.TrimPrefix(l, "LEFT ontime "), "%d", &ns)
++			sawLeft = true
++			if ns <= 0 || time.Duration(ns) > maxLeft {
++				t.Fatalf("the plugin saw %v left on the shared clock; at most %v can be", time.Duration(ns), maxLeft)
++			}
++		}
++	}
++	if deadlineMS < 1 || time.Duration(deadlineMS)*time.Millisecond > maxLeft {
++		t.Fatalf("ontime was written with deadline_ms %d; at most %v was left", deadlineMS, maxLeft)
++	}
++	if _, ok := monotonicNS(); ok != sawLeft {
++		t.Fatalf("expires_at_mono_ns sent %v, shared clock %v", sawLeft, ok)
++	}
++}
++
  func TestFaultState(t *testing.T) {
  	fx := start(t, "fault")
  	fx.waitState(StateFaulted)
ddiff --git a/plugins/sdk/openvibe_plugin/__init__.py b/plugins/sdk/openvibe_plugin/__init__.py
iindex 911a234..70af7ed 100644
---- a/plugins/sdk/openvibe_plugin/__init__.py
++++ b/plugins/sdk/openvibe_plugin/__init__.py
@@@ -21,8 +21,10 @@ import time
  
  __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"]
++           "RUNTIME_CAUSES", "SHARED_CLOCK", "__version__"]
  
++# The Node release this runtime ships with (pyproject.toml carries the same).
++__version__ = "0.18.0"
  PROTOCOL_VERSION = 1
  HEARTBEAT_TIMEOUT_S = 1.0
  DEFAULT_DEADLINE_MS = 300
@@@ -36,6 +38,16 @@ MOTION_KINDS = frozenset({"drive"})
  GUARDED_KINDS = frozenset({"drive", "ptz", "actuator"})
  
  
++def _monotonic_ns():
++    return time.clock_gettime_ns(time.CLOCK_MONOTONIC)
++
++
++# The clock this process shares with the core: CLOCK_MONOTONIC, one clock for every process on the machine, which a
++# step of the wall clock does not move. A command's expires_at_mono_ns is an instant on it (Node 0.18). None where
++# there is none to share (Windows): deadline_ms alone then, which the core counts down to the moment it wrote the line.
++SHARED_CLOCK = _monotonic_ns if hasattr(time, "clock_gettime_ns") and hasattr(time, "CLOCK_MONOTONIC") else None
++
++
  class Fault(Exception):
      """Raised by a driver to answer a command with a nack."""
  
@@@ -71,14 +83,18 @@ class Command:
      (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")."""
++    again; the runtime already refuses one whose seq is not above the last on its session (plugins.md, "Provenance").
++
++    deadline_ms is the whole milliseconds the command had left when its line was read, and deadline the instant on the
++    runtime's clock it expires: the core counts down to the moment it wrote the line, and the runtime takes off the time
++    the line then sat in the pipe (expires_at_mono_ns, Node 0.18), so no delay on the way gives a command more time."""
      __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 {}
--        self.deadline_ms = deadline_ms
++        self.deadline_ms = int(deadline_ms)
          self.deadline = now + deadline_ms / 1000.0
          self.target = target
          self.operator = operator if isinstance(operator, dict) else None
@@@ -89,6 +105,13 @@ class Command:
          return "Command(id=%r, kind=%r, value=%r, deadline_ms=%r)" % (self.id, self.kind, self.value, self.deadline_ms)
  
  
++def _left_ms(expires_ns, shared_at):
++    """Milliseconds from shared_at to expires_ns (both on the shared clock), or None when either is missing."""
++    if shared_at is None or not isinstance(expires_ns, int) or isinstance(expires_ns, bool):
++        return None
++    return (expires_ns - shared_at) / 1e6
++
++
  def _provenance(p):
      """The core's provenance dict with only well-typed fields, or None."""
      if not isinstance(p, dict):
@@@ -289,12 +312,14 @@ class Runtime:
      """Drives one Plugin from a line-oriented input and output. Tests construct it with in-memory streams."""
  
      def __init__(self, plugin, infile=None, outfile=None, clock=time.monotonic,
--                 heartbeat_timeout_s=HEARTBEAT_TIMEOUT_S):
++                 heartbeat_timeout_s=HEARTBEAT_TIMEOUT_S, shared_clock=SHARED_CLOCK):
          self.plugin = plugin
          plugin.runtime = self
          self.infile = infile if infile is not None else sys.stdin
          self.outfile = outfile if outfile is not None else sys.stdout
          self.clock = clock
++        # The clock shared with the core (SHARED_CLOCK, ns), read with clock when each line arrives; None: not shared.
++        self.shared_clock = shared_clock
          self.heartbeat_timeout_s = heartbeat_timeout_s
          self.lock = threading.RLock()
          self.out_lock = threading.Lock()
@@@ -378,10 +403,23 @@ class Runtime:
                  self.send(dict(op="telemetry", **t))
  
      # Input.
--    def dispatch(self, msg, at=None):
++    def shared_now(self):
++        """The shared clock now (ns), or None."""
++        if self.shared_clock is None:
++            return None
++        try:
++            return self.shared_clock()
++        except (OSError, ValueError):
++            return None
++
++    def dispatch(self, msg, at=None, shared_at=None):
          """Handle one message. at is when the line arrived (defaults to now): deadlines and heartbeats count from
--        arrival, so a backlog after a slow handler cannot make stale commands or heartbeats look fresh."""
--        at = self.clock() if at is None else at
++        arrival, so a backlog after a slow handler cannot make stale commands or heartbeats look fresh. shared_at is
++        the shared clock at that moment (read with at when at defaults): a command's expires_at_mono_ns is measured
++        against it, so time in the pipe before arrival counts too."""
++        if at is None:
++            at = self.clock()
++            shared_at = self.shared_now()
          op = msg.get("op")
          if op == "hello":
              self.on_hello(msg)
@@@ -390,7 +428,7 @@ class Runtime:
              if self.heartbeat_lost:
                  self.heartbeat_lost = False
          elif op == "command":
--            self.on_command(msg, at)
++            self.on_command(msg, at, shared_at)
          elif op == "stop":
              epoch = _epoch(msg.get("turn_epoch"))
              if epoch and epoch > self.halt_epoch:
@@@ -479,16 +517,23 @@ class Runtime:
          if self.plugin.retry_setup_s:
              self.setup_retry_at = self.clock() + float(self.plugin.retry_setup_s)
  
--    def on_command(self, msg, now=None):
++    def on_command(self, msg, now=None, shared_at=None):
          cid = msg.get("id")
          kind = msg.get("kind")
--        now = self.clock() if now is None else now
++        if now is None:
++            now = self.clock()
++            shared_at = self.shared_now()
          try:
              dl = int(msg.get("deadline_ms") or DEFAULT_DEADLINE_MS)
          except (TypeError, ValueError):
              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))
++        # What the command had left when its line arrived, on the clock shared with the core (Node 0.18): deadline_ms
++        # was counted down to the moment the core wrote the line, so this takes off the time it then sat in the pipe.
++        left = _left_ms(msg.get("expires_at_mono_ns"), shared_at) if moves else None
++        if left is not None and left >= 1:
++            dl = min(dl, left)
          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":
@@@ -509,8 +554,14 @@ class Runtime:
                             "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"]
--        # A motion command whose time ran out while it waited behind a slow handler (its deadline counts from when its
--        # line arrived) never reaches the driver: the runtime holds nothing past its deadline.
++        # A motion command whose time ran out in the pipe, before its line was read, never reaches the driver.
++        if left is not None and left < 1:
++            self.send({"op": "nack", "id": cid, "fault_code": "expired",
++                       "message": "its deadline passed before the runtime read it (%d ms late, counted on the clock it "
++                                  "shares with the core)" % max(0, round(-left))})
++            return
++        # Nor does one whose time ran out while it waited behind a slow handler (its deadline counts from when its line
++        # arrived): the runtime holds nothing past its deadline.
          if moves and self.clock() >= cmd.deadline:
              self.send({"op": "nack", "id": cid, "fault_code": "expired",
                         "message": "its deadline passed while it waited behind earlier input (%d ms after it arrived)"
@@@ -543,7 +594,7 @@ class Runtime:
              self.safe_stop("handler_error")
              self.send({"op": "nack", "id": cid, "fault_code": "hardware", "message": str(e)})
  
--    def feed_line(self, line, at=None):
++    def feed_line(self, line, at=None, shared_at=None):
          line = line.strip()
          if not line:
              return
@@@ -553,7 +604,7 @@ class Runtime:
              self.plugin.log("bad json from core: %r", line[:200])
              return
          if isinstance(msg, dict):
--            self.dispatch(msg, at)
++            self.dispatch(msg, at, shared_at)
  
      def shutdown(self, reason="eof"):
          if self.closed:
@@@ -569,7 +620,9 @@ class Runtime:
      def _reader(self):
          try:
              for line in self.infile:
--                self.inbox.put((line, self.clock()))
++                # The runtime's clock first, then the shared one: the time left can only come out short.
++                at = self.clock()
++                self.inbox.put((line, at, self.shared_now()))
          except Exception:
              pass
          self.inbox.put(None)
@@@ -582,12 +635,12 @@ class Runtime:
                  try:
                      item = self.inbox.get(timeout=0.02)
                  except queue.Empty:
--                    item = ("", None)
++                    item = ("", None, None)
                  if item is None:
                      break
--                line, at = item
++                line, at, shared_at = item
                  if line:
--                    self.feed_line(line, at)
++                    self.feed_line(line, at, shared_at)
                  self.tick()
          finally:
              self.shutdown("eof")
ddiff --git a/plugins/sdk/pyproject.toml b/plugins/sdk/pyproject.toml
iindex e6f5b99..7f93ec6 100644
---- a/plugins/sdk/pyproject.toml
++++ b/plugins/sdk/pyproject.toml
@@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
  
  [project]
  name = "openvibe-plugin"
--version = "0.1.0"
++version = "0.18.0"
  description = "Runtime for OpenVibe Node plugins (JSON lines over stdin/stdout, deadlines, heartbeat, e-stop)"
  requires-python = ">=3.9"
  license = { text = "AGPL-3.0-or-later" }
ddiff --git a/plugins/sdk/tests/test_runtime.py b/plugins/sdk/tests/test_runtime.py
iindex afa46d0..4e825f6 100644
---- a/plugins/sdk/tests/test_runtime.py
++++ b/plugins/sdk/tests/test_runtime.py
@@@ -3,13 +3,15 @@ import json
  import os
  import subprocess
  import sys
++import threading
  import time
  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, StopInfo  # noqa: E402
++import openvibe_plugin  # noqa: E402
++from openvibe_plugin import SHARED_CLOCK, Deferred, Fault, Plugin, Runtime, StopInfo  # noqa: E402
  
  
  class FakeClock:
@@@ -26,6 +28,7 @@ class Recorder(Plugin):
  
      def __init__(self):
          self.calls = []
++        self.handled = []
          self.moving = False
  
      def describe(self, config):
@@@ -36,6 +39,7 @@ class Recorder(Plugin):
          self.calls.append("setup")
  
      def handle(self, cmd):
++        self.handled.append(cmd)
          if cmd.kind == "drive":
              self.moving = True
              return
@@@ -48,11 +52,20 @@ class Recorder(Plugin):
          self.moving = False
  
  
--def make(plugin=None):
++class FakeShared:
++    """The clock shared with the core, in ns (a different origin from the runtime's own, as in life)."""
++    def __init__(self):
++        self.ns = 7_000_000_000_000
++
++    def __call__(self):
++        return self.ns
++
++
++def make(plugin=None, shared=None):
      out = io.StringIO()
      clock = FakeClock()
      p = plugin or Recorder()
--    rt = Runtime(p, infile=io.StringIO(""), outfile=out, clock=clock)
++    rt = Runtime(p, infile=io.StringIO(""), outfile=out, clock=clock, shared_clock=shared)
      return rt, p, out, clock
  
  
@@@ -306,6 +319,130 @@ class RuntimeTest(unittest.TestCase):
          self.assertFalse(p.moving)
          self.assertIn("deadline", rt.stop_reasons)
  
++    def test_pipe_delay_counts_against_the_deadline(self):
++        # Node 0.18 (Maticus, 2026-10-27): the core writes deadline_ms as what was left when it wrote the line, and
++        # expires_at_mono_ns as the instant on the clock it shares with this process. Time the line then sat in the pipe
++        # counts: one that expired there never reaches handle(), and one still valid gets only what is left.
++        shared = FakeShared()
++        rt, p, out, clock = make(shared=shared)
++        self.hello(rt)
++        rt.feed_line(json.dumps({"op": "heartbeat", "t": 1}))
++        # Written with 100 ms left; read 200 ms later: expired in the pipe.
++        written = shared.ns
++        shared.ns += 200_000_000
++        clock.t += 0.2
++        rt.feed_line(json.dumps({"op": "command", "id": "late", "kind": "drive", "deadline_ms": 100,
++                                 "expires_at_mono_ns": written + 100_000_000}), clock.t, shared.ns)
++        last = lines(out)[-1]
++        self.assertEqual((last["op"], last["id"], last["fault_code"]), ("nack", "late", "expired"))
++        self.assertIn("before the runtime read it (100 ms late", last["message"])
++        self.assertEqual(p.handled, [])
++        self.assertFalse(p.moving)
++        # Written with 300 ms left; read 260 ms later: 40 ms left, and the motion stops when they run out.
++        written = shared.ns
++        shared.ns += 260_000_000
++        clock.t += 0.26
++        rt.feed_line(json.dumps({"op": "heartbeat", "t": 2}))
++        rt.feed_line(json.dumps({"op": "command", "id": "ontime", "kind": "drive", "deadline_ms": 300,
++                                 "expires_at_mono_ns": written + 300_000_000}), clock.t, shared.ns)
++        self.assertEqual(lines(out)[-1], {"op": "ack", "id": "ontime"})
++        cmd = p.handled[-1]
++        self.assertEqual(cmd.deadline_ms, 40)
++        self.assertAlmostEqual(cmd.deadline, clock.t + 0.040, places=9)
++        clock.t += 0.039
++        rt.tick()
++        self.assertTrue(p.moving)
++        clock.t += 0.002
++        rt.tick()
++        self.assertFalse(p.moving)
++        self.assertEqual(rt.stop_reasons[-1], "deadline")
++        # Less than a millisecond left counts as none (deadline_ms is whole milliseconds, at least 1).
++        rt.feed_line(json.dumps({"op": "command", "id": "edge", "kind": "drive", "deadline_ms": 300,
++                                 "expires_at_mono_ns": shared.ns + 999_999}), clock.t, shared.ns)
++        self.assertEqual(lines(out)[-1]["fault_code"], "expired")
++        # A line that waited in the pipe and then in the inbox: both count, the inbox on the runtime's clock as before.
++        written = shared.ns
++        rt.feed_line(json.dumps({"op": "heartbeat", "t": 3}))
++        arrived, shared_arrived = clock.t, shared.ns + 150_000_000
++        clock.t += 0.2
++        rt.feed_line(json.dumps({"op": "command", "id": "both", "kind": "drive", "deadline_ms": 300,
++                                 "expires_at_mono_ns": written + 300_000_000}), arrived, shared_arrived)
++        last = lines(out)[-1]
++        self.assertEqual((last["id"], last["fault_code"]), ("both", "expired"))
++        self.assertEqual([c.id for c in p.handled], ["ontime"])
++
++    def test_motion_kinds_only_and_deadline_ms_alone_without_a_shared_clock(self):
++        # say is not motion: its deadline is a reply window, and expires_at_mono_ns does not apply to it.
++        shared = FakeShared()
++        rt, p, out, clock = make(shared=shared)
++        self.hello(rt)
++        rt.feed_line(json.dumps({"op": "command", "id": "s", "kind": "say", "deadline_ms": 300,
++                                 "expires_at_mono_ns": shared.ns - 1}), clock.t, shared.ns)
++        self.assertEqual(lines(out)[-1], {"op": "ack", "id": "s"})
++        # No clock shared with the core (Windows): deadline_ms alone, from arrival.
++        rt, p, out, clock = make()
++        self.hello(rt)
++        rt.feed_line(json.dumps({"op": "heartbeat", "t": 1}))
++        rt.feed_line(json.dumps({"op": "command", "id": "d", "kind": "drive", "deadline_ms": 300,
++                                 "expires_at_mono_ns": 1}), clock.t)
++        self.assertEqual(lines(out)[-1], {"op": "ack", "id": "d"})
++        self.assertEqual(p.handled[-1].deadline_ms, 300)
++
++    @unittest.skipIf(SHARED_CLOCK is None, "no clock shared with the core on this platform")
++    def test_a_real_pipe_delay_never_extends_a_deadline(self):
++        # The real thing: an OS pipe, the real clocks, the runtime's own reader thread. Lines are written as the core
++        # writes them and left in the pipe for 200 ms before the runtime starts reading.
++        r, w = os.pipe()
++        infile = os.fdopen(r, "r", encoding="utf-8")
++        out = io.StringIO()
++        p = Recorder()
++        rt = Runtime(p, infile=infile, outfile=out)
++        now = SHARED_CLOCK()
++        msgs = [
++            {"op": "hello", "config": {}},
++            {"op": "heartbeat", "t": 1},
++            {"op": "command", "id": "late", "kind": "drive", "deadline_ms": 100, "expires_at_mono_ns": now + 100_000_000},
++            {"op": "command", "id": "ontime", "kind": "drive", "deadline_ms": 1000, "expires_at_mono_ns": now + 1_000_000_000},
++        ]
++        os.write(w, "".join(json.dumps(m) + "\n" for m in msgs).encode())
++        time.sleep(0.2)
++        t = threading.Thread(target=rt.run)
++        t.start()
++        time.sleep(0.1)
++        os.close(w)
++        t.join(5)
++        infile.close()
++        self.assertFalse(t.is_alive())
++        got = {m["id"]: m for m in lines(out) if m.get("op") in ("ack", "nack")}
++        self.assertEqual((got["late"]["op"], got["late"]["fault_code"]), ("nack", "expired"))
++        self.assertEqual(got["ontime"]["op"], "ack")
++        self.assertEqual([c.id for c in p.handled], ["ontime"])
++        # It got only what was left of its original second: at most 800 ms after 200 ms in the pipe.
++        self.assertLessEqual(p.handled[0].deadline_ms, 800)
++        self.assertGreater(p.handled[0].deadline_ms, 0)
++
++    def test_a_replayed_estop_is_applied_before_any_motion(self):
++        # A restarted process hears, in this order: hello (with the newest halt epoch), the latched e-stop replayed,
++        # and only then commands (the core routes none to it before its ready, and lines reach it in the order they
++        # were queued). No motion handler runs while the replayed latch holds; resume is information only.
++        rt, p, out, clock = make()
++        rt.feed_line(json.dumps({"op": "hello", "config": {}, "halt_epoch": 1792900000500}))
++        rt.feed_line(json.dumps({"op": "estop", "cause": "latched", "first_cause": "remote", "reason": "staff: it wobbled",
++                                 "replayed": True}))
++        rt.feed_line(json.dumps({"op": "heartbeat", "t": 1}))
++        rt.feed_line(json.dumps({"op": "command", "id": "m1", "kind": "drive", "deadline_ms": 300,
++                                 "turn_epoch": 1792900000500}))
++        self.assertEqual(lines(out)[-1], {"op": "nack", "id": "m1", "fault_code": "estopped"})
++        self.assertEqual(p.handled, [])
++        self.assertFalse(p.moving)
++        self.assertTrue(rt.stop_infos[-1].emergency and rt.stop_infos[-1].replayed and rt.stop_infos[-1].latched)
++        self.assertEqual(rt.stop_infos[-1].halt_epoch, 1792900000500)
++
++    def test_the_package_reports_the_node_release(self):
++        self.assertEqual(openvibe_plugin.__version__, "0.18.0")
++        with open(os.path.join(os.path.dirname(HERE), "pyproject.toml")) as f:
++            self.assertIn('version = "%s"' % openvibe_plugin.__version__, f.read())
++
      def test_only_motion_expires_in_the_backlog(self):
          # say has no motion deadline to miss (its deadline is a reply window): a late one still runs.
          rt, p, out, clock = make()