From 7edb9d07a07a1be3a17852cae2b0dae03260b558 Mon Sep 17 00:00:00 2001 From: Michael Johnson Date: Wed, 2 Sep 2026 15:49:00 +0100 Subject: [PATCH 1/3] Read a robot's live state over HTTP, so a client need not join the broker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit M3's split — reads over MQTT, writes over HTTP — is right for the dashboard, which wants a live stream, and wrong for a client that asks once and acts: an MCP front door or a script then has to track the topic tree, retention semantics and a broker credential, three contracts where one would do. The fleet server now holds its own subscription to mote/v2/+/{presence,health,pose,capabilities,mission/status} (BrokerFeed into RobotState) and answers GET /v1/robots/ with the registry row plus what it last saw. GET /v1/robots carries the presence payload per row, so picking an online robot costs one request rather than N. Payloads are forwarded, never rebuilt — no field added, renamed or reinterpreted, the rule the agent follows. Absent state is null per field and a robot never heard from is 200 with every field null, 404 being reserved for one that is not enrolled; broker_connected sits beside them because with the feed down every field is null however healthy the fleet is. Nothing is persisted: every topic read is retained, so a restart is repopulated by the broker within a second, and a cleared retained topic clears the field. mission_status is the last status, not a history — watch and dispatch keep the broker for the rest. The route takes an operator token where the roster does not, checked before the robot is looked up so an anonymous caller cannot enumerate ids by reading 401 against 404. fleetctl robots gains a presence column and `fleetctl robots ` prints the detail, both over HTTP rather than MQTT. Verified: 279 mote_fleet tests pass, including the new test_a_mission_can_be_followed_over_http_alone, which dispatches and follows a real mission to a terminal state against a real mosquitto with no MQTT client in the test; pre-commit passes across the tree. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01UKPz7azUramGmdoySHZh57 --- CLAUDE.md | 37 ++++ docs/fleet/README.md | 34 ++- docs/fleet/control-plane.md | 12 +- docs/fleet/fleet-api.md | 123 ++++++++++- mote_fleet/README.md | 4 +- mote_fleet/deploy/Dockerfile | 5 +- mote_fleet/server/fleet_server.py | 316 +++++++++++++++++++++++++-- mote_fleet/server/fleetctl.py | 82 ++++++- mote_fleet/test/api_harness.py | 39 +++- mote_fleet/test/test_e2e_fleet.py | 162 +++++++++++++- mote_fleet/test/test_fleet_server.py | 140 +++++++++++- 11 files changed, 905 insertions(+), 49 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 8880ce8..6b2030c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -114,6 +114,43 @@ Milestone M2 of `docs/design/fleet.md`; the operator flow is `docs/fleet/README. Milestone M3 of `docs/design/fleet.md`, and the end of v0. The **HTTP** wire is specified as its own versioned contract in **`docs/fleet/fleet-api.md`** (M1's MQTT one is `control-plane.md`); the operator flow is `docs/fleet/README.md` §6–9 and the measurements are `m3-verification.md`. **The two directions of the loop take different paths on purpose.** *Reads* ride MQTT: the browser subscribes to `mote/v2/+/{presence,health,pose,capabilities,mission/status}` over WebSockets, and because all of those are retained it has the whole fleet's state within a second of loading — no polling, no service in the middle. *Writes* ride HTTP: `POST /v1/robots//dispatch` authorizes an operator token (`fleetctl operator new --name `; the name is what the audit row records), writes the audit row, then publishes to the same `mission/command` topic. **The topic tree did not change — only who publishes to it**, and `fleetctl dispatch` moved to the API too, so there is one write path rather than one per client. The mission's `input` is validated only by the robot, against the schema its own capability declared: a copy in the server would be a second contract to keep in step, and it would refuse missions a newer robot understands. **The browser cannot publish**: `server/ui/mqtt.mjs` is a hand-rolled subscribe-only MQTT 3.1.1 client that implements no PUBLISH packet, so the split is enforced by omission (M7 makes it structural with a subscribe-only broker credential). The UI is static ES modules — no bundler, no npm, no vendored library — served by the same stdlib `http.server`; `map.mjs` holds the Q5 world→pixel transform (`px = (wx-origin_x)/res`, `py = height - (wy-origin_y)/res`) and a pan/zoom/follow canvas, and only draws robots on the *same* site+floor as the selected one because a pose from another floor is a different map frame. **Basemaps come from site bundles on the fleet box** (`--maps-dir`, default `$MOTE_FLEET_HOME/sites`, the layout `sites.py` writes, seeded by rsync until **M4** makes the registry canonical behind the same two routes). **M1's websockets blocker is settled**: `pixi run fleet-broker` runs `eclipse-mosquitto` under docker with the repo's own `mosquitto.conf`, because conda-forge's build has none; `pixi run -e fleet fleet-broker-local` is the conda binary for a box without docker, and it strips the WS stanza and says so. Two things that run in the same file (`test_ui.py` → `ui_test.mjs`) are the MQTT codec and the transform, tested under node against the very files the browser loads; `browser_check.mjs` drives a real headless Chrome over CDP against a running stack and is an operator's tool, not a CI test — `pixi run fleet-ui-check` is that stack in one command (broker on ephemeral ports, server, a temp `MOTE_FLEET_HOME`, the sim's `office_world` bundle as the basemap, and `test/fake_robots.py`, which publishes `protocol.py` and `spec/` payloads, imports `mote_tasks`' own capability set rather than writing one, and is *not* a second robot implementation), torn down afterwards; `-- --keep` leaves it up for UI work. It stays out of CI because it needs docker (conda's mosquitto still has no websockets) *and* a chrome, which the arm runner has not — the decision, and what wiring it in would take, are recorded in `m3-verification.md` §2 rather than left looking like coverage. **A fourth pane, `review`, is where a candidate map is looked at and promoted** (`server/ui/review.mjs`; routes and rationale under the map registry below). It is a *mode*, not a column: opening it stands the operations panes down at every width, because two canvases — one canonical with robots on it, one a candidate without — is the confusion a dedicated view exists to remove. **The phone is the realistic off-LAN client**, so below 760 px the panes become one at a time behind a bottom tab bar (`server/ui/layout.mjs`), selecting a robot in the roster navigates to the map — what the desktop layout gets for free by showing both — and the canvas gained pinch-to-zoom (`pinchSpan`/`pinchUpdate` in `map.mjs`, pure and tested, because a division by a zero span puts NaN in the view scale and blanks the map for good) plus a fingertip-sized hit target. The breakpoint is a **silent** seam — CSS decides what is displayed, JS decides when a selection navigates, and disagreement yields a tab bar over stacked panes rather than an error — so it lives in `layout.mjs` and `ui_test.mjs` reads the stylesheet and holds it there, as it does for every pane having a tab and for `touch-action: none` on the canvas (without which the browser eats the drag and the pinch before a single pointer event arrives). Dispatch's form is **generated from the robot's own capability set** (retained on the broker): a select of the keys it offers, one field per input property, and a **zone picker** exactly where a property's schema `$ref`s zone/v0's zone reference — so the page holds no list of capabilities and no list of which inputs are places, and a keyboard is needed only where the schema really wants free text. Three pre-existing bugs fell out, all of which a desk hides: `hidden` does not hide an element whose class sets `display` (the empty promote picker), the canvas backing store was resized on width alone so a height change left the previous frame's scale bar under the new one, and the scale bar was drawn in the dark theme's near-white on a white basemap — a canvas gets no cascade, so it now reads `--dim` off the element. Measurements, including `browser_check.mjs`'s phone pass, are `m3-verification.md` §9; **a real device is still the acceptance** — emulation gets the viewport and the touch points right and the thumb wrong. +## Fleet: reading a robot's state over HTTP + +M3's split — reads over MQTT, writes over HTTP — is right for the dashboard, +which wants a live stream, and wrong for a client that asks once and acts (an +MCP front door, `fleetctl`, a script): coupling that client to the broker makes +it track the topic tree, retention semantics and, after M7, a broker +credential — three contracts where one would do, which is what the abandoned +July MCP front door was rewritten under twice. So the fleet server now holds +**its own subscription** to the retained half of the tree (`BrokerFeed` into +`RobotState` in `fleet_server.py`, `mote/v2/+/{presence,health,pose, +capabilities,mission/status}`) and `GET /v1/robots/` answers with the +registry row plus what it last saw; `GET /v1/robots` carries the `presence` +payload per row, so picking an online robot costs one request rather than N. +Contract in `docs/fleet/fleet-api.md`, operator flow `README.md` §8. Five things +are load-bearing. **The payloads are forwarded, never rebuilt** — each is the +publisher's own document, no field added, renamed or reinterpreted, the rule +the agent follows so there is one definition of this wire and not a second one +here. **Absent state is null per field**, and a robot the server has never heard +from is `200` with every field null rather than `404`, which stays reserved for +a robot that is not enrolled — so `broker_connected` sits beside them, because +with the feed down every field is null however healthy the fleet is and null +alone cannot say which. **Nothing is persisted**: every topic read is retained, +so a restarted server is repopulated by the broker within a second and a stored +copy could only be the staler answer — and a cleared retained topic (a +zero-length payload) clears the field, or this server would assert a state the +broker has stopped serving. **`mission_status` is the last status, not a +history**, since one transition is all that is retained; anything wanting every +transition still subscribes, which is what `watch` and `dispatch` keep the +broker for. And **the route takes an operator token where the roster does not**, +because this is where the coordinates are — checked *before* the robot is looked +up, so an anonymous caller cannot enumerate ids by reading 401 against 404. +`publisher` and `feed` are injected as a pair in `serve()`, since a live +subscription beside a stubbed publisher would have a test dialling a broker it +does not have; the acceptance is `test_e2e_fleet.py`'s +`test_a_mission_can_be_followed_over_http_alone`, which dispatches and follows a +real mission to terminal with no MQTT client in the test at all. + ## Fleet: the map registry (M4) Milestone M4 of `docs/design/fleet.md`: the fleet server becomes the **canonical diff --git a/docs/fleet/README.md b/docs/fleet/README.md index 84b7f67..e6b6293 100644 --- a/docs/fleet/README.md +++ b/docs/fleet/README.md @@ -553,7 +553,8 @@ it logs and retries — enrolling later brings it up without a restart. pixi run -e fleet fleetctl -- operator new --name michael export MOTE_FLEET_TOKEN= -pixi run fleetctl -- robots # the registry roster +pixi run fleetctl -- robots # the roster: who is enrolled, who is online +pixi run fleetctl -- robots mote-01 # one robot: health, pose, capabilities, last mission pixi run fleetctl -- watch # live: presence, health, pose, status pixi run fleetctl -- dispatch mote-01 goto target=kitchen # send a mission pixi run fleetctl -- dispatch mote-01 fetch target=lab destination=kitchen @@ -630,6 +631,37 @@ on request. A robot that loses power is marked offline by the broker itself, within the keepalive, via its Last Will — not after somebody notices the heartbeats stopped. +**Asking once needs no broker.** `robots` and `robots ` read the fleet +server's own copy of those retained topics over HTTP and print it: + +```console +$ pixi run fleetctl -- robots +ID NAME SITE PRESENCE ENROLLED FINGERPRINT +mote-01 Scout home online 2026-07-26T18:41:02Z serial:d25bff05 +mote-02 Rover - unknown 2026-07-26T18:44:10Z serial:8f21ac90 + +$ pixi run fleetctl -- robots mote-01 +mote-01 Scout (home) + enrolled 2026-07-26T18:41:02Z serial:d25bff05 + presence online + health ok: 6 subsystems ok + pose x=1.5 y=-2.25 yaw=0.75 (home/ground) + can do goto, fetch + mission succeeded (goto, id 3e99cf44d1294ab5) +``` + +`unknown` is a third state, not a synonym for offline: a robot that is switched +off publishes `online: false` through its Last Will and reads `offline`, while +`unknown` means nothing has ever been heard from it — or that the fleet server +is not connected to the broker, which both commands say outright when it is so. +The detail view needs the operator token; the roster does not. What it shows of +a mission is the **last** status, because one is all the broker retains; `watch` +and `dispatch` are what show every transition, and they keep the broker for +exactly that reason. The route is +[`fleet-api.md`](fleet-api.md#get-v1robotsrobot_id), and it is the same answer +an HTTP-only client — an MCP front door, a script — gets without speaking MQTT +at all. + --- ## 9. The dashboard diff --git a/docs/fleet/control-plane.md b/docs/fleet/control-plane.md index 05a8ab7..38941c0 100644 --- a/docs/fleet/control-plane.md +++ b/docs/fleet/control-plane.md @@ -426,14 +426,22 @@ supersedes M0's operator-set id: the server owns the id space. | Route | Purpose | |---|---| | `GET /healthz` | liveness, contract version, robot count | -| `GET /v1/robots` | the roster | -| `GET /v1/robots/` | one row | +| `GET /v1/robots` | the roster, each row with its `presence` payload | +| `GET /v1/robots/` | one row, plus the retained payloads below as the server last saw them | | `POST /v1/enroll` | allocate (or return) a robot id | The map registry's routes — upload a candidate, pull a revision, promote one — are the same server and are specified in [`fleet-api.md`](fleet-api.md), with the retained `current` topic above as their only MQTT half. +**A client that reads these topics need not subscribe to them.** The fleet +server holds its own subscription to `presence`, `health`, `pose`, +`capabilities` and `mission/status` and forwards what it last saw over +`GET /v1/robots/` — the same payloads, unchanged, for the client that +asks once and acts rather than following the stream (specified in +[`fleet-api.md`](fleet-api.md#get-v1robotsrobot_id)). Subscribing is still what +a dashboard does, and the only way to see every transition. + ### `POST /v1/enroll` ```json diff --git a/docs/fleet/fleet-api.md b/docs/fleet/fleet-api.md index 1dc51fa..2478d15 100644 --- a/docs/fleet/fleet-api.md +++ b/docs/fleet/fleet-api.md @@ -22,6 +22,15 @@ sees the whole fleet's current state the moment it connects, with no polling and no service in the middle. The browser speaks the same protocol as everything else, over WebSockets. +**Except for the client that asks once.** A stream is the right shape for a +dashboard and the wrong one for a tool that reads a robot's state, acts, and +exits: it would have to speak MQTT, track the topic tree, know which topics are +retained, and hold a broker credential — three contracts to keep up with where +one would do. So this server subscribes on such a client's behalf and serves +what it last saw at [`GET /v1/robots/`](#get-v1robotsrobot_id). The +distinction is *how often*, not *what*: the payloads are the same documents, +forwarded, and anything wanting every transition still joins the broker. + **Writes ride HTTP.** A command has to be attributed to somebody, recorded, and refusable. Broker ACLs can express "may publish" but not "who did", so dispatch is a request to this API, which authorizes the operator, writes the audit row, @@ -47,7 +56,7 @@ Status codes are part of the contract: a client may switch on them. | Route | Credential | |---|---| | `POST /v1/enroll` | an **enrollment token** in the body (single-use by default) | -| `POST /v1/robots//dispatch`, `GET /v1/audit` | an **operator token** as `Authorization: Bearer ` | +| `POST /v1/robots//dispatch`, `GET /v1/audit`, `GET /v1/robots/` | an **operator token** as `Authorization: Bearer ` | | `POST …/revisions//promote` | an **operator token** | | `POST …/revisions/` (map upload) | none, but the `robot_id` must be enrolled — see [the registry](#the-map-registry-m4) | | everything else | none — see the security note below | @@ -67,13 +76,15 @@ is refused. Revocation keeps the row: who *had* access is part of the record. Bearer header only — never a query parameter, which would put the credential in every access log between here and the browser. -**Security posture, plainly.** The read routes are unauthenticated and the +**Security posture, plainly.** Most read routes are unauthenticated and the broker is anonymous, exactly as M1 left them. M3 adds a credential on the -*write* path and a record of who used it, which is the milestone's brief; it is -proportionate only while the tailnet is the boundary. M7 adds operator auth on -the read routes, per-robot broker credentials, and the Tailscale ACLs. Until -then, do not expose this port to a network the robots are not already trusted -on. +*write* path and a record of who used it, which is the milestone's brief; since +then two reads have been gated as well, the audit log and one robot's live +state, because both carry something an anonymous caller has no business with. +It is proportionate only while the tailnet is the boundary. M7 adds operator +auth on the rest of the read routes, per-robot broker credentials, and the +Tailscale ACLs. Until then, do not expose this port to a network the robots are +not already trusted on. --- @@ -82,8 +93,8 @@ on. ``` GET /healthz liveness, contract, robot count GET /v1/config what the browser needs to bootstrap -GET /v1/robots the roster -GET /v1/robots/ one row +GET /v1/robots the roster, each row with its presence +GET /v1/robots/ one row + its live state (an operator) POST /v1/enroll allocate (or return) a robot id POST /v1/robots//dispatch authorize, audit, publish a command GET /v1/audit[?limit=&robot_id=] what was dispatched, by whom @@ -110,6 +121,100 @@ GET / the operator UI (static files) [`control-plane.md`](control-plane.md#enrollment--registry-api) and unchanged by M3; the rest are below. +### `GET /v1/robots` + +Every enrolled robot, each row carrying the **`presence` payload** the server +last saw on that robot's retained topic. + +```json +{"schema":1,"broker_connected":true,"robots":[ + {"robot_id":"mote-01","name":"Scout","site":"home","fingerprint":"serial:aaa", + "facts":{},"enrolled_at":"2026-07-26T18:41:02Z","last_enrolled_at":"2026-07-26T18:41:02Z", + "presence":{"schema":1,"robot_id":"mote-01","online":true, + "stamp":"2026-09-02T09:14:03.221Z","version":"0.4.1"}}, + {"robot_id":"mote-02","name":"Rover","site":"","fingerprint":"serial:bbb", + "facts":{},"enrolled_at":"2026-07-26T18:44:10Z","last_enrolled_at":"2026-07-26T18:44:10Z", + "presence":null}]} +``` + +Presence and nothing more: a client picking a robot to dispatch to asks one +question — which of these is online — and it should not cost a request per +robot. Everything else about one robot is on that robot's own route. + +**`presence: null` means the server has heard nothing, not that the robot is +offline.** A robot that is off publishes `online: false` through its Last Will +and the payload is there, retained, saying so; null is the answer for a robot +that has never connected — or for a server that cannot hear, which is what +`broker_connected` distinguishes. + +### `GET /v1/robots/` + +One robot's registry row **and the retained state as this server last saw it**. + +```json +{"schema":1,"robot_id":"mote-01","name":"Scout","site":"home", + "fingerprint":"serial:aaa","facts":{},"enrolled_at":"2026-07-26T18:41:02Z", + "last_enrolled_at":"2026-07-26T18:41:02Z","broker_connected":true, + "presence":{"schema":1,"robot_id":"mote-01","online":true,"stamp":"…"}, + "health":{"schema":1,"robot_id":"mote-01","state":"ok","summary":"…","subsystems":[…]}, + "pose":{"schema":1,"robot_id":"mote-01","frame_id":"map","x":1.5,"y":-2.25, + "yaw":0.75,"site":"home","floor":"ground","stamp":"…"}, + "capabilities":{"schema":1,"platform_id":"mote-01","capabilities":[…]}, + "mission_status":{"schema":1,"id":"3e99cf44d1294ab5","platform_id":"mote-01", + "capability":"goto","state":"succeeded","terminal":true, + "source":"fleet","stamp":"…"}} +``` + +| Field | Source | +|---|---| +| the row | the registry: `robot_id`, `name`, `site`, `fingerprint`, `facts`, `enrolled_at`, `last_enrolled_at` | +| `broker_connected` | whether *this server's* subscription is live | +| `presence` · `health` · `pose` · `capabilities` | the retained payload on `mote/v2//`, or `null` | +| `mission_status` | the retained payload on `mote/v2//mission/status`, or `null` | + +| Status | Meaning | +|---|---| +| `200` | the row, with whatever state the server holds | +| `401` | missing, unknown, or revoked operator token | +| `404` | no such robot in the registry | + +**This route exists so a client can discover and follow a mission without +joining the broker.** M3's split — reads over MQTT, writes over HTTP — is right +for the dashboard, which wants a live stream. It is wrong for a client that asks +once and acts: coupling that client to the broker makes it track the topic tree, +retention semantics and a broker credential, which is three contracts where one +would do. An HTTP-only client depends on this document alone. + +**The payloads are forwarded, not rebuilt.** Each is the publisher's own +document — `presence`/`health`/`pose` per +[`control-plane.md`](control-plane.md), `capabilities` per capability/v0, +`mission_status` per mission/v0 — served with no field added, renamed or +reinterpreted, which is the rule the agent follows so that there is one +definition of these payloads and not a second one here. (They are re-serialised, +so key formatting is JSON's; the fields are the robot's.) + +**Absent state is `null`, per field.** A robot the server has never heard from +answers `200` with the row and every state field null — not `404`, which is +reserved for a robot that is not enrolled, and not an invented payload. What +`null` cannot say on its own is *why*, which is why `broker_connected` is +beside it: with the feed down, every field is null however healthy the fleet is. + +**`mission_status` is the last status, not a history.** One transition is all +that is retained — a status is a snapshot rather than a delta, so the latest one +is the whole truth about that mission — and a client that wants every transition +subscribes to `mission/status` the way the dashboard does. Polling this route +follows a mission perfectly well and will miss intermediate states on a fast +mission, which is a property of asking rather than of listening. + +**Nothing here is persisted.** Every topic it reads is retained, so a restarted +fleet server is repopulated by the broker within about a second of connecting; +a stored copy could only ever be the staler answer. + +An operator token because this is where the coordinates are: a pose says where +in a building a robot is and the mission status says what it was told to do +there. The token is checked *before* the robot is looked up, so an +unauthenticated caller cannot enumerate ids by reading 401 against 404. + ### `POST /v1/robots//dispatch` Send one mission to one robot. diff --git a/mote_fleet/README.md b/mote_fleet/README.md index 362a67a..7afaac6 100644 --- a/mote_fleet/README.md +++ b/mote_fleet/README.md @@ -59,10 +59,10 @@ pixi run -e fleet fleetctl -- dispatch mote-01 goto target=kitchen | Script | | |---|---| -| [`server/fleet_server.py`](server/fleet_server.py) | the fleet API: enrollment, roster, dispatch, audit, basemaps, and the UI — stdlib `http.server` | +| [`server/fleet_server.py`](server/fleet_server.py) | the fleet API: enrollment, roster, one robot's live state, dispatch, audit, basemaps, and the UI — stdlib `http.server` | | [`server/registry.py`](server/registry.py) | the SQLite row store: robots, enrollment tokens, operators, the audit log, transactional id allocation | | [`server/bundle_store.py`](server/bundle_store.py) | the map registry's byte store: candidate revisions, validation on the way in, the atomic flip that publishes one | -| [`server/fleetctl.py`](server/fleetctl.py) | operator CLI: tokens, roster, dispatch, audit, watch | +| [`server/fleetctl.py`](server/fleetctl.py) | operator CLI: tokens, roster, one robot's state, dispatch, audit, watch | | [`server/ui/`](server/ui/) | the dashboard: `index.html`, `app.mjs`, `map.mjs` (basemap + the Q5 transform, pan/pinch), `mqtt.mjs` (a subscribe-only MQTT client), `layout.mjs` (one pane at a time on a phone), `review.mjs` (see a candidate before promoting it), `zone_editor.mjs` (name its zones, saved as a new candidate), `theme.mjs` (the palette, for the canvases a stylesheet cannot reach) | | [`server/mosquitto.conf`](server/mosquitto.conf), [`broker.sh`](server/broker.sh) | the broker, its WebSocket listener, and where its state goes | | [`deploy/`](deploy/) | the deployed shape: an image for the API+UI, a compose file that runs it beside the broker, and `fleet-deploy.sh` (gated update, rollback, backup, restore) | diff --git a/mote_fleet/deploy/Dockerfile b/mote_fleet/deploy/Dockerfile index 7ca5166..8e5cca7 100644 --- a/mote_fleet/deploy/Dockerfile +++ b/mote_fleet/deploy/Dockerfile @@ -5,8 +5,9 @@ # through the prefix.dev channel (fleet.md, "The other pipelines"). It is a # container image and a compose file, versioned in this repo. # -# python:3.12-slim plus three pure-python wheels: paho for the mediated-dispatch -# path (BrokerLink), and PyYAML + Pillow for the map registry's validator, which +# python:3.12-slim plus three pure-python wheels: paho for both halves of the +# broker hop (BrokerLink's mediated dispatch, BrokerFeed's read of the fleet's +# retained state), and PyYAML + Pillow for the map registry's validator, which # reads bundle YAML and occupancy PNGs with the same libraries that write them. # No ROS and no checkout — that is what the design asks of this box (fleet.md # Q4), not a dependency count. Measured 152 MB, and it builds in seconds, so a diff --git a/mote_fleet/server/fleet_server.py b/mote_fleet/server/fleet_server.py index 9d5e548..c03c622 100644 --- a/mote_fleet/server/fleet_server.py +++ b/mote_fleet/server/fleet_server.py @@ -18,8 +18,8 @@ GET /healthz liveness + how many robots GET /v1/config what the browser needs to bootstrap - GET /v1/robots the roster - GET /v1/robots/ one row + GET /v1/robots the roster + each robot's presence + GET /v1/robots/ one row + its live state (operator) POST /v1/enroll allocate (or return) a robot id POST /v1/robots//dispatch authorize, audit, then publish a mission (capability + typed input) @@ -62,6 +62,17 @@ the mission that holds the lane — so a dispatcher decides what to do without reading prose. +**A robot's state is readable without joining the broker.** M3's split — reads +over MQTT, writes over HTTP — is right for the dashboard, which wants a live +stream, and wrong for a client that asks once and acts. So this server holds +its own subscription to the retained half of the topic tree (:class:`BrokerFeed` +into :class:`RobotState`) and answers ``GET /v1/robots/`` with what it last +saw: presence, health, pose, capabilities and the last mission status, each +forwarded as published or null. Such a client then depends on +``docs/fleet/fleet-api.md`` alone rather than on the topic tree, its retention +rules and a broker credential — three contracts to track instead of one, which +is what the abandoned MCP front door was rewritten under twice. + **The registry is the source of truth for maps (M4).** A robot uploads a saved revision as a *candidate*, which changes nothing; an operator promotes one, which flips the floor's ``map`` symlink and publishes the retained @@ -83,13 +94,15 @@ false for the one beside it; the name is true for both. The binding stays where it was, under ``/v1/maps``, served to the client that also has the basemap. -**Security posture for M3:** the read routes are still unauthenticated, exactly +**Security posture for M3:** most read routes are still unauthenticated, exactly as M1 left them, and the broker is still anonymous. What M3 adds is a credential -on the *write* path and a record of who used it. That stays proportionate only -while the tailnet is the boundary; M7 is where operator auth reaches the read -routes, per-robot broker credentials land, and Tailscale ACLs stop robots -reaching each other. Until then, do not expose this port to a network the robots -are not already trusted on. +on the *write* path and a record of who used it. Two reads take one as well — +the audit log, and one robot's live state — because both carry something an +anonymous caller has no business with. That stays proportionate only +while the tailnet is the boundary; M7 is where operator auth reaches the rest of +the read routes, per-robot broker credentials land, and Tailscale ACLs stop +robots reaching each other. Until then, do not expose this port to a network the +robots are not already trusted on. """ import argparse @@ -227,6 +240,181 @@ def close(self): pass +#: The leaves this server keeps the last retained payload of — everything the +#: dashboard subscribes to, so a client that asks once over HTTP is told what a +#: client following the stream would see. +STATE_LEAVES = ( + protocol.PRESENCE, + protocol.HEALTH, + protocol.POSE, + protocol.CAPABILITIES, + protocol.STATUS, +) + +#: `mission/status` is not a JSON key. A field name is its topic leaf with the +#: separator flattened, so the route's payload maps onto the topic tree without +#: a table to look either up in. +STATE_FIELDS = {leaf: leaf.replace("/", "_") for leaf in STATE_LEAVES} + + +class RobotState: + """The last retained payload this server saw on each robot's topics. + + This is the read half of the control plane held in memory, for the client + that asks once and acts rather than following a stream — an MCP front door, + ``fleetctl``, a script. Such a client should not have to speak MQTT, track + the topic tree, know which topics are retained and, after M7, hold a broker + credential, to answer *is this robot online and what is it doing*. + + Payloads are stored and served as published: this server subscribes to the + same topics the dashboard does and forwards what it was handed, never + adding, renaming or reinterpreting a field. That is the rule the agent + follows, for the same reason — there is one definition of these payloads, + in ``mission/v0`` and ``control-plane.md``, and a second reading of them + here would be a contract nobody declared. + + Nothing is persisted. Every topic it reads is retained, so a restarted + server is repopulated by the broker within a second of connecting, and a + persisted copy could only ever be the staler answer. + """ + + def __init__(self): + self._lock = threading.Lock() + self._robots: dict[str, dict] = {} + #: Whether the subscription is live. Without it a client cannot tell a + #: robot that has said nothing from a server that cannot hear. + self.connected = False + + def apply(self, topic: str, raw: bytes) -> bool: + """Take one message off the wire. True if it is one this server holds + state from — a robot's topic, one of the leaves above, a JSON object or + the empty payload that clears one.""" + parsed = protocol.parse_topic(topic) + if parsed is None: + return False + robot_id, leaf = parsed + if leaf not in STATE_LEAVES or not protocol.valid_id(robot_id): + return False + if not raw: + # A zero-length payload is how a publisher *clears* a retained + # topic. Keeping the last value would leave this server asserting a + # state the broker has stopped serving. + with self._lock: + self._robots.get(robot_id, {}).pop(leaf, None) + return True + try: + payload = json.loads(raw) + except ValueError: + return False + if not isinstance(payload, dict): + return False + with self._lock: + self._robots.setdefault(robot_id, {})[leaf] = payload + return True + + def of(self, robot_id: str) -> dict: + """Every state field for one robot, the ones never heard of null. + + Null is the honest answer and the only one available: a payload this + server has not been handed cannot be invented, and a robot that has + never connected differs from one that has only in that its topics are + empty. + """ + with self._lock: + held = dict(self._robots.get(robot_id) or {}) + return {field: held.get(leaf) for leaf, field in STATE_FIELDS.items()} + + def presence(self, robot_id: str): + """Just the presence payload — what the roster carries per row.""" + with self._lock: + return (self._robots.get(robot_id) or {}).get(protocol.PRESENCE) + + +class BrokerFeed: + """This server's subscription to the fleet's retained state. + + Separate from :class:`BrokerLink` because the two halves fail differently. + A publish that does not land must be reported to the operator who asked for + it, synchronously; a subscription that drops must reconnect and resubscribe + on its own, forever, with nobody asking. + + **The subscribe lives in ``on_connect``.** A broker restart hands the client + a clean session, so one that subscribed beside its connect comes back + subscribed to nothing: it stays connected and goes silent, which is + indistinguishable from a fleet with nothing to say. The agent and + ``fleetctl`` are arranged the same way, for the same reason. + + paho is imported on first use, which keeps this module's import surface the + standard library plus the wire contract. + """ + + def __init__( + self, + host: str, + port: int, + state: RobotState, + *, + keepalive: int = 30, + client_id: str = "mote-fleet-api-feed", + ): + self.host = host + self.port = int(port) + self.state = state + self.keepalive = keepalive + self.client_id = client_id + self._client = None + + def start(self) -> bool: + """Connect in the background. A broker that is down is not an error + here: the API must answer while the robots are unreachable, and say so + rather than wait.""" + try: + import paho.mqtt.client as mqtt + except ImportError as exc: + print(f"no MQTT client, robot state unavailable: {exc}", file=sys.stderr) + return False + try: + client = mqtt.Client( + mqtt.CallbackAPIVersion.VERSION2, client_id=self.client_id + ) + except AttributeError: # paho 1.x + client = mqtt.Client(client_id=self.client_id) + client.on_connect = self._on_connect + client.on_disconnect = self._on_disconnect + client.on_message = self._on_message + client.reconnect_delay_set(min_delay=1, max_delay=30) + self._client = client + try: + client.connect_async(self.host, self.port, keepalive=self.keepalive) + client.loop_start() + except (OSError, ValueError) as exc: + print(f"robot state feed not started: {exc}", file=sys.stderr) + self._client = None + return False + return True + + def _on_connect(self, client, _userdata, *_args): + for leaf in STATE_LEAVES: + client.subscribe(protocol.any_robot(leaf), qos=protocol.QOS) + self.state.connected = True + + def _on_disconnect(self, _client, _userdata, *_args): + self.state.connected = False + + def _on_message(self, _client, _userdata, message): + self.state.apply(message.topic, bytes(message.payload)) + + def close(self): + client, self._client = self._client, None + self.state.connected = False + if client is not None: + try: + client.loop_stop() + client.disconnect() + except Exception: + pass + + def _describe(capability: str, payload_input) -> str: """One line naming a mission, for the audit row and the operator's log. @@ -324,16 +512,9 @@ def do_GET(self): elif path == "/v1/config": self._send(200, self.server.ui_config()) elif path == "/v1/robots": - self._send( - 200, - {"schema": protocol.SCHEMA, "robots": self.server.registry.robots()}, - ) + self._roster() elif path.startswith("/v1/robots/"): - robot = self.server.registry.robot(path.rsplit("/", 1)[-1]) - if robot is None: - self._error(404, "no such robot") - else: - self._send(200, {"schema": protocol.SCHEMA, **robot}) + self._robot(path[len("/v1/robots/") :]) elif path == "/v1/audit": self._audit(params) elif path == "/v1/maps": @@ -437,6 +618,67 @@ def _enroll(self, body: dict): }, ) + # -- the roster and one robot's live state ---------------------------- + + def _roster(self): + """Every enrolled robot, each with its ``presence`` payload. + + Presence rather than the whole state: a client picking a robot to + dispatch to asks one question — which of these is online — and the + answer to it should not cost a request per robot. Everything else about + one robot is on that robot's own route. + """ + state = self.server.state + self._send( + 200, + { + "schema": protocol.SCHEMA, + "broker_connected": state.connected, + "robots": [ + {**row, "presence": state.presence(row["robot_id"])} + for row in self.server.registry.robots() + ], + }, + ) + + def _robot(self, rest: str): + """One robot: its registry row, and the retained state as last seen. + + This is the whole of what the dashboard's MQTT subscription gets, + answered over HTTP for the client that asks once and acts. What it is + not is a stream: ``mission_status`` is the *last* status, one + transition, and a caller that wants every transition subscribes to the + topic the way the browser does. + + Behind an operator token where the roster is not, because this is where + the coordinates are — a pose says where in a building the robot is, and + the mission status says what it was told to do there. The token is + checked before the robot is looked up, so an unauthenticated caller + cannot enumerate ids by reading 401 against 404. + """ + if "/" in rest: + self._error(404, f"no route /v1/robots/{rest}") + return + try: + self._operator() + except Unauthorized as exc: + self._error(401, str(exc)) + return + robot = self.server.registry.robot(rest) + if robot is None: + self._error(404, "no such robot") + return + state = self.server.state + self._send( + 200, + { + "schema": protocol.SCHEMA, + **robot, + "broker_connected": state.connected, + **state.of(rest), + }, + ) + # -- dispatch + audit ------------------------------------------------- def _dispatch(self, robot_id: str, body: dict): @@ -1006,6 +1248,23 @@ def __init__( self.broker_ws_host = broker_ws_host self.broker_ws_port = broker_ws_port self.foxglove_url = foxglove_url + #: What the fleet's robots last said, for the HTTP read routes. The + #: feed that fills it is attached by :func:`serve`, because it + #: subscribes on behalf of this object and so cannot exist before it. + self.state = RobotState() + self.feed = None + + def server_close(self): + """Give back the sockets, the subscription and the publisher together. + + The two broker connections outlive ``shutdown()`` — they are paho's own + threads, not this server's — so a test that starts a server per case + leaks a client per case unless they are closed here. + """ + super().server_close() + if self.feed is not None: + self.feed.close() + self.publisher.close() # -- what the browser needs to bootstrap ------------------------------ @@ -1196,15 +1455,22 @@ def serve( publish_port=None, id_prefix=ID_PREFIX, publisher=None, + feed=None, maps_dir=None, ui_dir=UI_DIR, broker_ws_host=None, broker_ws_port=9001, foxglove_url=FOXGLOVE_URL, ) -> FleetServer: - """Build a listening server. The caller runs it (or its ``serve_forever``).""" + """Build a listening server. The caller runs it (or its ``serve_forever``). + + ``publisher`` and ``feed`` are the two halves of the broker hop and are + injected as a pair: supplying a ``publisher`` is what a test does to stand + the broker down, and building a live subscription beside a stubbed + publisher would have the server dialling a broker the test does not have. + """ broker_host = broker_host or socket.gethostname() - return FleetServer( + server = FleetServer( (host, port), Registry(db), broker_host=broker_host, @@ -1218,6 +1484,16 @@ def serve( broker_ws_port=broker_ws_port, foxglove_url=foxglove_url, ) + if feed is None and publisher is None: + feed = BrokerFeed( + publish_host or broker_host, + publish_port or broker_port, + server.state, + ) + if feed is not None: + server.feed = feed + feed.start() + return server def main(argv=None): @@ -1314,7 +1590,7 @@ def main(argv=None): thread.join() except KeyboardInterrupt: server.shutdown() - server.publisher.close() + server.server_close() if __name__ == "__main__": diff --git a/mote_fleet/server/fleetctl.py b/mote_fleet/server/fleetctl.py index 15d6fa1..d528bd5 100644 --- a/mote_fleet/server/fleetctl.py +++ b/mote_fleet/server/fleetctl.py @@ -6,7 +6,9 @@ fleetctl token new mint an enrollment token fleetctl operator new --name michael mint an operator token - fleetctl robots the registry roster + fleetctl robots the roster, with who is online + fleetctl robots mote-01 one robot: presence, health, + pose, capabilities, last mission fleetctl dispatch mote-01 goto target=kitchen send a mission, follow it to terminal fleetctl audit who dispatched what @@ -23,6 +25,13 @@ directly from the broker, which is the cheap, live, no-service-in-the-middle read path the design asks for. +**One-shot reads go through the API.** ``robots`` and ``robots `` ask the +fleet server what it last saw on the retained topics rather than opening an MQTT +connection to find out, because a question with an answer does not need a +stream: no broker credential, no waiting to see whether a topic is going to +arrive, and an exit status. ``watch`` and ``dispatch`` keep the broker, which is +what following every transition actually requires. + The token for that lives in ``--token`` or ``$MOTE_FLEET_TOKEN``. ``token``/``operator`` talk to the registry file directly rather than over @@ -139,16 +148,73 @@ def cmd_token(args): def cmd_robots(args): - robots = _get(args.server, "/v1/robots").get("robots", []) + """The roster, or one robot's live state — both from the API, not the + broker. ``watch`` is what a live stream is for; this is the one-shot + answer, and it costs no MQTT connection to get.""" + if args.robot_id: + _print_robot(_get(args.server, f"/v1/robots/{args.robot_id}", _token(args))) + return + body = _get(args.server, "/v1/robots") + robots = body.get("robots", []) if not robots: print("no robots enrolled") return - print(f"{'ID':12} {'NAME':16} {'SITE':10} {'ENROLLED':21} FINGERPRINT") + if not body.get("broker_connected"): + # Otherwise every robot reads as offline and the reason is invisible. + print( + "the fleet server is not connected to the broker — presence is " + "unknown, not offline", + file=sys.stderr, + ) + print( + f"{'ID':12} {'NAME':16} {'SITE':10} {'PRESENCE':8} {'ENROLLED':21} FINGERPRINT" + ) for robot in robots: print( f"{robot['robot_id']:12} {robot['name'][:16]:16} " - f"{(robot['site'] or '-')[:10]:10} {robot['enrolled_at']:21} " - f"{robot['fingerprint']}" + f"{(robot['site'] or '-')[:10]:10} {_presence(robot.get('presence')):8} " + f"{robot['enrolled_at']:21} {robot['fingerprint']}" + ) + + +def _presence(payload) -> str: + """Three states, not two: a robot nobody has heard from is not offline.""" + if not payload: + return "unknown" + return "online" if payload.get("online") else "offline" + + +def _print_robot(body: dict): + print(f"{body['robot_id']} {body['name']} ({body['site'] or 'no site'})") + print(f" enrolled {body['enrolled_at']} {body['fingerprint']}") + print(f" presence {_presence(body.get('presence'))}") + health = body.get("health") + if health: + print(f" health {health['state']}: {health['summary']}") + pose = body.get("pose") + if pose: + print( + f" pose x={pose['x']} y={pose['y']} yaw={pose['yaw']} " + f"({pose.get('site')}/{pose.get('floor')})" + ) + capabilities = body.get("capabilities") + if capabilities: + keys = ", ".join(item["key"] for item in capabilities["capabilities"]) + print(f" can do {keys}") + status = body.get("mission_status") + if status: + # The *last* transition, not a history: one retained status is all the + # broker holds, and inventing the rest here would be a fiction. + line = f"{status['state']} ({status['capability']}, id {status['id']})" + failure = status.get("failure") + if failure: + retry = "retryable" if failure["recoverable"] else "not retryable" + line += f" [{failure['class']}, {retry}] {failure['detail']}" + print(f" mission {line}") + if not body.get("broker_connected"): + print( + " (the fleet server is not connected to the broker — every field " + "above is the last it saw, or nothing)" ) @@ -474,6 +540,12 @@ def main(argv=None): p_operator.set_defaults(func=cmd_operator) p_robots = sub.add_parser("robots", help="list enrolled robots") + p_robots.add_argument( + "robot_id", + nargs="?", + default="", + help="one robot's live state instead of the roster (needs an operator token)", + ) p_robots.set_defaults(func=cmd_robots) p_audit = sub.add_parser("audit", help="what was dispatched, by whom") diff --git a/mote_fleet/test/api_harness.py b/mote_fleet/test/api_harness.py index a362fea..d2a8c02 100644 --- a/mote_fleet/test/api_harness.py +++ b/mote_fleet/test/api_harness.py @@ -26,17 +26,29 @@ class FakeBroker: - """Stands in for ``BrokerLink``. ``fail`` is how "the broker is down" is - tested without taking a broker down.""" + """Stands in for the broker — both halves of it. + + ``publish`` is ``BrokerLink``'s, and ``fail`` is how "the broker is down" is + tested without taking a broker down. It is also a *loopback*: a retained + publish is handed to whatever has subscribed, which is how a test puts a + robot's presence or status in front of the server without a broker. Only + retained messages are delivered, because only retained messages are what a + subscriber connecting later would be given, and that is what the read + routes serve. + """ def __init__(self): self.published = [] self.fail = "" + self.subscribers = [] def publish(self, topic, payload, retain=False): if self.fail: return False, self.fail self.published.append((topic, json.loads(payload), retain)) + if retain: + for subscriber in self.subscribers: + subscriber(topic, payload) return True, "" def retained(self, topic): @@ -159,16 +171,24 @@ def start_server(tmp_path, **kwargs): """A listening fleet server with a stub broker. Caller shuts it down.""" maps = Path(tmp_path) / "sites" maps.mkdir(exist_ok=True) + publisher = kwargs.pop("publisher", None) or FakeBroker() httpd = serve( db=Path(tmp_path) / "registry.db", host="127.0.0.1", port=0, broker_host="fleet-box", broker_port=1883, - publisher=kwargs.pop("publisher", None) or FakeBroker(), + publisher=publisher, maps_dir=maps, **kwargs, ) + # The stub's retained messages go where a real subscription's would. The + # feed itself is not started (there is no broker to dial); what is under + # test here is the state the server keeps and the route that serves it, and + # the real MQTT hop is tested against a real mosquitto in the e2e pair. + if hasattr(publisher, "subscribers"): + publisher.subscribers.append(httpd.state.apply) + httpd.state.connected = True thread = threading.Thread(target=httpd.serve_forever, daemon=True) thread.start() httpd.url = f"http://127.0.0.1:{httpd.server_address[1]}" @@ -228,6 +248,19 @@ def post_bytes(server, path, blob, content_type="application/gzip"): return response.status, json.loads(response.read()) +def retain(server, robot_id, leaf, payload): + """Publish one retained payload on a robot's topic, through the broker. + + Not straight into the server's state: the route's promise is that what a + robot published is what comes back, so the test's payload has to travel the + way a robot's does — encoded, decoded, and never touched in between. + """ + server.publisher.publish( + protocol.topic(robot_id, leaf), protocol.encode(payload), retain=True + ) + return payload + + def enroll(server, fingerprint, **extra): token = extra.pop("token", None) or server.registry.new_token() return post( diff --git a/mote_fleet/test/test_e2e_fleet.py b/mote_fleet/test/test_e2e_fleet.py index c907363..8b0f788 100644 --- a/mote_fleet/test/test_e2e_fleet.py +++ b/mote_fleet/test/test_e2e_fleet.py @@ -338,6 +338,20 @@ def test_the_registry_survives_a_server_restart(tmp_path, broker): second.server_close() +def api_get(url, path, token=""): + import urllib.error + import urllib.request + + request = urllib.request.Request(url + path) + if token: + request.add_header("Authorization", f"Bearer {token}") + try: + with urllib.request.urlopen(request, timeout=15) as response: + return response.status, json.loads(response.read()) + except urllib.error.HTTPError as exc: + return exc.code, json.loads(exc.read()) + + def api_post(url, path, payload, token=""): import urllib.error import urllib.request @@ -445,7 +459,153 @@ def test_dispatch_through_the_fleet_api(tmp_path, monkeypatch, broker, fleet_api finally: agent.close() operator.close() - fleet_api.publisher.close() + executor.shutdown() + for node in (agent, tasks, nav): + node.destroy_node() + rclpy.shutdown() + + +def test_a_mission_can_be_followed_over_http_alone( + tmp_path, monkeypatch, broker, fleet_api +): + """The acceptance for the read route: discover a robot, dispatch to it, and + follow the mission to a terminal state with no MQTT client in this test. + + Everything else is as real as the tests above — a mosquitto, the fleet + server with its own subscription, the agent, and the ``mote_tasks`` tree — + and the only thing the test speaks is HTTP. That is the whole claim: a + client which asks once and acts (an MCP front door, a script) depends on + ``fleet-api.md`` and not additionally on the topic tree, its retention rules + and a broker credential. + """ + monkeypatch.setenv("MOTE_HOME", str(tmp_path / "mote")) + monkeypatch.setenv("ROS_DOMAIN_ID", str(random.randint(60, 100))) + + from mote_bringup import identity + + from mote_fleet import enroll + + enroll.main(["--server", fleet_api.url, "--token", fleet_api.registry.new_token()]) + robot_id = identity.robot_id() + token = fleet_api.registry.new_operator(name="mcp") + + rclpy.init(args=["--ros-args", "-r", f"__ns:=/test_{os.getpid()}"]) + zones_file = tmp_path / "zones.yaml" + zones_file.write_text(ZONES) + + from mote_fleet.agent import MoteAgent + from mote_tasks.task_server import TaskServer + + agent = MoteAgent( + parameter_overrides=[ + Parameter("health_period", value=0.5), + Parameter("pose_period", value=0.5), + Parameter("keepalive", value=2), + ] + ) + tasks = TaskServer( + parameter_overrides=[ + Parameter("zones_file", value=str(zones_file)), + Parameter("tick_period", value=0.05), + ] + ) + nav = MockNav() + executor = SingleThreadedExecutor() + for node in (agent, tasks, nav): + executor.add_node(node) + + def robot_state(): + code, body = api_get(fleet_api.url, f"/v1/robots/{robot_id}", token=token) + assert code == 200, body + return body + + def until(condition, timeout=60.0): + """Spin the robot's executor, then ask the API. The request is blocking + and must not be made from inside a spin callback.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + for _ in range(10): + executor.spin_once(timeout_sec=0.05) + state = robot_state() + if condition(state): + return state + return None + + try: + assert spin_until(executor, lambda: agent.connected), "agent never connected" + + # ---- discovery: which robots there are, and which are online ---- + state = until(lambda state: (state["presence"] or {}).get("online")) + assert state, "presence never reached the API" + code, roster = api_get(fleet_api.url, "/v1/robots") + assert code == 200, roster + assert roster["broker_connected"] is True + row = next(r for r in roster["robots"] if r["robot_id"] == robot_id) + assert row["presence"]["online"] is True + + # ---- and what this robot can be asked to do ---- + state = until(lambda state: state["capabilities"] is not None) + assert state, "the capability set never reached the API" + keys = [item["key"] for item in state["capabilities"]["capabilities"]] + assert keys == ["goto", "fetch"] + # Forwarded, not rebuilt: this is the robot's own document. + assert state["capabilities"]["platform_id"] == robot_id + + # ---- dispatch, and follow it to terminal through the same route ---- + code, answer = api_post( + fleet_api.url, + f"/v1/robots/{robot_id}/dispatch", + { + "schema": mission.SCHEMA, + "capability": "goto", + "input": {"target": "kitchen"}, + }, + token=token, + ) + assert code == 202, answer + + state = until( + lambda state: ( + (state["mission_status"] or {}).get("id") == answer["id"] + and state["mission_status"]["terminal"] + ) + ) + assert state, robot_state()["mission_status"] + assert state["mission_status"]["state"] == mission.SUCCEEDED + assert state["mission_status"]["source"] == mission.SOURCE_FLEET + assert len(nav.goals) == 1 + assert nav.goals[0].pose.position.x == pytest.approx(-1.5) # kitchen + + # ---- a refusal is readable here too, and it is typed: the class is + # what a client acts on, not a sentence ---- + code, refused = api_post( + fleet_api.url, + f"/v1/robots/{robot_id}/dispatch", + { + "schema": mission.SCHEMA, + "capability": "goto", + "input": {"target": "nowhere"}, + }, + token=token, + ) + assert code == 202, refused + state = until( + lambda state: ( + (state["mission_status"] or {}).get("id") == refused["id"] + and state["mission_status"]["terminal"] + ) + ) + assert state, robot_state()["mission_status"] + assert state["mission_status"]["state"] == mission.REJECTED + assert state["mission_status"]["failure"]["class"] == mission.UNRESOLVED_ZONE + assert len(nav.goals) == 1 + + # ---- health arrives on the same route ---- + state = until(lambda state: state["health"] is not None) + assert state, "health never reached the API" + assert state["health"]["robot_id"] == robot_id + finally: + agent.close() executor.shutdown() for node in (agent, tasks, nav): node.destroy_node() diff --git a/mote_fleet/test/test_fleet_server.py b/mote_fleet/test/test_fleet_server.py index 6d6f15a..1ab9165 100644 --- a/mote_fleet/test/test_fleet_server.py +++ b/mote_fleet/test/test_fleet_server.py @@ -19,6 +19,7 @@ get_bytes, post, post_raw, + retain, ) from mote_bringup.spec import mission @@ -74,20 +75,151 @@ def test_the_roster_lists_what_was_enrolled(server): assert [r["name"] for r in body["robots"]] == ["Scout", "Rover"] -def test_one_robot_can_be_fetched_by_id(server): +def test_one_robot_can_be_fetched_by_id(server, operator): enroll(server, "serial:aaa") - status, body = get(server, "/v1/robots/mote-01") + status, body = get(server, "/v1/robots/mote-01", token=operator) assert (status, body["robot_id"]) == (200, "mote-01") -def test_unknown_robot_is_404(server): - expect_error(lambda: get(server, "/v1/robots/nope"), 404) +def test_unknown_robot_is_404(server, operator): + expect_error(lambda: get(server, "/v1/robots/nope", token=operator), 404) def test_unknown_route_is_404(server): expect_error(lambda: get(server, "/v1/nothing"), 404) +# -- one robot's live state, over HTTP -------------------------------------- +# +# The route exists so a client can discover and follow a mission without +# joining the broker. What it must never do is invent: a field it has not been +# handed is null, and a field it has been handed comes back as published. + + +def presence_payload(robot_id="mote-01", online=True): + return protocol.presence(robot_id, online, version="test") + + +def status_payload(robot_id="mote-01", state=mission.ACCEPTED, mission_id="abc123"): + return mission.status(robot_id, mission_id, "goto", state) + + +def test_a_robot_nothing_has_been_heard_from_has_every_field_null(server, operator): + enroll(server, "serial:aaa", name="Scout") + status, body = get(server, "/v1/robots/mote-01", token=operator) + assert status == 200 + assert body["name"] == "Scout" + assert body["presence"] is None + assert body["health"] is None + assert body["pose"] is None + assert body["capabilities"] is None + assert body["mission_status"] is None + + +def test_retained_state_comes_back_verbatim(server, operator, robot): + presence = retain(server, robot, protocol.PRESENCE, presence_payload()) + pose = retain( + server, + robot, + protocol.POSE, + protocol.pose(robot, 1.5, -2.25, 0.75, site="home", floor="ground"), + ) + running = retain(server, robot, protocol.STATUS, status_payload()) + + status, body = get(server, f"/v1/robots/{robot}", token=operator) + assert status == 200 + # Not "looks like": the whole payload, field for field, as the robot + # published it. A server that reinterpreted one would be a second + # definition of a wire that already has one. + assert body["presence"] == presence + assert body["pose"] == pose + assert body["mission_status"] == running + assert body["health"] is None + + +def test_the_latest_payload_on_a_topic_is_the_one_served(server, operator, robot): + retain(server, robot, protocol.STATUS, status_payload(state=mission.ACCEPTED)) + finished = retain( + server, robot, protocol.STATUS, status_payload(state=mission.SUCCEEDED) + ) + _, body = get(server, f"/v1/robots/{robot}", token=operator) + assert body["mission_status"] == finished + assert body["mission_status"]["terminal"] is True + + +def test_state_is_kept_per_robot(server, operator): + enroll(server, "serial:aaa") + enroll(server, "serial:bbb") + retain(server, "mote-01", protocol.PRESENCE, presence_payload("mote-01", True)) + retain(server, "mote-02", protocol.PRESENCE, presence_payload("mote-02", False)) + _, first = get(server, "/v1/robots/mote-01", token=operator) + _, second = get(server, "/v1/robots/mote-02", token=operator) + assert first["presence"]["online"] is True + assert second["presence"]["online"] is False + + +def test_clearing_a_retained_topic_clears_the_field(server, operator, robot): + retain(server, robot, protocol.PRESENCE, presence_payload()) + # A zero-length payload is how MQTT clears retention. Holding the old value + # would leave this server asserting a state the broker no longer serves. + server.state.apply(protocol.topic(robot, protocol.PRESENCE), b"") + _, body = get(server, f"/v1/robots/{robot}", token=operator) + assert body["presence"] is None + + +def test_the_route_needs_an_operator_token(server, robot): + body = expect_error(lambda: get(server, f"/v1/robots/{robot}"), 401) + assert "operator token" in body["error"] + + +def test_an_unknown_robot_without_a_token_is_401_not_404(server): + # Otherwise an anonymous caller enumerates the fleet's ids by reading one + # status code against the other. + expect_error(lambda: get(server, "/v1/robots/mote-99"), 401) + + +def test_the_roster_carries_presence_per_row(server, operator): + enroll(server, "serial:aaa", name="Scout") + enroll(server, "serial:bbb", name="Rover") + retain(server, "mote-01", protocol.PRESENCE, presence_payload("mote-01", True)) + status, body = get(server, "/v1/robots") + assert status == 200 + assert body["broker_connected"] is True + rows = {row["robot_id"]: row for row in body["robots"]} + assert rows["mote-01"]["presence"]["online"] is True + # Never invented: the second robot has published nothing. + assert rows["mote-02"]["presence"] is None + + +def test_a_server_that_cannot_hear_says_so(server, operator, robot): + server.state.connected = False + _, body = get(server, f"/v1/robots/{robot}", token=operator) + # All-null state means "the robot has said nothing" only if the server can + # hear; this is the field that separates the two. + assert body["broker_connected"] is False + assert body["presence"] is None + + +def test_what_the_state_refuses_to_hold(server): + # A map announcement is not a robot; a command is not state (it is the + # write path, and a retained one would re-fire anyway); a payload that is + # not a JSON object cannot go into a JSON envelope. + for topic, payload in ( + ("mote/v2/registry/site/home/floor/ground/current", b"{}"), + (protocol.topic("mote-01", protocol.COMMAND), b"{}"), + (protocol.topic("mote-01", protocol.POSE), b"not json"), + (protocol.topic("mote-01", protocol.POSE), b"[1, 2]"), + ): + assert not server.state.apply(topic, payload), topic + assert server.state.of("mote-01") == { + "presence": None, + "health": None, + "pose": None, + "capabilities": None, + "mission_status": None, + } + + def test_a_missing_token_is_401(server): body = expect_error( lambda: post( From b00f0e4a30b5f60431f53e8dec4894502bbdd569 Mon Sep 17 00:00:00 2001 From: Michael Johnson Date: Wed, 2 Sep 2026 16:32:05 +0100 Subject: [PATCH 2/3] Docs: keep words in heading slugs, as GitHub does pymdownx's slugifier strips anything angle-bracketed as an HTML tag, so `GET /v1/robots/` slugged to `get-v1robots_1` and every link written against GitHub's `#get-v1robotsrobot_id` failed the strict build. Drop the brackets before slugging; the toc extension has already turned real tags into text by then. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01TeNdP1SdpcegFhMaFiuHi4 --- docs/hooks/repo_links.py | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/docs/hooks/repo_links.py b/docs/hooks/repo_links.py index cfa8176..a921720 100644 --- a/docs/hooks/repo_links.py +++ b/docs/hooks/repo_links.py @@ -140,15 +140,26 @@ def replace(match: re.Match) -> str: return "\n".join(out) +def _github_slug(text, sep): + # GitHub keeps the word inside a `` and drops the brackets; + # pymdownx would strip it as a tag. Real tags never reach here — the toc + # extension has already reduced them to text. + return _slugify(text.replace("<", "").replace(">", ""), sep) + + +_slugify = slugify(case="lower") + + def on_config(config): """Slug headings the way GitHub does. Every heading link already written in the repo was written against a file rendered on GitHub, and the two sluggers disagree — an em dash leaves - GitHub a double hyphen where Python-Markdown leaves a single one. Matching - GitHub is what lets those links keep working here unchanged. + GitHub a double hyphen where Python-Markdown leaves a single one, and a + `` survives on GitHub but not in pymdownx. Matching GitHub is + what lets those links keep working here unchanged. """ - config.mdx_configs.setdefault("toc", {})["slugify"] = slugify(case="lower") + config.mdx_configs.setdefault("toc", {})["slugify"] = _github_slug return config From 688471c987f301ebaef54257591ef3a626c8db26 Mon Sep 17 00:00:00 2001 From: Michael Johnson Date: Fri, 11 Sep 2026 13:24:59 +0100 Subject: [PATCH 3/3] Robot state feed: survive a failed bind, read CONNACK refusals, scope the token claim Three review points on #115. A failed bind (port in use) raised AttributeError from server_close(): TCPServer calls it from inside super().__init__, before feed and publisher were assigned. They are now assigned before the socket is bound, so the OSError reaches the operator. The feed treated a refused CONNACK as connected, because paho calls on_connect for both. It now reads the reason code (paho 2's ReasonCode, paho 1's int) and logs the refusal. A real mosquitto with anonymous access off reproduced the defect against the previous code and passes now. The operator token on GET /v1/robots/ was documented as stopping id enumeration. It does not: the roster lists every id anonymously, and the broker carries the same payloads until M7. The docstring, fleet-api.md and CLAUDE.md now say what the token and the 401-before-404 order actually buy. Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 7 ++- docs/fleet/fleet-api.md | 18 +++++--- mote_fleet/server/fleet_server.py | 44 ++++++++++++------ mote_fleet/test/test_e2e_fleet.py | 46 +++++++++++++++++++ mote_fleet/test/test_fleet_server.py | 68 +++++++++++++++++++++++++++- 5 files changed, 159 insertions(+), 24 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 6b2030c..3e88a4c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -143,8 +143,11 @@ broker has stopped serving. **`mission_status` is the last status, not a history**, since one transition is all that is retained; anything wanting every transition still subscribes, which is what `watch` and `dispatch` keep the broker for. And **the route takes an operator token where the roster does not**, -because this is where the coordinates are — checked *before* the robot is looked -up, so an anonymous caller cannot enumerate ids by reading 401 against 404. +because this is where the coordinates are — which hides nothing until M7, since +the same payloads are on the anonymous broker and every id is in the anonymous +roster; the token gives the route M7's shape now. It is checked *before* the +lookup, so an unauthenticated answer does not depend on the id, which matters +once M7 gates the roster. `publisher` and `feed` are injected as a pair in `serve()`, since a live subscription beside a stubbed publisher would have a test dialling a broker it does not have; the acceptance is `test_e2e_fleet.py`'s diff --git a/docs/fleet/fleet-api.md b/docs/fleet/fleet-api.md index 2478d15..cf747b8 100644 --- a/docs/fleet/fleet-api.md +++ b/docs/fleet/fleet-api.md @@ -79,9 +79,9 @@ every access log between here and the browser. **Security posture, plainly.** Most read routes are unauthenticated and the broker is anonymous, exactly as M1 left them. M3 adds a credential on the *write* path and a record of who used it, which is the milestone's brief; since -then two reads have been gated as well, the audit log and one robot's live -state, because both carry something an anonymous caller has no business with. -It is proportionate only while the tailnet is the boundary. M7 adds operator +then two reads have been gated as well: the audit log, which nothing else +serves, and one robot's live state, whose payloads the anonymous broker also +carries until M7. It is proportionate only while the tailnet is the boundary. M7 adds operator auth on the rest of the read routes, per-robot broker credentials, and the Tailscale ACLs. Until then, do not expose this port to a network the robots are not already trusted on. @@ -210,10 +210,14 @@ mission, which is a property of asking rather than of listening. fleet server is repopulated by the broker within about a second of connecting; a stored copy could only ever be the staler answer. -An operator token because this is where the coordinates are: a pose says where -in a building a robot is and the mission status says what it was told to do -there. The token is checked *before* the robot is looked up, so an -unauthenticated caller cannot enumerate ids by reading 401 against 404. +**The operator token hides nothing yet.** This route carries the coordinates — +a pose says where in a building a robot is, the mission status what it was told +to do there — so it takes the credential M7 will require of every read. Until +M7, the same payloads are on the anonymous broker and every `robot_id` is in the +anonymous roster, whose `presence` column shows nothing a broker subscriber +cannot already see. The token is checked *before* the robot is looked up, so an +unauthenticated request gets `401` whatever id it names; that stops ids leaking +through this route once M7 gates the roster, and not before. ### `POST /v1/robots//dispatch` diff --git a/mote_fleet/server/fleet_server.py b/mote_fleet/server/fleet_server.py index c03c622..9a7b73f 100644 --- a/mote_fleet/server/fleet_server.py +++ b/mote_fleet/server/fleet_server.py @@ -96,9 +96,9 @@ **Security posture for M3:** most read routes are still unauthenticated, exactly as M1 left them, and the broker is still anonymous. What M3 adds is a credential -on the *write* path and a record of who used it. Two reads take one as well — -the audit log, and one robot's live state — because both carry something an -anonymous caller has no business with. That stays proportionate only +on the *write* path and a record of who used it. Two reads take one as well: +the audit log, which nothing else serves, and one robot's live state, whose +payloads the anonymous broker also carries until M7. That stays proportionate only while the tailnet is the boundary; M7 is where operator auth reaches the rest of the read routes, per-robot broker credentials land, and Tailscale ACLs stop robots reaching each other. Until then, do not expose this port to a network the @@ -393,7 +393,18 @@ def start(self) -> bool: return False return True - def _on_connect(self, client, _userdata, *_args): + def _on_connect(self, client, _userdata, _flags, reason_code, *_args): + # paho calls this for a refused CONNACK too (after M7, a bad broker + # credential). paho 2 passes a ReasonCode, paho 1 an int that is 0 on + # success. + if getattr(reason_code, "is_failure", reason_code != 0): + self.state.connected = False + print( + f"broker {self.host}:{self.port} refused the robot state feed: " + f"{reason_code}", + file=sys.stderr, + ) + return for leaf in STATE_LEAVES: client.subscribe(protocol.any_robot(leaf), qos=protocol.QOS) self.state.connected = True @@ -652,9 +663,13 @@ def _robot(self, rest: str): Behind an operator token where the roster is not, because this is where the coordinates are — a pose says where in a building the robot is, and - the mission status says what it was told to do there. The token is - checked before the robot is looked up, so an unauthenticated caller - cannot enumerate ids by reading 401 against 404. + the mission status says what it was told to do there. The token hides + nothing yet: until M7 the same payloads are on the anonymous broker and + every id is in the anonymous roster. It gives the route the shape M7 + will require, so an HTTP client written now already carries the + credential. It is checked before the lookup so that an unauthenticated + answer does not depend on the id, which matters once M7 gates the + roster and not before. """ if "/" in rest: self._error(404, f"no route /v1/robots/{rest}") @@ -1236,23 +1251,26 @@ def __init__( broker_ws_port, foxglove_url, ): + # Assigned before the socket is bound: when the bind fails, TCPServer + # calls server_close() from inside super().__init__, and it must reach + # these rather than raise AttributeError over the real OSError. + self.publisher = publisher + #: What the fleet's robots last said, for the HTTP read routes. The + #: feed that fills it is attached by :func:`serve`, because it + #: subscribes on behalf of this object and so cannot exist before it. + self.state = RobotState() + self.feed = None super().__init__(address, FleetHandler) self.registry = registry self.broker_host = broker_host self.broker_port = broker_port self.id_prefix = id_prefix - self.publisher = publisher self.maps_dir = Path(maps_dir).expanduser() if maps_dir else None self.store = BundleStore(self.maps_dir) self.ui_dir = Path(ui_dir).resolve() if ui_dir else None self.broker_ws_host = broker_ws_host self.broker_ws_port = broker_ws_port self.foxglove_url = foxglove_url - #: What the fleet's robots last said, for the HTTP read routes. The - #: feed that fills it is attached by :func:`serve`, because it - #: subscribes on behalf of this object and so cannot exist before it. - self.state = RobotState() - self.feed = None def server_close(self): """Give back the sockets, the subscription and the publisher together. diff --git a/mote_fleet/test/test_e2e_fleet.py b/mote_fleet/test/test_e2e_fleet.py index 8b0f788..c26d30f 100644 --- a/mote_fleet/test/test_e2e_fleet.py +++ b/mote_fleet/test/test_e2e_fleet.py @@ -338,6 +338,52 @@ def test_the_registry_survives_a_server_restart(tmp_path, broker): second.server_close() +def test_a_broker_that_refuses_the_feed_is_reported_as_not_connected(tmp_path): + """A real refusal, not a stubbed reason code: mosquitto with anonymous + access off answers the feed's CONNECT with "not authorized", which is what + a bad credential will look like once M7 issues them.""" + from fleet_server import BrokerFeed, BrokerLink, serve + + refusing = Broker(tmp_path) + refusing.conf.write_text( + f"listener {refusing.port} 127.0.0.1\nallow_anonymous false\n" + "persistence false\n" + ) + refusing.start() + answered = threading.Event() + + class ObservedFeed(BrokerFeed): + def _on_connect(self, *args): + super()._on_connect(*args) + answered.set() + + # A supplied publisher means serve() builds no feed of its own, so the + # observed one is the only client the broker sees. + server = serve( + db=tmp_path / "registry.db", + host="127.0.0.1", + port=0, + broker_host="127.0.0.1", + broker_port=refusing.port, + publisher=BrokerLink("127.0.0.1", refusing.port), + ) + server.feed = ObservedFeed("127.0.0.1", refusing.port, server.state) + server.feed.start() + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + url = f"http://127.0.0.1:{server.server_address[1]}" + try: + assert answered.wait(10), "the broker never answered the feed's CONNECT" + assert server.state.connected is False + code, roster = api_get(url, "/v1/robots") + assert code == 200, roster + assert roster["broker_connected"] is False + finally: + server.shutdown() + server.server_close() + refusing.stop() + + def api_get(url, path, token=""): import urllib.error import urllib.request diff --git a/mote_fleet/test/test_fleet_server.py b/mote_fleet/test/test_fleet_server.py index 1ab9165..91cfdc4 100644 --- a/mote_fleet/test/test_fleet_server.py +++ b/mote_fleet/test/test_fleet_server.py @@ -12,7 +12,12 @@ ``api_harness.py``; the map registry's own routes are ``test_map_registry.py``. """ +import errno +import socket + +import pytest from api_harness import ( + FakeBroker, enroll, expect_error, get, @@ -21,6 +26,7 @@ post_raw, retain, ) +from fleet_server import STATE_LEAVES, BrokerFeed, RobotState, serve from mote_bringup.spec import mission from mote_fleet import protocol @@ -173,8 +179,9 @@ def test_the_route_needs_an_operator_token(server, robot): def test_an_unknown_robot_without_a_token_is_401_not_404(server): - # Otherwise an anonymous caller enumerates the fleet's ids by reading one - # status code against the other. + # The unauthenticated answer does not depend on the id. That hides nothing + # while the roster is anonymous; it stops this route leaking ids once M7 + # gates the roster. expect_error(lambda: get(server, "/v1/robots/mote-99"), 401) @@ -220,6 +227,63 @@ def test_what_the_state_refuses_to_hold(server): } +def test_a_port_already_in_use_is_the_error_raised(tmp_path): + # TCPServer calls server_close() itself when the bind fails, before + # FleetServer.__init__ has run past its super() call. + with socket.socket() as holder: + holder.bind(("127.0.0.1", 0)) + holder.listen() + with pytest.raises(OSError) as raised: + serve( + db=tmp_path / "registry.db", + host="127.0.0.1", + port=holder.getsockname()[1], + broker_host="fleet-box", + publisher=FakeBroker(), + ) + assert raised.value.errno == errno.EADDRINUSE + + +class RecordingClient: + def __init__(self): + self.subscribed = [] + + def subscribe(self, topic, qos=0): + self.subscribed.append(topic) + + +def test_a_refused_connection_is_not_a_connected_one(): + """paho calls on_connect for a refused CONNACK as well as an accepted one. + Once M7 issues broker credentials a bad one is the likely refusal, and + reporting it as connected is the ambiguity `broker_connected` removes.""" + reasoncodes = pytest.importorskip("paho.mqtt.reasoncodes") + packettypes = pytest.importorskip("paho.mqtt.packettypes") + state = RobotState() + feed = BrokerFeed("fleet-box", 1883, state) + client = RecordingClient() + + refused = reasoncodes.ReasonCode(packettypes.PacketTypes.CONNACK, "Not authorized") + feed._on_connect(client, None, {}, refused, None) + assert state.connected is False + assert client.subscribed == [] + + accepted = reasoncodes.ReasonCode(packettypes.PacketTypes.CONNACK, "Success") + feed._on_connect(client, None, {}, accepted, None) + assert state.connected is True + assert len(client.subscribed) == len(STATE_LEAVES) + + +def test_a_paho_1_return_code_is_read_the_same_way(): + # paho 1.x passes a bare int: 0 accepted, anything else refused. + state = RobotState() + feed = BrokerFeed("fleet-box", 1883, state) + client = RecordingClient() + feed._on_connect(client, None, {}, 5) + assert (state.connected, client.subscribed) == (False, []) + feed._on_connect(client, None, {}, 0) + assert state.connected is True + + def test_a_missing_token_is_401(server): body = expect_error( lambda: post(