diff --git a/.gitignore b/.gitignore index 8bc02f5f..6377b52f 100644 --- a/.gitignore +++ b/.gitignore @@ -20,3 +20,6 @@ secret* # Playground and scratch state (see updater/examples/playground.rs). /verify + +# Owner-specific notes and locally captured robot data never belong in this public repository. +/.private/ diff --git a/docs/README.md b/docs/README.md index 4fcf2330..d0b01183 100644 --- a/docs/README.md +++ b/docs/README.md @@ -11,6 +11,7 @@ one in front of you and want to drive it, start at the [cheat sheet](robot/cheat | [`pair-a-gamepad.md`](robot/pair-a-gamepad.md) | Once per pad: pairing mode, `pad pair`, and what to do when it will not bond. | | [`cheatsheet-dev.md`](robot/cheatsheet-dev.md) | The commands that need a dev board: branch builds, candidates, dev pushes. | | [`dev-push.md`](robot/dev-push.md) | Build on your machine and install on the board over ssh, with no CI run. | +| [`model-channel.md`](robot/model-channel.md) | Package, verify and trial a policy bundle. | | [`duckctl.md`](robot/duckctl.md) | Every `duckctl` command — the robot from a laptop, over Bluetooth. | | [`install-dev.md`](robot/install-dev.md) | Setting up a board for development, from nothing. | | [`install-by-hand.md`](robot/install-by-hand.md) | The same install as separate commands, for testing one step at a time. | diff --git a/docs/project/npu-bringup.md b/docs/project/npu-bringup.md index 326f4c8a..c09858e8 100644 --- a/docs/project/npu-bringup.md +++ b/docs/project/npu-bringup.md @@ -99,15 +99,18 @@ that as the price of perception, the two should be measured apart. ## What is still missing -**Nothing on the robot can get a frame.** `mediad` has a raw NV12 tee branch that exists precisely -for this — `architecture.md` §5.3 — but no IPC exposes it, which is also why capturing a dataset has -to stop `mediad` to take the camera. Two ways forward, and they are not exclusive: - -- **`media.frame`**: a call that answers with one frame. Useful for far more than perception (a - snapshot in the console, a still for a bug report), and it makes capture stop fighting the daemon. -- **The detector inside `mediad`**: subscribe to the raw branch, run the model at a few Hz, and - publish detections on the state stream. This is where it ends up — perception next to the sensor, - deriving features rather than shipping pixels — and it is what a behaviour would consume. +`media.frame` now exposes the raw **UYVY** tee branch locally at `/run/mediad/media.sock`. +`robotctl media frame --output frame.uyvy` asks for the latest frame without stopping `mediad` or +back-pressuring its encoder. The response starts with a JSON-RPC header (geometry, format, byte +count and capture time), followed by exactly that many binary bytes. It is deliberately not a +WebRTC method: a frame is about 1.8 MiB at the default geometry, whereas the control channel has to +stay prompt. A recorder can therefore join camera observations to robot state on the board without +fighting the daemon or base64-encoding pixels. + +What remains is publishing the detector's output on the robot state stream, so a behaviour can +consume it without opening a second observation channel. The detector already subscribes to the +raw branch and runs at a paced rate; publishing keeps perception next to the sensor, deriving +features rather than shipping pixels. Once detections exist as state, the behaviours in `docs/ideas/autonomous_behavior.md` that currently key on Bluetooth ("a duck is *nearby*") can key on sight ("a duck is *there*"): approaching, diff --git a/docs/project/roadmap.md b/docs/project/roadmap.md index 72ead9b1..481b8a90 100644 --- a/docs/project/roadmap.md +++ b/docs/project/roadmap.md @@ -252,17 +252,20 @@ version line. **What is missing is at the two ends, not in the middle:** -- **`robotd` cannot reload.** There is no SIGHUP handler and no way to swap an `ort` session - under a running 50 Hz loop. This is the milestone's real engineering: the swap must not drop a - tick, and a model whose shape is not `obs[1,61] → actions[1,14]` has to be refused *before* it - goes live rather than at the first inference. -- **Nothing publishes a bundle.** `xtask package --channel model-walk` is close — it checks - `--version` against the crate version, which a model does not have — and the HF repo layout - and naming do not exist. +- **Safe reload is implemented, but needs board evidence.** `robotd` coalesces SIGHUP requests, + waits until policy control is disabled, builds the candidate off the 50 Hz loop, and keeps the + current controller if loading fails. The remaining proof is a board run measuring the reload + boundary and refusing a malformed ONNX model before it is used. +- **A bundle can now be made, but nothing publishes one.** `cargo xtask package --channel + model-walk --model-dir --model-api 1 --version ` makes an independently-versioned + artifact whose manifest carries its compatibility API; it does not accept daemon hooks or + binaries. The HF repo layout, release workflow and naming still do not exist. - **A third signing key.** `release-1` is CI's and `team.dev` installs nothing on a customer robot, so *who may publish a policy a robot will run* is a new custody question, not a reuse of an existing one. -- **`model_api`** (§5.5) is designed and unimplemented on both sides. +- **`model_api`** (§5.5) is checked by the updater against the running daemon and is required + when packaging a model bundle. It still needs a real published model to exercise that boundary + on a board. - **The training loop.** `microduck_rl` trains and exports to ONNX; nothing carries the result to a board without a daemon release. The model equivalent of `dev-push.sh` is what makes "train it and try it" a minute rather than a CI run. diff --git a/docs/robot/cheatsheet.md b/docs/robot/cheatsheet.md index b49adb28..f0890712 100644 --- a/docs/robot/cheatsheet.md +++ b/docs/robot/cheatsheet.md @@ -31,6 +31,18 @@ Hardware and software in one report. Exits non-zero when the robot is unhealthy it can gate a script — a hot motor or a pinned component is reported, not judged, and does not affect the exit code. `--json` for a support bundle. +### A support bundle + +``` +robotctl support +``` + +Writes `/var/tmp/microduck-support.txt`: health and version reports, relevant unit status, the +latest daemon journal, and update history. It does not change the robot. Lines that may carry a +credential are removed before the file is written, but review the file before sharing it because +it still identifies the robot and describes its software state. Choose another destination with +`--output path/to/report.txt`. + ### Watching the loop ``` @@ -790,4 +802,3 @@ eval "$(robotctl completions bash)" ``` `zsh`, `fish`, `elvish` and `powershell` work in place of `bash`. - diff --git a/docs/robot/lerobot-local-recording.md b/docs/robot/lerobot-local-recording.md new file mode 100644 index 00000000..eccd1ffb --- /dev/null +++ b/docs/robot/lerobot-local-recording.md @@ -0,0 +1,47 @@ +# Local LeRobot recording + +Before hardware arrives, run `bash scripts/record-lerobot-preflight.sh`. It uses synthetic data only. + +On a robot, record one 30-second local episode, validate it, then inspect the report. Do not export +or train on an episode with missing frames, non-monotonic capture times, or malformed actions. +The recorder never sends a control command and never uploads data. Images remain under +`/var/lib/robot/datasets/` until an operator deliberately copies them for local development. + +Recording stops at the first configured budget: 300 frames, 512 MiB of raw images, or a 1 GiB +free-space reserve by default. It also refuses a 21st local episode until the existing ones have +been reviewed or archived. These are deliberate experiment-cost guardrails, not a retention +policy: choose explicit `--max-*` values for a larger, reviewed run. The validator rejects a +dataset that exceeds its recorded frame or byte budget. + +Before converting, run the exporter's `--dry-run`. It reports policy-labelled frames and the +estimated uncompressed RGB footprint, refusing more than 300 frames or 1 GiB by default. Creating +the local LeRobot dataset then requires `--confirm-export`, and its destination must be empty: + +``` +python3 scripts/export-lerobot-local.py /var/lib/robot/datasets/ \ + --root /var/lib/robot/lerobot/ --dry-run +python3 scripts/export-lerobot-local.py /var/lib/robot/datasets/ \ + --root /var/lib/robot/lerobot/ --confirm-export +``` + +## Before trying a model update + +With policy control disabled, run a read-only update dry run. It refuses an armed policy, a fallen +or limp robot, an unhealthy daemon, and an older daemon that cannot report its policy state. The +dry run then uses the normal updater path to verify the candidate's signature, hash and +`model_api`, but stops before moving `current` or signalling `robotd`: + +``` +python3 scripts/model-update-preflight.py model-walk +# for a signed local artifact directory: +python3 scripts/model-update-preflight.py model-walk --from /path/to/signed-artifacts +``` + +After an operator applies the model through the normal updater flow, observe the disabled-policy +control loop for ten seconds before re-enabling policy control. This command does not signal, +reload, or change robot settings; it writes a local report and rejects a policy-enabled, fallen, +limp, unhealthy, or missed-tick observation: + +``` +python3 scripts/observe-model-reload.py +``` diff --git a/docs/robot/model-channel.md b/docs/robot/model-channel.md new file mode 100644 index 00000000..2dac0c6c --- /dev/null +++ b/docs/robot/model-channel.md @@ -0,0 +1,54 @@ +# Model channel contract + +A policy is a separately versioned `model-*` component. It is not a daemon release: it carries +weights and model metadata only, and it cannot carry daemon binaries or install hooks. This keeps +a model trial inside the updater's ordinary signature, hash, compatibility, rollback and pinning +boundaries without giving a policy artifact a second way to alter a robot. + +## Bundle + +Build a bundle from a directory containing the files for one named policy slot. Files are installed +at the root of that model component's release, so a configured slot can name, for example, +`/opt/robot/model/walk/current/walk.onnx` directly. + +``` +cargo xtask package \ + --channel model-walk \ + --version 1.2.3 \ + --model-dir /path/to/walk-bundle \ + --model-api 1 \ + --out dist/ +``` + +`--model-api` is required. A robot accepts the artifact only when its running daemon implements at +least that API. The package command rejects non-`model-*` channels, daemon hooks, and daemon binary +layout for model bundles. Sign the resulting artifact and manifests with a policy signing key whose +public half is already trusted by the target robot; never put a private key or passphrase in this +repository, a command history, or a model bundle. + +The bundle is deliberately shallow: package the ONNX file and the small metadata needed to run it, +not recordings, checkpoints, training logs, notebooks, or dependencies. Those belong to the +training environment, not a robot update. + +## Trial sequence + +1. Disable policy control and leave the robot upright, not limp or fallen. +2. Run `python3 scripts/model-update-preflight.py model-walk` (add `--from` for a signed local + artifact directory). It verifies the stopped state, health, signature, hash and compatibility + through the updater's dry-run path; it does not swap `current`. +3. An operator applies the configured model component with the normal `robotctl update apply` + workflow. This is the only step that may move `current` or request a reload. +4. Before re-enabling policy control, run `python3 scripts/observe-model-reload.py`. It records a + ten-second disabled-policy window and rejects missed ticks, a safety event, or unhealthy state. +5. Inspect the local observation report and update transcript. Only then make a deliberate, + supervised decision to re-enable policy control. + +An older daemon that does not report `policy_enabled` fails step 2 rather than being assumed safe. +An incompatible or malformed model must remain non-current; do not bypass the preflight by copying +files into a component's `current` path. + +## Experiment budget + +Use short, local-only recordings for the first trial. The recorder and converter default to bounded +frames, storage, episode count and export size; see [local LeRobot recording](lerobot-local-recording.md). +Increase a `--max-*` limit only for a reviewed run with a stated metric and a stopping condition. diff --git a/duck-ipc-proto/src/lib.rs b/duck-ipc-proto/src/lib.rs index 33224abe..436a5dab 100644 --- a/duck-ipc-proto/src/lib.rs +++ b/duck-ipc-proto/src/lib.rs @@ -202,6 +202,10 @@ pub mod socket { /// Under `/run/tofd/` for the same reason as the pad's: it is that unit's /// `RuntimeDirectory=`, so systemd removes the socket when the daemon stops. pub const TOF: &str = "/run/tofd/tof.sock"; + + /// `mediad`'s on-demand raw-frame endpoint. It is local-only: a raw camera frame is for a + /// recorder or perception process on the robot, not a multi-megabyte WebRTC control reply. + pub const MEDIA: &str = "/run/mediad/media.sock"; } /// Where each daemon publishes what it is running: `/run//identity.json`. @@ -259,6 +263,10 @@ pub const JOINT_NAMES: [&str; 15] = [ pub mod method { pub const HELLO: &str = "hello"; + /// One raw camera frame. `mediad` answers the JSON-RPC header, followed immediately by the + /// bytes named in that header, on its local Unix socket. + pub const MEDIA_FRAME: &str = "media.frame"; + pub const CHECK: &str = "update.check"; pub const APPLY: &str = "update.apply"; pub const ROLLBACK: &str = "update.rollback"; @@ -2566,6 +2574,12 @@ pub struct RobotState { pub head: [f64; 4], /// Which policy drove this tick: `walk`, `stand`, or `held` when none did. pub policy: String, + /// Whether policy control is armed for this tick. `held` alone is not enough to decide that: + /// an enabled policy can be holding on a zero command. This lets an update preflight require + /// the disabled state before a model reload. Absent from older daemons, which callers must + /// treat as unknown rather than safe. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub policy_enabled: Option, pub safety: SafetyState, #[serde(rename = "loop")] pub control_loop: LoopState, @@ -2573,6 +2587,11 @@ pub struct RobotState { pub joints: Vec, /// What was commanded, so a viewer can show tracking error rather than guessing at it. pub targets: Vec, + /// Raw output of the policy which drove this tick, in its 14-wide policy order. It is the + /// action before scaling and filtering, so an imitation-learning recorder does not have to + /// reverse engineer an action from joint targets. Empty when no policy drove this tick. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub policy_action: Vec, /// Where contact odometry believes the robot is. `default` so a frame from /// a `robotd` predating the estimator still parses — zeros, like a robot /// that has not moved. @@ -4416,6 +4435,7 @@ mod tests { }, head: [0.0; 4], policy: "stand".into(), + policy_enabled: None, safety: SafetyState { fallen: false, limp: false, @@ -4428,6 +4448,7 @@ mod tests { }, joints: vec![0.0; 15], targets: vec![0.0; 15], + policy_action: vec![], odom: OdomState::default(), theremin: None, chorale: None, @@ -4473,6 +4494,7 @@ mod tests { }, head: [0.0; 4], policy: "walk".into(), + policy_enabled: Some(false), safety: SafetyState { fallen: false, limp: false, @@ -4485,6 +4507,7 @@ mod tests { }, joints: vec![0.0; 15], targets: vec![0.0; 15], + policy_action: vec![], odom: OdomState::default(), theremin: None, chorale: None, @@ -4493,6 +4516,7 @@ mod tests { let line = serde_json::to_string(&Request::notify_state(&state)).unwrap(); assert!(line.contains(r#""method":"robot.state""#), "{line}"); assert!(line.contains(r#""move":"#), "{line}"); + assert!(line.contains(r#""policy_enabled":false"#), "{line}"); assert!(line.contains(r#""loop":"#), "{line}"); assert!(!line.contains("movement"), "{line}"); assert!(!line.contains("control_loop"), "{line}"); @@ -4833,4 +4857,53 @@ mod tests { released ); } + + #[test] + fn policy_action_is_present_only_for_a_policy_driven_tick() { + let state = RobotState { + t: 1.0, + movement: MoveState { + requested: [0.0; 3], + applied: [0.0; 3], + limited_by: vec![], + }, + head: [0.0; 4], + policy: "walk".into(), + policy_enabled: None, + safety: SafetyState { + fallen: false, + limp: false, + gravity: [0.0, 0.0, -1.0], + gain: None, + }, + control_loop: LoopState { + hz: 50.0, + missed: 0, + }, + joints: vec![0.0; 15], + targets: vec![0.0; 15], + policy_action: vec![0.0; 14], + odom: OdomState::default(), + theremin: None, + chorale: None, + }; + let line = serde_json::to_string(&state).unwrap(); + assert!(line.contains("\"policy_action\":[0.0,0.0,0.0"), "{line}"); + assert_eq!( + serde_json::from_str::(&line) + .unwrap() + .policy_action + .len(), + 14 + ); + let held = RobotState { + policy_action: vec![], + ..state + }; + assert!( + !serde_json::to_string(&held) + .unwrap() + .contains("policy_action") + ); + } } diff --git a/mediad/src/exposure.rs b/mediad/src/exposure.rs index e84d3cc3..662102a8 100644 --- a/mediad/src/exposure.rs +++ b/mediad/src/exposure.rs @@ -420,6 +420,7 @@ mod tests { width: 1280, height: 720, format: CAPTURE_FORMAT, + captured_at: std::time::SystemTime::UNIX_EPOCH, data, } } diff --git a/mediad/src/frame.rs b/mediad/src/frame.rs new file mode 100644 index 00000000..b0fe32c4 --- /dev/null +++ b/mediad/src/frame.rs @@ -0,0 +1,235 @@ +//! A local `media.frame` endpoint for a recorder or perception process on the robot. +//! +//! A frame stays out of the WebRTC control channel: at the default geometry the UYVY payload is +//! about 1.8 MiB, so JSON/base64 would make a control request several MiB and let a slow peer tie +//! camera data to the network. The Unix socket is group-readable like the other observation +//! sockets. It sends one JSON-RPC response header, then precisely `bytes` raw bytes; that keeps +//! the metadata inspectable without copying pixels through a text encoding. + +use std::os::unix::fs::PermissionsExt; +use std::path::Path; +use std::time::UNIX_EPOCH; + +use anyhow::{Context, Result}; +use duck_ipc_proto as proto; +use serde::Serialize; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{UnixListener, UnixStream}; + +use crate::pipeline::Frames; + +const SOCKET_MODE: u32 = 0o660; +const MAX_REQUEST_BYTES: usize = 4096; + +#[derive(Debug, Serialize)] +struct Header { + width: u32, + height: u32, + format: &'static str, + bytes: usize, + /// Wall time makes the snapshot joinable to a separately sampled robot state. It is not used + /// to pace capture, so an NTP adjustment cannot affect the pipeline. + captured_at_unix_us: u128, +} + +/// Serve snapshots until the daemon exits. A client gets the latest frame immediately, or an +/// explicit error before the first camera buffer arrives; it never waits in a queue for a future +/// frame. +pub async fn serve(socket: &Path, frames: Frames) -> Result<()> { + if let Some(parent) = socket.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("creating {}", parent.display()))?; + } + if socket.exists() { + std::fs::remove_file(socket) + .with_context(|| format!("removing stale {}", socket.display()))?; + } + let listener = + UnixListener::bind(socket).with_context(|| format!("binding {}", socket.display()))?; + std::fs::set_permissions(socket, std::fs::Permissions::from_mode(SOCKET_MODE)) + .with_context(|| format!("setting permissions on {}", socket.display()))?; + tracing::info!(path = %socket.display(), mode = format!("{SOCKET_MODE:o}"), "serving media.frame locally"); + + loop { + match listener.accept().await { + Ok((stream, _)) => { + let frames = frames.clone(); + tokio::spawn(async move { + if let Err(error) = handle(stream, frames).await { + tracing::debug!(error = %error, "media.frame client ended"); + } + }); + } + Err(error) => tracing::warn!(error = %error, "media.frame accept failed"), + } + } +} + +async fn handle(stream: UnixStream, frames: Frames) -> Result<()> { + let (read, mut write) = stream.into_split(); + let mut reader = BufReader::new(read); + let mut line = String::new(); + reader.read_line(&mut line).await?; + if line.len() > MAX_REQUEST_BYTES { + write_response( + &mut write, + proto::Response::err( + None, + proto::Error::new(proto::code::INVALID_PARAMS, "request is too large"), + ), + ) + .await?; + return Ok(()); + } + let request: proto::Request = match serde_json::from_str(line.trim()) { + Ok(request) => request, + Err(error) => { + write_response( + &mut write, + proto::Response::err( + None, + proto::Error::new(proto::code::PARSE_ERROR, error.to_string()), + ), + ) + .await?; + return Ok(()); + } + }; + if request.method != proto::method::MEDIA_FRAME { + write_response( + &mut write, + proto::Response::err( + request.id, + proto::Error::new( + proto::code::METHOD_NOT_FOUND, + format!("{} is not served by mediad", request.method), + ), + ), + ) + .await?; + return Ok(()); + } + let Some(frame) = frames.latest() else { + write_response( + &mut write, + proto::Response::err( + request.id, + proto::Error::new( + proto::code::INTERNAL_ERROR, + "camera has not produced a frame yet", + ), + ), + ) + .await?; + return Ok(()); + }; + let captured_at_unix_us = frame + .captured_at + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_micros(); + let header = Header { + width: frame.width, + height: frame.height, + format: frame.format, + bytes: frame.data.len(), + captured_at_unix_us, + }; + write_response(&mut write, proto::Response::ok(request.id, &header)).await?; + write.write_all(&frame.data).await?; + write.flush().await?; + Ok(()) +} + +async fn write_response( + write: &mut tokio::net::unix::OwnedWriteHalf, + response: proto::Response, +) -> Result<()> { + let mut line = serde_json::to_vec(&response)?; + line.push(b'\n'); + write.write_all(&line).await?; + write.flush().await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::time::{Duration, UNIX_EPOCH}; + + use super::*; + use crate::pipeline::Frame; + use tokio::io::AsyncReadExt; + + async fn reply(frames: Frames, request: &str) -> proto::Response { + let (mut client, server) = UnixStream::pair().unwrap(); + let task = tokio::spawn(handle(server, frames)); + client.write_all(request.as_bytes()).await.unwrap(); + client.shutdown().await.unwrap(); + let mut text = String::new(); + BufReader::new(client).read_line(&mut text).await.unwrap(); + task.await.unwrap().unwrap(); + serde_json::from_str(text.trim()).unwrap() + } + + #[tokio::test] + async fn no_camera_frame_is_an_explicit_error() { + let response = reply( + Frames::default(), + "{\"jsonrpc\":\"2.0\",\"id\":7,\"method\":\"media.frame\",\"params\":{}}\n", + ) + .await; + assert_eq!(response.id, Some(proto::Id::Number(7))); + assert_eq!(response.error.unwrap().code, proto::code::INTERNAL_ERROR); + } + + #[tokio::test] + async fn an_unknown_method_is_refused_without_reading_a_frame() { + let response = reply( + Frames::default(), + "{\"jsonrpc\":\"2.0\",\"id\":\"request\",\"method\":\"media.other\"}\n", + ) + .await; + assert_eq!(response.id, Some(proto::Id::String("request".into()))); + assert_eq!(response.error.unwrap().code, proto::code::METHOD_NOT_FOUND); + } + + #[tokio::test] + async fn an_oversized_request_is_rejected_before_it_is_parsed() { + let request = format!("{}\n", "x".repeat(MAX_REQUEST_BYTES + 1)); + let response = reply(Frames::default(), &request).await; + assert_eq!(response.id, None); + assert_eq!(response.error.unwrap().code, proto::code::INVALID_PARAMS); + } + + #[tokio::test] + async fn a_frame_reply_names_and_follows_with_exactly_its_pixels() { + let frames = Frames::default(); + frames.publish(Frame { + width: 2, + height: 1, + format: "UYVY", + captured_at: UNIX_EPOCH + Duration::from_secs(1), + data: vec![128, 32, 128, 64], + }); + let (mut client, server) = UnixStream::pair().unwrap(); + let task = tokio::spawn(handle(server, frames)); + client + .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"media.frame\"}\n") + .await + .unwrap(); + let mut reader = BufReader::new(client); + let mut line = String::new(); + reader.read_line(&mut line).await.unwrap(); + let response: proto::Response = serde_json::from_str(line.trim()).unwrap(); + let result = response.result.unwrap(); + assert_eq!(result["width"], 2); + assert_eq!(result["height"], 1); + assert_eq!(result["format"], "UYVY"); + assert_eq!(result["bytes"], 4); + assert_eq!(result["captured_at_unix_us"], 1_000_000); + let mut pixels = [0; 4]; + reader.read_exact(&mut pixels).await.unwrap(); + assert_eq!(pixels, [128, 32, 128, 64]); + task.await.unwrap().unwrap(); + } +} diff --git a/mediad/src/lib.rs b/mediad/src/lib.rs index cf7ee95a..9e3c2179 100644 --- a/mediad/src/lib.rs +++ b/mediad/src/lib.rs @@ -38,3 +38,8 @@ pub mod exposure; /// reads the same raw branch, in the same pixel format the pipeline names. #[cfg(target_os = "linux")] pub mod detect; + +/// The local, on-demand raw-frame endpoint. Linux only because it shares the pipeline's raw +/// frame store; the WebRTC control channel deliberately does not carry camera-sized replies. +#[cfg(target_os = "linux")] +pub mod frame; diff --git a/mediad/src/main.rs b/mediad/src/main.rs index 6289656d..64de69b3 100644 --- a/mediad/src/main.rs +++ b/mediad/src/main.rs @@ -253,6 +253,17 @@ fn main() -> ExitCode { } }; + // A recorder asks the local Unix socket for a single latest raw frame. It is intentionally + // separate from the datachannel: snapshots can be camera-sized, while control must remain + // prompt even when a recorder is writing slowly. + let frame_socket = std::path::PathBuf::from(duck_ipc_proto::socket::MEDIA); + let frame_source = frames.clone(); + tokio::spawn(async move { + if let Err(error) = mediad::frame::serve(&frame_socket, frame_source).await { + tracing::error!(error = %format!("{error:#}"), "media.frame endpoint stopped"); + } + }); + // After the pipeline, because it meters the pipeline's own frames — and only with a real // camera, since a test pattern has no sensor to write and the loop would spend the daemon's // life reporting that it cannot. diff --git a/mediad/src/pipeline.rs b/mediad/src/pipeline.rs index 6c6c4c6a..70072ec1 100644 --- a/mediad/src/pipeline.rs +++ b/mediad/src/pipeline.rs @@ -79,6 +79,7 @@ //! naming the arity rather than as an abort. use std::sync::{Arc, Mutex}; +use std::time::SystemTime; use anyhow::{Context, Result, anyhow, bail}; use duck_ipc_proto as proto; @@ -215,6 +216,9 @@ pub struct Frame { /// The GStreamer format name — [`CAPTURE_FORMAT`], carried rather than assumed so a consumer /// reading this cannot silently misinterpret the bytes if the capture format changes again. pub format: &'static str, + /// Captured when this buffer reached the raw branch. This is observation time for a recorder; + /// it is intentionally separate from GStreamer's scheduling timestamps. + pub captured_at: SystemTime, /// Tightly packed as the caps describe it, in `format`. pub data: Vec, } @@ -229,6 +233,12 @@ pub struct Frame { pub struct Frames(Arc>>); impl Frames { + /// Publish the newest frame. Replacing rather than queueing is what prevents a slow local + /// observer from back-pressuring capture or receiving stale observations. + pub(crate) fn publish(&self, frame: Frame) { + *self.0.lock().expect("frame lock") = Some(frame); + } + /// Read the latest frame in place, without copying it. `None` until the first one arrives. /// /// For a reader that wants a number out of a frame rather than the frame — the auto-exposure @@ -612,10 +622,11 @@ fn wire_frames(appsink: &gst_app::AppSink, frames: Frames, width: u32, height: u width, height, format: CAPTURE_FORMAT, + captured_at: SystemTime::now(), data: map.as_slice().to_vec(), }; // Replaced, not queued: last-value-wins is the contract. - *frames.0.lock().expect("frame lock") = Some(frame); + frames.publish(frame); Ok(gst::FlowSuccess::Ok) }) diff --git a/robotctl/src/main.rs b/robotctl/src/main.rs index c7f0fc59..5bbc7c08 100644 --- a/robotctl/src/main.rs +++ b/robotctl/src/main.rs @@ -30,7 +30,7 @@ //! while progress stays visible. //! - Works when `robotd` is dead. It talks to `updaterd`, not to `robotd`. -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader, Read, Write}; use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; use std::process::ExitCode; @@ -43,6 +43,7 @@ mod duck; mod monitor; mod path_map; mod show; +mod support; /// Exit codes. Stable — CI asserts on these. mod exit { @@ -100,6 +101,11 @@ struct Cli { #[arg(long, global = true, default_value = proto::socket::TOF)] tof_socket: PathBuf, + /// Path to mediad's local raw-frame endpoint. This is deliberately separate from the WebRTC + /// control channel: a camera frame is too large to make a good network control reply. + #[arg(long, global = true, default_value = proto::socket::MEDIA)] + media_socket: PathBuf, + #[command(subcommand)] namespace: Namespace, } @@ -200,6 +206,13 @@ enum Namespace { command: PadCommand, }, + /// Camera observations for a local recorder or perception program. + #[command(subcommand_required = true, arg_required_else_help = true)] + Media { + #[command(subcommand)] + command: MediaCommand, + }, + /// Update and release management. #[command(subcommand_required = true, arg_required_else_help = true)] Update { @@ -262,6 +275,13 @@ enum Namespace { json: bool, }, + /// Write a redacted snapshot for support. It never changes the robot. + Support { + /// Destination file. Safe to attach after reviewing the redacted contents. + #[arg(long, default_value = "/var/tmp/microduck-support.txt")] + output: PathBuf, + }, + /// Print a shell completion script on stdout. /// /// Generated from this binary's own command tree, so the completions a robot offers @@ -311,6 +331,17 @@ enum NetCommand { }, } +#[derive(Subcommand, Debug)] +enum MediaCommand { + /// Save the newest raw UYVY camera frame. It does not alter camera streaming or wait for a + /// future frame. + Frame { + /// Output file for the raw UYVY bytes. + #[arg(long)] + output: PathBuf, + }, +} + #[derive(Subcommand, Debug)] enum SystemCommand { /// Name, serial and uptime. Changes nothing. @@ -2790,6 +2821,169 @@ fn resolve_from_dir(dir: &std::path::Path) -> Result { Ok(resolved.to_string_lossy().into_owned()) } +/// Make the artifact somebody can attach after a fault. It asks this exact binary for the +/// JSON reports rather than rebuilding their IPC calls here, so support and the human-facing +/// commands cannot drift into two different diagnoses. +fn run_support( + socket: &Path, + robot_socket: &Path, + config_socket: &Path, + output: &Path, +) -> Result<(), Failure> { + let exe = std::env::current_exe() + .map_err(|e| Failure::new(exit::FAILED, format!("locate robotctl: {e}")))?; + let base = |command: &str| { + vec![ + "--socket".to_owned(), + socket.display().to_string(), + "--robot-socket".to_owned(), + robot_socket.display().to_string(), + "--config-socket".to_owned(), + config_socket.display().to_string(), + command.to_owned(), + "--json".to_owned(), + ] + }; + let health = support::command(&exe.display().to_string(), &base("health")); + let version = support::command(&exe.display().to_string(), &base("version")); + let units = support::command( + "systemctl", + &[ + "--no-pager".to_owned(), + "--full".to_owned(), + "--plain".to_owned(), + "status".to_owned(), + "robotd.service".to_owned(), + "updaterd.service".to_owned(), + "mediad.service".to_owned(), + "configd.service".to_owned(), + ], + ); + let journal = support::command( + "journalctl", + &[ + "--utc".to_owned(), + "--no-pager".to_owned(), + "--unit=robotd.service".to_owned(), + "--unit=updaterd.service".to_owned(), + "--unit=mediad.service".to_owned(), + "--unit=configd.service".to_owned(), + "-n".to_owned(), + "300".to_owned(), + ], + ); + let update_history = support::file(Path::new("/var/lib/robot/updater/update-log.jsonl")); + let bundle = support::render(&[ + ("robotctl health --json", &health), + ("robotctl version --json", &version), + ("unit status", &units), + ("recent daemon journal", &journal), + ("update history", &update_history), + ]); + std::fs::write(output, bundle) + .map_err(|e| Failure::new(exit::FAILED, format!("write {}: {e}", output.display())))?; + println!("support bundle written to {}", output.display()); + Ok(()) +} + +/// Fetch one latest frame from mediad's local endpoint. Its JSON-RPC header is followed by the +/// exact binary byte count it names, so pixels never pass through a base64 control message. +fn run_media(socket: &Path, command: MediaCommand) -> Result<(), Failure> { + let MediaCommand::Frame { output } = command; + let stream = UnixStream::connect(socket) + .map_err(|e| Failure::new(exit::UNREACHABLE, unreachable_hint("mediad", socket, &e)))?; + let mut writer = stream.try_clone().map_err(|e| { + Failure::new( + exit::FAILED, + format!("could not split the media socket: {e}"), + ) + })?; + let request = proto::Request { + jsonrpc: proto::JSONRPC_VERSION.to_owned(), + id: Some(proto::Id::Number(1)), + method: proto::method::MEDIA_FRAME.to_owned(), + params: Some(serde_json::json!({})), + }; + let mut request = serde_json::to_vec(&request) + .map_err(|e| Failure::new(exit::FAILED, format!("could not encode media.frame: {e}")))?; + request.push(b'\n'); + writer + .write_all(&request) + .and_then(|()| writer.flush()) + .map_err(|e| Failure::new(exit::UNREACHABLE, format!("could not request a frame: {e}")))?; + + let mut reader = BufReader::new(stream); + let mut line = String::new(); + if reader.read_line(&mut line).map_err(|e| { + Failure::new( + exit::UNREACHABLE, + format!("could not read frame header: {e}"), + ) + })? == 0 + { + return Err(Failure::new( + exit::UNREACHABLE, + "mediad closed the media socket".into(), + )); + } + let response: proto::Response = serde_json::from_str(line.trim()).map_err(|e| { + Failure::new( + exit::FAILED, + format!("mediad sent an invalid frame header: {e}"), + ) + })?; + if let Some(error) = response.error { + return Err(Failure::new( + exit::FAILED, + format!("media.frame: {}", error.message), + )); + } + let result = response + .result + .ok_or_else(|| Failure::new(exit::FAILED, "mediad returned no frame metadata".into()))?; + let bytes = result + .get("bytes") + .and_then(serde_json::Value::as_u64) + .and_then(|bytes| usize::try_from(bytes).ok()) + .ok_or_else(|| { + Failure::new( + exit::FAILED, + "mediad returned an invalid frame length".into(), + ) + })?; + let mut data = vec![0; bytes]; + reader.read_exact(&mut data).map_err(|e| { + Failure::new( + exit::UNREACHABLE, + format!("frame ended before {bytes} bytes: {e}"), + ) + })?; + std::fs::write(&output, data) + .map_err(|e| Failure::new(exit::FAILED, format!("write {}: {e}", output.display())))?; + println!( + "frame written to {} ({}×{}, {}, {} bytes, captured {})", + output.display(), + result + .get("width") + .and_then(serde_json::Value::as_u64) + .unwrap_or(0), + result + .get("height") + .and_then(serde_json::Value::as_u64) + .unwrap_or(0), + result + .get("format") + .and_then(serde_json::Value::as_str) + .unwrap_or("unknown"), + bytes, + result + .get("captured_at_unix_us") + .and_then(serde_json::Value::as_u64) + .unwrap_or(0), + ); + Ok(()) +} + fn run(cli: Cli) -> Result<(), Failure> { let command = match cli.namespace { Namespace::Health { json } => { @@ -2798,6 +2992,10 @@ fn run(cli: Cli) -> Result<(), Failure> { Namespace::Version { json } => { return run_version(&cli.socket, &cli.robot_socket, &cli.config_socket, json); } + Namespace::Support { output } => { + return run_support(&cli.socket, &cli.robot_socket, &cli.config_socket, &output); + } + Namespace::Media { command } => return run_media(&cli.media_socket, command), Namespace::Monitor { hz, json } => { return monitor::run( &cli.robot_socket, diff --git a/robotctl/src/monitor.rs b/robotctl/src/monitor.rs index 232767ca..27f8328f 100644 --- a/robotctl/src/monitor.rs +++ b/robotctl/src/monitor.rs @@ -4199,6 +4199,7 @@ mod tests { }, head: [0.0; 4], policy: "stand".to_owned(), + policy_enabled: Some(false), safety: proto::SafetyState { fallen: false, limp: false, @@ -4211,6 +4212,7 @@ mod tests { }, joints: vec![0.0; proto::JOINT_NAMES.len()], targets: vec![0.0; proto::JOINT_NAMES.len()], + policy_action: vec![], odom: proto::OdomState::default(), theremin: None, chorale: None, diff --git a/robotctl/src/support.rs b/robotctl/src/support.rs new file mode 100644 index 00000000..321dbc80 --- /dev/null +++ b/robotctl/src/support.rs @@ -0,0 +1,116 @@ +//! A bounded, redacted support report for a robot somebody cannot inspect over SSH. + +use std::fmt::Write as _; +use std::path::Path; +use std::process::Command; + +const MAX_SECTION_BYTES: usize = 256 * 1024; + +/// Run a local diagnostic command without a shell. Failure is evidence, not a reason to +/// abandon the bundle: the useful case is precisely when a daemon is absent. +pub fn command(program: &str, args: &[String]) -> String { + match Command::new(program).args(args).output() { + Ok(output) => { + let mut text = String::from_utf8_lossy(&output.stdout).into_owned(); + text.push_str(&String::from_utf8_lossy(&output.stderr)); + if !output.status.success() { + let _ = writeln!(text, "[command exited {}]", output.status); + } + bounded(text) + } + Err(error) => format!("[could not run {program}: {error}]"), + } +} + +pub fn file(path: &Path) -> String { + match std::fs::read_to_string(path) { + Ok(text) => bounded(text), + Err(error) => format!("[could not read {}: {error}]", path.display()), + } +} + +fn bounded(mut text: String) -> String { + if text.len() > MAX_SECTION_BYTES { + text.truncate(MAX_SECTION_BYTES); + text.push_str("\n[truncated]\n"); + } + text +} + +/// Lines that can plausibly carry a credential. A support bundle must be useful to send +/// to somebody else, so failing closed here beats preserving one more journal line. +pub fn redact(text: &str) -> String { + text.lines() + .map(|line| { + let lower = line.to_ascii_lowercase(); + if [ + "password", + "passphrase", + "psk", + "token", + "authorization", + "secret", + "credential", + "api-key", + "api_key", + "access_key", + "sessionid", + "set-cookie", + ] + .iter() + .any(|needle| lower.contains(needle)) + { + "[redacted: possible credential]".to_owned() + } else { + line.to_owned() + } + }) + .collect::>() + .join("\n") +} + +/// Assemble labelled, already-bounded command output into one portable text report. +pub fn render(sections: &[(&str, &str)]) -> String { + let mut out = String::from("microduck support bundle v1\n"); + for (name, body) in sections { + let _ = writeln!(out, "\n===== {name} ====="); + out.push_str(&redact(body)); + out.push('\n'); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn credentials_are_removed_case_insensitively() { + let out = redact( + "normal\nPSK=hunter2\nAuthorization: Bearer abc\n\ + Credential: value\nX-Api-Key: value\naccess_key=value\n\ + Set-Cookie: session=value\nother", + ); + assert_eq!( + out, + "normal\n[redacted: possible credential]\n[redacted: possible credential]\n\ + [redacted: possible credential]\n[redacted: possible credential]\n\ + [redacted: possible credential]\n[redacted: possible credential]\nother" + ); + } + + #[test] + fn ordinary_diagnostics_are_preserved() { + assert_eq!( + redact("robotd active\nbattery 7.4 V\nupdate completed"), + "robotd active\nbattery 7.4 V\nupdate completed" + ); + } + + #[test] + fn sections_are_labelled_and_redacted() { + let out = render(&[("health", "ok"), ("journal", "token=abc")]); + assert!(out.contains("===== health =====\nok")); + assert!(out.contains("===== journal =====\n[redacted: possible credential]")); + } +} diff --git a/robotd/src/control.rs b/robotd/src/control.rs index bd258ddd..375415e2 100644 --- a/robotd/src/control.rs +++ b/robotd/src/control.rs @@ -119,6 +119,9 @@ impl Default for SkillTuning { #[derive(Debug, Clone, Copy, PartialEq)] pub struct Step { pub targets: [f64; NUM_JOINTS], + /// Raw network output before action scaling or target filtering. This is distinct from + /// `targets`: training on filtered joint targets would label actuator dynamics as policy. + pub action: [f32; ACTION_LEN], /// Which network drove, as the wire label: `walk`, `stand`, `ground_pick`, `kick_left`, /// `kick_right`, `sit`, `rise`. pub label: &'static str, @@ -473,6 +476,7 @@ impl Controller { Ok(Step { targets, + action, label, gain, busy: self.busy(), diff --git a/robotd/src/main.rs b/robotd/src/main.rs index aa7c46bc..14cb7e2f 100644 --- a/robotd/src/main.rs +++ b/robotd/src/main.rs @@ -28,8 +28,8 @@ mod theremin; use std::path::PathBuf; use std::process::ExitCode; -use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, Ordering}; +use std::sync::{Arc, mpsc}; use std::time::{Duration, Instant}; use arc_swap::{ArcSwap, ArcSwapOption}; @@ -52,6 +52,18 @@ use params::{Mode, Params}; /// call instead of powering off the machine running them. type PowerOff = Arc; +/// An imitation-learning label exists only for a tick driven by an ONNX policy. Holding, homing +/// and limp-fall targets are generated by the daemon, not demonstrations to train back into it. +fn recorded_policy_action(action: Option<[f32; duck_control::ACTION_LEN]>) -> Vec { + action.map_or_else(Vec::new, |action| action.to_vec()) +} + +/// A reload may start only after the policy is disabled and no earlier candidate is loading. +/// Keeping this as a pure predicate makes the safety boundary testable without an ONNX runtime. +fn may_start_reload(policy_enabled: bool, worker_running: bool, requested: bool) -> bool { + !policy_enabled && !worker_running && requested +} + /// Model API version this build implements (`updater-design.md` §5.5). Bump when the /// sensor-input / actuator-output contract a model sees changes. const MODEL_API: u32 = 1; @@ -372,6 +384,9 @@ struct RobotState { imu_stale_run: AtomicU64, imu_ready: AtomicBool, shutdown: AtomicBool, + /// SIGHUP asks for a policy reload. It is a level so repeated signals coalesce while a + /// candidate is loading; the control loop alone clears it when it starts that work. + reload_requested: AtomicBool, /// Fan-out for `robot.state`. Bounded and lossy by design — see [`STATE_BUFFER`]. state_tx: tokio::sync::broadcast::Sender, /// What `btd` should be advertising, published when it changes. @@ -457,6 +472,7 @@ impl RobotState { imu_stale_run: AtomicU64::new(0), imu_ready: AtomicBool::new(false), shutdown: AtomicBool::new(false), + reload_requested: AtomicBool::new(false), state_tx: tokio::sync::broadcast::Sender::new(STATE_BUFFER), chorale_tx: tokio::sync::broadcast::Sender::new(8), policy_error: ArcSwapOption::empty(), @@ -718,6 +734,7 @@ async fn main() -> ExitCode { Arc::clone(&intents), args.socket.clone(), ); + let reload_watcher = tokio::spawn(watch_reload_requests(Arc::clone(&state))); let mut code = ExitCode::SUCCESS; tokio::select! { result = serving => { @@ -732,6 +749,7 @@ async fn main() -> ExitCode { // Ask the loop to stop and let it finish the tick it is in, rather than aborting // mid-transaction and leaving a half-written packet on the bus. state.shutdown.store(true, Ordering::Relaxed); + reload_watcher.abort(); let _ = control.join(); let _ = std::fs::remove_file(&args.socket); code @@ -1057,68 +1075,61 @@ async fn adopt_startup_pose( /// A policy that was *not wanted* is healthy; one that was wanted and could not be loaded is not. /// Collapsing those two would either make a bench robot look broken or let a release with an /// unusable bundle pass the health gate. +fn load_controller(policy_cfg: ¶ms::ResolvedPolicy) -> Result, String> { + if !policy_cfg.enabled { + return Ok(None); + } + let tuning = Tuning { + action_scale: policy_cfg.action_scale, + standing_action_scale: policy_cfg.standing_action_scale, + standing_gain_ratio: policy_cfg.standing_gain_ratio, + gain: policy_cfg.gain, + head_lowpass: policy_cfg.head_lowpass, + legs_lowpass: policy_cfg.legs_lowpass, + }; + let skills = SkillTuning { + ground_pick_period: policy_cfg.ground_pick_period, + ground_pick_action_scale: policy_cfg.ground_pick_action_scale, + ground_pick_gain_ratio: policy_cfg.ground_pick_gain_ratio, + kick_duration: policy_cfg.kick_duration, + roulade_duration: policy_cfg.roulade_duration, + roulade_action_scale: policy_cfg.roulade_action_scale, + roulade_gain_ratio: policy_cfg.roulade_gain_ratio, + }; + let paths = PolicyPaths { + walk: policy_cfg.walk.clone(), + stand: policy_cfg.stand.clone(), + sitstand: policy_cfg.sitstand.clone(), + ground_pick: policy_cfg.ground_pick.clone(), + kick_left: policy_cfg.kick_left.clone(), + kick_right: policy_cfg.kick_right.clone(), + roulade: policy_cfg.roulade.clone(), + }; + let mut policy = Policy::load(&paths, DEFAULT_STANDING_THRESHOLD).map_err(|e| e.to_string())?; + if policy_cfg.mode == Mode::Roller { + policy.set_standing_disabled(true); + } + Ok(Some(Controller::new(policy, tuning, skills))) +} + fn build_controller( policy_cfg: ¶ms::ResolvedPolicy, limp_fall: bool, state: &RobotState, ) -> Option { - if !policy_cfg.enabled { - tracing::warn!("policy disabled; holding the startup pose"); - return None; - } - { - let tuning = Tuning { - action_scale: policy_cfg.action_scale, - standing_action_scale: policy_cfg.standing_action_scale, - standing_gain_ratio: policy_cfg.standing_gain_ratio, - gain: policy_cfg.gain, - head_lowpass: policy_cfg.head_lowpass, - legs_lowpass: policy_cfg.legs_lowpass, - }; - let skills = SkillTuning { - ground_pick_period: policy_cfg.ground_pick_period, - ground_pick_action_scale: policy_cfg.ground_pick_action_scale, - ground_pick_gain_ratio: policy_cfg.ground_pick_gain_ratio, - kick_duration: policy_cfg.kick_duration, - roulade_duration: policy_cfg.roulade_duration, - roulade_action_scale: policy_cfg.roulade_action_scale, - roulade_gain_ratio: policy_cfg.roulade_gain_ratio, - }; - let paths = PolicyPaths { - walk: policy_cfg.walk.clone(), - stand: policy_cfg.stand.clone(), - sitstand: policy_cfg.sitstand.clone(), - ground_pick: policy_cfg.ground_pick.clone(), - kick_left: policy_cfg.kick_left.clone(), - kick_right: policy_cfg.kick_right.clone(), - roulade: policy_cfg.roulade.clone(), - }; - match Policy::load(&paths, DEFAULT_STANDING_THRESHOLD) { - Ok(mut policy) => { - // Roller mode has no standing network — command magnitude stops selecting - // it. Nothing else reserves it: limp-fall hands back by *letting* the - // standing network be selected, which is what stands the robot up. - if policy_cfg.mode == Mode::Roller { - policy.set_standing_disabled(true); - } - tracing::warn!( - mode = policy_cfg.mode.as_str(), - walk = %policy_cfg.walk.display(), - stand = ?policy_cfg.stand.as_ref().map(|p| p.display().to_string()), - sitstand = ?policy_cfg.sitstand.as_ref().map(|p| p.display().to_string()), - ground_pick = ?policy_cfg.ground_pick.as_ref().map(|p| p.display().to_string()), - kicks = policy_cfg.kick_left.is_some() || policy_cfg.kick_right.is_some(), - roulade = ?policy_cfg.roulade.as_ref().map(|p| p.display().to_string()), - limp_fall, - "policy loaded" - ); - Some(Controller::new(policy, tuning, skills)) - } - Err(e) => { - tracing::error!(error = %e, "policy unavailable; holding the pose"); - state.policy_error.store(Some(Arc::new(e.to_string()))); - None - } + match load_controller(policy_cfg) { + Ok(Some(controller)) => { + tracing::warn!(mode = policy_cfg.mode.as_str(), walk = %policy_cfg.walk.display(), limp_fall, "policy loaded"); + Some(controller) + } + Ok(None) => { + tracing::warn!("policy disabled; holding the startup pose"); + None + } + Err(error) => { + tracing::error!(error = %error, "policy unavailable; holding the pose"); + state.policy_error.store(Some(Arc::new(error))); + None } } } @@ -1169,6 +1180,9 @@ async fn control_loop( // Loaded once here and again on a mode switch — see `build_controller`. let mut controller = build_controller(&policy_cfg, params.safety.limp_fall, &state); + // The candidate loads off this thread. A one-slot channel is enough: SIGHUP is a level and + // a newer request after a worker starts will schedule one more load, never a backlog. + let mut reload_worker: Option, String>>> = None; tracing::warn!( joints = NUM_JOINTS, @@ -1377,6 +1391,45 @@ async fn control_loop( state.fallen.store(safety.fallen(), Ordering::Relaxed); let snapshot = intents.snapshot(); + + // A model reload is deliberately more conservative than a mode switch: do not begin it + // while the policy is enabled. The candidate's ONNX sessions are built on another thread; + // only the finished controller crosses into this loop, at this tick boundary. + if !snapshot.enabled { + if let Some(worker) = reload_worker.as_ref() { + match worker.try_recv() { + Ok(Ok(candidate)) => { + controller = candidate; + state.policy_error.store(None); + reload_worker = None; + tracing::warn!("policy reload accepted while disabled"); + } + Ok(Err(error)) => { + reload_worker = None; + tracing::error!(error = %error, "policy reload rejected; keeping current controller"); + } + Err(mpsc::TryRecvError::Empty) => {} + Err(mpsc::TryRecvError::Disconnected) => { + reload_worker = None; + tracing::error!("policy reload worker stopped without a result"); + } + } + } + if may_start_reload( + snapshot.enabled, + reload_worker.is_some(), + state.reload_requested.load(Ordering::Relaxed), + ) && state.reload_requested.swap(false, Ordering::Relaxed) + { + let candidate_cfg = policy_cfg.clone(); + let (tx, rx) = mpsc::sync_channel(1); + std::thread::spawn(move || { + let _ = tx.send(load_controller(&candidate_cfg)); + }); + reload_worker = Some(rx); + tracing::info!("loading policy reload candidate while disabled"); + } + } let (gated, deadman) = safety.gate(snapshot.command, snapshot.twist_age); let mut limits: Vec = deadman.into_iter().collect(); @@ -1933,7 +1986,7 @@ async fn control_loop( 1.0 }; - let (mut targets, gain, moving, policy_label) = match (driving, sensors.as_ref()) { + let (mut targets, action, gain, moving, policy_label) = match (driving, sensors.as_ref()) { // The limp-fall sequence, before anything else — `driving` is false throughout, // so without this it would fall through to the hold branch and the robot would // be commanded its pre-fall pose at walking gain, which is precisely the thing @@ -1949,6 +2002,7 @@ async fn control_loop( // gain is a motor pushing back against the floor. LimpFall::Limp { .. } => ( coast.known_positions(hold), + None, params.safety.gain_limp, true, "limp_fall", @@ -1960,6 +2014,7 @@ async fn control_loop( limp_fall .pose_target(tick_start, limp_fall_pose) .unwrap_or(DEFAULT_POSITION), + None, params.safety.limp_fall_pose_gain, true, "limp_pose", @@ -1971,6 +2026,7 @@ async fn control_loop( match controller.step(sensors, &command, snapshot.pose.active, dt, scale_mult) { Ok(step) => ( step.targets, + Some(step.action), step.gain, // A scripted move is motion whatever the twist says; so is walking. step.busy || command.twist_magnitude() > 0.0, @@ -1978,7 +2034,7 @@ async fn control_loop( ), Err(e) => { tracing::warn!(error = %e, "inference failed; holding"); - (hold, policy_cfg.gain, false, "held") + (hold, None, policy_cfg.gain, false, "held") } } } @@ -1988,11 +2044,12 @@ async fn control_loop( bringup .homing_target(tick_start) .expect("just checked it is Some"), + None, policy_cfg.gain, true, "homing", ), - _ => (hold, policy_cfg.gain, false, "held"), + _ => (hold, None, policy_cfg.gain, false, "held"), }; state.moving.store(moving, Ordering::Relaxed); @@ -2212,6 +2269,8 @@ async fn control_loop( }, joints: sensors.positions.to_vec(), targets: targets.to_vec(), + policy_enabled: Some(snapshot.enabled), + policy_action: recorded_policy_action(action), odom: proto::OdomState { position: odometry.position(), yaw: odometry.yaw(), @@ -3040,10 +3099,46 @@ async fn shutdown() { } } +/// Coalesce SIGHUP into one request the control loop can handle at a safe tick boundary. +async fn watch_reload_requests(state: Arc) { + use tokio::signal::unix::{SignalKind, signal}; + let Ok(mut hup) = signal(SignalKind::hangup()) else { + tracing::warn!("cannot listen for SIGHUP"); + return; + }; + while hup.recv().await.is_some() { + state.reload_requested.store(true, Ordering::Relaxed); + tracing::info!("policy reload requested; waiting until policy is disabled"); + } +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn only_policy_driven_ticks_have_imitation_labels() { + assert!(recorded_policy_action(None).is_empty()); + assert_eq!( + recorded_policy_action(Some([0.0; duck_control::ACTION_LEN])), + vec![0.0; duck_control::ACTION_LEN] + ); + } + + #[test] + fn a_reload_is_deferred_until_policy_control_is_off_and_the_worker_is_free() { + assert!( + !may_start_reload(true, false, true), + "never reload while driving" + ); + assert!(!may_start_reload(false, true, true), "one worker at a time"); + assert!( + !may_start_reload(false, false, false), + "signals are requests, not polls" + ); + assert!(may_start_reload(false, false, true)); + } + /// The limp-fall pose ramp: starts where the robot landed, ends at the standing pose, /// and reports itself finished rather than pinning at the end — the state machine reads /// `None` as "hand back to the policy". diff --git a/scripts/export-lerobot-local.py b/scripts/export-lerobot-local.py new file mode 100644 index 00000000..08de5adb --- /dev/null +++ b/scripts/export-lerobot-local.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +"""Convert validated MicroDuck staging data to a local LeRobot dataset (no Hub upload).""" +import argparse, json, subprocess, sys +from pathlib import Path + +def rgb_uyvy(raw, w, h): + import numpy as np + x = np.frombuffer(raw, dtype=np.uint8).reshape(h, w // 2, 4).astype(np.float32) + u, y0, v, y1 = (x[..., i] for i in range(4)) + y = np.stack((y0, y1), -1).reshape(h, w); u = np.repeat(u, 2, -1); v = np.repeat(v, 2, -1) + return np.clip(np.stack((y + 1.402*(v-128), y - .344136*(u-128)-.714136*(v-128), y + 1.772*(u-128)), -1), 0, 255).astype(np.uint8) + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("staging", type=Path) + p.add_argument("--root", type=Path, required=True) + p.add_argument("--repo-id", default="local/microduck") + p.add_argument("--max-frames", type=int, default=300, + help="refuse an export with more labelled frames (default: 300)") + p.add_argument("--max-estimated-bytes", type=int, default=1024 * 1024 * 1024, + help="refuse when uncompressed RGB would exceed this budget (default: 1 GiB)") + p.add_argument("--dry-run", action="store_true", + help="validate and print the export budget without importing LeRobot") + p.add_argument("--confirm-export", action="store_true", + help="required to create the local dataset after reviewing --dry-run") + a = p.parse_args() + if a.max_frames <= 0 or a.max_estimated_bytes <= 0: + p.error("export limits must be positive") + validator = Path(__file__).with_name("validate-lerobot-staging.py") + subprocess.run([sys.executable, str(validator), str(a.staging)], check=True) + rows = [json.loads(x) for x in (a.staging / "samples.jsonl").read_text().splitlines()] + labelled = [row for row in rows if len(row["state"].get("policy_action", [])) == 14] + if not labelled: + p.error("no policy-labelled frames; refusing an empty training dataset") + estimated_bytes = sum( + row["camera"]["width"] * row["camera"]["height"] * 3 for row in labelled + ) + print(f"labelled frames: {len(labelled)}; estimated uncompressed RGB: {estimated_bytes} bytes") + if len(labelled) > a.max_frames: + p.error(f"labelled frame count exceeds --max-frames ({len(labelled)}/{a.max_frames})") + if estimated_bytes > a.max_estimated_bytes: + p.error(f"estimated RGB bytes exceed --max-estimated-bytes ({estimated_bytes}/{a.max_estimated_bytes})") + if a.dry_run: + print("dry run only; no LeRobot dataset was created") + return + if not a.confirm_export: + p.error("run with --dry-run, then pass --confirm-export to create the local dataset") + if a.root.exists() and any(a.root.iterdir()): + p.error(f"destination {a.root} is not empty; choose a new local directory") + + from lerobot.datasets.lerobot_dataset import LeRobotDataset + import numpy as np + first = labelled[0]; c = first["camera"]; features = { + "observation.image": {"dtype":"image", "shape":(c["height"], c["width"], 3), "names":["height","width","channel"]}, + "observation.state": {"dtype":"float32", "shape":(15,), "names":None}, + "action": {"dtype":"float32", "shape":(14,), "names":None}, } + ds = LeRobotDataset.create(repo_id=a.repo_id, root=a.root, fps=json.loads((a.staging/"meta.json").read_text())["fps"], features=features, robot_type="microduck", use_videos=False) + for row in labelled: + s, c = row["state"], row["camera"]; action = s.get("policy_action", []) + image = rgb_uyvy((a.staging/row["frame"]).read_bytes(), c["width"], c["height"]) + ds.add_frame({"observation.image":image, "observation.state":np.asarray(s["joints"], dtype=np.float32), "action":np.asarray(action, dtype=np.float32), "task":json.loads((a.staging/"meta.json").read_text())["task"]}) + ds.save_episode(); ds.finalize() + print(f"wrote local LeRobot dataset to {a.root}; no upload was requested") +if __name__ == "__main__": main() diff --git a/scripts/model-update-preflight.py b/scripts/model-update-preflight.py new file mode 100644 index 00000000..74aeeae3 --- /dev/null +++ b/scripts/model-update-preflight.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python3 +"""Refuse an unsafe model-update dry run; never changes robot state itself.""" +import argparse +import json +import socket +import subprocess +import sys + + +def robot_state(socket_path): + with socket.socket(socket.AF_UNIX) as conn: + conn.settimeout(3) + conn.connect(socket_path) + request = { + "jsonrpc": "2.0", "id": 1, "method": "robot.subscribe", "params": {"hz": 1}, + } + conn.sendall((json.dumps(request, separators=(",", ":")) + "\n").encode()) + reader = conn.makefile("rb") + acknowledgement = json.loads(reader.readline()) + if "error" in acknowledgement: + raise RuntimeError(acknowledgement["error"].get("message", "subscribe refused")) + state = json.loads(reader.readline()).get("params") + if not isinstance(state, dict): + raise RuntimeError("robotd did not send a state frame") + return state + + +def require_safe_state(state): + # Missing is deliberately a refusal: an older daemon cannot prove this is the new, explicit + # policy-disabled state. `held` is insufficient because a live policy can hold on zero input. + if state.get("policy_enabled") is not False: + raise RuntimeError("policy is enabled or this robotd cannot report policy_enabled; disable policy control first") + safety = state.get("safety", {}) + if safety.get("fallen") or safety.get("limp"): + raise RuntimeError("robot is fallen or limp; recover it before changing a policy") + + +def require_healthy(robotctl): + result = subprocess.run([robotctl, "health", "--json"], text=True, capture_output=True) + if result.returncode: + raise RuntimeError(f"robotctl health failed: {result.stderr.strip() or result.stdout.strip()}") + report = json.loads(result.stdout) + if report.get("robot", {}).get("healthy") is not True: + raise RuntimeError("robotd is not healthy; model update dry run is not safe") + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("component", help="configured model component, for example model-walk") + parser.add_argument("--robotctl", default="robotctl") + parser.add_argument("--robot-socket", default="/run/robotd.sock") + parser.add_argument("--from", dest="from_dir", help="signed local artifact directory to dry-run") + args = parser.parse_args() + if not args.component.startswith("model-"): + parser.error("component must start with model-") + + try: + require_safe_state(robot_state(args.robot_socket)) + require_healthy(args.robotctl) + except (OSError, ValueError, RuntimeError) as error: + print(f"REFUSED: {error}", file=sys.stderr) + return 2 + + command = [args.robotctl, "update", "apply", args.component, "--dry-run"] + if args.from_dir: + command.extend(["--from", args.from_dir]) + print("preflight passed: policy is disabled and robotd is healthy; verifying signed artifact") + return subprocess.run(command).returncode + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/observe-model-reload.py b/scripts/observe-model-reload.py new file mode 100644 index 00000000..37dbd544 --- /dev/null +++ b/scripts/observe-model-reload.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python3 +"""Record a read-only safety observation around a model update or reload.""" +import argparse +import json +import socket +import subprocess +import sys +import time +from pathlib import Path + + +def subscribe(socket_path, seconds): + deadline = time.monotonic() + seconds + frames = [] + with socket.socket(socket.AF_UNIX) as conn: + conn.settimeout(3) + conn.connect(socket_path) + request = { + "jsonrpc": "2.0", "id": 1, "method": "robot.subscribe", "params": {"hz": 10}, + } + conn.sendall((json.dumps(request, separators=(",", ":")) + "\n").encode()) + reader = conn.makefile("rb") + acknowledgement = json.loads(reader.readline()) + if "error" in acknowledgement: + raise RuntimeError(acknowledgement["error"].get("message", "subscribe refused")) + while time.monotonic() < deadline: + state = json.loads(reader.readline()).get("params") + if not isinstance(state, dict): + raise RuntimeError("robotd sent an invalid state frame") + frames.append(state) + if not frames: + raise RuntimeError("robotd sent no state frames") + return frames + + +def health(robotctl): + result = subprocess.run([robotctl, "health", "--json"], text=True, capture_output=True) + if result.returncode: + raise RuntimeError(f"robotctl health failed: {result.stderr.strip() or result.stdout.strip()}") + return json.loads(result.stdout) + + +def evaluate(frames, report, max_missed): + violations = [] + missed = [frame.get("loop", {}).get("missed") for frame in frames] + hz = [frame.get("loop", {}).get("hz") for frame in frames] + if any(frame.get("policy_enabled") is not False for frame in frames): + violations.append("policy control was enabled or could not be verified as disabled") + if any(frame.get("safety", {}).get("fallen") for frame in frames): + violations.append("robot reported fallen") + if any(frame.get("safety", {}).get("limp") for frame in frames): + violations.append("robot reported limp") + if not all(isinstance(value, int) for value in missed): + violations.append("control-loop missed-tick count was unavailable") + elif max(missed) - min(missed) > max_missed: + violations.append(f"missed ticks increased by {max(missed) - min(missed)} (limit {max_missed})") + if not all(isinstance(value, (int, float)) for value in hz): + violations.append("control-loop rate was unavailable") + report.update({ + "frames": len(frames), + "policy_enabled": False, + "missed_start": missed[0] if missed else None, + "missed_end": missed[-1] if missed else None, + "min_hz": min(hz) if hz and all(isinstance(value, (int, float)) for value in hz) else None, + "violations": violations, + }) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--seconds", type=float, default=10, help="observation window (default: 10)") + parser.add_argument("--max-missed", type=int, default=0, + help="maximum allowed increase in missed ticks (default: 0)") + parser.add_argument("--robotctl", default="robotctl") + parser.add_argument("--robot-socket", default="/run/robotd.sock") + parser.add_argument("--output", type=Path, default=Path("/var/tmp/microduck-model-reload-observation.json")) + args = parser.parse_args() + if args.seconds <= 0 or args.max_missed < 0: + parser.error("--seconds must be positive and --max-missed cannot be negative") + + report = {"format": "microduck-model-reload-observation-v1", "started_at_unix": time.time()} + try: + frames = subscribe(args.robot_socket, args.seconds) + evaluate(frames, report, args.max_missed) + report["health"] = health(args.robotctl) + if report["health"].get("robot", {}).get("healthy") is not True: + report["violations"].append("robotd health was not healthy") + except (OSError, ValueError, RuntimeError) as error: + report["violations"] = [str(error)] + + args.output.write_text(json.dumps(report, indent=2) + "\n") + if report["violations"]: + print(f"REJECTED: {args.output}", *report["violations"], sep="\n- ", file=sys.stderr) + return 1 + print(f"OK: observed {report['frames']} disabled-policy frames; report: {args.output}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/record-lerobot-local.py b/scripts/record-lerobot-local.py new file mode 100644 index 00000000..dcb50671 --- /dev/null +++ b/scripts/record-lerobot-local.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 +"""Record MicroDuck observations locally; never uploads or sends control commands.""" +import argparse, json, os, shutil, socket, time +from pathlib import Path + +MEDIA = "/run/mediad/media.sock" +ROBOT = "/run/robotd.sock" + +def line(sock, value): + sock.sendall((json.dumps(value, separators=(",", ":")) + "\n").encode()) + +def frame(sock): + line(sock, {"jsonrpc":"2.0", "id":1, "method":"media.frame", "params":{}}) + f = sock.makefile("rb") + header = json.loads(f.readline()) + if "error" in header: raise RuntimeError(header["error"]["message"]) + meta = header["result"] + data = f.read(meta["bytes"]) + if len(data) != meta["bytes"]: raise RuntimeError("short camera frame") + return meta, data + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--task", required=True) + p.add_argument("--seconds", type=float, default=30) + p.add_argument("--hz", type=float, default=5) + p.add_argument("--root", type=Path, default=Path("/var/lib/robot/datasets")) + p.add_argument("--max-samples", type=int, default=300, + help="hard cap on frames written (default: 300)") + p.add_argument("--max-bytes", type=int, default=512 * 1024 * 1024, + help="hard cap on raw frame bytes written (default: 512 MiB)") + p.add_argument("--min-free-bytes", type=int, default=1024 * 1024 * 1024, + help="stop before free space drops below this reserve (default: 1 GiB)") + p.add_argument("--max-episodes", type=int, default=20, + help="refuse to create another episode once this many exist (default: 20)") + a = p.parse_args() + if not 0 < a.hz <= 10: p.error("--hz must be between 0 and 10") + if a.seconds <= 0: p.error("--seconds must be positive") + if min(a.max_samples, a.max_bytes, a.min_free_bytes, a.max_episodes) <= 0: + p.error("all recording limits must be positive") + a.root.mkdir(parents=True, exist_ok=True) + episodes = [path for path in a.root.iterdir() if path.is_dir() and (path / "meta.json").is_file()] + if len(episodes) >= a.max_episodes: + p.error(f"episode cap reached ({len(episodes)}/{a.max_episodes}); review or archive recordings first") + if shutil.disk_usage(a.root).free < a.min_free_bytes: + p.error("free-space reserve is already below --min-free-bytes; free space before recording") + out = a.root / time.strftime("microduck-%Y%m%d-%H%M%S") + out.mkdir(parents=True, exist_ok=False) + (out / "frames").mkdir() + manifest = {"format":"microduck-lerobot-staging-v1", "task":a.task, "fps":a.hz, + "upload":"disabled", "camera":"UYVY", "action":"policy_action", + "guardrails":{"max_samples":a.max_samples, "max_bytes":a.max_bytes, + "min_free_bytes":a.min_free_bytes, "max_episodes":a.max_episodes}} + (out / "meta.json").write_text(json.dumps(manifest, indent=2) + "\n") + end, n, frame_bytes, stop_reason = time.monotonic() + a.seconds, 0, 0, "duration" + with (out / "samples.jsonl").open("x") as samples: + while time.monotonic() < end: + if n >= a.max_samples: + stop_reason = "max_samples" + break + started = time.monotonic() + with socket.socket(socket.AF_UNIX) as media: + media.connect(MEDIA); meta, pixels = frame(media) + if frame_bytes + len(pixels) > a.max_bytes: + stop_reason = "max_bytes" + break + if shutil.disk_usage(a.root).free - len(pixels) < a.min_free_bytes: + stop_reason = "min_free_bytes" + break + name = f"frames/{n:06d}.uyvy" + (out / name).write_bytes(pixels) + # State is intentionally sampled after the image and carries its own monotonic t. + # The converter pairs by the image's capture timestamp and preserves this skew. + with socket.socket(socket.AF_UNIX) as robot: + robot.connect(ROBOT) + line(robot, {"jsonrpc":"2.0", "id":1, "method":"robot.subscribe", "params":{"hz":1}}) + r = robot.makefile("rb") + r.readline() # subscription acknowledgement + state = json.loads(r.readline())["params"] + samples.write(json.dumps({"frame":name, "camera":meta, "state":state}, separators=(",", ":")) + "\n") + samples.flush(); os.fsync(samples.fileno()); n += 1; frame_bytes += len(pixels) + time.sleep(max(0, 1/a.hz - (time.monotonic()-started))) + manifest["samples"] = n + manifest["frame_bytes"] = frame_bytes + manifest["stop_reason"] = stop_reason + (out / "meta.json").write_text(json.dumps(manifest, indent=2) + "\n") + print(f"recorded {n} local samples ({frame_bytes} bytes, stopped by {stop_reason}) in {out}") + +if __name__ == "__main__": main() diff --git a/scripts/record-lerobot-preflight.sh b/scripts/record-lerobot-preflight.sh new file mode 100644 index 00000000..cdd063f2 --- /dev/null +++ b/scripts/record-lerobot-preflight.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +# Verify the local-only data path before the first real MicroDuck recording. +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +python3 "$ROOT/scripts/test-lerobot-staging.py" +printf '%s\n' 'Synthetic preflight passed. On the robot, collect a short episode with:' +printf '%s\n' ' sudo python3 scripts/record-lerobot-local.py --task "" --seconds 30 --hz 5' +printf '%s\n' 'Then validate and inspect it before exporting to LeRobot:' +printf '%s\n' ' python3 scripts/validate-lerobot-staging.py /var/lib/robot/datasets/' +printf '%s\n' ' python3 scripts/report-lerobot-staging.py /var/lib/robot/datasets/' diff --git a/scripts/report-lerobot-staging.py b/scripts/report-lerobot-staging.py new file mode 100644 index 00000000..8a5d23ce --- /dev/null +++ b/scripts/report-lerobot-staging.py @@ -0,0 +1,19 @@ +#!/usr/bin/env python3 +"""Print a small quality report for a validated local MicroDuck recording.""" +import argparse, json +from pathlib import Path + +def main(): + p = argparse.ArgumentParser(description=__doc__); p.add_argument("dataset", type=Path); a = p.parse_args() + rows = [json.loads(x) for x in (a.dataset / "samples.jsonl").read_text().splitlines()] + times = [x["camera"]["captured_at_unix_us"] for x in rows] + actions = [x["state"].get("policy_action", []) for x in rows] + labelled = [x for x in actions if len(x) == 14] + gaps = [b-a for a,b in zip(times, times[1:])] + print(f"samples: {len(rows)}") + print(f"policy-labelled: {len(labelled)} ({len(labelled)/len(rows):.0%})") + if gaps: print(f"camera interval us: min={min(gaps)} median={sorted(gaps)[len(gaps)//2]} max={max(gaps)}") + if labelled: + flat = [v for action in labelled for v in action] + print(f"action range: {min(flat):.3f} .. {max(flat):.3f}") +if __name__ == "__main__": main() diff --git a/scripts/test-lerobot-staging.py b/scripts/test-lerobot-staging.py new file mode 100644 index 00000000..a5df140f --- /dev/null +++ b/scripts/test-lerobot-staging.py @@ -0,0 +1,23 @@ +#!/usr/bin/env python3 +"""Exercise the local recording validator with synthetic, non-camera data.""" +import json, subprocess, sys, tempfile +from pathlib import Path + +def main(): + repo = Path(__file__).resolve().parents[1] + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp); (root / "frames").mkdir() + (root / "meta.json").write_text(json.dumps({"format":"microduck-lerobot-staging-v1", "upload":"disabled", "guardrails":{"max_samples":2, "max_bytes":8, "min_free_bytes":1, "max_episodes":1}})) + rows = [] + for i in range(2): + name = f"frames/{i:06d}.uyvy"; data = bytes([128, 32, 128, 32]) + (root / name).write_bytes(data) + rows.append({"frame":name, "camera":{"format":"UYVY", "bytes":len(data), "width":2, "height":1, "captured_at_unix_us":1000+i}, "state":{"joints":[0.0]*15, "policy_action":[0.0]*14}}) + (root / "samples.jsonl").write_text("\n".join(json.dumps(x) for x in rows)+"\n") + subprocess.run([sys.executable, str(repo / "scripts/validate-lerobot-staging.py"), str(root)], check=True) + subprocess.run([ + sys.executable, str(repo / "scripts/export-lerobot-local.py"), str(root), + "--root", str(root / "lerobot-output"), "--dry-run", + ], check=True) + print("synthetic staging test passed") +if __name__ == "__main__": main() diff --git a/scripts/validate-lerobot-staging.py b/scripts/validate-lerobot-staging.py new file mode 100644 index 00000000..5d9a98f6 --- /dev/null +++ b/scripts/validate-lerobot-staging.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python3 +"""Validate a local MicroDuck recording before converting it to LeRobot.""" +import argparse, json, math, sys +from pathlib import Path + +def bad(errors, text): errors.append(text) + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("dataset", type=Path) + a = p.parse_args(); root = a.dataset; errors = []; previous = -1; count = 0 + try: meta = json.loads((root / "meta.json").read_text()) + except Exception as e: print(f"invalid meta.json: {e}", file=sys.stderr); return 2 + if meta.get("format") != "microduck-lerobot-staging-v1": bad(errors, "unknown staging format") + if meta.get("upload") != "disabled": bad(errors, "dataset is not explicitly local-only") + guardrails = meta.get("guardrails") + if not isinstance(guardrails, dict): bad(errors, "missing recording guardrails") + else: + for key in ("max_samples", "max_bytes", "min_free_bytes", "max_episodes"): + if not isinstance(guardrails.get(key), int) or guardrails[key] <= 0: + bad(errors, f"invalid guardrail {key}") + max_gap = int(2_000_000 / meta.get("fps", 1)) + frame_bytes = 0 + for number, raw in enumerate((root / "samples.jsonl").read_text().splitlines(), 1): + try: sample = json.loads(raw); camera = sample["camera"]; state = sample["state"] + except Exception as e: bad(errors, f"line {number}: invalid JSON: {e}"); continue + path = root / sample.get("frame", "") + if camera.get("format") != "UYVY": bad(errors, f"line {number}: expected UYVY") + if not path.is_file() or path.stat().st_size != camera.get("bytes"): + bad(errors, f"line {number}: missing or truncated {path.name}") + elif isinstance(camera.get("bytes"), int): frame_bytes += camera["bytes"] + stamp = camera.get("captured_at_unix_us") + if not isinstance(stamp, int) or stamp <= previous: bad(errors, f"line {number}: non-monotonic camera time") + if previous >= 0 and isinstance(stamp, int) and stamp - previous > max_gap: bad(errors, f"line {number}: camera gap exceeds two sample periods") + previous = stamp if isinstance(stamp, int) else previous + action = state.get("policy_action", []) + if action and len(action) != 14: bad(errors, f"line {number}: policy_action has {len(action)}, expected 14") + if action and not all(isinstance(x, (int, float)) and math.isfinite(x) for x in action): bad(errors, f"line {number}: non-finite action") + if len(state.get("joints", [])) != 15: bad(errors, f"line {number}: joints must have 15 values") + if state.get("safety", {}).get("fallen") or state.get("safety", {}).get("limp"): bad(errors, f"line {number}: unsafe robot state") + count += 1 + if not count: bad(errors, "no samples") + if isinstance(guardrails, dict): + if count > guardrails.get("max_samples", 0): bad(errors, "sample count exceeds recording cap") + if frame_bytes > guardrails.get("max_bytes", 0): bad(errors, "frame bytes exceed recording cap") + if errors: + print("REJECTED", *errors, sep="\n- ", file=sys.stderr); return 1 + print(f"OK: {count} samples, local-only, ready for LeRobot conversion") + return 0 +if __name__ == "__main__": raise SystemExit(main()) diff --git a/updater/src/config.rs b/updater/src/config.rs index 77a370d6..0c57790f 100644 --- a/updater/src/config.rs +++ b/updater/src/config.rs @@ -159,6 +159,15 @@ pub struct ComponentConfig { /// Refuse anything but this version. Set by `robotctl pin`. #[serde(default)] pub pinned: Option, + + /// Files which must be present in an extracted artifact before it may become current. Model + /// components use this to reject a signed but incomplete weights bundle before the swap. + #[serde(default)] + pub required_files: Vec, + + /// Refuse an artifact above this reviewed compressed-size budget before downloading it. + #[serde(default)] + pub max_artifact_bytes: Option, } fn default_keep_previous() -> usize { @@ -422,6 +431,27 @@ impl Config { rollback target" )); } + + for required in &component.required_files { + if required.is_absolute() + || required.components().any(|part| { + matches!( + part, + std::path::Component::ParentDir | std::path::Component::RootDir + ) + }) + { + return bad(format!( + "component {name}: required_files entry {} must be a relative path inside the artifact", + required.display() + )); + } + } + if component.max_artifact_bytes == Some(0) { + return bad(format!( + "component {name}: max_artifact_bytes must be positive" + )); + } } if !self.state_dir.is_absolute() { @@ -475,7 +505,16 @@ mod tests { ); // One component per model, each independently versioned (§5.5). - assert!(config.component("model-walk").is_ok()); + let model_walk = config.component("model-walk").unwrap(); + assert!(matches!( + model_walk.on_apply, + ApplyAction::Reload { ref unit, ref signal } if unit == "robotd" && signal == "SIGHUP" + )); + assert_eq!( + model_walk.required_files, + [std::path::PathBuf::from("walk.onnx")] + ); + assert_eq!(model_walk.max_artifact_bytes, Some(128 * 1024 * 1024)); assert!(config.component("model-jump").is_ok()); } diff --git a/updater/src/engine.rs b/updater/src/engine.rs index 365b1663..3e414dac 100644 --- a/updater/src/engine.rs +++ b/updater/src/engine.rs @@ -419,6 +419,7 @@ impl Engine { self.verify_manifest(&signed)?; let manifest = signed.parsed; Self::check_channel(&manifest, component)?; + require_artifact_budget(manifest.size, cfg.max_artifact_bytes)?; if Some(&manifest.version) == installed.as_ref() { return Ok(CheckResult::UpToDate { @@ -946,6 +947,8 @@ impl Engine { } })?; + require_files(extract_dir, &cfg.required_files)?; + // 5b. Would this release leave an installed unit with nothing to exec? See // [`crate::orphan`] — a downgrade past the release that introduced a daemon leaves // that daemon's unit behind, and it then fails with `203/EXEC`, which fails the @@ -2072,9 +2075,7 @@ impl Engine { Ok(()) } ApplyAction::Reload { unit, signal } => { - let mut c = tokio::process::Command::new(SYSTEMCTL); - c.arg("kill").arg(format!("--signal={signal}")).arg(unit); - let result = run_systemctl(c, "apply action").await; + let result = reload_one(SYSTEMCTL, unit, signal).await; rec.note(RunEvent::Unit { unit: unit.clone(), action: format!("reload ({signal})"), @@ -2966,6 +2967,51 @@ async fn restart_one(systemctl: &str, unit: &str) -> Result<(), Error> { } } +/// Ask systemd to deliver one named signal without restarting the unit. Models use this for +/// SIGHUP so their candidate session can be swapped only at `robotd`'s safe control boundary. +/// Kept separate from [`Engine::run_apply_action`] because the exact argv is a safety contract and +/// needs a stubbed test rather than a board running systemd. +async fn reload_one(systemctl: &str, unit: &str, signal: &str) -> Result<(), Error> { + let mut command = tokio::process::Command::new(systemctl); + command + .arg("kill") + .arg(format!("--signal={signal}")) + .arg(unit); + run_systemctl(command, "apply action").await +} + +/// Reject an otherwise valid artifact which omitted a file the configured component must load. +/// This runs against staging, before a dry run succeeds or `current` can move. +fn require_files(root: &Path, required: &[PathBuf]) -> Result<(), Error> { + for relative in required { + let path = root.join(relative); + if !path.is_file() { + return Err(Error::Incompatible(format!( + "artifact is missing required file {}", + relative.display() + ))); + } + } + Ok(()) +} + +fn require_artifact_budget(size: Option, maximum: Option) -> Result<(), Error> { + let Some(maximum) = maximum else { + return Ok(()); + }; + let Some(size) = size else { + return Err(Error::Incompatible( + "artifact size is required for this component's budget".into(), + )); + }; + if size > maximum { + return Err(Error::Incompatible(format!( + "artifact is {size} bytes, above configured budget {maximum}" + ))); + } + Ok(()) +} + /// One `systemctl restart`, with the unit named in the error. /// /// Named, because the caller restarts up to six of them and the bare message — systemd's own @@ -3566,6 +3612,51 @@ exit 1 path } + #[tokio::test] + async fn a_model_reload_delivers_sighup_without_restarting_robotd() { + let dir = tempfile::tempdir().unwrap(); + let systemctl = stub_recorder(dir.path(), "systemctl"); + + reload_one(systemctl.to_str().unwrap(), "robotd", "SIGHUP") + .await + .expect("the stub accepts the reload"); + + let log = calls(dir.path()); + assert_eq!(log, "kill --signal=SIGHUP robotd\n"); + assert!( + !log.contains("restart"), + "a model reload must not restart motor control" + ); + } + + #[test] + fn a_model_bundle_missing_its_onnx_file_is_refused_before_swap() { + let dir = tempfile::tempdir().unwrap(); + let error = + require_files(dir.path(), &[std::path::PathBuf::from("walk.onnx")]).unwrap_err(); + assert!(format!("{error}").contains("missing required file walk.onnx")); + std::fs::write(dir.path().join("walk.onnx"), b"weights").unwrap(); + require_files(dir.path(), &[std::path::PathBuf::from("walk.onnx")]).unwrap(); + } + + #[test] + fn a_model_artifact_must_declare_and_fit_its_budget() { + require_artifact_budget(Some(128), Some(128)).unwrap(); + assert!( + require_artifact_budget(Some(129), Some(128)) + .unwrap_err() + .to_string() + .contains("above configured budget") + ); + assert!( + require_artifact_budget(None, Some(128)) + .unwrap_err() + .to_string() + .contains("size is required") + ); + require_artifact_budget(None, None).unwrap(); + } + /// The deferred restarts, as an actual command line. Until this existed nothing in the repository /// could observe that call: the program name was hardcoded, so `--on-active` could have been /// wrong and every test would still pass — while on a board the only symptom is `btd` quietly diff --git a/updater/tests/apply.rs b/updater/tests/apply.rs index 923947c0..ac961cd9 100644 --- a/updater/tests/apply.rs +++ b/updater/tests/apply.rs @@ -166,11 +166,27 @@ impl Fixture { } fn publish_model(&self, version: &str) { + self.publish_model_for("model", version); + } + + /// Publish a model artifact for the named component. A manifest's channel is a component + /// identity, not display text: the updater refuses a signed artifact for the wrong slot. + fn publish_model_for(&self, component: &str, version: &str) { + self.publish_model_for_with(component, version, |_| {}); + } + + fn publish_model_for_with( + &self, + component: &str, + version: &str, + edit: impl FnOnce(&mut serde_json::Value), + ) { self.publisher .release(version) - .channel("model") + .channel(component) .dir(self.model_releases()) .file("walk.onnx", b"weights", 0o644) + .manifest(edit) .write(); } @@ -1529,6 +1545,98 @@ health = {{ probe = "none" }} assert_eq!(fx.live_version().as_deref(), Some("1.0.0")); } +/// A model uses the ordinary signed-component path: its own source and install directory, then +/// `select` to repoint an already verified local version. This is the development analogue of a +/// Hub model trial and must never need a daemon artifact or a relaxed signature check. +#[tokio::test] +async fn a_signed_local_model_installs_updates_and_selects_an_older_version() { + let fx = Fixture::new(); + let install = fx.root.join("opt/robot/model/walk"); + fx.publish_model_for("model-walk", "1.0.0"); + + let extra = format!( + r#" +[component.model-walk] +install_dir = "{}" +source = {{ type = "local_dir", path = "{}" }} +on_apply = {{ action = "none" }} +health = {{ probe = "none" }} +"#, + install.display(), + fx.model_releases().display(), + ); + + let mut engine = fx.engine(Box::new(FakeRobot::healthy()), Faults::none(), &extra); + let (tx, _rx) = progress_channel(); + engine + .apply( + "model-walk", + Target::Exact(semver::Version::new(1, 0, 0)), + ApplyOptions::default(), + tx, + ) + .await + .unwrap(); + assert_eq!( + test_support::live_version(&install).as_deref(), + Some("1.0.0") + ); + + fx.publish_model_for("model-walk", "1.1.0"); + let (tx, _rx) = progress_channel(); + engine + .apply("model-walk", Target::Latest, ApplyOptions::default(), tx) + .await + .unwrap(); + assert_eq!( + test_support::live_version(&install).as_deref(), + Some("1.1.0") + ); + + engine + .select("model-walk", &semver::Version::new(1, 0, 0)) + .await + .unwrap(); + assert_eq!( + test_support::live_version(&install).as_deref(), + Some("1.0.0") + ); +} + +/// A valid signature does not make a model compatible. The updater checks the candidate's model +/// API before extracting or repointing it, so an older daemon keeps the working model it has. +#[tokio::test] +async fn an_incompatible_local_model_never_becomes_current() { + let fx = Fixture::new(); + let install = fx.root.join("opt/robot/model/walk"); + fx.publish_model_for_with("model-walk", "2.0.0", |manifest| { + manifest["model_api"] = serde_json::json!(2); + }); + let extra = format!( + r#" +[component.model-walk] +install_dir = "{}" +source = {{ type = "local_dir", path = "{}" }} +on_apply = {{ action = "none" }} +health = {{ probe = "none" }} +"#, + install.display(), + fx.model_releases().display(), + ); + let mut engine = fx.engine(Box::new(FakeRobot::healthy()), Faults::none(), &extra); + let (tx, _rx) = progress_channel(); + let error = engine + .apply("model-walk", Target::Latest, ApplyOptions::default(), tx) + .await + .unwrap_err(); + + assert!( + matches!(error, updater::Error::Incompatible(_)), + "{error:?}" + ); + assert_eq!(test_support::live_version(&install), None); +} + /// **#5** `transition_to` armed the boot counter before validating the target, so a /// failed swap left a trial for a version that was never live — self-healing later /// only via a spurious rollback and a bogus log entry. diff --git a/updater/updater.example.toml b/updater/updater.example.toml index 0b608518..6ee88377 100644 --- a/updater/updater.example.toml +++ b/updater/updater.example.toml @@ -164,6 +164,9 @@ timeout = "30s" [component.model-walk] install_dir = "/opt/robot/model/walk" keep_previous = 3 +# Refuse a signed but incomplete artifact before it can replace `current`. +required_files = ["walk.onnx"] +max_artifact_bytes = 134217728 # 128 MiB; raise only for a reviewed model bundle [component.model-walk.source] type = "hf_hub" @@ -171,12 +174,12 @@ repo = "ORG/gait-walk" revision = "main" manifest_file = "manifest.json" -# Bootstrap-inert until robotd exists — see the daemon component above. [component.model-walk.on_apply] -action = "none" -# action = "reload" -# unit = "robotd" -# signal = "SIGHUP" +# Model bundles never restart motor control. `robotd` accepts the SIGHUP only after policy +# control is disabled, and keeps the current controller if the candidate cannot load. +action = "reload" +unit = "robotd" +signal = "SIGHUP" [component.model-walk.health] probe = "none" diff --git a/xtask/src/main.rs b/xtask/src/main.rs index f5006a48..55c98cbc 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -12,6 +12,7 @@ //! //! ```text //! cargo xtask package --version 1.2.3 --channel daemon --bin-dir --out dist/ +//! cargo xtask package --version 1.2.3 --channel model-walk --model-dir --model-api 1 --out dist/ //! cargo xtask sign --dir dist/ --key secret.key //! cargo xtask promote --version 1.2.3 --staging-tag daemon-staging-v1.2.3 \ //! --stable-tag daemon-v1.2.3 \ @@ -60,7 +61,8 @@ struct Cli { enum Command { /// Assemble a `.tar.zst` artifact and its unsigned manifest. Package { - /// Release version. Must match the crate version — see `--allow-version-drift`. + /// Release version. Must match the crate version for daemon artifacts. Model artifacts + /// have their own version line when `--model-dir` is used. #[arg(long)] version: semver::Version, @@ -68,9 +70,23 @@ enum Command { #[arg(long, default_value = "daemon")] channel: String, - /// Directory holding the built binaries to ship. - #[arg(long)] - bin_dir: PathBuf, + /// Directory holding the built daemon binaries to ship. + #[arg( + long, + conflicts_with = "model_dir", + required_unless_present = "model_dir" + )] + bin_dir: Option, + + /// Directory holding one model bundle (for example `walk.onnx` and its metadata). Model + /// files are placed at the root of the installed model release, never under `bin/`. + #[arg(long, conflicts_with = "bin_dir", required_unless_present = "bin_dir")] + model_dir: Option, + + /// Model API required by this bundle. Required with `--model-dir`; this is the + /// compatibility gate checked before the updater makes a model current. + #[arg(long, requires = "model_dir")] + model_api: Option, /// Where to write the artifact and manifest. #[arg(long, default_value = "dist")] @@ -217,6 +233,8 @@ fn run() -> Result<(), Box> { version, channel, bin_dir, + model_dir, + model_api, out, base_url, revision, @@ -229,6 +247,8 @@ fn run() -> Result<(), Box> { version, channel, bin_dir, + model_dir, + model_api, out, base_url, revision, @@ -273,7 +293,9 @@ fn run() -> Result<(), Box> { struct PackageArgs { version: semver::Version, channel: String, - bin_dir: PathBuf, + bin_dir: Option, + model_dir: Option, + model_api: Option, out: PathBuf, base_url: Option, revision: Option, @@ -285,9 +307,32 @@ struct PackageArgs { } fn package(args: PackageArgs) -> Result<(), Box> { + let is_model = args.model_dir.is_some(); + if is_model { + if !args.channel.starts_with("model-") { + return Err(format!( + "model bundles must use a `model-…` channel, got {:?}", + args.channel + ) + .into()); + } + if args.model_api.is_none() { + return Err("--model-api is required with --model-dir".into()); + } + if !args.includes.is_empty() { + return Err( + "model bundles take every file from --model-dir; --include is daemon-only".into(), + ); + } + } else if args.model_api.is_some() { + return Err("--model-api requires --model-dir".into()); + } + // Catch the classic mistake: tagging a release without bumping Cargo.toml, so the - // robot reports a version that doesn't match what it's running. - if !args.allow_version_drift { + // robot reports a version that doesn't match what it's running. A model deliberately has a + // separate version line, so applying that comparison to it would turn the testing-only escape + // hatch into the normal publishing path. + if !is_model && !args.allow_version_drift { let crate_version = workspace_version()?; // A dev build is the crate version plus a prerelease tag — `0.2.0-dev.17.abc1234` // against a crate at `0.2.0` — so its release triple must match while its prerelease @@ -330,8 +375,13 @@ fn package(args: PackageArgs) -> Result<(), Box> { let encoder = zstd::Encoder::new(file, args.zstd_level)?.auto_finish(); let mut builder = tar::Builder::new(encoder); + let source_dir = args + .model_dir + .as_ref() + .or(args.bin_dir.as_ref()) + .ok_or("one of --bin-dir or --model-dir is required")?; let mut shipped = Vec::new(); - for entry in std::fs::read_dir(&args.bin_dir)? { + for entry in std::fs::read_dir(source_dir)? { let path = entry?.path(); if !path.is_file() { continue; @@ -341,12 +391,20 @@ fn package(args: PackageArgs) -> Result<(), Box> { .and_then(|n| n.to_str()) .ok_or("binary has an unreadable name")? .to_owned(); - // Executable: the robot runs these straight out of the release directory. - append_file(&mut builder, &path, &format!("bin/{name}"), 0o755)?; + let destination = if is_model { + // `robotd`'s policy paths name files from the model release directly. Keeping + // the model at its root makes that path unambiguous and prevents a model from + // carrying an executable daemon payload by accident. + name.clone() + } else { + format!("bin/{name}") + }; + let mode = if is_model { 0o644 } else { 0o755 }; + append_file(&mut builder, &path, &destination, mode)?; shipped.push(name); } if shipped.is_empty() { - return Err(format!("no binaries found in {}", args.bin_dir.display()).into()); + return Err(format!("no files found in {}", source_dir.display()).into()); } shipped.sort(); @@ -365,28 +423,32 @@ fn package(args: PackageArgs) -> Result<(), Box> { append_file(&mut builder, Path::new(src), dest, mode)?; } - // The preinstall hook, always, generated from its template. + // The preinstall hook is a daemon-artifact prerequisite, not code a model release may + // execute. Keeping it out of model bundles is part of the channel boundary. // // Not an `--include` the release workflow has to remember: the board prerequisites it // asserts are a property of every release, and a check that ships only when someone // adds a flag is a check that will one day be missing from the release that needed it. - const PREINSTALL_TEMPLATE: &str = "hooks/preinstall.in"; - if args - .includes - .iter() - .any(|i| i.ends_with("=hooks/preinstall")) - { - return Err("hooks/preinstall is generated; remove the --include for it".into()); + if !is_model { + const PREINSTALL_TEMPLATE: &str = "hooks/preinstall.in"; + if args + .includes + .iter() + .any(|i| i.ends_with("=hooks/preinstall")) + { + return Err("hooks/preinstall is generated; remove the --include for it".into()); + } + let template = std::fs::read_to_string(PREINSTALL_TEMPLATE) + .map_err(|e| format!("reading {PREINSTALL_TEMPLATE}: {e}"))?; + let hook = render_preinstall_hook(&template)?; + append_bytes(&mut builder, "hooks/preinstall", hook.as_bytes(), 0o755)?; } - let template = std::fs::read_to_string(PREINSTALL_TEMPLATE) - .map_err(|e| format!("reading {PREINSTALL_TEMPLATE}: {e}"))?; - let hook = render_preinstall_hook(&template)?; - append_bytes(&mut builder, "hooks/preinstall", hook.as_bytes(), 0o755)?; // Recorded inside the release so a robot can identify what it is running even // with no network and no manifest. + let kind = if is_model { "model" } else { "daemon" }; let version_toml = format!( - "version = \"{}\"\nchannel = \"{}\"\nrevision = \"{}\"\nbinaries = {:?}\n", + "version = \"{}\"\nchannel = \"{}\"\nkind = \"{kind}\"\nrevision = \"{}\"\nfiles = {:?}\n", args.version, args.channel, args.revision.as_deref().unwrap_or("unknown"), @@ -427,6 +489,9 @@ fn package(args: PackageArgs) -> Result<(), Box> { if let Some(floor) = &args.min_supported { manifest["min_supported"] = serde_json::json!(floor); } + if let Some(model_api) = args.model_api { + manifest["model_api"] = serde_json::json!(model_api); + } let manifest_path = args.out.join("manifest.json"); std::fs::write(&manifest_path, serde_json::to_vec_pretty(&manifest)?)?; @@ -848,6 +913,90 @@ fn sha256_hex(bytes: &[u8]) -> String { #[cfg(test)] mod tests { + use super::*; + + #[test] + fn model_bundle_has_an_independent_version_line_and_required_compatibility_api() { + let scratch = tempfile::tempdir().unwrap(); + let model_dir = scratch.path().join("model"); + let out = scratch.path().join("out"); + std::fs::create_dir(&model_dir).unwrap(); + std::fs::write(model_dir.join("walk.onnx"), b"not-a-real-model").unwrap(); + std::fs::write(model_dir.join("normalization.json"), b"{}\n").unwrap(); + + // This is intentionally unrelated to Cargo.toml's daemon version: a policy is an + // independently versioned component. Requiring --model-api makes that independence + // safe for the daemon that will load it. + package(PackageArgs { + version: semver::Version::new(7, 2, 1), + channel: "model-walk".into(), + bin_dir: None, + model_dir: Some(model_dir), + model_api: Some(1), + out: out.clone(), + base_url: None, + revision: None, + min_hw_rev: 0, + min_supported: None, + includes: Vec::new(), + allow_version_drift: false, + zstd_level: 1, + }) + .unwrap(); + + let manifest: serde_json::Value = + serde_json::from_slice(&std::fs::read(out.join("manifest.json")).unwrap()).unwrap(); + assert_eq!(manifest["channel"], "model-walk"); + assert_eq!(manifest["version"], "7.2.1"); + assert_eq!(manifest["model_api"], 1); + + let artifact = std::fs::File::open(out.join("model-walk-7.2.1.tar.zst")).unwrap(); + let decoder = zstd::Decoder::new(artifact).unwrap(); + let mut archive = tar::Archive::new(decoder); + let paths: Vec<_> = archive + .entries() + .unwrap() + .map(|entry| entry.unwrap().path().unwrap().into_owned()) + .collect(); + assert!(paths.contains(&PathBuf::from("walk.onnx"))); + assert!(paths.contains(&PathBuf::from("normalization.json"))); + assert!(paths.contains(&PathBuf::from(VERSION_FILE))); + assert!( + !paths.iter().any(|path| path.starts_with("bin")), + "a model bundle must not look like an executable daemon release" + ); + assert!( + !paths.iter().any(|path| path.starts_with("hooks")), + "a model bundle must not execute daemon install hooks" + ); + } + + #[test] + fn model_bundle_refuses_missing_compatibility_api() { + let scratch = tempfile::tempdir().unwrap(); + let model_dir = scratch.path().join("model"); + std::fs::create_dir(&model_dir).unwrap(); + std::fs::write(model_dir.join("walk.onnx"), b"weights").unwrap(); + + let error = package(PackageArgs { + version: semver::Version::new(1, 0, 0), + channel: "model-walk".into(), + bin_dir: None, + model_dir: Some(model_dir), + model_api: None, + out: scratch.path().join("out"), + base_url: None, + revision: None, + min_hw_rev: 0, + min_supported: None, + includes: Vec::new(), + allow_version_drift: false, + zstd_level: 1, + }) + .unwrap_err(); + assert!(error.to_string().contains("--model-api is required")); + } + /// Every file that packages a release, which is where the `--include` list and the staged /// binaries live. Repository paths, because one of them is not a workflow. ///