feat(robotd): execute timed action chunks for VLA runners

README.md

@@ -4,6 +4,8 @@
 
 <h1 align="center">Microduck</h1>
 
+Duckmind adds [native Action Chunk execution](docs/design/action-chunks.md) to this Microduck runtime, currently available on fake and MuJoCo bodies.
+
 <p align="center">
   <em>A tiny biped robot that moves using reinforcement learning policies.</em>
 </p>

btd/src/route.rs

@@ -234,6 +234,8 @@ fn permits(call: &proto::Call) -> bool {
         // not exist for the first ~73s of a boot is not a control transport. The body pose and
         // the mouth ride with it: all of these are a stream of small updates, and the argument is
         // about the stream, not about any one of them.
+        // First deployment uses a local fake/sim policy runner, not remote joint ownership.
+        RobotActionsBegin | RobotActionsSubmit(_) | RobotActionsEnd(_) => false,
         RobotMove(_) | RobotHead(_) | RobotLook(_) | RobotPose(_) | RobotMouth(_) => false,
 
         // **Not teleop either, and it sat in that group for the same reason a skill did**: it was

docs/README.md

@@ -1,5 +1,7 @@
 # Docs
 
+Duckmind model execution: [Action Chunk API and timing contract](design/action-chunks.md).
+
 The [README](../README.md) is the front door — what a microduck is, and where to go. If you have
 one in front of you and want to drive it, start at the [cheat sheet](robot/cheatsheet.md).
 

docs/design/action-chunks.md

@@ -0,0 +1,79 @@
+# Action Chunk execution
+
+Duckmind accepts model-generated joint trajectories through `robot.actions.*`. A Python policy runner owns language, images, history, inference and model-specific action decoding. `robotd` owns the action timeline and the only motor-writing path.
+
+The API is enabled only with `--fake` or `--sim`, at a 50 Hz control rate. Physical hardware rejects these calls. BLE and WebRTC do not forward them; a policy runner on the simulation host uses the existing Unix JSON-RPC socket. The runner may call a remote model service.
+
+## Start and observe
+
+```bash
+cargo run -p robotd -- --fake --no-policy --socket /tmp/duckmind.sock
+# In another terminal; wait for homing to complete before acquiring the joints.
+cargo run -p robotctl -- --robot-socket /tmp/duckmind.sock robot init
+cargo run -p robotctl -- --robot-socket /tmp/duckmind.sock robot actions begin
+```
+
+`begin` requires fresh sensor feedback, warmed IMU and completed bring-up, with no shutdown, mode change, policy swap or limp-fall sequence in progress. It returns `session_id`, robot-clock `t_ns`, `step_ns`, `joint_names` and measured `positions`.
+
+Acquisition disables the on-board movement policy and clears its velocity intent. Theremin and chorale joint writers are deactivated. Another movement writer receives `BUSY` while the session owns the body. Stop, disable, init, relax, motor reboot and shutdown can preempt the session.
+
+## Submit a chunk
+
+Use `robotctl robot actions submit chunk.json`, or send a JSON-RPC request with method `robot.actions.submit` and these parameters:
+
+| Field | Meaning |
+|---|---|
+| `session_id` | ID returned by `begin`; invalid after cancellation, exhaustion or daemon restart. |
+| `sequence` | Increasing inference-request number within this session. |
+| `observation_t_ns` | Robot-clock timestamp of the state used for inference. |
+| `start_t_ns` | Start of the first action interval, on the same clock. |
+| `step_ns` | Exactly `20000000` (20 ms). |
+| `positions` | 1–100 arrays of 15 absolute joint angles in radians, ordered as `joint_names`. |
+
+The 15-joint body contract includes the mouth. The upstream locomotion model produces 14 normalized offsets; a model adapter must decode those and supply a mouth target before submitting. `robotd` does not guess normalization, HOME offsets or joint mappings.
+
+Angles must be finite and within the actuator travel range. This is not a complete anatomical or collision constraint model. A chunk's end and its source observation must fall within the bounded two-second admission window; timestamps that overflow or use a future observation are refused.
+
+Admission happens on the motor loop. A successful RPC means the timeline accepted the chunk, not that joints reached their targets. The IPC caller waits at most 250 ms for admission. A request whose reply was abandoned before processing is not executed later.
+
+## Time and replacement
+
+Action `i` applies during `[start_t_ns + i * step_ns, start_t_ns + (i + 1) * step_ns)`. Cloud wall time is irrelevant: use the clock already reported by `robot.state.t_ns`. The client maintains alignment to that clock from received robot state.
+
+Past intervals are discarded. A late tick selects the current target instead of replaying missed targets rapidly. A future replacement preserves the old prefix before its start and discards the old tail from that start onward. An entirely expired or out-of-order response cannot replace the current timeline.
+
+Waiting for a future first action holds the measured pose captured at acquisition. Replacements must overlap or continue the buffered timeline; a gap is rejected without modifying the old buffer. There is no automatic interpolation, action averaging or RTC inpainting. Plan continuous boundaries in the policy runner and validate them in simulation.
+
+## Feedback and termination
+
+`robot.subscribe` adds an `actions` block on eligible backends:
+
+- `phase`: `idle`, `waiting`, `running` or `ended`.
+- `session_id`, latest accepted `sequence` and current robot-clock `t_ns`.
+- `selected_sequence` / `selected_index`: most recently selected command, not measured task completion.
+- `remaining`: buffered intervals, including the currently active interval.
+- `last_write_ok`: whether the session's most recent bus write succeeded.
+- `reason`: why control ended, such as `cancelled`, `operator_preempted`, `buffer_exhausted`, `body_not_ready` or `bus_write_failed`.
+
+Measured motion remains in `robot.state.joints` and `velocities`; `targets` reports the selected target. A successful write is not proof that the mechanism moved, and chunk exhaustion is not task success.
+
+```bash
+robotctl --robot-socket /tmp/duckmind.sock robot actions end SESSION_ID
+```
+
+End, stop, buffer exhaustion or loss of fresh/ready body state invalidates the session and discards its timeline. A session with no first chunk expires after two seconds. Except for explicit power/mode operations, the loop captures the last valid measured pose and holds it; the old policy is not automatically resumed. Holding a pose does not guarantee dynamic balance on a biped. A lost client can execute only the remaining bounded timeline, not an unbounded backlog.
+
+## Ownership and implementation
+
+`duck-ipc-proto/src/actions.rs` owns the wire types. `robotd/src/action_chunk.rs` owns admission, session generations and timed replacement. `main.rs` selects the source before the shared `Safety::apply → RobotIo` write; IPC never writes motors. Stop generations invalidate commands that were queued before the stop as well as already active sessions.
+
+This follows [LeRobot asynchronous inference](https://huggingface.co/docs/lerobot/async) and [OpenPI's action-chunk broker](https://github.com/Physical-Intelligence/openpi/blob/main/packages/openpi-client/src/openpi_client/action_chunk_broker.py) in separating inference from action consumption. [RTC](https://huggingface.co/docs/lerobot/main/rtc) additionally conditions generation on the previous action prefix; it belongs in the policy runner and is not implemented by this queue.
+
+## Verification
+
+```bash
+cargo test -p duck-ipc-proto -p robotd -p robotctl
+cargo test -p robotd --test action_chunk_ipc
+```
+
+Unit tests cover late responses, tick skips, future replacement, stale sessions, queue admission and invalid actions. The integration test launches the real `robotd --fake`, sends JSON-RPC chunks, observes joint feedback, preempts execution and verifies the cancelled tail never moves the joint. MuJoCo checks exercise the same `RobotIo` path against physics; neither test establishes a learned language skill or balance policy.

duck-ipc-proto/src/actions.rs

@@ -0,0 +1,57 @@
+//! Robot-clock action chunks. Model normalization belongs to the caller.
+use serde::{Deserialize, Serialize};
+
+pub const ACTION_STEP_NS: u64 = 20_000_000;
+pub const MAX_ACTION_STEPS: usize = 100;
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ActionChunkParams {
+    pub session_id: String,
+    pub sequence: u64,
+    /// Timestamp of the observation used for this inference, on robot.state's clock.
+    pub observation_t_ns: u64,
+    pub start_t_ns: u64,
+    pub step_ns: u64,
+    /// Absolute radians in JOINT_NAMES order, including the mouth.
+    pub positions: Vec<[f64; 15]>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ActionEndParams {
+    pub session_id: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct ActionSession {
+    pub session_id: String,
+    pub t_ns: u64,
+    pub step_ns: u64,
+    pub joint_names: Vec<String>,
+    pub positions: [f64; 15],
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ActionPhase {
+    #[default]
+    Idle,
+    Waiting,
+    Running,
+    Ended,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+pub struct ActionStatus {
+    pub session_id: Option<String>,
+    pub phase: ActionPhase,
+    pub sequence: Option<u64>,
+    /// Sequence/index selected for the most recent write attempt; not measured completion.
+    pub selected_sequence: Option<u64>,
+    pub selected_index: Option<usize>,
+    pub remaining: usize,
+    pub t_ns: u64,
+    pub reason: Option<String>,
+    pub last_write_ok: Option<bool>,
+}

duck-ipc-proto/src/lib.rs

@@ -29,6 +29,9 @@
 //! including the ones on the recovery path, so nothing here may pull in http, tar, crypto
 //! or an async runtime.
 
+mod actions;
+pub use actions::*;
+
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
@@ -423,7 +426,8 @@ pub const JSONRPC_VERSION: &str = "2.0";
 /// an `updaterd` that has not run its first check yet — every board for the minute after it
 /// starts, including the one right after the update that brought v35 in. Both warned. The attempt
 /// tells them apart, and its error is what the warning was pointing at the journal for.
-pub const API_VERSION: u32 = 37;
+// v38: robot.actions.* and robot.state.actions; fake/sim only.
+pub const API_VERSION: u32 = 38;
 
 /// The observation width every policy this robot family runs is built against.
 ///
@@ -540,6 +544,9 @@ pub const JOINT_NAMES: [&str; 15] = [
 /// Method names, as they go on the wire. Namespaced so a new namespace cannot collide
 /// with `update.*`. [`Call`] is the typed form.
 pub mod method {
+    pub const ROBOT_ACTIONS_BEGIN: &str = "robot.actions.begin";
+    pub const ROBOT_ACTIONS_SUBMIT: &str = "robot.actions.submit";
+    pub const ROBOT_ACTIONS_END: &str = "robot.actions.end";
     pub const HELLO: &str = "hello";
 
     /// One raw camera frame. `mediad` answers the JSON-RPC header, followed immediately by the
@@ -985,6 +992,11 @@ pub enum Call {
     RobotModelApi,
     RobotRemoteSessionActive,
 
+    /// Acquire timed joint execution. Action calls require request IDs for admission feedback.
+    RobotActionsBegin,
+    RobotActionsSubmit(ActionChunkParams),
+    RobotActionsEnd(ActionEndParams),
+
     // ── intents ──────────────────────────────────────────────────────────────
     /// Continuous. Send as a notification.
     RobotMove(MoveParams),
@@ -1174,6 +1186,9 @@ impl Call {
             Call::RobotHealth => method::ROBOT_HEALTH,
             Call::RobotModelApi => method::ROBOT_MODEL_API,
             Call::RobotRemoteSessionActive => method::ROBOT_SESSION_ACTIVE,
+            Call::RobotActionsBegin => method::ROBOT_ACTIONS_BEGIN,
+            Call::RobotActionsSubmit(_) => method::ROBOT_ACTIONS_SUBMIT,
+            Call::RobotActionsEnd(_) => method::ROBOT_ACTIONS_END,
             Call::RobotMove(_) => method::ROBOT_MOVE,
             Call::RobotHead(_) => method::ROBOT_HEAD,
             Call::RobotLook(_) => method::ROBOT_LOOK,
@@ -1355,6 +1370,10 @@ impl Call {
             | Call::RobotPolicies
             | Call::RobotModel
             | Call::RobotMode => (Robot, Prompt),
+            // Action requests wait for bounded loop admission, not for motion completion.
+            Call::RobotActionsBegin | Call::RobotActionsSubmit(_) | Call::RobotActionsEnd(_) => {
+                (Robot, Prompt)
+            }
             // Intents and one-shot skills. All fast: they store a value the control loop reads on
             // its next tick, and none of them waits for the robot to finish anything.
             Call::RobotMove(_)
@@ -1469,6 +1488,9 @@ impl Call {
             Call::Pin(p) => encode(p),
             Call::Log(p) => encode(p),
             Call::Show(p) => encode(p),
+            Call::RobotActionsBegin => Value::Object(serde_json::Map::new()),
+            Call::RobotActionsSubmit(p) => encode(p),
+            Call::RobotActionsEnd(p) => encode(p),
             Call::RobotMove(p) => encode(p),
             Call::RobotHead(p) => encode(p),
             Call::RobotLook(p) => encode(p),
@@ -1563,6 +1585,9 @@ impl Call {
             method::ROBOT_HEALTH => Call::RobotHealth,
             method::ROBOT_MODEL_API => Call::RobotModelApi,
             method::ROBOT_SESSION_ACTIVE => Call::RobotRemoteSessionActive,
+            method::ROBOT_ACTIONS_BEGIN => Call::RobotActionsBegin,
+            method::ROBOT_ACTIONS_SUBMIT => Call::RobotActionsSubmit(decode(params)?),
+            method::ROBOT_ACTIONS_END => Call::RobotActionsEnd(decode(params)?),
             method::ROBOT_MOVE => Call::RobotMove(decode(params)?),
             method::ROBOT_HEAD => Call::RobotHead(decode(params)?),
             method::ROBOT_LOOK => Call::RobotLook(decode(params)?),
@@ -1699,6 +1724,18 @@ pub mod test_support {
             Call::RobotHealth,
             Call::RobotModelApi,
             Call::RobotRemoteSessionActive,
+            Call::RobotActionsBegin,
+            Call::RobotActionsSubmit(ActionChunkParams {
+                session_id: "sample".into(),
+                sequence: 1,
+                observation_t_ns: 0,
+                start_t_ns: 0,
+                step_ns: ACTION_STEP_NS,
+                positions: vec![[0.0; 15]],
+            }),
+            Call::RobotActionsEnd(ActionEndParams {
+                session_id: "sample".into(),
+            }),
             Call::RobotMove(MoveParams {
                 vx: 0.2,
                 vy: -0.1,
@@ -3706,6 +3743,9 @@ impl IntentResult {
 /// `requested` rather than the stream carrying only outcomes.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct RobotState {
+    /// External action execution, independent of whether the task has succeeded.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub actions: Option<ActionStatus>,
     /// Seconds since the daemon started. Monotonic: it is for correlating samples, not for
     /// telling the time.
     pub t: f64,
@@ -5560,7 +5600,7 @@ mod tests {
     fn every_call_covers_every_variant() {
         assert_eq!(
             every_call().len(),
-            67,
+            70,
             "a Call variant was added or removed — update every_call() and this count"
         );
     }
@@ -6262,6 +6302,7 @@ mod tests {
     /// — so an additive field is one line here rather than one line per test.
     fn a_state() -> RobotState {
         RobotState {
+            actions: None,
             t: 1.5,
             movement: MoveState {
                 requested: [0.0; 3],

mediad/src/route.rs

@@ -60,6 +60,8 @@ fn permits(call: &proto::Call) -> bool {
         // budget and a link that does not exist for the first ~73s of a boot is not a control
         // transport". A datachannel is a control transport, so this is the transport those
         // refusals were pointing at.
+        // First deployment uses a local fake/sim policy runner, not remote joint ownership.
+        RobotActionsBegin | RobotActionsSubmit(_) | RobotActionsEnd(_) => false,
         RobotMove(_) | RobotHead(_) | RobotLook(_) | RobotPose(_) | RobotMouth(_) => true,
         // The theremin rides with the sounds: it is one, and a browser that can quack a duck
         // may pick its instrument up too.

robotctl/src/actions.rs

@@ -0,0 +1,36 @@
+//! Thin action-session client; the daemon owns timing and admission.
+use std::path::PathBuf;
+
+use clap::Subcommand;
+use duck_ipc_proto as proto;
+
+use crate::{Client, Failure, compact, exit, result_of};
+
+#[derive(Debug, Subcommand)]
+pub enum Command {
+    /// Acquire all joints on a ready fake/sim body; prints session and robot clock.
+    Begin,
+    /// Submit an ActionChunkParams JSON file with robot-clock timestamps.
+    Submit { file: PathBuf },
+    /// Cancel queued actions and release this session.
+    End { session_id: String },
+}
+
+pub fn run(client: &mut Client, command: &Command) -> Result<(), Failure> {
+    let call = match command {
+        Command::Begin => proto::Call::RobotActionsBegin,
+        Command::Submit { file } => {
+            let bytes = std::fs::read(file)
+                .map_err(|e| Failure::new(exit::USAGE, format!("{}: {e}", file.display())))?;
+            let params = serde_json::from_slice(&bytes)
+                .map_err(|e| Failure::new(exit::USAGE, format!("invalid action chunk: {e}")))?;
+            proto::Call::RobotActionsSubmit(params)
+        }
+        Command::End { session_id } => proto::Call::RobotActionsEnd(proto::ActionEndParams {
+            session_id: session_id.clone(),
+        }),
+    };
+    let result = result_of(client.call(&call)?)?;
+    println!("{}", compact(&result));
+    Ok(())
+}

robotctl/src/main.rs

@@ -40,6 +40,7 @@ use clap::{Args, CommandFactory, Parser, Subcommand};
 use duck_ipc_proto as proto;
 use robotd_params::Slot;
 
+mod actions;
 mod camera;
 mod cells;
 mod configure;
@@ -461,6 +462,11 @@ enum SystemCommand {
 
 #[derive(Subcommand, Debug)]
 enum RobotCommand {
+    /// Execute model action chunks on a fake or simulated body.
+    Actions {
+        #[command(subcommand)]
+        command: actions::Command,
+    },
     /// Power the joints and ramp to the home pose, over about two seconds.
     ///
     /// **This moves every joint.** Have the robot on its stand, or hold it. Needs no policy — a
@@ -3153,7 +3159,12 @@ fn run_robot(socket: &Path, command: RobotCommand) -> Result<(), Failure> {
     let mut client = Client::connect_to("robotd", socket)?;
     client.hello()?;
 
+    if let RobotCommand::Actions { command } = &command {
+        return actions::run(&mut client, command);
+    }
+
     let (call, json) = match &command {
+        RobotCommand::Actions { .. } => unreachable!("handled above"),
         RobotCommand::Init { json } => (proto::Call::RobotInit, *json),
         RobotCommand::Relax { json, .. } => (proto::Call::RobotRelax, *json),
         RobotCommand::Enable { off, toggle, json } => (
@@ -3232,6 +3243,7 @@ fn run_robot(socket: &Path, command: RobotCommand) -> Result<(), Failure> {
         return Err(Failure::new(exit::REFUSED, reason));
     }
     match command {
+        RobotCommand::Actions { .. } => unreachable!("handled above"),
         RobotCommand::Init { .. } => println!("standing up — about two seconds to the home pose"),
         RobotCommand::Relax { .. } => println!("torque off"),
         // The daemon's own `reason` names the state it ended in, which is the only trustworthy

robotctl/src/monitor.rs

@@ -4818,6 +4818,7 @@ mod tests {
 
     fn a_state() -> proto::RobotState {
         proto::RobotState {
+            actions: None,
             t: 1.0,
             movement: proto::MoveState {
                 requested: [0.0; 3],