openvibers/openvibe.link Commits

Node 0.30.0: Windows keeps the original deadline across the pipe (QPC shared clock); strict timing refuses instead of downgrading (#78)

OP OpenVibers committed 16 days ago · d4aa3ad72f · 29 files, +1057 −116

* Node 0.30.0: Windows keeps the original deadline across the pipe; strict timing refuses instead of downgrading

Maticus (2026-10-27, Windows 10, against v0.18.0 399ef83): on Windows the plugin got deadline_ms alone and counted it
from when its reader got the line, so pipe time could extend a command; a strict deployment needs refusal, not a
silent fallback; the tests must run on Windows.

- internal/mono: one precise clock for the core's whole deadline path (the link's estimate of Bot's clock, round
  trips, receipt, expiry, the writer's check): CLOCK_MONOTONIC on Linux and macOS, the raw QueryPerformanceCounter on
  Windows. Go's own monotonic clock there is the interrupt time, a tick (15.625 ms) coarse.
- The writer passes the instant a command expires unchanged as expires_at_mono_ns; hello names the clock
  (shared_clock: qpc | clock_monotonic). The Python runtime reads the same raw counter through kernel32 with the same
  integer conversion (identical vectors in Go and Python), and compares only on the clock the core named.
- Strict timing: Plugin.strict_timing, OPENVIBE_STRICT_TIMING=1, or strict_timing on the plugin in config.json (sent in
  hello). A motion command whose original expiry cannot be checked (no clock, a failed read, another clock, no
  expires_at_mono_ns) is refused with timing_unavailable before handle(); the core does not write it to a strict
  plugin. Plugins that do not ask keep working on deadline_ms.
- Tests with a real Python plugin over real pipes on every OS (Windows and macOS CI log them verbosely): the clock read
  across processes, pipe delay (A, B), writer queue then pipe (D), missing and failed clocks under strict timing (C),
  the SDK suite; Python boundaries (at the deadline, 0.999999 ms, 1 ms, the cap, restart). With the stamp removed the
  late command runs with 99 ms and the tests fail.
- looi runs as python -m openvibe_looi without a command: install.sh --driver looi and the documented Windows
  config.json wrote {"name": "looi"}, which the Node refused to load.
- docs/plugins.md answer 9; protocol.md lists timing_unavailable; hardware-looi.md sets strict_timing.

* TestWriterQueueThenPipeDelay: the bound counts from when the plugin's hold began, not from queued

The hold starts when the plugin reads it, a few ms before queued (the 1 MB filler is marshalled in between, ~50 ms
under -race), so up to that much more than 1000 ms can rightly be left: Windows CI saw 1003, the race job 1033.
The bound is now (queued + 1500 ms) - (held + 500 ms), as TestWriterQueueDelayCountsAgainstTheDeadline does.

Changes

ddiff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
iindex a778ba0..0f6d644 100644
---- a/.github/workflows/ci.yml
++++ b/.github/workflows/ci.yml
@@@ -76,6 +76,10 @@ jobs:
- uses: actions/setup-python@v5
with:
python-version: "3.11"
++ # The deadline across the real pipe to a Python plugin, the clock it shares with the core, strict timing and the
++ # Python SDK's own suite, verbose so the log shows each ran on this OS (and its clock readings), never skipped.
++ - name: deadlines across the pipe (verbose)
++ run: go test -count=1 -v -run "TestQPC|TestTheClock|TestUntilAndSince|TestPython|TestHelloNamesTheSharedClock|TestPipeDelay|TestWriterQueue|TestStrictTiming" ./internal/mono ./internal/plugins
- run: go test -count=1 ./...
python:
ddiff --git a/STATUS.json b/STATUS.json
iindex 8877507..12e84f6 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.29.0",
-- "release": "v0.29.0, published 2026-10-27 as a GitHub release (the ai plugin says how fast each model answers here: ai_tokens_per_s for the last model, averaged with the earlier answers, and ai_busy while one answers, shown beside the machine's models on openvibe.bot; v0.28.0: install.sh --ai sets up your own models: it adds the ai plugin, which a driver chosen later keeps, says which model server answers (Ollama, LM Studio, llama.cpp) or how to get one, and --ai-endpoint names another port; v0.27.0: your own models for your OpenVibe work: the ai plugin relays kind ai (contracts 0.219.0), one chat completion, to a local OpenAI-compatible model server (Ollama, LM Studio, llama.cpp), lists its models, runs one at a time, and reaches nothing else; the core routes ai to it like a computer command that never acts; v0.26.0: programs on a desk: exec takes desk, running in that desk's session so a program's windows open there, and detach, starting a program that keeps running on a whole desk (an office suite, a terminal) and ends with its desk; desk templates follow: research, office and dev desks; v0.25.0: the egress guard: every connection every agent browser makes, a page's own scripts included, goes through a proxy the plugin runs on 127.0.0.1 that resolves, checks and connects to the address it checked, so a page never reaches the machine's own network below level full; browser_egress: open turns it off; v0.24.1: a kept profile carries session cookies too: Chrome drops cookies that end with the browser as it quits, and a sign-in is often one, so a desk keeps them beside the profile and the next desk's browser gets them back as it starts, found by the production proof; v0.24.0: kept desk profiles: desk start takes profile, so a desk's browser starts from the person's kept profile (sign in once, on any desk) and saves it back when it stops, closing the browser gracefully so its cookies are written, the last desk to stop wins, caches never kept, and each desk's own copy removed; the browser follows a page its page opens (a sign-in popup, a new tab) and comes back when it closes; the agent's clicks count as a person's, so sites open their sign-in popups; v0.23.0: browser desks, the Desks plan's step D2: desk start takes kind browser, a headless browser alone with no display at about half the cost of a whole desk, on Linux or macOS; its page is its screen, mouse and keyboard and live view over the browser's pipe, exactly the desk's size; a desk host, which has no screen or input of its own, gets screen, screen_stream and input on its desks; a click on a named desk scales to that desk's size; v0.22.0: desks on a desk host, found by the first production proof: a desk's temp folder is short (Chrome's SingletonSocket must fit a Unix socket path, and the desks folder on a desk host was too deep, so Chrome closed as it started), and a desk no command used for desk_idle_s (15 minutes) stops by itself, so a desk an agent left behind when it crashed or restarted ends; v0.21.0: element refs and receipts, the Desks plan's step D3: the agent's browser lists a page's interactive elements as compact refs (e7 button \"Sign in\"), kept on the element while it stays, and click, press and back answer with what changed (url, lines that appeared, lines gone), so agents act on the page's structure before screenshots; v0.20.0: desk hosts, the Desks plan's step D5: install.sh --desk-host turns an old laptop, a mini PC or a spare workstation into a host of desks for its owner's agents, run as its own openvibe-desk account with Xvfb, openbox, D-Bus, xdotool, ImageMagick and Chromium installed and max_desks sized to its memory; v0.19.0: several desks per machine, the Desks plan's step D1: the computer plugin runs named desks (desk-<name>, private kept), each with its own X display, D-Bus, browser profile, downloads and temp folder, up to max_desks (one per 2 GiB of memory by default), so two agents work at once without seeing or driving each other's screen; proven with two real Chrome desks that never share a cookie; v0.18.0: 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",
++ "version": "0.30.0",
++ "release": "v0.30.0, published 2026-10-27 as a GitHub release (Windows keeps the original deadline across the pipe too: the core measures every deadline on one precise clock shared with its plugins, CLOCK_MONOTONIC on Linux and macOS and the raw QueryPerformanceCounter on Windows, named in hello as shared_clock, and passes the instant a command expires on it; the Python runtime reads the same raw counter with the same integer conversion; strict timing (Plugin.strict_timing, OPENVIBE_STRICT_TIMING=1, or strict_timing in the Node's plugin config) refuses a motion command whose original expiry cannot be checked with timing_unavailable instead of running it on deadline_ms; the deadline tests run a real Python plugin over real pipes on Linux, macOS and Windows; looi runs as python -m openvibe_looi without a command; v0.29.0: the ai plugin says how fast each model answers here: ai_tokens_per_s for the lastthe ai plugin says how fast each model answers here: ai_tokens_per_s for the last model, averaged with the earlier answers, and ai_busy while one answers, shown beside the machine's models on openvibe.bot; v0.28.0: install.sh --ai sets up your own models: it adds the ai plugin, which a driver chosen later keeps, says which model server answers (Ollama, LM Studio, llama.cpp) or how to get one, and --ai-endpoint names another port; v0.27.0: your own models for your OpenVibe work: the ai plugin relays kind ai (contracts 0.219.0), one chat completion, to a local OpenAI-compatible model server (Ollama, LM Studio, llama.cpp), lists its models, runs one at a time, and reaches nothing else; the core routes ai to it like a computer command that never acts; v0.26.0: programs on a desk: exec takes desk, running in that desk's session so a program's windows open there, and detach, starting a program that keeps running on a whole desk (an office suite, a terminal) and ends with its desk; desk templates follow: research, office and dev desks; v0.25.0: the egress guard: every connection every agent browser makes, a page's own scripts included, goes through a proxy the plugin runs on 127.0.0.1 that resolves, checks and connects to the address it checked, so a page never reaches the machine's own network below level full; browser_egress: open turns it off; v0.24.1: a kept profile carries session cookies too: Chrome drops cookies that end with the browser as it quits, and a sign-in is often one, so a desk keeps them beside the profile and the next desk's browser gets them back as it starts, found by the production proof; v0.24.0: kept desk profiles: desk start takes profile, so a desk's browser starts from the person's kept profile (sign in once, on any desk) and saves it back when it stops, closing the browser gracefully so its cookies are written, the last desk to stop wins, caches never kept, and each desk's own copy removed; the browser follows a page its page opens (a sign-in popup, a new tab) and comes back when it closes; the agent's clicks count as a person's, so sites open their sign-in popups; v0.23.0: browser desks, the Desks plan's step D2: desk start takes kind browser, a headless browser alone with no display at about half the cost of a whole desk, on Linux or macOS; its page is its screen, mouse and keyboard and live view over the browser's pipe, exactly the desk's size; a desk host, which has no screen or input of its own, gets screen, screen_stream and input on its desks; a click on a named desk scales to that desk's size; v0.22.0: desks on a desk host, found by the first production proof: a desk's temp folder is short (Chrome's SingletonSocket must fit a Unix socket path, and the desks folder on a desk host was too deep, so Chrome closed as it started), and a desk no command used for desk_idle_s (15 minutes) stops by itself, so a desk an agent left behind when it crashed or restarted ends; v0.21.0: element refs and receipts, the Desks plan's step D3: the agent's browser lists a page's interactive elements as compact refs (e7 button \"Sign in\"), kept on the element while it stays, and click, press and back answer with what changed (url, lines that appeared, lines gone), so agents act on the page's structure before screenshots; v0.20.0: desk hosts, the Desks plan's step D5: install.sh --desk-host turns an old laptop, a mini PC or a spare workstation into a host of desks for its owner's agents, run as its own openvibe-desk account with Xvfb, openbox, D-Bus, xdotool, ImageMagick and Chromium installed and max_desks sized to its memory; v0.19.0: several desks per machine, the Desks plan's step D1: the computer plugin runs named desks (desk-<name>, private kept), each with its own X display, D-Bus, browser profile, downloads and temp folder, up to max_desks (one per 2 GiB of memory by default), so two agents work at once without seeing or driving each other's screen; proven with two real Chrome desks that never share a cookie; v0.18.0: 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,
ddiff --git a/cmd/openvibe-node/main.go b/cmd/openvibe-node/main.go
iindex 950be0f..01bb0fa 100644
---- a/cmd/openvibe-node/main.go
++++ b/cmd/openvibe-node/main.go
@@@ -246,7 +246,7 @@ func pluginSpecs(cfg *config.Config, paths config.Paths) []plugins.Spec {
for k, v := range p.Env {
env = append(env, k+"="+v)
}
-- out = append(out, plugins.Spec{Name: p.Name, Argv: cfg.PluginCommand(p, paths), Env: env, Dir: p.Dir, Config: p.Config})
++ out = append(out, plugins.Spec{Name: p.Name, Argv: cfg.PluginCommand(p, paths), Env: env, Dir: p.Dir, Config: p.Config, StrictTiming: p.StrictTiming})
}
return out
}
ddiff --git a/docs/hardware-looi.md b/docs/hardware-looi.md
iindex 676a4bb..81674fc 100644
---- a/docs/hardware-looi.md
++++ b/docs/hardware-looi.md
@@@ -47,6 +47,11 @@ held control every 150 ms). With the Python runtime (`openvibe_plugin`, [plugins
`motion_kinds = {"drive"}`, and it enforces the deadline, the 1 s heartbeat, stdin closing, `stop`, `estop` and
`halt` for you.
++The deadline is the original one on every OS, Windows included (Node 0.30): the Node passes the instant a command
++expires on the clock it shares with the driver, so time the line spends in the pipe counts against it. For live motion,
++also set `strict_timing = True` on the plugin (or `"strict_timing": true` on it in `config.json`): a command whose
++original expiry cannot be checked is then refused, never run on `deadline_ms` alone ([plugins.md](plugins.md), answer 9).
++
The driver should send telemetry with `battery` (`{"percent": 0..100}` or `{"volts": …}`) about twice a second.
## Stopping
@@@ -87,7 +92,7 @@ openvibe-node install --user-session --home "%OVHOME%"
"server": "https://openvibe.bot",
"device_kind": "bridge",
"python": "C:\\Users\\you\\AppData\\Local\\Programs\\Python\\Python312\\python.exe",
-- "plugins": [{"name": "looi"}],
++ "plugins": [{"name": "looi", "strict_timing": true}],
"video": {"source": "off"}
}
```
ddiff --git a/docs/plugins.md b/docs/plugins.md
iindex b9e19d8..21cee55 100644
---- a/docs/plugins.md
++++ b/docs/plugins.md
@@@ -47,7 +47,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) |
++| `hello` | `config` {…}, `probe` bool, `halt_epoch`? (the newest a halt carried, 0.17), `shared_clock`? (`clock_monotonic`, `qpc`; 0.30), `strict_timing`? (0.30) | answer `describe`; then set up (unless probe) |
| `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 |
@@@ -60,9 +60,14 @@ describes itself and waits for EOF without opening any hardware.
`value` has already been validated and clamped to the owner's limits by the core ([protocol.md](protocol.md#commands)).
`deadline_ms` is relative to receipt and already capped. 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.
++the line, and adds `expires_at_mono_ns`: the instant the command expires on the clock every process on the machine
++shares, `CLOCK_MONOTONIC` on Linux and macOS and, since 0.30, the raw `QueryPerformanceCounter` on Windows. `hello` names
++that clock in `shared_clock` (`clock_monotonic` or `qpc`, 0.30). A plugin that reads it when the line arrives takes off
++the time the line sat in the pipe too (answers 8 and 9); the Python runtime does, on every OS.
++
++`hello` also carries `strict_timing: true` when the Node's plugin configuration sets it (0.30). A plugin under strict
++timing refuses (`timing_unavailable`) any motion command whose original expiry it cannot check on that clock, instead of
++running it on `deadline_ms` alone (answer 9).
`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
@@@ -286,8 +291,8 @@ Reviewed again by the same authors against v0.17.0 (2026-10-27, `632207f`), and
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.
++ assumes both read the same clock id. In 0.18 Windows had no shared clock: `expires_at_mono_ns` was absent, and only
++ the core's queue time was counted, up to the write. 0.30 shares one there too (answer 9).
- **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
@@@ -339,11 +344,104 @@ Reviewed again by the same authors against v0.17.0 (2026-10-27, `632207f`), and
- **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.
++Reviewed by the same authors against v0.18.0 for Windows 10 (2026-10-27, `399ef83`), and changed in Node 0.30.0:
++
++9. **Windows: the original deadline across the pipe, and strict timing.**
++ - **The issue (confirmed, fixed).**
++ - In 0.18 `internal/plugins/mono_windows.go` had no clock to share, so on Windows a command carried `deadline_ms`
++ alone. The Python runtime counted it from when its reader got the line. A line that sat 200 ms in the pipe with
++ 100 ms left could still reach `handle()` with 100 ms.
++ - The real-pipe test skipped itself there, so a green Windows job proved nothing about it.
++ - A second gap, found while fixing the first: the core measured its own side on Go's monotonic clock. On Windows
++ that is the interrupt time, which moves once per clock tick (15.625 ms unless a program raises the timer
++ resolution; Go no longer does, since it sleeps on high-resolution timers). The link's estimate of Bot's clock,
++ its round trips and a command's receipt could each read up to a tick off: a command could get up to about two
++ ticks more than its deadline allowed.
++ - **What we found about QueryPerformanceCounter (QPC).**
++ - QPC is one counter for the whole machine, counting from boot at a frequency fixed at boot. Every process reads
++ the same value, and a step of the wall clock does not move it (Microsoft, "Acquiring high-resolution time
++ stamps"). `QueryPerformanceCounter` and `QueryPerformanceFrequency` are documented never to fail on Windows XP
++ and later.
++ - Go's own clock is not QPC. Its monotonic reading on Windows is the interrupt time (the runtime's `nanotime1`
++ reads `KUSER_SHARED_DATA`), with its own epoch and a tick's resolution.
++ - Python's own clocks are not reliably QPC either. `time.monotonic()` is QPC only from Python 3.13; before that it
++ is `GetTickCount64` (milliseconds, a different epoch). `time.perf_counter()` reads QPC but does not document its
++ reference point.
++ - So neither side uses its language's clock. Both read the raw counter: Go through `kernel32.dll` from System32
++ (`golang.org/x/sys/windows`), Python through `ctypes.WinDLL("kernel32")`, each with the same frequency.
++ - Both convert with the same integer arithmetic: whole seconds and the remainder apart, rounded down, so it never
++ overflows: `ns = (ticks / f) × 10⁹ + (ticks mod f) × 10⁹ / f`. The values are identical in both processes, and
++ the same six vectors (up to a hundred years of uptime at 10 MHz) are checked in Go and in Python.
++ - **The fix: one precise clock, shared.**
++ - `internal/mono` is the clock the core measures every deadline on: `CLOCK_MONOTONIC` on Linux and macOS, the raw
++ QPC on Windows. It covers the link's estimate of Bot's clock, its round trips, a command's receipt, the instant
++ it expires, and the writer's check. On Windows the core no longer measures anything on the tick clock.
++ - The core fixes the instant a command expires when it arrives, as before, now on that clock. The writer sends
++ that instant unchanged as `expires_at_mono_ns` (0.18 re-derived it when writing), with `deadline_ms` the whole
++ milliseconds then left. It never writes a command with less than 1 ms left.
++ - `hello` names the clock: `shared_clock` is `qpc` on Windows and `clock_monotonic` elsewhere.
++ - The runtime compares `expires_at_mono_ns` only with a reading of the clock the core named. Another clock's
++ instant counts as absent: it is never compared, and never read as expired or as open. (A 0.18 to 0.29 core names
++ none; it sent the field only on Linux and macOS, on `CLOCK_MONOTONIC`, which the runtime reads there.)
++ - On arrival the runtime reads its own clock, then the shared one, so the time left can only come out short. It
++ refuses (`expired`) a motion command whose instant has passed, before `handle()`.
++ - Python's version does not matter: 3.9 to 3.13 read the same raw counter.
++ - **Strict timing: refuse instead of downgrading.**
++ - Asked for in any of three ways:
++ - by the plugin: `strict_timing = True` on the class;
++ - by the plugin's environment: `OPENVIBE_STRICT_TIMING=1`;
++ - by the Node's configuration: `"strict_timing": true` on the plugin, which `hello` passes on.
++ - Under strict timing, a motion command whose original expiry cannot be checked is refused with
++ `timing_unavailable`, before `handle()`. That covers:
++ - no shared clock in the runtime (a missing clock);
++ - a read that fails when the line arrives (a failed clock);
++ - a core that named a different clock;
++ - a line without `expires_at_mono_ns` (an older core, or one without the clock).
++ - The core holds the same line. A plugin configured strict, or one that describes `"timing": "strict"`, never gets
++ a motion command without its expiry: the core answers `timing_unavailable` and does not write it.
++ - A plugin that does not ask for it runs as before: without the clock, on `deadline_ms`.
++ - A restarted plugin starts over. Its new `hello` names the clock again, and the rule applies from its first line.
++ - **The tests, and where they run.**
++ - On Linux, macOS and Windows: CI runs `go test` on all three, and the Windows and macOS jobs have a verbose step
++ "deadlines across the pipe" whose log shows each of these ran there. Each starts a real Python 3.11 plugin
++ process over real OS pipes, and fails rather than skips when CI has no Python:
++ - `TestPythonRuntimeReadsTheCoresClock`: the runtime's reading of the shared clock lies between two of the
++ core's, so they read one clock across the two processes. The log prints the three readings.
++ - `TestHelloNamesTheSharedClock`.
++ - `TestPipeDelayNeverExtendsADeadline` (your A and B): the plugin waits 200 ms before reading each line. The
++ command with 100 ms is refused `expired` and never reaches `handle()`. The one with 1,000 ms reaches it with
++ at most 800.
++ - `TestWriterQueueThenPipeDelay` (D, writer queue then pipe): the plugin is held 300 ms behind a 1 MB line, then
++ reads with a 200 ms delay. The command with 250 ms is refused. The one with 1,500 ms gets at most 1,000.
++ - `TestStrictTimingRefusesWithoutTheClock` (C):
++ - the runtime's read fails: refused;
++ - the core has no clock: a strict plugin refuses, and a strict core does not send the line;
++ - a plugin that is not strict still runs.
++ - `TestPythonSDKSuiteOnThisOS`: the SDK's own `test_runtime` and `test_strict_timing` on that OS.
++ - `internal/mono`: the conversion vectors; that the clock is there, named right and monotonic; and that it
++ resolves well under a millisecond. Its log line counts Go's own clock in the same 5 ms, for comparison.
++ - In the Python suite (Python 3.9, 3.11, 3.12 and 3.13 on Linux, plus the runs above), `test_strict_timing` covers
++ D's boundaries:
++ - exactly at the deadline, 0.999999 ms left and past it: all expired;
++ - 1 ms left: handled with exactly 1;
++ - never more than `deadline_ms`;
++ - a missing, a failed and a mismatched clock;
++ - strict timing from `hello` and from the environment;
++ - a restart naming another clock.
++ - With the shared clock taken away, the 100 ms command reaches `handle()` with 99 ms, and the tests fail.
++ - **What it still assumes.**
++ - The plugin runs on the same machine, started by the Node.
++ - QPC is consistent across processes, as Windows documents.
++ - The host is not suspended mid-command. The link is lost on resume anyway, and that loss is not a promise of an
++ immediate physical stop.
++ - As before, the guarantee ends at the runtime's check. The time inside `handle()` and the driver's own write to
++ the hardware (BLE, for P7) are the driver's.
++
### Plugin → core
| op | fields |
|-------------|---------------------------------------------------------------------------------------------|
--| `describe` | `driver`, `version`, `protocol` (1), `capabilities` {…}, `motion_kinds` [kinds] |
++| `describe` | `driver`, `version`, `protocol` (1), `capabilities` {…}, `motion_kinds` [kinds], `timing`? (`strict`, 0.30: the core refuses motion it cannot stamp with `expires_at_mono_ns`) |
| `ready` | — setup finished, every actuator stopped |
| `fault` | `fault_code`, `message` — the plugin runs but cannot drive (commands get this code as nack); `ready` again when it can (the robot came back) ends the fault |
| `ack` | `id` |
ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex b248baf..fc2d833 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -344,7 +344,8 @@ this mapping:
`bad_frame`, `bad_value`, `unsupported`, `not_allowed`, `estopped`, `local_stop`, `expired`, `no_heartbeat`,
`not_ready`, `plugin_down`, `plugin_timeout`, `hardware`, `firmware_unsupported`, `not_connected`, `cliff`,
--`picked_up`, `shutting_down`. Plugins may add their own; they are passed through.
++`picked_up`, `shutting_down`, `timing_unavailable` (0.30: a driver under strict timing could not check a motion command's
++original expiry, [plugins.md](plugins.md) answer 9). Plugins may add their own; they are passed through.
### Heartbeat, deadman, reconnect
ddiff --git a/internal/config/config.go b/internal/config/config.go
iindex da7ec61..16ffa03 100644
---- a/internal/config/config.go
++++ b/internal/config/config.go
@@@ -317,6 +317,10 @@ type PluginConfig struct {
Dir string `json:"dir,omitempty"`
Config map[string]any `json:"config,omitempty"` // sent to the plugin in hello
Disabled bool `json:"disabled,omitempty"`
++ // StrictTiming: every motion command must reach this plugin with its original expiry on the clock they share, else
++ // it is refused (timing_unavailable), never downgraded to deadline_ms alone. For deployments that require it
++ // (Windows 10 robots: the shared clock is QueryPerformanceCounter from Node v0.30.0).
++ StrictTiming bool `json:"strict_timing,omitempty"`
}
// VideoConfig selects the camera source published over WHIP.
@@@ -335,7 +339,9 @@ type VideoConfig struct {
WHIPURL string `json:"whip_url,omitempty"`
}
--// Bundled lists the plugins that ship with the Node.
++// Bundled lists the plugins the Node runs as <python> -m <module> without a command: the ones that ship with it, and
++// looi, whose driver is installed into the same Python on its own (docs/hardware-looi.md; install.sh --driver looi
++// writes {"name": "looi"}, which was refused as "needs a command" until 0.30).
var Bundled = map[string]string{
"dryrun": "openvibe_dryrun",
"adeept_adr036": "openvibe_adeept_adr036",
@@@ -345,6 +351,7 @@ var Bundled = map[string]string{
"computer": "openvibe_computer",
"voice": "openvibe_voice",
"ai": "openvibe_ai",
++ "looi": "openvibe_looi",
}
// Default is the config written by `openvibe-node pair` when none exists: the dry-run plugin and its test pattern.
ddiff --git a/internal/config/config_test.go b/internal/config/config_test.go
iindex aa02c1d..cab7f09 100644
---- a/internal/config/config_test.go
++++ b/internal/config/config_test.go
@@@ -66,6 +66,26 @@ func TestPluginCommand(t *testing.T) {
}
}
++// strict_timing on a plugin is read from config.json (the LOOI bridge's, docs/hardware-looi.md), and is off by default.
++func TestPluginStrictTiming(t *testing.T) {
++ p := filepath.Join(t.TempDir(), "config.json")
++ body := `{"server": "https://openvibe.bot", "plugins": [{"name": "looi", "strict_timing": true}, {"name": "dryrun"}]}`
++ if err := os.WriteFile(p, []byte(body), 0o600); err != nil {
++ t.Fatal(err)
++ }
++ c, err := Load(p)
++ if err != nil {
++ t.Fatal(err)
++ }
++ if len(c.Plugins) != 2 || !c.Plugins[0].StrictTiming || c.Plugins[1].StrictTiming {
++ t.Fatalf("%+v", c.Plugins)
++ }
++ // looi needs no command: the Node runs the driver installed into its Python (install.sh --driver looi).
++ if got := c.PluginCommand(c.Plugins[0], HomePaths(t.TempDir())); len(got) != 3 || got[2] != "openvibe_looi" {
++ t.Fatalf("%v", got)
++ }
++}
++
// TestWorkerVMLimitRequired: RLIMIT_AS is always set: max_vm_bytes has no "none" value and defaults when zero.
func TestWorkerVMLimitRequired(t *testing.T) {
c := Default()
ddiff --git a/internal/link/link.go b/internal/link/link.go
iindex 9e3c6e2..7d86d60 100644
---- a/internal/link/link.go
++++ b/internal/link/link.go
@@@ -34,6 +34,7 @@ import (
"github.com/gorilla/websocket"
"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
++ "github.com/OpenVibers/OpenVibe.Node/internal/mono"
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
)
@@@ -85,7 +86,7 @@ const HelloAllowance = 10 * time.Second
// hbSend is one heartbeat the link sent: its monotonic send time and the t it carried, so an ack can be matched
// back either by the echoed t (Bot) or by the body seq (older servers).
type hbSend struct {
-- at time.Time
++ at mono.Instant
t int64
}
@@@ -107,7 +108,7 @@ type Link struct {
// hasSkew is false until a frame arrives.
skew int64
hasSkew bool
-- clockBase time.Time
++ clockBase mono.Instant
connected atomic.Bool
lastRTT atomic.Int64
@@@ -162,7 +163,7 @@ func (l *Link) timing() (time.Duration, time.Duration) {
}
// observe records one server frame's ts, read at local time at.
--func (l *Link) observe(ts int64, at time.Time) {
++func (l *Link) observe(ts int64, at mono.Instant) {
if ts <= 0 {
return
}
@@@ -172,7 +173,7 @@ func (l *Link) observe(ts int64, at time.Time) {
l.clockBase, l.skew, l.hasSkew = at, ts*int64(time.Millisecond), true
return
}
-- l.skew = max(l.skew, ts*int64(time.Millisecond)-int64(at.Sub(l.clockBase)))
++ l.skew = max(l.skew, ts*int64(time.Millisecond)-int64(at-l.clockBase))
}
// TSResolution is the resolution of the server's ts: Bot stamps whole milliseconds (Date.now(), rounded down), so a
@@@ -187,13 +188,13 @@ const TSResolution = time.Millisecond
// 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) {
++func (l *Link) ServerNow(at mono.Instant) (ns int64, ok bool) {
l.mu.Lock()
defer l.mu.Unlock()
if !l.hasSkew {
return 0, false
}
-- return int64(at.Sub(l.clockBase)) + l.skew, true
++ return int64(at-l.clockBase) + l.skew, true
}
// ClockUncertainty bounds how far ServerNow can trail the server's real clock on this connection: the fastest
@@@ -254,7 +255,7 @@ func (l *Link) Send(m protocol.Message) error {
delete(l.hbSent, s)
}
}
-- l.hbSent[l.seq] = hbSend{at: now, t: hb.T}
++ l.hbSent[l.seq] = hbSend{at: mono.Now(), t: hb.T}
}
b, err := protocol.Encode(l.seq, now.UnixMilli(), m)
if err != nil {
@@@ -492,6 +493,7 @@ func (l *Link) session(ctx context.Context) error {
}
return endErr
}
++ rxAt := mono.Now() // the frame's arrival on the precise clock, for the estimate of Bot's
rx := time.Now()
lastRx.Store(rx.UnixNano())
f, err := protocol.Decode(data)
@@@ -503,7 +505,7 @@ func (l *Link) session(ctx context.Context) error {
}
continue
}
-- l.observe(f.TS, rx)
++ l.observe(f.TS, rxAt)
switch m := f.Msg.(type) {
case protocol.HeartbeatAck:
// Bot's envelope seq is its own counter, so the ack is matched to a stored send time by the echoed t
@@@ -514,7 +516,7 @@ func (l *Link) session(ctx context.Context) error {
echo = *m.Echo
}
l.mu.Lock()
-- var sent time.Time
++ var sent mono.Instant
if echo > 0 {
for s, hb := range l.hbSent {
if hb.t == echo {
@@@ -524,16 +526,16 @@ func (l *Link) session(ctx context.Context) error {
}
}
}
-- if sent.IsZero() && m.Seq != 0 {
++ if sent == 0 && m.Seq != 0 {
if hb, ok := l.hbSent[m.Seq]; ok {
sent = hb.at
delete(l.hbSent, m.Seq)
}
}
l.mu.Unlock()
-- if !sent.IsZero() {
-- // A coarse clock (Windows) can read 0 on a fast loopback; 0 means "not measured yet", so floor at 1 ns.
-- rtt := int64(max(time.Since(sent), 1))
++ if sent != 0 {
++ // 0 means "not measured yet", so a round trip is at least 1 ns.
++ rtt := int64(max(mono.Since(sent), 1))
l.lastRTT.Store(rtt)
for {
cur := l.minRTT.Load()
ddiff --git a/internal/link/link_test.go b/internal/link/link_test.go
iindex 7c78105..c058e10 100644
---- a/internal/link/link_test.go
++++ b/internal/link/link_test.go
@@@ -15,6 +15,7 @@ import (
"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
"github.com/OpenVibers/OpenVibe.Node/internal/fakebot"
++ "github.com/OpenVibers/OpenVibe.Node/internal/mono"
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
)
@@@ -493,32 +494,33 @@ func TestReplacedReconnects(t *testing.T) {
// ServerNow reads the server's clock from the frames' ts: the skew less the fastest transit.
func TestServerNow(t *testing.T) {
l := New(Options{URL: "ws://127.0.0.1:1/device"}, newRecorder())
-- if _, ok := l.ServerNow(time.Now()); ok {
++ if _, ok := l.ServerNow(mono.Now()); ok {
t.Fatal("an estimate before any frame")
}
-- ms := func(at time.Time) int64 {
++ ms := func(at mono.Instant) 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 _, 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)
++ local := func(ms int64) mono.Instant { return mono.Instant(ms * int64(time.Millisecond)) }
++ l.observe(10_000, local(9_900)) // server 100 ms ahead, fast frame
++ l.observe(10_000, local(10_500)) // a frame that sat 600 ms in a queue
++ if _, ok := l.ServerNow(local(20_000)); !ok || ms(local(20_000)) != 20_100 {
++ t.Fatalf("server now %d %v", ms(local(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))
++ l.observe(int64(30_000+i*10), local(int64(30_000+i*10)+400))
}
-- if got := ms(time.UnixMilli(40_000)); got != 40_100 {
++ if got := ms(local(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 got := ms(time.UnixMilli(60_000)); got != 60_000 {
++ l.observe(50_000, local(50_000))
++ if got := ms(local(60_000)); got != 60_000 {
t.Fatalf("estimate across connections: %d", got)
}
}
@@@ -527,7 +529,7 @@ func TestServerNow(t *testing.T) {
// 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)
++ base := mono.Instant(1_000 * int64(time.Second))
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))
ddiff --git a/internal/mono/mono.go b/internal/mono/mono.go
nnew file mode 100644
iindex 0000000..85d86ea
---- /dev/null
++++ b/internal/mono/mono.go
@@@ -0,0 +1,48 @@
++// Package mono is the machine's precise monotonic clock, in nanoseconds. Every deadline on this machine is measured on
++// it: in the core (the link's estimate of Bot's clock, its round trips, a command's receipt and the instant it
++// expires) and, through expires_at_mono_ns, in the plugins the core starts. It is one clock for every process on the
++// machine, which a step of the wall clock does not move: CLOCK_MONOTONIC on Linux and macOS (mono_unix.go), the raw
++// QueryPerformanceCounter on Windows (mono_windows.go).
++//
++// It is not Go's own monotonic clock (Maticus, 2026-10-27). On Windows that one is the interrupt time, which moves once
++// per clock tick: every 15.625 ms unless a program has raised the timer resolution, and Go no longer does since it
++// sleeps on high-resolution timers. A deadline measured on it could look up to a tick further away than it is.
++package mono
++
++import "time"
++
++// Instant is a reading of the clock: nanoseconds since an arbitrary point fixed at boot.
++type Instant int64
++
++// Read is the clock now, or ok false when it cannot be read: never on Linux or macOS, and never on Windows XP or later
++// by Microsoft's documentation of QueryPerformanceCounter. Plugins are told the clock only when it can be read.
++func Read() (int64, bool) { return read() }
++
++// startNS and startGo are one reading of the clock and of Go's, taken together, for Now's fallback.
++var (
++ startNS, _ = read()
++ startGo = time.Now()
++)
++
++// Now is the clock now. Should a read ever fail, it is the first reading plus the time Go's clock has run since: the
++// same timeline, at Go's resolution.
++func Now() Instant {
++ if ns, ok := read(); ok {
++ return Instant(ns)
++ }
++ return Instant(startNS + int64(time.Since(startGo)))
++}
++
++// Since is how long ago i was; Until how long until it is.
++func Since(i Instant) time.Duration { return time.Duration(Now() - i) }
++func Until(i Instant) time.Duration { return time.Duration(i - Now()) }
++
++func (i Instant) Add(d time.Duration) Instant { return i + Instant(d) }
++func (i Instant) Sub(j Instant) time.Duration { return time.Duration(i - j) }
++
++// qpcNS turns a QueryPerformanceCounter reading into nanoseconds, rounded down: whole seconds and the remainder apart,
++// so nothing overflows int64 (ticks × 1e9 would after about 15 minutes of uptime at 10 MHz). The Python runtime does the
++// same integer arithmetic on the same raw counter (openvibe_plugin._qpc_ns), so the two values compare exactly.
++func qpcNS(ticks, freq int64) int64 {
++ return (ticks/freq)*1_000_000_000 + (ticks%freq)*1_000_000_000/freq
++}
ddiff --git a/internal/mono/mono_test.go b/internal/mono/mono_test.go
nnew file mode 100644
iindex 0000000..5b229f0
---- /dev/null
++++ b/internal/mono/mono_test.go
@@@ -0,0 +1,81 @@
++package mono
++
++import (
++ "runtime"
++ "testing"
++ "time"
++)
++
++// The same vectors as plugins/sdk/tests/test_strict_timing.py QPC_VECTORS: Node and the Python runtime turn a raw
++// QueryPerformanceCounter reading into the same nanoseconds (rounded down, no overflow at a hundred years of uptime).
++func TestQPCNSMatchesThePythonRuntime(t *testing.T) {
++ for _, v := range [][3]int64{
++ {0, 10_000_000, 0},
++ {9_999_999, 10_000_000, 999_999_900},
++ {30_000_005, 10_000_000, 3_000_000_500},
++ {3_579_545*3600 + 1, 3_579_545, 3_600_000_000_279},
++ {31_536_000_000_000_007, 10_000_000, 3_153_600_000_000_000_700},
++ {2_441_406*86_400 + 2_441_405, 2_441_406, 86_400_999_999_590},
++ } {
++ if got := qpcNS(v[0], v[1]); got != v[2] {
++ t.Errorf("qpcNS(%d, %d) = %d, want %d", v[0], v[1], got, v[2])
++ }
++ }
++}
++
++// Every OS has the clock (Windows: QueryPerformanceCounter, from v0.30.0), it never goes back, and Now reads it.
++func TestTheClockIsThereAndMonotonic(t *testing.T) {
++ a, ok := Read()
++ if !ok {
++ t.Fatalf("no clock on %s", runtime.GOOS)
++ }
++ n := Now()
++ b, _ := Read()
++ if Instant(a) > n || n > Instant(b) {
++ t.Fatalf("Now is not the clock: %d ≤ %d ≤ %d does not hold", a, n, b)
++ }
++ want := map[string]string{"windows": "qpc"}[runtime.GOOS]
++ if want == "" {
++ want = "clock_monotonic"
++ }
++ if Name != want {
++ t.Fatalf("the clock on %s is %q, want %q", runtime.GOOS, Name, want)
++ }
++}
++
++// The clock resolves well under a millisecond on every OS, so no deadline is measured a clock tick long. On Windows
++// Go's own monotonic clock is the interrupt time: the log line shows it moving a few times in the same window (or not
++// at all), where this clock moves thousands of times.
++func TestTheClockResolvesMicroseconds(t *testing.T) {
++ distinct, goDistinct := 0, 0
++ var last Instant
++ var lastGo time.Duration
++ t0, start := time.Now(), Now()
++ for Since(start) < 5*time.Millisecond {
++ if n := Now(); n != last {
++ distinct, last = distinct+1, n
++ }
++ if g := time.Since(t0); g != lastGo {
++ goDistinct, lastGo = goDistinct+1, g
++ }
++ }
++ t.Logf("%s: %s took %d distinct readings in 5 ms; Go's monotonic clock %d", runtime.GOOS, Name, distinct, goDistinct)
++ if distinct < 50 {
++ t.Fatalf("%s moved only %d times in 5 ms: coarser than 100 µs", Name, distinct)
++ }
++}
++
++// Until and Since are the two sides of the same reading.
++func TestUntilAndSince(t *testing.T) {
++ now := Now()
++ later := now.Add(time.Hour)
++ if d := Until(later); d <= 59*time.Minute || d > time.Hour {
++ t.Fatalf("until an hour from now: %v", d)
++ }
++ if d := later.Sub(now); d != time.Hour {
++ t.Fatalf("sub: %v", d)
++ }
++ if d := Since(now); d < 0 || d > time.Minute {
++ t.Fatalf("since: %v", d)
++ }
++}
ddiff --git a/internal/mono/mono_unix.go b/internal/mono/mono_unix.go
nnew file mode 100644
iindex 0000000..54c3d3e
---- /dev/null
++++ b/internal/mono/mono_unix.go
@@@ -0,0 +1,17 @@
++//go:build !windows
++
++package mono
++
++import "golang.org/x/sys/unix"
++
++// Name is the clock as hello names it to a plugin (shared_clock): the plugin reads the same one, or says it cannot.
++const Name = "clock_monotonic"
++
++// read is CLOCK_MONOTONIC (a plugin reads it in Python with time.clock_gettime_ns(time.CLOCK_MONOTONIC)).
++func read() (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/mono/mono_windows.go b/internal/mono/mono_windows.go
nnew file mode 100644
iindex 0000000..29deb44
---- /dev/null
++++ b/internal/mono/mono_windows.go
@@@ -0,0 +1,42 @@
++//go:build windows
++
++package mono
++
++import (
++ "unsafe"
++
++ "golang.org/x/sys/windows"
++)
++
++// Name is the clock as hello names it to a plugin (shared_clock): the plugin reads the same one, or says it cannot.
++const Name = "qpc"
++
++// QueryPerformanceCounter is one counter for the whole machine, counting since boot at a frequency fixed at boot, which
++// a step of the wall clock does not move: every process reads the same value (Microsoft: "Acquiring high-resolution
++// time stamps"). Read raw, from kernel32 in System32.
++var (
++ procQPC = windows.NewLazySystemDLL("kernel32.dll").NewProc("QueryPerformanceCounter")
++ procQPF = windows.NewLazySystemDLL("kernel32.dll").NewProc("QueryPerformanceFrequency")
++ qpcFreq = func() int64 {
++ var f int64
++ if procQPF.Find() != nil {
++ return 0
++ }
++ if r, _, _ := procQPF.Call(uintptr(unsafe.Pointer(&f))); r == 0 || f <= 0 {
++ return 0
++ }
++ return f
++ }()
++)
++
++// read is QueryPerformanceCounter in nanoseconds (qpcNS), or ok false when it cannot be read.
++func read() (int64, bool) {
++ if qpcFreq <= 0 || procQPC.Find() != nil {
++ return 0, false
++ }
++ var t int64
++ if r, _, _ := procQPC.Call(uintptr(unsafe.Pointer(&t))); r == 0 {
++ return 0, false
++ }
++ return qpcNS(t, qpcFreq), true
++}
ddiff --git a/internal/node/clock_test.go b/internal/node/clock_test.go
iindex d217a45..6e315d1 100644
---- a/internal/node/clock_test.go
++++ b/internal/node/clock_test.go
@@@ -1,6 +1,7 @@
package node
import (
++ "github.com/OpenVibers/OpenVibe.Node/internal/mono"
"testing"
"time"
)
@@@ -37,12 +38,12 @@ func TestCeilMSRoundsUpFromNanoseconds(t *testing.T) {
}
}
--// Without a link (local commands) the time left is exact to the nanosecond: nothing rounds a deadline later.
++// Without a link (local commands) the frame's own ts is the server's clock, and the time left is exact: nothing rounds
++// a deadline later. (The link's estimate keeps nanoseconds: link.TestServerNowKeepsNanoseconds.)
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 {
++ left, unc, synced := n.timeLeft(1_000_100, 1_000_000, mono.Now())
++ if left != 100*time.Millisecond || 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 57b0240..800dc63 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -29,6 +29,7 @@ import (
"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
"github.com/OpenVibers/OpenVibe.Node/internal/link"
"github.com/OpenVibers/OpenVibe.Node/internal/localctl"
++ "github.com/OpenVibers/OpenVibe.Node/internal/mono"
"github.com/OpenVibers/OpenVibe.Node/internal/plugins"
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
"github.com/OpenVibers/OpenVibe.Node/internal/safety"
@@@ -172,7 +173,7 @@ func New(opt Options) (*Node, error) {
if _, ok := cfg["frame_dir"]; !ok {
cfg["frame_dir"] = opt.Paths.FramesDir() + string(os.PathSeparator) + p.Name
}
-- specs = append(specs, plugins.Spec{Name: p.Name, Argv: opt.Config.PluginCommand(p, opt.Paths), Env: env, Dir: p.Dir, Config: cfg})
++ specs = append(specs, plugins.Spec{Name: p.Name, Argv: opt.Config.PluginCommand(p, opt.Paths), Env: env, Dir: p.Dir, Config: cfg, StrictTiming: p.StrictTiming})
}
n.mgr = plugins.NewManager(specs, opt.Log)
@@@ -324,7 +325,7 @@ func (n *Node) Frame(f protocol.Frame) {
case protocol.Config:
n.applyConfig(m)
case protocol.Command:
-- n.command(m, f.TS, f.Seq, time.Now())
++ n.command(m, f.TS, f.Seq, mono.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 {
@@@ -466,7 +467,7 @@ func (n *Node) voiceAnswer(a protocol.VoiceAnswer) {
// 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) {
++func (n *Node) command(c protocol.Command, sentMS int64, seq uint64, received mono.Instant) {
if c.ID == "" {
n.log.Warn("command without id ignored", "kind", c.Kind)
return
@@@ -504,7 +505,7 @@ func (n *Node) command(c protocol.Command, sentMS int64, seq uint64, received ti
if c.Kind == protocol.KindHalt {
n.noteHaltEpoch(c.RobotID, c.TurnEpoch)
n.mgr.HaltAll(c.TurnEpoch)
-- r := protocol.Ack{ID: c.ID, LatencyMS: time.Since(received).Milliseconds()}
++ r := protocol.Ack{ID: c.ID, LatencyMS: mono.Since(received).Milliseconds()}
n.dedup.Complete(entry, r)
n.send(r)
return
@@@ -598,7 +599,7 @@ func (n *Node) command(c protocol.Command, sentMS int64, seq uint64, received ti
r := wait()
var msg protocol.Message
if r.OK {
-- msg = protocol.Ack{ID: c.ID, LatencyMS: time.Since(received).Milliseconds()}
++ msg = protocol.Ack{ID: c.ID, LatencyMS: mono.Since(received).Milliseconds()}
} else {
msg = protocol.Nack{ID: c.ID, FaultCode: r.FaultCode, Message: r.Message}
}
@@@ -610,7 +611,7 @@ func (n *Node) command(c protocol.Command, sentMS int64, seq uint64, received ti
// computer runs one exec, fs or screen command: no clamping, a long reply window (the server's deadline, at most
// MaxComputerMS, else DefaultComputerMS), refused while the local stop or an e-stop is latched when it would change the
// machine, and answered with the plugin's data. The plugin applies its owner's control level.
--func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, received time.Time, nack func(code, msg string)) {
++func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, received mono.Instant, nack func(code, msg string)) {
if len(c.Value) > protocol.MaxComputerValue {
nack(protocol.FaultBadValue, fmt.Sprintf("the value is larger than %d KB", protocol.MaxComputerValue/1024))
return
@@@ -670,7 +671,7 @@ func (n *Node) computer(c protocol.Command, entry *safety.Entry, sentMS int64, r
case r.OK && len(r.Data) > protocol.MaxAckData:
msg = protocol.Nack{ID: c.ID, FaultCode: protocol.FaultBadValue, Message: fmt.Sprintf("the answer is larger than %d KB", protocol.MaxAckData/1024)}
case r.OK:
-- msg = protocol.AckData{ID: c.ID, LatencyMS: time.Since(received).Milliseconds(), Data: r.Data}
++ msg = protocol.AckData{ID: c.ID, LatencyMS: mono.Since(received).Milliseconds(), Data: r.Data}
default:
msg = protocol.Nack{ID: c.ID, FaultCode: r.FaultCode, Message: r.Message}
}
@@@ -709,7 +710,7 @@ func (n *Node) haltEpoch(robotID string) int64 {
// 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) {
++func (n *Node) timeLeft(deadlineMS, sentMS int64, received mono.Instant) (left time.Duration, unc int64, synced bool) {
end := deadlineMS * int64(time.Millisecond)
now := n.serverNow(sentMS, received)
if n.link == nil {
@@@ -736,8 +737,8 @@ func latchReason(st safety.LatchState) string {
}
// 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 {
++// frame's own ts, else the wall clock now (later than the receipt, so never more time).
++func (n *Node) serverNow(sentMS int64, received mono.Instant) int64 {
if n.link != nil {
if ns, ok := n.link.ServerNow(received); ok {
return ns
@@@ -746,7 +747,7 @@ func (n *Node) serverNow(sentMS int64, received time.Time) int64 {
if sentMS > 0 {
return sentMS * int64(time.Millisecond)
}
-- return received.UnixNano()
++ return time.Now().UnixNano()
}
func (n *Node) send(m protocol.Message) {
ddiff --git a/internal/node/screen.go b/internal/node/screen.go
iindex a2467b1..334ae08 100644
---- a/internal/node/screen.go
++++ b/internal/node/screen.go
@@@ -10,6 +10,7 @@ import (
"time"
"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
++ "github.com/OpenVibers/OpenVibe.Node/internal/mono"
"github.com/OpenVibers/OpenVibe.Node/internal/plugins"
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
"github.com/OpenVibers/OpenVibe.Node/internal/safety"
@@@ -66,7 +67,7 @@ type screenValue struct {
// screenCommand answers screen_stream: the plugin is asked first (it applies the owner's level and has the capture
// tool), and only once it said yes does the core publish.
--func (n *Node) screenCommand(c protocol.Command, entry *safety.Entry, received time.Time, nack func(code, msg string)) {
++func (n *Node) screenCommand(c protocol.Command, entry *safety.Entry, received mono.Instant, nack func(code, msg string)) {
var v screenValue
if err := json.Unmarshal(c.Value, &v); err != nil {
nack(protocol.FaultBadValue, "value must be {action, …}")
@@@ -82,7 +83,7 @@ func (n *Node) screenCommand(c protocol.Command, entry *safety.Entry, received t
go wait()
}
data, _ := json.Marshal(map[string]any{"streaming": false, "stopped": stopped})
-- r := protocol.AckData{ID: c.ID, LatencyMS: time.Since(received).Milliseconds(), Data: data}
++ r := protocol.AckData{ID: c.ID, LatencyMS: mono.Since(received).Milliseconds(), Data: data}
n.dedup.Complete(entry, r)
n.send(r)
return
@@@ -133,7 +134,7 @@ func (n *Node) screenCommand(c protocol.Command, entry *safety.Entry, received t
_ = json.Unmarshal(r.Data, &got)
until := n.startScreen(v.WHIPURL, key, seconds, fps, p.Name(), got.SID)
data, _ := json.Marshal(map[string]any{"streaming": true, "seconds": seconds, "fps": fps, "until_ms": until.UnixMilli()})
-- msg = protocol.AckData{ID: c.ID, LatencyMS: time.Since(received).Milliseconds(), Data: data}
++ msg = protocol.AckData{ID: c.ID, LatencyMS: mono.Since(received).Milliseconds(), Data: data}
} else {
msg = protocol.Nack{ID: c.ID, FaultCode: r.FaultCode, Message: r.Message}
}
ddiff --git a/internal/plugins/clock.go b/internal/plugins/clock.go
nnew file mode 100644
iindex 0000000..bcfcf6d
---- /dev/null
++++ b/internal/plugins/clock.go
@@@ -0,0 +1,13 @@
++package plugins
++
++import "github.com/OpenVibers/OpenVibe.Node/internal/mono"
++
++// The clock a plugin shares with the core (internal/mono: CLOCK_MONOTONIC on Unix, the raw QueryPerformanceCounter on
++// Windows, named in hello as shared_clock): a command's expires_at_mono_ns is an instant on it, so the runtime can take
++// off the time a line spent in the pipe and its reader (Maticus, 2026-10-27).
++
++// monotonicNS reads it; a variable so tests can take it away (a machine without one, or a read that fails).
++var monotonicNS = mono.Read
++
++// sharedClockName is the clock hello names.
++const sharedClockName = mono.Name
ddiff --git a/internal/plugins/clock_test.go b/internal/plugins/clock_test.go
nnew file mode 100644
iindex 0000000..ef142f8
---- /dev/null
++++ b/internal/plugins/clock_test.go
@@@ -0,0 +1,26 @@
++package plugins
++
++import (
++ "os"
++ "os/exec"
++ "path/filepath"
++ "runtime"
++ "strings"
++ "testing"
++)
++
++// The Python SDK's own tests, on every OS a CI job runs go test on (Linux, macOS, Windows): the runtime behaves the
++// same everywhere a driver runs (owner, 2026-10-27: "work more in harmony and unified across all operating systems").
++func TestPythonSDKSuiteOnThisOS(t *testing.T) {
++ py := pythonForTest(t)
++ dir, _ := filepath.Abs(filepath.Join("..", "..", "plugins", "sdk", "tests"))
++ cmd := exec.Command(py, "-m", "unittest", "-v", "test_runtime", "test_strict_timing")
++ cmd.Dir = dir
++ cmd.Env = append(os.Environ(), "PYTHONDONTWRITEBYTECODE=1")
++ out, err := cmd.CombinedOutput()
++ if err != nil {
++ t.Fatalf("the Python SDK's tests failed on %s:\n%s", runtime.GOOS, out)
++ }
++ lines := strings.Split(strings.TrimSpace(string(out)), "\n")
++ t.Logf("%s: %s", runtime.GOOS, lines[len(lines)-1])
++}
ddiff --git a/internal/plugins/mono_unix.go b/internal/plugins/mono_unix.go
ddeleted file mode 100644
iindex 057a325..0000000
---- a/internal/plugins/mono_unix.go
++++ /dev/null
@@@ -1,15 +0,0 @@
--//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
ddeleted file mode 100644
iindex 33dfc2f..0000000
---- a/internal/plugins/mono_windows.go
++++ /dev/null
@@@ -1,7 +0,0 @@
--//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 0851090..5d1311c 100644
---- a/internal/plugins/plugin.go
++++ b/internal/plugins/plugin.go
@@@ -18,6 +18,7 @@ import (
"sync"
"time"
++ "github.com/OpenVibers/OpenVibe.Node/internal/mono"
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
)
@@@ -48,6 +49,10 @@ type Spec struct {
Env []string // extra KEY=VALUE entries
Dir string
Config map[string]any
++ // StrictTiming: a motion command is never written to this plugin without its expiry on the clock they share
++ // (expires_at_mono_ns); with no shared clock it is answered timing_unavailable instead. Also set by the plugin's
++ // describe ("timing": "strict"), and passed to it in hello so its runtime refuses the same way.
++ StrictTiming bool
}
// Describe is the plugin's first message.
@@@ -57,6 +62,8 @@ type Describe struct {
Protocol int `json:"protocol"`
Capabilities map[string]any `json:"capabilities"`
MotionKinds []string `json:"motion_kinds"`
++ // Timing "strict": the plugin requires every motion command's original expiry on the shared clock (Spec.StrictTiming).
++ Timing string `json:"timing,omitempty"`
}
// Reply is a plugin's answer to one command.
@@@ -174,43 +181,78 @@ type line struct {
type timed struct {
id string
msg map[string]any
-- expires time.Time // on this process's monotonic clock
-- queued time.Time
++ expires mono.Instant // on the machine's precise clock, the one the plugin shares
++ queued mono.Instant
++ motion bool // a guarded or motion kind: strict timing applies to it
}
--// 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)
++// render is the command's line as it is written now, or the fault it is refused with instead: expired when less than a
++// millisecond is left, timing_unavailable when strict timing applies to it and there is no clock shared with the plugin
++// (or it could not be read) to carry its original expiry.
++func (c *timed) render(strict bool) (b []byte, refused string) {
++ left := mono.Until(c.expires)
if left < time.Millisecond {
-- return nil, false
++ return nil, protocol.FaultExpired
++ }
++ _, hasMono := monotonicNS()
++ if strict && c.motion && !hasMono {
++ return nil, protocol.FaultTimingUnavailable
}
c.msg["deadline_ms"] = int64(left / time.Millisecond)
if hasMono {
-- c.msg["expires_at_mono_ns"] = mono + int64(left)
++ // The instant itself: the core keeps it on the clock it shares with the plugin, so it is passed exactly.
++ c.msg["expires_at_mono_ns"] = int64(c.expires)
}
b, err := json.Marshal(c.msg)
if err != nil {
-- return nil, false
++ return nil, protocol.FaultExpired
}
-- return append(b, '\n'), true
++ return append(b, '\n'), ""
}
--// 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) {
++// refuse answers a timed command that was never written: its time ran out in the queue to the plugin (expired), or
++// strict timing could not carry its original expiry to the plugin (timing_unavailable).
++func (p *Plugin) refuse(c *timed, code string) {
p.mu.Lock()
ch := p.pending[c.id]
p.mu.Unlock()
if ch == nil {
return
}
++ msg := fmt.Sprintf("its deadline passed while it waited %d ms in the queue to the driver; it was never sent", mono.Since(c.queued).Milliseconds())
++ if code == protocol.FaultTimingUnavailable {
++ msg = "strict timing: this machine has no clock shared with the driver to carry the command's original expiry, so it was not sent"
++ }
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())}:
++ case ch <- Reply{FaultCode: code, Message: msg}:
default:
}
}
++// strictTiming is whether motion commands to this plugin need the shared clock: its configuration or its describe.
++func (p *Plugin) strictTiming() bool {
++ p.mu.Lock()
++ defer p.mu.Unlock()
++ return p.spec.StrictTiming || (p.desc != nil && p.desc.Timing == "strict")
++}
++
++// motionKind is whether kind moves something on this plugin: a guarded kind or one of its describe's motion kinds.
++func (p *Plugin) motionKind(kind string) bool {
++ if protocol.GuardedKinds[kind] {
++ return true
++ }
++ p.mu.Lock()
++ defer p.mu.Unlock()
++ if p.desc != nil {
++ for _, k := range p.desc.MotionKinds {
++ if k == kind {
++ return true
++ }
++ }
++ }
++ return false
++}
++
// 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 {
@@@ -339,7 +381,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.begin(ctx, id, kind, value, deadlineMS, time.Time{}, op, turnEpoch, nil)
++ return p.begin(ctx, id, kind, value, deadlineMS, 0, op, turnEpoch, nil)
}
// Provenance is where a command came from, passed unchanged from Bot to the driver as `provenance` (plugins.md): the
@@@ -361,12 +403,12 @@ type Provenance struct {
// 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 {
++func (p *Plugin) BeginWith(ctx context.Context, id, kind string, value json.RawMessage, expires mono.Instant, 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 {
++func (p *Plugin) begin(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int, expires mono.Instant, 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 {
@@@ -406,10 +448,10 @@ func (p *Plugin) begin(ctx context.Context, id, kind string, value json.RawMessa
msg["provenance"] = prov
}
var err error
-- if expires.IsZero() {
++ if expires == 0 {
err = p.send(msg)
} else {
-- err = p.enqueue(line{cmd: &timed{id: id, msg: msg, expires: expires, queued: time.Now()}})
++ err = p.enqueue(line{cmd: &timed{id: id, msg: msg, expires: expires, queued: mono.Now(), motion: p.motionKind(kind)}})
}
if err != nil {
done()
@@@ -418,8 +460,8 @@ func (p *Plugin) begin(ctx context.Context, id, kind string, value json.RawMessa
return func() Reply {
defer done()
wait := time.Duration(deadlineMS) * time.Millisecond
-- if !expires.IsZero() {
-- wait = time.Until(expires)
++ if expires != 0 {
++ wait = mono.Until(expires)
}
if wait < MinReplyTimeout {
wait = MinReplyTimeout
@@@ -531,9 +573,9 @@ func (p *Plugin) runOnce(ctx context.Context) error {
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)
++ var refused string
++ if b, refused = l.cmd.render(p.strictTiming()); refused != "" {
++ p.refuse(l.cmd, refused)
return nil
}
}
@@@ -574,6 +616,13 @@ func (p *Plugin) runOnce(ctx context.Context) error {
}()
hello := map[string]any{"op": "hello", "config": nonNil(p.spec.Config), "probe": false}
++ // The clock expires_at_mono_ns is on: the runtime checks it reads the same one before trusting the field.
++ if _, ok := monotonicNS(); ok {
++ hello["shared_clock"] = sharedClockName
++ }
++ if p.spec.StrictTiming {
++ hello["strict_timing"] = true
++ }
if haltEpoch > 0 {
hello["halt_epoch"] = haltEpoch
}
ddiff --git a/internal/plugins/plugins_test.go b/internal/plugins/plugins_test.go
iindex a0bba90..ed30bc6 100644
---- a/internal/plugins/plugins_test.go
++++ b/internal/plugins/plugins_test.go
@@@ -15,6 +15,7 @@ import (
"testing"
"time"
++ "github.com/OpenVibers/OpenVibe.Node/internal/mono"
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
)
@@@ -387,12 +388,12 @@ func TestWriterQueueDelayCountsAgainstTheDeadline(t *testing.T) {
fx.waitState(StateReady)
p := fx.m.Get("fake")
ctx := context.Background()
-- held := time.Now()
++ held := mono.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()
++ queued := mono.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") {
ddiff --git a/internal/plugins/python_ipc_test.go b/internal/plugins/python_ipc_test.go
nnew file mode 100644
iindex 0000000..4f8418e
---- /dev/null
++++ b/internal/plugins/python_ipc_test.go
@@@ -0,0 +1,277 @@
++package plugins
++
++// The original-deadline guarantee across the real IPC boundary, with the real Python runtime (plugins/sdk), on every OS
++// CI runs (Linux, macOS and Windows: each job has Python; Maticus, 2026-10-27: "these tests must actually execute on
++// Windows. A skipped test or a Unix-only equivalent is not sufficient"). Under GitHub Actions a missing Python fails the
++// test instead of skipping it.
++//
++// The plugin script holds each drive line it has read for IPC_DELAY_MS before the runtime's reader sees it (time in the
++// pipe and the reader), and holds the whole pipe for HOLD_MS on a "hold" command (so the core's writer queue backs up
++// behind a full pipe). handle() records every command it is given in a file: an expired command must never be in it.
++
++import (
++ "context"
++ "encoding/json"
++ "io"
++ "log/slog"
++ "os"
++ "os/exec"
++ "path/filepath"
++ "runtime"
++ "strings"
++ "testing"
++ "time"
++
++ "github.com/OpenVibers/OpenVibe.Node/internal/mono"
++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++const ipcPlugin = `
++import json, os, sys, time
++from openvibe_plugin import Plugin, Runtime, SHARED_CLOCK, SHARED_CLOCK_NAME
++
++DELAY = int(os.environ.get("IPC_DELAY_MS", "0")) / 1000.0
++HOLD = int(os.environ.get("HOLD_MS", "0")) / 1000.0
++HANDLED = os.environ["HANDLED_FILE"]
++
++
++class Slow:
++ """stdin as the runtime reads it: a drive line reaches the runtime IPC_DELAY_MS after it was read off the pipe, and
++ a hold command stops reading the pipe for HOLD_MS (the pipe fills and the core's writer waits)."""
++ def __init__(self, f):
++ self.f = f
++
++ def __iter__(self):
++ for line in self.f:
++ if '"kind":"drive"' in line:
++ time.sleep(DELAY)
++ if '"kind":"hold"' in line:
++ time.sleep(HOLD)
++ yield line
++
++
++class Probe(Plugin):
++ driver = "ipc-probe"
++ version = "1"
++ motion_kinds = frozenset({"drive"})
++ strict_timing = os.environ.get("STRICT") == "1"
++
++ def describe(self, config):
++ # This process's reading of the shared clock: the core checks it falls between two readings of its own.
++ return {"shared_clock_name": SHARED_CLOCK_NAME, "shared_clock_ns": SHARED_CLOCK() if SHARED_CLOCK else None}
++
++ def handle(self, cmd):
++ with open(HANDLED, "a") as f:
++ f.write(cmd.id + "\n")
++ return {"deadline_ms": cmd.deadline_ms}
++
++
++shared = SHARED_CLOCK
++if os.environ.get("BREAK_CLOCK") == "1":
++ def shared():
++ raise OSError("the shared clock read failed (test)")
++Runtime(Probe(), infile=Slow(sys.stdin), shared_clock=shared).run()
++`
++
++// pythonForTest finds a Python 3, or skips (fails under GitHub Actions, where every job installs one).
++func pythonForTest(t *testing.T) string {
++ t.Helper()
++ for _, name := range []string{"python3", "python"} {
++ if p, err := exec.LookPath(name); err == nil {
++ if out, err := exec.Command(p, "-c", "import sys; print(sys.version_info[0])").Output(); err == nil && strings.TrimSpace(string(out)) == "3" {
++ return p
++ }
++ }
++ }
++ if os.Getenv("GITHUB_ACTIONS") == "true" {
++ t.Fatal("no Python 3 on this CI runner: the cross-process deadline test must run here, not be skipped")
++ }
++ t.Skip("no Python 3")
++ return ""
++}
++
++type ipcFixture struct {
++ t *testing.T
++ m *Manager
++ p *Plugin
++ handled string
++ before int64
++}
++
++func startIPC(t *testing.T, strictNode bool, env ...string) *ipcFixture {
++ t.Helper()
++ py := pythonForTest(t)
++ dir := t.TempDir()
++ script := filepath.Join(dir, "probe.py")
++ if err := os.WriteFile(script, []byte(ipcPlugin), 0o644); err != nil {
++ t.Fatal(err)
++ }
++ sdk, _ := filepath.Abs(filepath.Join("..", "..", "plugins", "sdk"))
++ handled := filepath.Join(dir, "handled.txt")
++ spec := Spec{Name: "py", Argv: []string{py, script}, StrictTiming: strictNode,
++ Env: append([]string{"PYTHONPATH=" + sdk, "HANDLED_FILE=" + handled, "PYTHONUNBUFFERED=1"}, env...)}
++ before, _ := monotonicNS()
++ m := NewManager([]Spec{spec}, slog.New(slog.NewTextHandler(io.Discard, nil)))
++ ctx, cancel := context.WithCancel(context.Background())
++ m.Start(ctx)
++ go func() {
++ for range m.Output() {
++ }
++ }()
++ t.Cleanup(func() { cancel(); m.Wait() })
++ fx := &ipcFixture{t: t, m: m, p: m.Get("py"), handled: handled, before: before}
++ deadline := time.Now().Add(20 * time.Second)
++ for time.Now().Before(deadline) && fx.p.Info().State != StateReady {
++ time.Sleep(10 * time.Millisecond)
++ }
++ if st := fx.p.Info().State; st != StateReady {
++ t.Fatalf("the Python plugin never became ready (state %s, fault %+v)", st, fx.p.Info().Fault)
++ }
++ return fx
++}
++
++func (fx *ipcFixture) drive(id string, left time.Duration) Reply {
++ return fx.p.BeginWith(context.Background(), id, protocol.KindDrive, json.RawMessage(`{"throttle":0.1,"steer":0}`), mono.Now().Add(left), nil, 0, nil)()
++}
++
++func (fx *ipcFixture) handledIDs() []string {
++ b, _ := os.ReadFile(fx.handled)
++ return strings.Fields(string(b))
++}
++
++// The plugin's reading of the shared clock lies between two of the core's: the same clock, epoch and unit in both
++// processes (on Windows, the raw QueryPerformanceCounter through qpcNS and _qpc_ns).
++func TestPythonRuntimeReadsTheCoresClock(t *testing.T) {
++ fx := startIPC(t, false)
++ after, ok := monotonicNS()
++ if !ok {
++ t.Fatalf("no shared clock on %s", runtime.GOOS)
++ }
++ var caps struct {
++ Name string `json:"shared_clock_name"`
++ NS *int64 `json:"shared_clock_ns"`
++ }
++ b, _ := json.Marshal(fx.p.Info().Describe.Capabilities)
++ _ = json.Unmarshal(b, &caps)
++ if caps.Name != sharedClockName || caps.NS == nil {
++ t.Fatalf("the runtime reads %q (%v), the core %q", caps.Name, caps.NS, sharedClockName)
++ }
++ if *caps.NS < fx.before || *caps.NS > after {
++ t.Fatalf("the runtime's clock read %d, outside the core's %d…%d: not the same clock", *caps.NS, fx.before, after)
++ }
++ t.Logf("%s on %s: core %d ≤ runtime %d ≤ core %d", sharedClockName, runtime.GOOS, fx.before, *caps.NS, after)
++}
++
++// hello names the clock expires_at_mono_ns is on, on every start (a restarted plugin hears it again), and passes the
++// Node's strict timing to the runtime.
++func TestHelloNamesTheSharedClock(t *testing.T) {
++ fx := start(t, "")
++ fx.waitState(StateReady)
++ hello := fx.lines()[0]
++ if !strings.Contains(hello, `"op":"hello"`) || !strings.Contains(hello, `"shared_clock":"`+sharedClockName+`"`) {
++ t.Fatalf("hello: %s", hello)
++ }
++}
++
++// Test A and B: 200 ms in the pipe and reader. A command with 100 ms left never reaches handle(); one with a second
++// left is handled with at most what remains of its original second.
++func TestPipeDelayNeverExtendsADeadline(t *testing.T) {
++ fx := startIPC(t, false, "IPC_DELAY_MS=200")
++ late := fx.drive("late", 100*time.Millisecond)
++ if late.FaultCode != protocol.FaultExpired || !strings.Contains(late.Message, "before the runtime read it") {
++ t.Fatalf("late: %+v", late)
++ }
++ start := time.Now()
++ ontime := fx.drive("ontime", 1000*time.Millisecond)
++ if !ontime.OK {
++ t.Fatalf("ontime: %+v", ontime)
++ }
++ var got struct {
++ DeadlineMS int `json:"deadline_ms"`
++ }
++ _ = json.Unmarshal(ontime.Data, &got)
++ // It waited at least the 200 ms delay: at most 800 ms can be left (plus nothing: the clocks only round down).
++ if got.DeadlineMS < 1 || got.DeadlineMS > 800 {
++ t.Fatalf("ontime was handled with deadline_ms %d after %v; at most 800 can be left", got.DeadlineMS, time.Since(start))
++ }
++ if ids := fx.handledIDs(); len(ids) != 1 || ids[0] != "ontime" {
++ t.Fatalf("handle() saw %v; the expired command must never reach it", ids)
++ }
++ t.Logf("%s: late refused (%s), ontime handled with %d ms of its 1000", runtime.GOOS, late.Message, got.DeadlineMS)
++}
++
++// Test D: the core's writer queue (a full pipe while the plugin is not reading) and then the pipe delay, together.
++func TestWriterQueueThenPipeDelay(t *testing.T) {
++ fx := startIPC(t, false, "IPC_DELAY_MS=200", "HOLD_MS=300")
++ ctx := context.Background()
++ held := mono.Now() // the plugin starts its 300 ms hold no earlier than this
++ hold := fx.p.Begin(ctx, "hold", "hold", nil, 3000)
++ filler, _ := json.Marshal(map[string]any{"text": strings.Repeat("x", 1<<20)})
++ fill := fx.p.Begin(ctx, "fill", "say", filler, 5000)
++ queued := mono.Now()
++ late := fx.p.BeginWith(ctx, "late", protocol.KindDrive, nil, queued.Add(250*time.Millisecond), nil, 0, nil)
++ combo := fx.p.BeginWith(ctx, "combo", protocol.KindDrive, nil, queued.Add(1500*time.Millisecond), nil, 0, nil)
++ if r := late(); r.FaultCode != protocol.FaultExpired {
++ t.Fatalf("late: %+v", r)
++ }
++ r := combo()
++ hold()
++ fill()
++ if !r.OK {
++ t.Fatalf("combo: %+v", r)
++ }
++ var got struct {
++ DeadlineMS int `json:"deadline_ms"`
++ }
++ _ = json.Unmarshal(r.Data, &got)
++ // The pipe was held for 300 ms from when the plugin read "hold" (after held), then combo's line waited 200 ms more
++ // before the runtime got it (late was never written: its 250 ms ran out in the writer queue). So at most
++ // (queued + 1500 ms) − (held + 500 ms) can be left: about 1000 ms, plus the few ms between held and queued.
++ maxLeft := queued.Add(1500 * time.Millisecond).Sub(held.Add(500 * time.Millisecond))
++ if got.DeadlineMS < 1 || time.Duration(got.DeadlineMS)*time.Millisecond > maxLeft {
++ t.Fatalf("combo was handled with deadline_ms %d; at most %v can be left", got.DeadlineMS, maxLeft)
++ }
++ for _, id := range fx.handledIDs() {
++ if id == "late" {
++ t.Fatal("the expired command reached handle()")
++ }
++ }
++ t.Logf("%s: writer queue then pipe: late refused, combo handled with %d ms of its 1500 (at most %v could be left)", runtime.GOOS, got.DeadlineMS, maxLeft)
++}
++
++// Test C: strict timing, when the shared clock is missing (on the core's side, or the runtime's read fails): timed
++// motion is refused, never run on deadline_ms alone; a plugin that does not ask for it keeps running as before.
++func TestStrictTimingRefusesWithoutTheClock(t *testing.T) {
++ real := monotonicNS
++ t.Cleanup(func() { monotonicNS = real })
++
++ // The runtime's read fails: the runtime refuses, before handle().
++ fx := startIPC(t, false, "STRICT=1", "BREAK_CLOCK=1")
++ r := fx.drive("broken", time.Second)
++ if r.FaultCode != protocol.FaultTimingUnavailable || !strings.Contains(r.Message, "could not be read") {
++ t.Fatalf("broken clock: %+v", r)
++ }
++ if ids := fx.handledIDs(); len(ids) != 0 {
++ t.Fatalf("handle() saw %v", ids)
++ }
++
++ // The core has no shared clock: a strict plugin refuses the line that comes without an expiry…
++ monotonicNS = func() (int64, bool) { return 0, false }
++ fx2 := startIPC(t, false, "STRICT=1")
++ if r := fx2.drive("bare", time.Second); r.FaultCode != protocol.FaultTimingUnavailable {
++ t.Fatalf("no clock, strict plugin: %+v", r)
++ }
++ // …and a core configured for strict timing does not write it at all.
++ fx3 := startIPC(t, true)
++ if r := fx3.drive("unsent", time.Second); r.FaultCode != protocol.FaultTimingUnavailable || !strings.Contains(r.Message, "was not sent") {
++ t.Fatalf("no clock, strict core: %+v", r)
++ }
++ if ids := append(fx2.handledIDs(), fx3.handledIDs()...); len(ids) != 0 {
++ t.Fatalf("handle() saw %v", ids)
++ }
++ // A plugin that does not ask for strict timing still runs on deadline_ms where there is no shared clock.
++ fx4 := startIPC(t, false)
++ if r := fx4.drive("compat", time.Second); !r.OK {
++ t.Fatalf("not strict, no clock: %+v", r)
++ }
++}
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex ac6cdd4..c43ed98 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -260,6 +260,9 @@ const (
FaultNotConnected = "not_connected" // a bridge driver cannot reach its robot
FaultCliff = "cliff" // refused because the robot is at a cliff edge
FaultShuttingDown = "shutting_down"
++ // Strict timing (0.30): the driver requires a motion command's original expiry on the clock it shares with the
++ // core, and there is none to carry it. Refused, never run on deadline_ms alone.
++ FaultTimingUnavailable = "timing_unavailable"
)
// DefaultDeadlineMS is used when a motion command carries no deadline.
ddiff --git a/plugins/sdk/openvibe_plugin/__init__.py b/plugins/sdk/openvibe_plugin/__init__.py
iindex d4d75a6..2f01059 100644
---- a/plugins/sdk/openvibe_plugin/__init__.py
++++ b/plugins/sdk/openvibe_plugin/__init__.py
@@@ -21,10 +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", "SHARED_CLOCK", "__version__"]
++ "RUNTIME_CAUSES", "SHARED_CLOCK", "SHARED_CLOCK_NAME", "TIMING_UNAVAILABLE", "__version__"]
# The Node release this runtime ships with (pyproject.toml carries the same).
--__version__ = "0.29.0"
++__version__ = "0.30.0"
PROTOCOL_VERSION = 1
HEARTBEAT_TIMEOUT_S = 1.0
DEFAULT_DEADLINE_MS = 300
@@@ -42,10 +42,54 @@ 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
++def _qpc_ns(ticks, freq):
++ """A QueryPerformanceCounter reading in nanoseconds, rounded down: whole seconds and the remainder apart, exactly as
++ Node computes it (internal/plugins/clock.go qpcNS), so the two values compare exactly."""
++ return (ticks // freq) * 1000000000 + (ticks % freq) * 1000000000 // freq
++
++
++def _qpc_clock():
++ """A reader of the raw QueryPerformanceCounter in ns (Windows), or None. Raw, through kernel32: Python's own
++ time.monotonic is GetTickCount64 before 3.13 and perf_counter's reference point is not specified, so neither is
++ relied on to match Node."""
++ try:
++ import ctypes
++ k32 = ctypes.WinDLL("kernel32")
++ f = ctypes.c_int64()
++ if not k32.QueryPerformanceFrequency(ctypes.byref(f)) or f.value <= 0:
++ return None
++ freq = f.value
++ except (AttributeError, OSError, ImportError):
++ return None
++
++ def read():
++ t = ctypes.c_int64()
++ if not k32.QueryPerformanceCounter(ctypes.byref(t)):
++ raise OSError("QueryPerformanceCounter failed")
++ return _qpc_ns(t.value, freq)
++ return read
++
++
++def _shared_clock():
++ """(name, reader) of the clock this process shares with the core, as hello names it (Node 0.30), or (None, None).
++ CLOCK_MONOTONIC on Unix; the raw QueryPerformanceCounter on Windows: 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)."""
++ if sys.platform == "win32":
++ read = _qpc_clock()
++ return ("qpc", read) if read else (None, None)
++ if hasattr(time, "clock_gettime_ns") and hasattr(time, "CLOCK_MONOTONIC"):
++ return "clock_monotonic", _monotonic_ns
++ return None, None
++
++
++SHARED_CLOCK_NAME, SHARED_CLOCK = _shared_clock()
++
++# The fault a strict-timing runtime answers a motion command with when its original expiry cannot be checked.
++TIMING_UNAVAILABLE = "timing_unavailable"
++
++
++def _truthy(v):
++ return str(v or "").strip().lower() in ("1", "true", "yes", "on")
class Fault(Exception):
@@@ -207,6 +251,11 @@ class Plugin:
# Seconds between attempts to set up again after setup raised (the hardware or the robot's computer was not up
# yet). None: a failed setup stays failed until the core restarts the plugin.
retry_setup_s = None
++ # Strict timing (Node 0.30): a motion command whose original expiry cannot be checked on the clock shared with the
++ # core (none on this machine, a failed read, a different clock, or a core that sent none) is refused before handle()
++ # with timing_unavailable, never run on deadline_ms alone. The deployment can also require it: OPENVIBE_STRICT_TIMING=1
++ # in the plugin's environment, or strict_timing in the Node's plugin configuration (hello).
++ strict_timing = False
runtime = None # set by Runtime
@@@ -312,7 +361,7 @@ 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, shared_clock=SHARED_CLOCK):
++ heartbeat_timeout_s=HEARTBEAT_TIMEOUT_S, shared_clock=SHARED_CLOCK, shared_clock_name=SHARED_CLOCK_NAME):
self.plugin = plugin
plugin.runtime = self
self.infile = infile if infile is not None else sys.stdin
@@@ -320,6 +369,11 @@ class Runtime:
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.shared_clock_name = shared_clock_name if shared_clock is not None else None
++ # The clock the core named in hello (Node 0.30), and whether it is ours: expires_at_mono_ns is trusted only then.
++ self.core_clock = None
++ self.clock_agrees = True
++ self.strict_timing = bool(getattr(plugin, "strict_timing", False)) or _truthy(os.environ.get("OPENVIBE_STRICT_TIMING"))
self.heartbeat_timeout_s = heartbeat_timeout_s
self.lock = threading.RLock()
self.out_lock = threading.Lock()
@@@ -403,6 +457,18 @@ class Runtime:
self.send(dict(op="telemetry", **t))
# Input.
++ def _timing_gap(self, msg, shared_at):
++ """Why a motion command's original expiry cannot be checked, in words."""
++ if self.shared_clock is None:
++ return "this machine has no clock this runtime shares with the core"
++ if not self.clock_agrees:
++ return "the core's clock (%s) is not this runtime's (%s)" % (self.core_clock, self.shared_clock_name)
++ if not isinstance(msg.get("expires_at_mono_ns"), int) or isinstance(msg.get("expires_at_mono_ns"), bool):
++ return "the core sent no expiry on the shared clock (expires_at_mono_ns)"
++ if shared_at is None:
++ return "the shared clock could not be read when the command arrived"
++ return "the command's expiry could not be checked"
++
def shared_now(self):
"""The shared clock now (ns), or None."""
if self.shared_clock is None:
@@@ -481,6 +547,13 @@ class Runtime:
def on_hello(self, msg):
self.config = msg.get("config") or {}
self.probe = bool(msg.get("probe"))
++ # The clock the core put expires_at_mono_ns on. A different one than ours (or ours missing) is never compared
++ # with: the field then counts as absent, and strict timing refuses motion.
++ core_clock = msg.get("shared_clock")
++ self.core_clock = core_clock if isinstance(core_clock, str) else None
++ self.clock_agrees = self.core_clock is None or self.core_clock == self.shared_clock_name
++ if msg.get("strict_timing") is True:
++ self.strict_timing = True
# The newest halt epoch the core has seen (Node v0.17.0): a restarted process keeps refusing older turns.
epoch = _epoch(msg.get("halt_epoch"))
if epoch and epoch > self.halt_epoch:
@@@ -490,9 +563,12 @@ class Runtime:
except Exception as e:
caps = {}
self.plugin.log("describe failed: %r", e)
-- self.send({"op": "describe", "driver": self.plugin.driver, "version": self.plugin.version,
-- "protocol": PROTOCOL_VERSION, "capabilities": caps,
-- "motion_kinds": sorted(self.plugin.motion_kinds)})
++ describe = {"op": "describe", "driver": self.plugin.driver, "version": self.plugin.version,
++ "protocol": PROTOCOL_VERSION, "capabilities": caps, "motion_kinds": sorted(self.plugin.motion_kinds)}
++ if self.strict_timing:
++ # The core refuses too, before writing: a motion command never leaves it without its original expiry.
++ describe["timing"] = "strict"
++ self.send(describe)
if self.probe:
return
self._setup()
@@@ -531,7 +607,8 @@ class Runtime:
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
++ # Only on the clock the core named (hello, Node 0.30): another clock's instant is never compared with ours.
++ left = _left_ms(msg.get("expires_at_mono_ns"), shared_at if self.clock_agrees else None) 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"),
@@@ -554,6 +631,11 @@ 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"]
++ # Strict timing: a motion command whose original expiry cannot be checked here never reaches the driver.
++ if moves and self.strict_timing and left is None:
++ self.send({"op": "nack", "id": cid, "fault_code": TIMING_UNAVAILABLE,
++ "message": "strict timing: " + self._timing_gap(msg, shared_at)})
++ return
# 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",
ddiff --git a/plugins/sdk/pyproject.toml b/plugins/sdk/pyproject.toml
iindex 3c15098..93934e0 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.29.0"
++version = "0.30.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 cfd2c28..5efc016 100644
---- a/plugins/sdk/tests/test_runtime.py
++++ b/plugins/sdk/tests/test_runtime.py
@@@ -439,7 +439,7 @@ class RuntimeTest(unittest.TestCase):
self.assertEqual(rt.stop_infos[-1].halt_epoch, 1792900000500)
def test_the_package_reports_the_node_release(self):
-- self.assertEqual(openvibe_plugin.__version__, "0.29.0")
++ self.assertEqual(openvibe_plugin.__version__, "0.30.0")
with open(os.path.join(os.path.dirname(HERE), "pyproject.toml")) as f:
self.assertIn('version = "%s"' % openvibe_plugin.__version__, f.read())
ddiff --git a/plugins/sdk/tests/test_strict_timing.py b/plugins/sdk/tests/test_strict_timing.py
nnew file mode 100644
iindex 0000000..2359701
---- /dev/null
++++ b/plugins/sdk/tests/test_strict_timing.py
@@@ -0,0 +1,182 @@
++"""Node 0.30 (Maticus, 2026-10-27, Windows 10): the clock shared with the core on Windows, the hello check that both read
++the same clock, and strict timing, which refuses a motion command before handle() whenever its original expiry cannot be
++checked, instead of running it on deadline_ms alone."""
++import io
++import json
++import os
++import sys
++import unittest
++
++sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
++import openvibe_plugin # noqa: E402
++from openvibe_plugin import Plugin, Runtime, TIMING_UNAVAILABLE, _qpc_ns # noqa: E402
++
++from test_runtime import FakeClock, FakeShared, Recorder, lines # noqa: E402
++
++# The same vectors as internal/plugins/clock_test.go TestQPCNSMatchesThePythonRuntime: both sides must give these.
++QPC_VECTORS = [
++ (0, 10_000_000, 0),
++ (9_999_999, 10_000_000, 999_999_900),
++ (30_000_005, 10_000_000, 3_000_000_500),
++ (3_579_545 * 3600 + 1, 3_579_545, 3_600_000_000_279),
++ (31_536_000_000_000_007, 10_000_000, 3_153_600_000_000_000_700), # a hundred years of uptime at 10 MHz
++ (2_441_406 * 86_400 + 2_441_405, 2_441_406, 86_400_999_999_590),
++]
++
++
++class Strict(Recorder):
++ strict_timing = True
++
++
++def make(plugin=None, shared=None, name="clock_monotonic"):
++ out = io.StringIO()
++ clock = FakeClock()
++ p = plugin or Recorder()
++ rt = Runtime(p, infile=io.StringIO(""), outfile=out, clock=clock, shared_clock=shared, shared_clock_name=name)
++ return rt, p, out, clock
++
++
++def drive(cid, shared_ns=None, left_ns=None, deadline_ms=1000):
++ m = {"op": "command", "id": cid, "kind": "drive", "deadline_ms": deadline_ms}
++ if shared_ns is not None:
++ m["expires_at_mono_ns"] = shared_ns + left_ns
++ return json.dumps(m)
++
++
++class QPCTest(unittest.TestCase):
++ def test_conversion_is_nodes_exactly(self):
++ for ticks, freq, ns in QPC_VECTORS:
++ self.assertEqual(_qpc_ns(ticks, freq), ns, (ticks, freq))
++
++ def test_windows_shares_the_raw_counter(self):
++ # On Windows the shared clock is the raw QueryPerformanceCounter (never time.monotonic, which is GetTickCount64
++ # before Python 3.13); elsewhere CLOCK_MONOTONIC.
++ if sys.platform == "win32":
++ self.assertEqual(openvibe_plugin.SHARED_CLOCK_NAME, "qpc")
++ a = openvibe_plugin.SHARED_CLOCK()
++ b = openvibe_plugin.SHARED_CLOCK()
++ self.assertTrue(0 < a <= b)
++ elif hasattr(__import__("time"), "CLOCK_MONOTONIC"):
++ self.assertEqual(openvibe_plugin.SHARED_CLOCK_NAME, "clock_monotonic")
++
++
++class StrictTimingTest(unittest.TestCase):
++ def ready(self, rt, **hello):
++ rt.feed_line(json.dumps({"op": "hello", "config": {}, **hello}))
++ rt.feed_line(json.dumps({"op": "heartbeat", "t": 1}))
++
++ def test_strict_needs_the_original_expiry(self):
++ shared = FakeShared()
++ rt, p, out, clock = make(Strict(), shared)
++ self.ready(rt, shared_clock="clock_monotonic")
++ describe = next(m for m in lines(out) if m["op"] == "describe")
++ self.assertEqual(describe.get("timing"), "strict", "the core refuses too, before writing")
++ # With its expiry on the shared clock: handled, with what is left.
++ rt.feed_line(drive("ok", shared.ns, 500_000_000), clock.t, shared.ns)
++ self.assertEqual(lines(out)[-1], {"op": "ack", "id": "ok"})
++ self.assertEqual(p.handled[-1].deadline_ms, 500)
++ # Without one (an older core, or a core with no shared clock): refused before handle(), never on deadline_ms.
++ rt.feed_line(drive("bare"), clock.t, shared.ns)
++ last = lines(out)[-1]
++ self.assertEqual((last["op"], last["id"], last["fault_code"]), ("nack", "bare", TIMING_UNAVAILABLE))
++ self.assertIn("no expiry on the shared clock", last["message"])
++ self.assertEqual([c.id for c in p.handled], ["ok"])
++
++ def test_strict_with_no_clock_here_refuses(self):
++ # Missing clock: this machine has none the runtime can read.
++ rt, p, out, clock = make(Strict(), shared=None)
++ self.ready(rt)
++ rt.feed_line(drive("m", 1_000, 500_000_000))
++ last = lines(out)[-1]
++ self.assertEqual(last["fault_code"], TIMING_UNAVAILABLE)
++ self.assertIn("no clock this runtime shares", last["message"])
++ self.assertEqual(p.handled, [])
++
++ def test_strict_with_a_failed_read_refuses(self):
++ # Failed clock: the read raised when the line arrived (shared_now answers None).
++ def broken():
++ raise OSError("QueryPerformanceCounter failed")
++ rt, p, out, clock = make(Strict(), shared=broken)
++ self.ready(rt)
++ rt.feed_line(drive("m", 1_000, 500_000_000))
++ last = lines(out)[-1]
++ self.assertEqual(last["fault_code"], TIMING_UNAVAILABLE)
++ self.assertIn("could not be read", last["message"])
++ self.assertEqual(p.handled, [])
++
++ def test_a_different_clock_is_never_compared(self):
++ # The core says its expiry is on QueryPerformanceCounter, this runtime reads CLOCK_MONOTONIC: the field counts
++ # as absent (strict refuses; a non-strict plugin runs on deadline_ms, as before Node 0.18).
++ shared = FakeShared()
++ rt, p, out, clock = make(Strict(), shared)
++ self.ready(rt, shared_clock="qpc")
++ rt.feed_line(drive("m", shared.ns, 500_000_000), clock.t, shared.ns)
++ last = lines(out)[-1]
++ self.assertEqual(last["fault_code"], TIMING_UNAVAILABLE)
++ self.assertIn("is not this runtime's", last["message"])
++ rt2, p2, out2, clock2 = make(Recorder(), shared)
++ self.ready(rt2, shared_clock="qpc")
++ rt2.feed_line(drive("m", shared.ns, -500_000_000, deadline_ms=300), clock2.t, shared.ns)
++ self.assertEqual(lines(out2)[-1], {"op": "ack", "id": "m"}, "an instant on another clock is not read as expired")
++ self.assertEqual(p2.handled[-1].deadline_ms, 300)
++
++ def test_strict_from_the_deployment(self):
++ shared = FakeShared()
++ # The Node's plugin configuration (strict_timing in hello)…
++ rt, p, out, clock = make(Recorder(), shared)
++ self.ready(rt, strict_timing=True)
++ rt.feed_line(drive("bare"), clock.t, shared.ns)
++ self.assertEqual(lines(out)[-1]["fault_code"], TIMING_UNAVAILABLE)
++ # …or the plugin's environment.
++ os.environ["OPENVIBE_STRICT_TIMING"] = "1"
++ try:
++ rt2, p2, out2, clock2 = make(Recorder(), shared)
++ finally:
++ del os.environ["OPENVIBE_STRICT_TIMING"]
++ self.ready(rt2)
++ rt2.feed_line(drive("bare"), clock2.t, shared.ns)
++ self.assertEqual(lines(out2)[-1]["fault_code"], TIMING_UNAVAILABLE)
++ self.assertEqual(p.handled + p2.handled, [])
++
++ def test_not_strict_keeps_working_without_a_clock(self):
++ # Compatibility: a plugin that does not ask for strict timing still runs on deadline_ms where there is no
++ # shared clock (an older core, a machine without one).
++ rt, p, out, clock = make(Recorder(), shared=None)
++ self.ready(rt)
++ rt.feed_line(drive("m", deadline_ms=250))
++ self.assertEqual(lines(out)[-1], {"op": "ack", "id": "m"})
++ self.assertEqual(p.handled[-1].deadline_ms, 250)
++
++ def test_boundaries(self):
++ shared = FakeShared()
++ rt, p, out, clock = make(Strict(), shared)
++ self.ready(rt, shared_clock="clock_monotonic")
++ # Exactly at the deadline, and less than a millisecond before it: expired, never handled.
++ for cid, left in (("at", 0), ("under", 999_999), ("past", -1)):
++ rt.feed_line(drive(cid, shared.ns, left), clock.t, shared.ns)
++ last = lines(out)[-1]
++ self.assertEqual((last["id"], last["fault_code"]), (cid, "expired"), cid)
++ # One millisecond left: handled, with exactly that.
++ rt.feed_line(drive("one", shared.ns, 1_000_000), clock.t, shared.ns)
++ self.assertEqual(lines(out)[-1], {"op": "ack", "id": "one"})
++ self.assertEqual(p.handled[-1].deadline_ms, 1)
++ # Never more than deadline_ms, whatever the expiry says: the core's count-down is a cap too.
++ rt.feed_line(drive("cap", shared.ns, 900_000_000, deadline_ms=400), clock.t, shared.ns)
++ self.assertEqual(p.handled[-1].deadline_ms, 400)
++ self.assertEqual([c.id for c in p.handled], ["one", "cap"])
++
++ def test_a_restart_needs_the_clock_again(self):
++ # A restarted process starts over: a new hello names the clock again, and a hello naming another clock (or none
++ # from an older core, then no expiry) is held to the same rule.
++ shared = FakeShared()
++ rt, p, out, clock = make(Strict(), shared)
++ self.ready(rt, shared_clock="qpc")
++ rt.feed_line(drive("m", shared.ns, 500_000_000), clock.t, shared.ns)
++ self.assertEqual(lines(out)[-1]["fault_code"], TIMING_UNAVAILABLE)
++ rt.feed_line(json.dumps({"op": "hello", "config": {}, "shared_clock": "clock_monotonic"}))
++ rt.feed_line(drive("n", shared.ns, 500_000_000), clock.t, shared.ns)
++ self.assertEqual(lines(out)[-1], {"op": "ack", "id": "n"})
++
++
++if __name__ == "__main__":
++ unittest.main()