diff --git a/CLAUDE.md b/CLAUDE.md index 8880ce8..3e88a4c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -114,6 +114,46 @@ 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 — 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 +`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..cf747b8 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, 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. --- @@ -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,104 @@ 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. + +**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` Send one mission to one robot. 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 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..9a7b73f 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, 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 +robots are not already trusted on. """ import argparse @@ -227,6 +240,192 @@ 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, _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 + + 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 +523,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 +629,71 @@ 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 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}") + 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): @@ -994,12 +1251,20 @@ 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 @@ -1007,6 +1272,18 @@ def __init__( self.broker_ws_port = broker_ws_port self.foxglove_url = foxglove_url + 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 ------------------------------ def ui_config(self) -> dict: @@ -1196,15 +1473,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 +1502,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 +1608,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..c26d30f 100644 --- a/mote_fleet/test/test_e2e_fleet.py +++ b/mote_fleet/test/test_e2e_fleet.py @@ -338,6 +338,66 @@ 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 + + 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 +505,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..91cfdc4 100644 --- a/mote_fleet/test/test_fleet_server.py +++ b/mote_fleet/test/test_fleet_server.py @@ -12,14 +12,21 @@ ``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, get_bytes, post, post_raw, + retain, ) +from fleet_server import STATE_LEAVES, BrokerFeed, RobotState, serve from mote_bringup.spec import mission from mote_fleet import protocol @@ -74,20 +81,209 @@ 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): + # 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) + + +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_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(