From bc662f1b9e3935e822c8173fdd5fc10f366fb578 Mon Sep 17 00:00:00 2001 From: phoxal-release-bot Date: Thu, 16 Jul 2026 13:07:07 +0200 Subject: [PATCH 1/2] feat(runtime): keep tools outside logical time Signed-off-by: phoxal-release-bot --- Cargo.lock | 1 + docs/CONVENTIONS.md | 21 +- docs/GETTING_STARTED.md | 18 +- docs/SIMULATION.md | 6 + examples/hello-rover/worlds/tool-boundary.wbt | 21 ++ phoxal/README.md | 11 + phoxal/src/lib.rs | 13 + phoxal/src/participant/api.rs | 11 +- phoxal/src/participant/context.rs | 10 - phoxal/src/participant/runner.rs | 56 ++++- phoxal/tests/runner.rs | 100 ++++++++ .../trybuild/fail/tool_cannot_access_clock.rs | 15 ++ .../fail/tool_cannot_access_clock.stderr | 5 + service/motion/Cargo.toml | 1 + service/motion/src/arbitration.rs | 236 +++++++++++------- service/motion/src/main.rs | 159 ++++++++++-- tool/joypad/src/main.rs | 165 +++++++++--- tool/router/src/main.rs | 15 +- tool/telemetry/src/main.rs | 12 +- xtask/src/workspace.rs | 85 +++++++ 20 files changed, 771 insertions(+), 190 deletions(-) create mode 100644 examples/hello-rover/worlds/tool-boundary.wbt create mode 100644 phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs create mode 100644 phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr diff --git a/Cargo.lock b/Cargo.lock index 506601b2..631ec03f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2522,6 +2522,7 @@ dependencies = [ "phoxal", "phoxal-api", "serde_json", + "tokio", ] [[package]] diff --git a/docs/CONVENTIONS.md b/docs/CONVENTIONS.md index 0b4f7881..bef9d7e8 100644 --- a/docs/CONVENTIONS.md +++ b/docs/CONVENTIONS.md @@ -72,11 +72,23 @@ the only source of keys, and the wire body never appears in the key ## Logical time -- Participants track time, watchdogs, and staleness with **logical time** - only - never wall time directly. +- Robot services, drivers, and simulators track state-transition time, + watchdogs, and synchronous-input staleness with **logical time** - never wall + time directly. The runner owns one `ClockSource` and stamps every `StepContext` and every - `produced_at_ns` from it, so all participants share one time domain + `produced_at_ns` from it, so participants in the robot clock share one domain ([`phoxal/src/participant/clock.rs`](../phoxal/src/participant/clock.rs)). +- Tools are outside that clock. They run from external events and host-monotonic + timers in Run, Webots, and Deploy, even if a generic launch requests + `ClockMode::Simulation`. Tool envelope metadata uses + `phoxal::raw::host_time()`; the runner never gives tools `StepContext` or + enrolls them in simulation scheduling. Official tool sources are checked + against simulation-clock imports; privileged user-authored raw-bus tools must + uphold the same rule and never decide freshness from simulator time. +- A logical-time consumer of asynchronous external input owns retention and + freshness. Keep only the latest bounded value, record its consumer-local + monotonic arrival instant, and sample that value at the logical step. A + logical pause must not accumulate a replay backlog. - `LogicalTime` is `{ epoch, time_ns }`. Within an epoch `time_ns` strictly increases; an epoch bump signals a reset. `RealClock` reads the host-wide UNIX-epoch domain (so cross-process staleness @@ -104,7 +116,8 @@ the only source of keys, and the wire body never appears in the key `#[phoxal::driver]` for per-component-instance participants that can call `ctx.component()`; `#[phoxal::tool]` for host-side utilities that inspect `ctx.robot()` (`Api` - defaults to `()` - tools stay raw-bus only); + defaults to `()` - tools stay raw-bus only and host/event driven, with no + logical clock accessor or scheduled step); `#[phoxal::simulator]` for simulation-only participants. Each kind embeds its own JSON metadata (id, kind, contract surface) as a static in a dedicated linker section on the compiled binary diff --git a/docs/GETTING_STARTED.md b/docs/GETTING_STARTED.md index 9b75cd51..8b801a17 100644 --- a/docs/GETTING_STARTED.md +++ b/docs/GETTING_STARTED.md @@ -23,6 +23,7 @@ examples/hello-rover/ robot.yaml # the manifest robot.dev.yaml # dev-only overlay: local path pins structure.urdf # the robot's own URDF (chassis + mounts) + worlds/tool-boundary.wbt # minimal framework-owned Webots world components/wheel_drive/ # one robot-local component definition component.yaml structure.urdf @@ -33,7 +34,7 @@ Both wheels are instances of the same `wheel_drive` component type. Motion is commanded through the official manual or navigation candidate contracts; the example intentionally carries no always-moving cruise service. It is deliberately small: no sensors beyond what the kinematic model needs, -no simulation world, and no hardware driver. +one minimal simulation world, and no hardware driver. Use it as the shape to copy from, not as a real robot. ## `robot.yaml` anatomy @@ -288,10 +289,17 @@ phoxal-cli validate --report --allow-user-service-drift # lower-level structur `check` resolves every official participant (services, tools, and any catalog-sourced component drivers) plus every user service and prints `ok: N participants validated` once the whole graph is coherent. -`hello-rover` has no Webots world checked in (its component uses primitive -URDF geometry, not meshes, so there is nothing to render yet); a project that -wants `phoxal-cli simulate ` needs a `worlds/.wbt` file, the -same way [`robot-v1`](https://github.com/phoxal/robot-v1) does. +`hello-rover` includes the minimal `tool-boundary` Webots world used to verify +that official tools remain host/event driven while robot services follow +logical simulation time: + +```sh +phoxal-cli simulation run tool-boundary --env dev +``` + +The CLI stages the authored world and injects the generated rover and Webots +controllers. Keep additional example worlds equally small and owned by the +behavior they validate; do not use a private robot as the framework harness. ## Where to go from here diff --git a/docs/SIMULATION.md b/docs/SIMULATION.md index 1c4f86c6..28b6fedb 100644 --- a/docs/SIMULATION.md +++ b/docs/SIMULATION.md @@ -46,6 +46,12 @@ compare timestamps in one domain. A pure-bus participant seeds its simulation scheduler at logical zero, because the feed publishes 0-based simulation time and logical time only advances forward. +Official tools are the deliberate exception: they remain host/event driven in +simulation exactly as in Run and Deploy. They do not subscribe to +`simulation/clock`, join the Webots step barrier, or derive external-input +freshness from logical time. Services retain and sample their latest tool input +at their own logical step boundary. + ## Acceptance record - single robot (2026-07-12) Command: `phoxal-cli simulation run default --env dev` in the `robot-v1` reference diff --git a/examples/hello-rover/worlds/tool-boundary.wbt b/examples/hello-rover/worlds/tool-boundary.wbt new file mode 100644 index 00000000..713658c4 --- /dev/null +++ b/examples/hello-rover/worlds/tool-boundary.wbt @@ -0,0 +1,21 @@ +#VRML_SIM R2025a utf8 + +EXTERNPROTO "https://raw.githubusercontent.com/cyberbotics/webots/R2025a/projects/objects/floors/protos/RectangleArena.proto" +EXTERNPROTO "https://raw.githubusercontent.com/cyberbotics/webots/R2025a/projects/objects/backgrounds/protos/TexturedBackground.proto" +EXTERNPROTO "https://raw.githubusercontent.com/cyberbotics/webots/R2025a/projects/objects/backgrounds/protos/TexturedBackgroundLight.proto" + +WorldInfo { + basicTimeStep 16 +} +Viewpoint { + orientation -0.36 -0.22 0.91 5.05 + position -2.5 3.5 2.2 +} +TexturedBackground { +} +TexturedBackgroundLight { +} +RectangleArena { + floorSize 4 4 + wallHeight 0.25 +} diff --git a/phoxal/README.md b/phoxal/README.md index 2cf3c545..9a424b66 100644 --- a/phoxal/README.md +++ b/phoxal/README.md @@ -80,6 +80,17 @@ Key rules the example shows: The runner also owns the rest of the lifecycle: `#[step(hz = ...)]` is the scheduled control loop, `#[server]` / `#[server_snapshot]` serve queries, and `#[shutdown]` runs graceful park/stop/flush before the bus closes. +`#[phoxal::tool]` is intentionally outside the robot clock. Tools use managed +host/event loops, host-monotonic timers for cadence and freshness, and +`phoxal::raw::host_time()` only when a generic bus envelope timestamp is +required. The macro rejects `#[step]`, the setup context exposes no clock, and +the runner never enrolls a tool in simulation scheduling or a logical-step +barrier. Official tool sources are additionally checked against importing or +subscribing to simulation-clock surfaces; user-authored tools must preserve the +same boundary when using the privileged raw bus. A service that consumes +asynchronous tool input keeps a bounded latest value with consumer-local +monotonic arrival time and samples it at its own logical step. + ## Modules A participant depends on two crates: the `phoxal` engine (the modules below) and the `phoxal-api` contract tree. diff --git a/phoxal/src/lib.rs b/phoxal/src/lib.rs index 63037d1c..14c86ac2 100644 --- a/phoxal/src/lib.rs +++ b/phoxal/src/lib.rs @@ -195,6 +195,19 @@ pub mod bus { pub mod raw { pub use crate::participant::runner::run_with_bus; pub use phoxal_bus::*; + + /// Current host wall time for privileged tools and bridges that need an + /// envelope timestamp without joining a robot's logical clock. + /// + /// This is metadata time only. Scheduling periodic work should use a host + /// monotonic timer such as [`std::time::Instant`] or + /// [`tokio::time::interval`]. + pub fn host_time() -> LogicalTime { + let elapsed = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default(); + LogicalTime::new(0, u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX)) + } } /// The framework result type (`anyhow`-backed). Authoring code uses bare diff --git a/phoxal/src/participant/api.rs b/phoxal/src/participant/api.rs index 7d8d4816..7476302c 100644 --- a/phoxal/src/participant/api.rs +++ b/phoxal/src/participant/api.rs @@ -660,23 +660,18 @@ impl SetupContextSimulatorExt for SetupContext } /// Tool-only `SetupContext` accessor (`R: Participant + IsTool`). Tools stay -/// raw-bus only (decided 2026-07-09), so this is their sole IO seam. +/// raw-bus only (decided 2026-07-09), so this is their sole IO seam. Tools do +/// not receive the runner clock: host timers and [`crate::raw::host_time`] keep +/// them outside robot logical/simulation time. pub trait SetupContextToolExt { /// Clone the runner-owned raw bus for privileged tool internals. The bus is /// already open from the launch contract, so a tool does not reparse launch /// env or open an unrelated session. fn raw_bus(&self) -> Bus; - - /// Clone the logical clock owned by the participant runner. - fn clock(&self) -> std::sync::Arc; } impl SetupContextToolExt for SetupContext { fn raw_bus(&self) -> Bus { self.bus().clone() } - - fn clock(&self) -> std::sync::Arc { - self.clock_source() - } } diff --git a/phoxal/src/participant/context.rs b/phoxal/src/participant/context.rs index 8fdb92bb..0e74d48a 100644 --- a/phoxal/src/participant/context.rs +++ b/phoxal/src/participant/context.rs @@ -8,7 +8,6 @@ use std::time::Duration; use crate::bus::{LogicalTime, OwnerCap}; use crate::model::v0::Robot; -use crate::participant::clock::ClockSource; use crate::participant::managed::{ManagedTaskPolicy, ManagedTasks}; use phoxal_bus::Bus; @@ -18,7 +17,6 @@ use phoxal_bus::Bus; /// enforces. pub struct SetupContext { bus: Bus, - clock: Arc, owner_cap: OwnerCap, robot: Option>, robot_root: Option, @@ -34,7 +32,6 @@ pub struct SetupContext { impl SetupContext { pub(crate) fn new( bus: Bus, - clock: Arc, owner_cap: OwnerCap, robot: Option>, robot_root: Option, @@ -42,7 +39,6 @@ impl SetupContext { ) -> Self { SetupContext { bus, - clock, owner_cap, robot, robot_root, @@ -114,12 +110,6 @@ impl SetupContext { &self.bus } - /// Clone the runner-owned logical clock so privileged raw-bus tools can - /// stamp messages in the active real or simulation time domain. - pub(crate) fn clock_source(&self) -> Arc { - Arc::clone(&self.clock) - } - /// The runner-minted owner capability (plan #00 L2). In-crate accessor the /// NEW model's `SetupContextApiExt::owner_capability` (`participant::api`) /// reads the raw field through. diff --git a/phoxal/src/participant/runner.rs b/phoxal/src/participant/runner.rs index ac218d02..8e227845 100644 --- a/phoxal/src/participant/runner.rs +++ b/phoxal/src/participant/runner.rs @@ -300,10 +300,19 @@ where S: Future, { let schedule = R::__step_schedule(); - let (scheduler, clock_handle) = step_scheduler_for(launch.clock, schedule, clock.now()); - let effective_clock = Arc::new(match &scheduler { - AnyStepScheduler::Simulation(sim) => RunnerClock::Simulation(sim.simulation_clock()), - AnyStepScheduler::Real(_) => RunnerClock::Delegated(clock), + let clock_mode = effective_clock_mode(R::KIND, launch.clock); + let tool_clock = (R::KIND == "tool").then(RealClock::new); + let scheduler_now = tool_clock + .as_ref() + .map_or_else(|| clock.now(), ClockSource::now); + let (scheduler, clock_handle) = step_scheduler_for(clock_mode, schedule, scheduler_now); + let effective_clock = Arc::new(if let Some(tool_clock) = tool_clock { + RunnerClock::Host(tool_clock) + } else { + match &scheduler { + AnyStepScheduler::Simulation(sim) => RunnerClock::Simulation(sim.simulation_clock()), + AnyStepScheduler::Real(_) => RunnerClock::Delegated(clock), + } }); // Subscribe before setup so every lifecycle heartbeat uses the effective // clock domain and the simulation clock can advance while setup runs. @@ -334,16 +343,31 @@ where result } +/// Tools are host/event driven in every launch mode. A generic orchestrator +/// may still pass `PHOXAL_CLOCK=simulation`, but the runner structurally +/// prevents a tool from creating a simulation scheduler, subscribing to the +/// simulation clock, or stamping lifecycle state in that domain. +fn effective_clock_mode(participant_kind: &str, requested: ClockMode) -> ClockMode { + if participant_kind == "tool" { + ClockMode::Real + } else { + requested + } +} + /// The runner's effective timestamp clock, chosen once the scheduler is built. /// -/// In real mode it simply delegates to the caller-provided [`ClockSource`] (the -/// host [`RealClock`], or a [`TestClock`](crate::participant::clock::TestClock) -/// under the test harness). In simulation mode it is the +/// Tools always use an internally owned host [`RealClock`], even if an embedding +/// caller injects a different source. For other participants, real mode delegates +/// to the caller-provided [`ClockSource`] (the host [`RealClock`], or a +/// [`TestClock`](crate::participant::clock::TestClock) under the test harness). +/// In simulation mode it is the /// [`SimulationClock`](crate::participant::clock::SimulationClock) that shares /// the `SimulationScheduler`'s live `simulation/clock` feed, so stamped time and /// step-release time stay in the one simulation domain (see the `SimulationClock` /// docs for why wall-stamping a simulated participant is wrong). enum RunnerClock { + Host(RealClock), Delegated(C), Simulation(crate::participant::clock::SimulationClock), } @@ -351,6 +375,7 @@ enum RunnerClock { impl ClockSource for RunnerClock { fn now(&self) -> LogicalTime { match self { + RunnerClock::Host(clock) => clock.now(), RunnerClock::Delegated(clock) => clock.now(), RunnerClock::Simulation(clock) => clock.now(), } @@ -391,7 +416,6 @@ where // `api::topic::internal::new(cap)`). The runner is the only minter. let mut ctx = SetupContext::::new( bus.clone(), - Arc::clone(&clock) as Arc, ::phoxal_bus::OwnerCap::__mint(), robot, launch.robot_root.clone(), @@ -908,6 +932,22 @@ mod tests { ); } + #[test] + fn tools_always_use_real_clock_mode() { + assert_eq!( + effective_clock_mode("tool", ClockMode::Simulation), + ClockMode::Real + ); + assert_eq!( + effective_clock_mode("tool", ClockMode::Real), + ClockMode::Real + ); + assert_eq!( + effective_clock_mode("service", ClockMode::Simulation), + ClockMode::Simulation + ); + } + #[test] fn logical_step_deadline_skips_collapsed_ticks() { let next = advance_logical_deadline(LogicalTime::new(0, 10), Duration::from_nanos(10), 3); diff --git a/phoxal/tests/runner.rs b/phoxal/tests/runner.rs index e1a0848e..e3b8c1c5 100644 --- a/phoxal/tests/runner.rs +++ b/phoxal/tests/runner.rs @@ -42,6 +42,8 @@ static COUNTER_STEPS: AtomicU64 = AtomicU64::new(0); static SHUTDOWN_CALLED: AtomicBool = AtomicBool::new(false); static SLOW_SHUTDOWN_COMPLETED: AtomicBool = AtomicBool::new(false); static SIM_CLOCK_STEPS: AtomicU64 = AtomicU64::new(0); +static HOST_TOOL_TICKS: AtomicU64 = AtomicU64::new(0); +static HOST_TOOL_MESSAGES: AtomicU64 = AtomicU64::new(0); static SIM_CLOCK_CONTEXTS: Mutex> = Mutex::new(Vec::new()); /// A fresh in-process namespace per test invocation, so concurrently-run @@ -695,6 +697,39 @@ impl RobotInspector { } } +#[phoxal::tool(id = "host-driven-tool")] +struct HostDrivenTool; + +#[phoxal::behavior] +impl HostDrivenTool { + #[setup] + async fn setup(ctx: &mut SetupContext) -> Result<(Self, Self::Api)> { + let bus = ctx.raw_bus(); + let manual = phoxal::raw::Subscriber::::new( + &bus, + &api::topic::internal::new(ctx.owner_capability()) + .motion() + .manual(), + 8, + ) + .await?; + ctx.spawn_managed("host-ticker", async { + let mut interval = tokio::time::interval(Duration::from_millis(10)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + HOST_TOOL_TICKS.fetch_add(1, Ordering::Relaxed); + } + }); + ctx.spawn_managed("raw-subscriber", async move { + while manual.recv().await.is_ok() { + HOST_TOOL_MESSAGES.fetch_add(1, Ordering::Relaxed); + } + }); + Ok((Self, ())) + } +} + #[derive(serde::Deserialize, phoxal::Config)] struct ConfiguredInspectorConfig { label: String, @@ -736,6 +771,71 @@ async fn configless_tool_accepts_absent_config_but_configured_tool_rejects_it() ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn tool_ignores_simulation_clock_mode_and_keeps_host_work_running() { + HOST_TOOL_TICKS.store(0, Ordering::Relaxed); + HOST_TOOL_MESSAGES.store(0, Ordering::Relaxed); + let participant_id = "host-driven-tool-1"; + let namespace = unique_namespace("host-driven-tool"); + let mut bus_config = BusConfig::in_process(namespace.clone(), "robot"); + bus_config.participant = participant_id.to_string(); + let bus = Bus::open(bus_config).await.expect("bus should open"); + let heartbeat_topic = api::topic::internal::new(OwnerCap::__mint()) + .presence() + .heartbeat(); + let heartbeats = Subscriber::::new(&bus, &heartbeat_topic, 8) + .await + .expect("heartbeat subscriber should attach"); + let manual = phoxal::raw::Publisher::new(bus.clone(), &api::topic::new().motion().manual()) + .expect("manual publisher should attach"); + + let mut launch = ParticipantLaunch::local(participant_id, "robot"); + launch.namespace = namespace; + launch.clock = ClockMode::Simulation; + let injected_clock = TestClock::new(); + run_with_bus::(&bus, launch, injected_clock, async move { + tokio::time::sleep(Duration::from_millis(20)).await; + manual + .publish_at( + LogicalTime::new(7, 42), + api::motion::ManualCommand { + linear_x_mps: 0.2, + angular_z_radps: 0.0, + }, + ) + .await + .expect("raw tool input should publish while logical time is paused"); + tokio::time::sleep(Duration::from_millis(80)).await; + }) + .await + .expect("tool should not wait for a simulation clock feed"); + + assert!( + HOST_TOOL_TICKS.load(Ordering::Relaxed) >= 2, + "host ticker must continue without simulation clock samples" + ); + assert_eq!( + HOST_TOOL_MESSAGES.load(Ordering::Relaxed), + 1, + "raw tool subscriptions must continue without simulation clock samples" + ); + let mut produced_at = Vec::new(); + while let Some(received) = heartbeats.try_recv() { + if received.body.participant == participant_id { + produced_at.push(received.metadata.produced_at_ns); + } + } + assert!( + !produced_at.is_empty(), + "tool heartbeats should be observed" + ); + assert!( + produced_at.iter().all(|at| *at > 1_000_000_000_000_000_000), + "tool lifecycle timestamps must stay in host time, got {produced_at:?}" + ); + bus.close().await.expect("bus should close"); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn runner_runs_steps_then_shuts_down_cleanly() { let launch = ParticipantLaunch::local("counter-1", "robot"); diff --git a/phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs new file mode 100644 index 00000000..933d3bab --- /dev/null +++ b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs @@ -0,0 +1,15 @@ +use phoxal::prelude::*; + +#[phoxal::tool(id = "clocked-tool")] +struct ClockedTool; + +#[phoxal::behavior] +impl ClockedTool { + #[setup] + async fn setup(ctx: &mut SetupContext) -> Result<(Self, Self::Api)> { + let _clock = ctx.clock(); + Ok((Self, ())) + } +} + +fn main() {} diff --git a/phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr new file mode 100644 index 00000000..125afeab --- /dev/null +++ b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr @@ -0,0 +1,5 @@ +error[E0599]: no method named `clock` found for mutable reference `&mut phoxal::participant::SetupContext` in the current scope + --> tests/trybuild/fail/tool_cannot_access_clock.rs:10:26 + | +10 | let _clock = ctx.clock(); + | ^^^^^ method not found in `&mut phoxal::participant::SetupContext` diff --git a/service/motion/Cargo.toml b/service/motion/Cargo.toml index 8932584e..1077527e 100644 --- a/service/motion/Cargo.toml +++ b/service/motion/Cargo.toml @@ -21,6 +21,7 @@ anyhow = { workspace = true } [dev-dependencies] serde_json = { workspace = true } +tokio = { workspace = true } [lints] workspace = true diff --git a/service/motion/src/arbitration.rs b/service/motion/src/arbitration.rs index b5af8433..90ff0286 100644 --- a/service/motion/src/arbitration.rs +++ b/service/motion/src/arbitration.rs @@ -1,12 +1,20 @@ //! Body-twist arbitration for manual and navigation candidates. +use std::time::{Duration, Instant}; + use phoxal::bus::LogicalTime; use phoxal::model::robot::v0::MotionLimits; use phoxal_api::v1 as api; -pub(crate) const MANUAL_STALE_NS: u64 = 500_000_000; +pub(crate) const MANUAL_STALE: Duration = Duration::from_millis(150); pub(crate) const AUTONOMOUS_STALE_NS: u64 = 500_000_000; +#[derive(Clone)] +pub(crate) struct ManualCandidate { + pub(crate) body: api::motion::ManualCommand, + pub(crate) received_at: Instant, +} + #[derive(Clone)] pub(crate) struct Timed { pub(crate) body: T, @@ -21,12 +29,13 @@ pub(crate) struct Arbitration { } pub(crate) fn arbitrate( - manual: Option<&Timed>, + manual: Option<&ManualCandidate>, autonomous: Option<&Timed>, emergency_stop_engaged: bool, safety: Option<&Timed>, limits: MotionLimits, now: LogicalTime, + host_now: Instant, ) -> Arbitration { if emergency_stop_engaged { return zero( @@ -35,16 +44,16 @@ pub(crate) fn arbitrate( ); } - if manual.is_some_and(|candidate| { - candidate.at.epoch() == now.epoch() && candidate.at.time_ns() > now.time_ns() - }) { + if manual.is_some_and(|candidate| candidate.received_at > host_now) { return zero( Some(api::motion::Source::Manual), api::motion::ZeroReason::ManualCandidateFromFuture, ); } - if let Some(manual) = manual.filter(|candidate| is_fresh(candidate.at, now, MANUAL_STALE_NS)) { + if let Some(manual) = + manual.filter(|candidate| host_now.duration_since(candidate.received_at) <= MANUAL_STALE) + { // Manual teleoperation is the recovery/commissioning path and must not // depend on the still-experimental world-safety provider being ready. // A valid protective stop still wins, and every manual command remains @@ -166,6 +175,16 @@ pub(crate) fn candidate_age_ns(candidate: Option<&Timed>, now: LogicalTime }) } +pub(crate) fn manual_candidate_age_ns( + candidate: Option<&ManualCandidate>, + now: Instant, +) -> Option { + candidate.and_then(|candidate| { + now.checked_duration_since(candidate.received_at) + .map(|age| u64::try_from(age.as_nanos()).unwrap_or(u64::MAX)) + }) +} + fn select( source: api::motion::Source, linear_x_mps: f64, @@ -235,6 +254,20 @@ mod tests { LogicalTime::new(0, NOW_NS) } + fn host_now() -> Instant { + Instant::now() + } + + fn manual(linear_x_mps: f64, angular_z_radps: f64, received_at: Instant) -> ManualCandidate { + ManualCandidate { + body: api::motion::ManualCommand { + linear_x_mps, + angular_z_radps, + }, + received_at, + } + } + fn safety() -> Timed { Timed { body: api::safety::MotionConstraints { @@ -285,7 +318,15 @@ mod tests { let mut safety = safety(); safety.body.constraints.push(safety_constraint(true)); let autonomous = autonomous(); - let result = arbitrate(None, Some(&autonomous), false, Some(&safety), LIMITS, now()); + let result = arbitrate( + None, + Some(&autonomous), + false, + Some(&safety), + LIMITS, + now(), + host_now(), + ); assert_eq!( result.zero_reason, Some(api::motion::ZeroReason::SafetyConstraintsUnavailable) @@ -310,7 +351,15 @@ mod tests { mutate(&mut constraint); safety.body.constraints.push(constraint); let autonomous = autonomous(); - let result = arbitrate(None, Some(&autonomous), false, Some(&safety), LIMITS, now()); + let result = arbitrate( + None, + Some(&autonomous), + false, + Some(&safety), + LIMITS, + now(), + host_now(), + ); assert_eq!( result.zero_reason, Some(api::motion::ZeroReason::SafetyConstraintsUnavailable) @@ -320,7 +369,15 @@ mod tests { #[test] fn missing_candidates_are_explained() { - let result = arbitrate(None, None, false, Some(&safety()), LIMITS, now()); + let result = arbitrate( + None, + None, + false, + Some(&safety()), + LIMITS, + now(), + host_now(), + ); assert_eq!(result.source, None); assert_eq!( result.zero_reason, @@ -330,14 +387,17 @@ mod tests { #[test] fn stale_manual_is_distinct_from_missing_manual() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.2, - angular_z_radps: 0.1, - }, - at: LogicalTime::new(0, NOW_NS - MANUAL_STALE_NS - 1), - }; - let result = arbitrate(Some(&manual), None, false, Some(&safety()), LIMITS, now()); + let host_now = host_now(); + let manual = manual(0.2, 0.1, host_now - MANUAL_STALE - Duration::from_nanos(1)); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&safety()), + LIMITS, + now(), + host_now, + ); assert_eq!( result.zero_reason, Some(api::motion::ZeroReason::ManualCandidateStale) @@ -346,14 +406,8 @@ mod tests { #[test] fn emergency_stop_overrides_manual_and_navigation() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.2, - angular_z_radps: 0.1, - }, - at: LogicalTime::new(0, NOW_NS), - }; - let result = arbitrate(Some(&manual), None, true, None, LIMITS, now()); + let manual = manual(0.2, 0.1, host_now()); + let result = arbitrate(Some(&manual), None, true, None, LIMITS, now(), host_now()); assert_eq!(result.selected, zero_target()); assert_eq!( result.zero_reason, @@ -363,14 +417,16 @@ mod tests { #[test] fn manual_wins_and_is_clamped_to_robot_limits() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 9.0, - angular_z_radps: -9.0, - }, - at: LogicalTime::new(0, NOW_NS), - }; - let result = arbitrate(Some(&manual), None, false, Some(&safety()), LIMITS, now()); + let manual = manual(9.0, -9.0, host_now()); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&safety()), + LIMITS, + now(), + host_now(), + ); assert_eq!(result.source, Some(api::motion::Source::Manual)); assert_eq!(result.selected.linear_x_mps, 0.6); assert_eq!(result.selected.angular_z_radps, -2.0); @@ -378,14 +434,16 @@ mod tests { #[test] fn non_finite_candidate_stops() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: f64::NAN, - angular_z_radps: 0.0, - }, - at: LogicalTime::new(0, NOW_NS), - }; - let result = arbitrate(Some(&manual), None, false, Some(&safety()), LIMITS, now()); + let manual = manual(f64::NAN, 0.0, host_now()); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&safety()), + LIMITS, + now(), + host_now(), + ); assert_eq!(result.selected, zero_target()); assert_eq!( result.zero_reason, @@ -395,14 +453,17 @@ mod tests { #[test] fn future_dated_manual_candidate_stops() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.2, - angular_z_radps: 0.0, - }, - at: LogicalTime::new(0, NOW_NS + 1), - }; - let result = arbitrate(Some(&manual), None, false, Some(&safety()), LIMITS, now()); + let host_now = host_now(); + let manual = manual(0.2, 0.0, host_now + Duration::from_nanos(1)); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&safety()), + LIMITS, + now(), + host_now, + ); assert_eq!( result.zero_reason, Some(api::motion::ZeroReason::ManualCandidateFromFuture) @@ -411,37 +472,33 @@ mod tests { #[test] fn large_finite_candidate_stays_finite_after_f32_conversion() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 1.0e300, - angular_z_radps: -1.0e300, - }, - at: LogicalTime::new(0, NOW_NS), - }; + let manual = manual(1.0e300, -1.0e300, host_now()); let limits = MotionLimits { max_linear_speed_mps: f64::from(f32::MAX), max_angular_speed_radps: f64::from(f32::MAX), }; - let result = arbitrate(Some(&manual), None, false, Some(&safety()), limits, now()); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&safety()), + limits, + now(), + host_now(), + ); assert!(result.selected.linear_x_mps.is_finite()); assert!(result.selected.angular_z_radps.is_finite()); } #[test] - fn prior_epoch_candidate_cannot_resurrect_after_clock_reset() { + fn logical_epoch_reset_does_not_expire_fresh_manual_input() { let limits = MotionLimits { max_linear_speed_mps: 0.6, max_angular_speed_radps: 2.0, }; - let prior_epoch = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.4, - angular_z_radps: 0.2, - }, - at: LogicalTime::new(1, NOW_NS), - }; + let manual = manual(0.4, 0.2, host_now()); let result = arbitrate( - Some(&prior_epoch), + Some(&manual), None, false, Some(&Timed { @@ -450,24 +507,16 @@ mod tests { }), limits, LogicalTime::new(2, NOW_NS), + host_now(), ); - assert_eq!( - result.zero_reason, - Some(api::motion::ZeroReason::ManualCandidateStale) - ); - assert_eq!(result.selected, zero_target()); + assert_eq!(result.source, Some(api::motion::Source::Manual)); + assert_eq!(result.selected.linear_x_mps, 0.4); } #[test] fn manual_works_without_safety_and_uses_valid_constraints_when_available() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.5, - angular_z_radps: 0.4, - }, - at: now(), - }; - let missing = arbitrate(Some(&manual), None, false, None, LIMITS, now()); + let manual = manual(0.5, 0.4, host_now()); + let missing = arbitrate(Some(&manual), None, false, None, LIMITS, now(), host_now()); assert_eq!(missing.source, Some(api::motion::Source::Manual)); assert_eq!(missing.selected.linear_x_mps, 0.5); @@ -483,6 +532,7 @@ mod tests { Some(&constrained), LIMITS, now(), + host_now(), ); assert_eq!(result.selected.linear_x_mps, 0.1); assert_eq!(result.selected.angular_z_radps, 0.4); @@ -491,7 +541,15 @@ mod tests { #[test] fn autonomous_still_requires_safety_constraints() { let autonomous = autonomous(); - let missing = arbitrate(None, Some(&autonomous), false, None, LIMITS, now()); + let missing = arbitrate( + None, + Some(&autonomous), + false, + None, + LIMITS, + now(), + host_now(), + ); assert_eq!( missing.zero_reason, Some(api::motion::ZeroReason::SafetyConstraintsUnavailable) @@ -500,17 +558,19 @@ mod tests { #[test] fn valid_safety_protective_stop_still_blocks_manual() { - let manual = Timed { - body: api::motion::ManualCommand { - linear_x_mps: 0.5, - angular_z_radps: 0.4, - }, - at: now(), - }; + let manual = manual(0.5, 0.4, host_now()); let mut stopped = safety(); stopped.body.stop = true; stopped.body.constraints.push(safety_constraint(true)); - let result = arbitrate(Some(&manual), None, false, Some(&stopped), LIMITS, now()); + let result = arbitrate( + Some(&manual), + None, + false, + Some(&stopped), + LIMITS, + now(), + host_now(), + ); assert_eq!(result.selected, zero_target()); assert_eq!( result.zero_reason, diff --git a/service/motion/src/main.rs b/service/motion/src/main.rs index f89c902a..9853bada 100644 --- a/service/motion/src/main.rs +++ b/service/motion/src/main.rs @@ -11,6 +11,9 @@ mod arbitration; +use std::sync::{Arc, Mutex}; +use std::time::Instant; + use anyhow::Result; use phoxal::model::component::v0::capability::Capability; use phoxal::model::robot::v0::MotionLimits; @@ -18,7 +21,9 @@ use phoxal::model::v0::Robot; use phoxal::prelude::*; use phoxal_api::v1 as api; -use crate::arbitration::{Timed, arbitrate, candidate_age_ns, safety_is_usable}; +use crate::arbitration::{ + ManualCandidate, Timed, arbitrate, candidate_age_ns, manual_candidate_age_ns, safety_is_usable, +}; const COMPONENT_ESTOP_STALE_NS: u64 = 1_000_000_000; @@ -88,6 +93,9 @@ impl EmergencyStopLatch { #[derive(phoxal::Api)] struct Api { + // Declares the typed graph subscription. The managed receiver owns the + // only draining clone so logical-step pauses cannot accumulate a replay + // backlog in this handle. manual: Subscriber, autonomous: Subscriber, software_estop: Subscriber, @@ -100,7 +108,7 @@ struct Api { #[phoxal::service(id = "motion", config = ())] struct Motion { limits: MotionLimits, - last_manual: Option>, + latest_manual: Arc>>, last_autonomous: Option>, estop: EmergencyStopLatch, last_safety_constraints: Option>, @@ -114,6 +122,15 @@ impl Motion { let robot = ctx.robot()?; let limits = robot.manifest.robot.motion_limits.validate()?; let estop_bindings = emergency_stop_bindings(robot); + let manual_subscriber = ctx + .subscriber(api::topic::internal::new(cap).motion().manual(), 32) + .await?; + let latest_manual = Arc::new(Mutex::new(None)); + let manual_slot = Arc::clone(&latest_manual); + let manual_receiver = manual_subscriber.clone(); + ctx.spawn_managed("manual-receiver", async move { + receive_manual_forever(manual_receiver, manual_slot).await; + }); let mut component_estops = Vec::with_capacity(estop_bindings.len()); for binding in &estop_bindings { @@ -132,15 +149,13 @@ impl Motion { Ok(( Self { limits, - last_manual: None, + latest_manual, last_autonomous: None, estop: EmergencyStopLatch::new(estop_bindings.len()), last_safety_constraints: None, }, Self::Api { - manual: ctx - .subscriber(api::topic::internal::new(cap).motion().manual(), 32) - .await?, + manual: manual_subscriber, autonomous: ctx .subscriber(api::topic::new().navigation().candidate(), 32) .await?, @@ -161,12 +176,8 @@ impl Motion { #[step(hz = 20)] async fn step(&mut self, api: &mut Self::Api, step: StepContext) -> Result<()> { - while let Some(received) = api.manual.try_recv() { - self.last_manual = Some(Timed { - body: received.body, - at: LogicalTime::new(received.metadata.epoch, received.metadata.produced_at_ns), - }); - } + let manual = latest_manual(&self.latest_manual); + let host_now = Instant::now(); while let Some(received) = api.autonomous.try_recv() { self.last_autonomous = Some(Timed { body: received.body, @@ -202,12 +213,13 @@ impl Motion { api::motion::SafetyRuntime::Absent }; let arbitration = arbitrate( - self.last_manual.as_ref(), + manual.as_ref(), self.last_autonomous.as_ref(), self.estop.engaged(step.time()), self.last_safety_constraints.as_ref(), self.limits, step.time(), + host_now, ); self.estop.finish_cycle(); @@ -218,10 +230,7 @@ impl Motion { .publish_at( step.time(), api::motion::State { - manual_candidate_age_ns: candidate_age_ns( - self.last_manual.as_ref(), - step.time(), - ), + manual_candidate_age_ns: manual_candidate_age_ns(manual.as_ref(), host_now), autonomous_candidate_age_ns: candidate_age_ns( self.last_autonomous.as_ref(), step.time(), @@ -253,6 +262,36 @@ impl Motion { } } +async fn receive_manual_forever( + subscriber: Subscriber, + latest: Arc>>, +) { + loop { + match subscriber.recv().await { + Ok(received) => store_manual(&latest, received.body, Instant::now()), + Err(_) => return, + } + } +} + +fn store_manual( + latest: &Mutex>, + body: api::motion::ManualCommand, + received_at: Instant, +) { + *latest + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = + Some(ManualCandidate { body, received_at }); +} + +fn latest_manual(latest: &Mutex>) -> Option { + latest + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() +} + fn state_target(target: &api::drive::Target) -> api::motion::Target { api::motion::Target { linear_x_mps: target.linear_x_mps, @@ -342,6 +381,92 @@ mod tests { assert_eq!(state.curvature_limit_radpm, drive.curvature_limit_radpm); } + #[test] + fn manual_receiver_slot_is_bounded_and_newest_wins() { + let latest = Mutex::new(None); + let first_at = Instant::now(); + store_manual( + &latest, + api::motion::ManualCommand { + linear_x_mps: 0.1, + angular_z_radps: 0.2, + }, + first_at, + ); + let second_at = first_at + std::time::Duration::from_millis(10); + store_manual( + &latest, + api::motion::ManualCommand { + linear_x_mps: 0.3, + angular_z_radps: 0.4, + }, + second_at, + ); + + let manual = latest_manual(&latest).expect("latest command should be retained"); + assert_eq!(manual.body.linear_x_mps, 0.3); + assert_eq!(manual.body.angular_z_radps, 0.4); + assert_eq!(manual.received_at, second_at); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn manual_receiver_runs_without_logical_steps_and_keeps_newest() { + let namespace = format!( + "test/motion-manual-receiver/{}-{}", + std::process::id(), + phoxal::raw::host_time().time_ns() + ); + let bus = phoxal::raw::Bus::open(phoxal::raw::BusConfig::in_process(namespace, "robot")) + .await + .expect("bus should open"); + let publisher = + phoxal::raw::Publisher::new(bus.clone(), &api::topic::new().motion().manual()) + .expect("manual publisher should attach"); + let subscriber = Subscriber::new( + &bus, + &api::topic::internal::new(phoxal::raw::OwnerCap::__mint()) + .motion() + .manual(), + 32, + ) + .await + .expect("manual subscriber should attach"); + let latest = Arc::new(Mutex::new(None)); + let receiver_slot = Arc::clone(&latest); + let receiver = tokio::spawn(async move { + receive_manual_forever(subscriber, receiver_slot).await; + }); + + for linear_x_mps in [0.1, 0.2, 0.3] { + publisher + .publish_at( + phoxal::raw::host_time(), + api::motion::ManualCommand { + linear_x_mps, + angular_z_radps: 0.0, + }, + ) + .await + .expect("manual command should publish"); + } + + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + if latest_manual(&latest).is_some_and(|manual| manual.body.linear_x_mps == 0.3) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("receiver should advance independently of logical steps"); + + let manual = latest_manual(&latest).expect("latest manual command should be retained"); + assert_eq!(manual.body.linear_x_mps, 0.3); + receiver.abort(); + bus.close().await.expect("bus should close"); + } + #[test] fn software_and_component_estops_latch_independently() { let mut latch = EmergencyStopLatch::new(2); diff --git a/tool/joypad/src/main.rs b/tool/joypad/src/main.rs index f33bca17..5389a5e6 100644 --- a/tool/joypad/src/main.rs +++ b/tool/joypad/src/main.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet}; use gilrs::{Button, EventType, Gamepad, GamepadId, Gilrs}; use phoxal::prelude::*; -use phoxal::raw::{Publisher, Subscriber}; +use phoxal::raw::{Publisher, Subscriber, host_time}; use phoxal_api::v1 as motion_api; use phoxal_api::v2 as api; @@ -29,7 +29,6 @@ impl ToolJoypad { async fn setup(ctx: &mut SetupContext) -> Result<(Self, Self::Api)> { let cap = ctx.owner_capability(); let bus = ctx.raw_bus(); - let clock = ctx.clock(); let manual_publisher = Publisher::new(bus.clone(), &motion_api::topic::new().motion().manual())?; @@ -48,7 +47,6 @@ impl ToolJoypad { devices_publisher, connect_subscriber, rescan_subscriber, - clock, ) .await }); @@ -93,7 +91,6 @@ async fn run_joypad( devices_publisher: Publisher, connect_subscriber: Subscriber, rescan_subscriber: Subscriber, - clock: std::sync::Arc, ) { let mut gilrs = match Gilrs::new() { Ok(gilrs) => Some(gilrs), @@ -109,29 +106,35 @@ async fn run_joypad( } else { registry.last_error = Some("gamepad backend unavailable".to_string()); } - publish_devices(&devices_publisher, ®istry, clock.now()).await; + publish_devices(&devices_publisher, ®istry, host_time()).await; let mut ticker = tokio::time::interval(std::time::Duration::from_secs_f64(1.0 / POLL_HZ)); - ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { tokio::select! { _ = ticker.tick() => { let mut changed = false; + let mut selected_disconnected = false; if let Some(gilrs) = gilrs.as_mut() { while let Some(event) = gilrs.next_event() { gilrs.update(&event); - changed |= apply_event(gilrs, &mut registry, event.id, &event.event); + let outcome = apply_event(gilrs, &mut registry, event.id, &event.event); + changed |= outcome.changed; + selected_disconnected |= outcome.selected_disconnected; } } + if selected_disconnected { + publish_zero(&manual_publisher).await; + } if changed { - publish_devices(&devices_publisher, ®istry, clock.now()).await; + publish_devices(&devices_publisher, ®istry, host_time()).await; } let Some(gilrs) = gilrs.as_ref() else { continue }; let Some(selected_id) = selected_gilrs_id(®istry) else { continue }; let command = command_from_gamepad(&gilrs.gamepad(selected_id)); - if let Err(error) = manual_publisher.publish_at(clock.now(), command).await { + if let Err(error) = manual_publisher.publish_at(host_time(), command).await { tracing::warn!(target: "tool_joypad", error = %error, "publish failed"); } } @@ -143,7 +146,7 @@ async fn run_joypad( } else { registry.last_error = Some("gamepad backend unavailable".to_string()); } - publish_devices(&devices_publisher, ®istry, clock.now()).await; + publish_devices(&devices_publisher, ®istry, host_time()).await; } Err(error) => { tracing::warn!(target: "tool_joypad", error = %error, "connect subscription failed"); @@ -155,11 +158,13 @@ async fn run_joypad( match received { Ok(_received) => { if let Some(gilrs) = gilrs.as_ref() { - rescan(gilrs, &mut registry); + if rescan(gilrs, &mut registry) { + publish_zero(&manual_publisher).await; + } } else { registry.last_error = Some("gamepad backend unavailable".to_string()); } - publish_devices(&devices_publisher, ®istry, clock.now()).await; + publish_devices(&devices_publisher, ®istry, host_time()).await; } Err(error) => { tracing::warn!(target: "tool_joypad", error = %error, "rescan subscription failed"); @@ -171,6 +176,16 @@ async fn run_joypad( } } +async fn publish_zero(publisher: &Publisher) { + let zero = motion_api::motion::ManualCommand { + linear_x_mps: 0.0, + angular_z_radps: 0.0, + }; + if let Err(error) = publisher.publish_at(host_time(), zero).await { + tracing::warn!(target: "tool_joypad", error = %error, "disconnect zero publish failed"); + } +} + fn has_compatible_mapping(gamepad: &Gamepad<'_>) -> bool { mapping_supports_manual_controls([ gamepad.button_code(Button::LeftTrigger).is_some(), @@ -221,7 +236,18 @@ fn devices_snapshot(registry: &Registry) -> api::joypad::Devices { /// Handle a decoded gilrs hotplug event, mutating `registry` in place. /// Returns `true` if the device set or selection changed (Devices should be /// republished). -fn apply_event(gilrs: &Gilrs, registry: &mut Registry, id: GamepadId, event: &EventType) -> bool { +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +struct RegistryChange { + changed: bool, + selected_disconnected: bool, +} + +fn apply_event( + gilrs: &Gilrs, + registry: &mut Registry, + id: GamepadId, + event: &EventType, +) -> RegistryChange { match event { EventType::Connected => { let stable_id = observe(gilrs, registry, id); @@ -233,10 +259,13 @@ fn apply_event(gilrs: &Gilrs, registry: &mut Registry, id: GamepadId, event: &Ev registry.selected = Some(stable_id); registry.last_error = None; } - true + RegistryChange { + changed: true, + selected_disconnected: false, + } } EventType::Disconnected => on_disconnected(registry, id), - _ => false, + _ => RegistryChange::default(), } } @@ -268,7 +297,8 @@ fn handle_connect(registry: &mut Registry, id: &str) { /// previously known set so a still-connected pad keeps its stable id, a /// newly seen pad is assigned one, and a pad no longer present is marked /// disconnected (not removed - it may reconnect later). -fn rescan(gilrs: &Gilrs, registry: &mut Registry) { +fn rescan(gilrs: &Gilrs, registry: &mut Registry) -> bool { + let selected_before = registry.selected.clone(); let mut seen: HashSet = HashSet::new(); for (id, _gamepad) in gilrs.gamepads() { seen.insert(id); @@ -283,6 +313,12 @@ fn rescan(gilrs: &Gilrs, registry: &mut Registry) { } } reconcile_selection(gilrs, registry); + selected_before.is_some_and(|selected| { + registry + .entries + .get(&selected) + .is_some_and(|entry| !entry.connected) + }) } /// If the current selection is no longer connected, clear it (reporting why) @@ -380,24 +416,32 @@ fn observe(gilrs: &Gilrs, registry: &mut Registry, id: GamepadId) -> String { stable_id } -fn on_disconnected(registry: &mut Registry, id: GamepadId) -> bool { - let mut disconnected: Option = None; - for (stable_id, entry) in registry.entries.iter_mut() { - if entry.gilrs_id == Some(id) { - entry.gilrs_id = None; - entry.connected = false; - disconnected = Some(stable_id.clone()); - break; - } - } - let Some(stable_id) = disconnected else { - return false; +fn on_disconnected(registry: &mut Registry, id: GamepadId) -> RegistryChange { + let stable_id = registry + .entries + .iter() + .find_map(|(stable_id, entry)| (entry.gilrs_id == Some(id)).then(|| stable_id.clone())); + let Some(stable_id) = stable_id else { + return RegistryChange::default(); + }; + disconnect_stable_id(registry, &stable_id) +} + +fn disconnect_stable_id(registry: &mut Registry, stable_id: &str) -> RegistryChange { + let Some(entry) = registry.entries.get_mut(stable_id) else { + return RegistryChange::default(); }; - if registry.selected.as_deref() == Some(stable_id.as_str()) { + entry.gilrs_id = None; + entry.connected = false; + let selected_disconnected = registry.selected.as_deref() == Some(stable_id); + if selected_disconnected { registry.selected = None; registry.last_error = Some(format!("selected device '{stable_id}' disconnected")); } - true + RegistryChange { + changed: true, + selected_disconnected, + } } /// Derive a STABLE wire id for a pad from its identity - NOT the @@ -626,6 +670,67 @@ mod tests { ); } + #[test] + fn selected_device_disconnect_requests_an_immediate_zero() { + let mut registry = Registry { + selected: Some("pad".to_string()), + ..Registry::default() + }; + registry.entries.insert( + "pad".to_string(), + PadEntry { + base_id: "pad".to_string(), + gilrs_id: None, + name: "Pad".to_string(), + connected: true, + mapped: true, + }, + ); + + let outcome = disconnect_stable_id(&mut registry, "pad"); + + assert_eq!( + outcome, + RegistryChange { + changed: true, + selected_disconnected: true, + } + ); + assert!(registry.selected.is_none()); + assert!(!registry.entries["pad"].connected); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn disconnect_zero_is_published_on_the_manual_contract() { + let bus = phoxal::raw::Bus::open(phoxal::raw::BusConfig::in_process( + format!("test/joypad-disconnect/{}", std::process::id()), + "robot", + )) + .await + .expect("bus should open"); + let publisher = Publisher::new(bus.clone(), &motion_api::topic::new().motion().manual()) + .expect("manual publisher should attach"); + let subscriber = Subscriber::new( + &bus, + &motion_api::topic::internal::new(phoxal::raw::OwnerCap::__mint()) + .motion() + .manual(), + 1, + ) + .await + .expect("manual subscriber should attach"); + + publish_zero(&publisher).await; + let received = tokio::time::timeout(std::time::Duration::from_secs(2), subscriber.recv()) + .await + .expect("zero command should arrive") + .expect("zero command should decode"); + + assert_eq!(received.body.linear_x_mps, 0.0); + assert_eq!(received.body.angular_z_radps, 0.0); + bus.close().await.expect("bus should close"); + } + #[test] fn all_manual_controls_are_required_for_compatibility() { assert!(!mapping_supports_manual_controls([true, true, true, false])); diff --git a/tool/router/src/main.rs b/tool/router/src/main.rs index 452567a5..c7e2f35e 100644 --- a/tool/router/src/main.rs +++ b/tool/router/src/main.rs @@ -3,7 +3,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use phoxal::prelude::*; -use phoxal::raw::{Bus, BusMetadata, LogicalTime, OwnerCap, Publisher}; +use phoxal::raw::{Bus, BusMetadata, OwnerCap, Publisher, host_time}; use phoxal_api::v1 as api; use phoxal_api::v2 as preview_api; @@ -240,7 +240,7 @@ async fn spawn_metrics(ctx: &mut SetupContext, router: &zenoh::Sessi ManagedTaskPolicy::FaultOnExit, async move { let mut ticker = tokio::time::interval(METRICS_WINDOW); - ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); // Tokio intervals yield their first tick immediately. Consume that // bootstrap tick so the first published snapshot covers a real // window instead of reporting a near-zero interval as one second. @@ -252,7 +252,7 @@ async fn spawn_metrics(ctx: &mut SetupContext, router: &zenoh::Sessi let samples = drain_window(&publish_counters); let metrics = build_metrics(&samples, window_ended.duration_since(window_started)); window_started = window_ended; - if let Err(error) = metrics_publisher.publish_at(now(), metrics).await { + if let Err(error) = metrics_publisher.publish_at(host_time(), metrics).await { tracing::warn!(target: "tool_router", error = %error, "router metrics publish failed"); } } @@ -460,17 +460,10 @@ async fn publish_uplink_state( publisher: &Publisher, state: api::bus::uplink::State, ) -> Result<()> { - publisher.publish_at(now(), state).await?; + publisher.publish_at(host_time(), state).await?; Ok(()) } -fn now() -> LogicalTime { - let elapsed = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default(); - LogicalTime::new(0, u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX)) -} - fn main() -> phoxal::Result<()> { phoxal::run::() } diff --git a/tool/telemetry/src/main.rs b/tool/telemetry/src/main.rs index 7de3f930..d85dc819 100644 --- a/tool/telemetry/src/main.rs +++ b/tool/telemetry/src/main.rs @@ -1,7 +1,7 @@ use std::time::Duration; use phoxal::prelude::*; -use phoxal::raw::{Bus, LogicalTime, OwnerCap, Publisher}; +use phoxal::raw::{Bus, OwnerCap, Publisher, host_time}; use phoxal_api::v2 as api; use sysinfo::{CpuRefreshKind, MemoryRefreshKind, RefreshKind, System}; @@ -51,6 +51,7 @@ async fn sample_host_forever(publisher: Publisher) { .with_memory(MemoryRefreshKind::nothing().with_ram()), ); let mut interval = tokio::time::interval(SAMPLE_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); // `sysinfo`'s CPU usage is a delta against the PREVIOUS refresh, so the // very first refresh has nothing to diff against and reads inaccurately @@ -64,7 +65,7 @@ async fn sample_host_forever(publisher: Publisher) { loop { interval.tick().await; let sample = sample_host(&mut system); - if let Err(error) = publisher.publish_at(now(), sample).await { + if let Err(error) = publisher.publish_at(host_time(), sample).await { tracing::warn!(target: "tool_telemetry", error = %error, "host telemetry publish failed"); } } @@ -86,13 +87,6 @@ fn sample_host(system: &mut System) -> api::telemetry::Host { } } -fn now() -> LogicalTime { - let elapsed = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default(); - LogicalTime::new(0, u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX)) -} - fn main() -> phoxal::Result<()> { phoxal::run::() } diff --git a/xtask/src/workspace.rs b/xtask/src/workspace.rs index 11c011d7..4c96c602 100644 --- a/xtask/src/workspace.rs +++ b/xtask/src/workspace.rs @@ -567,6 +567,8 @@ pub fn require_nonempty_artifacts(artifacts: &[OfficialArtifact]) -> Result<()> #[cfg(test)] mod tests { + use std::fs; + use serde_json::json; use super::*; @@ -579,6 +581,89 @@ mod tests { classify_manifest_path(&root(), &root().join(relative)) } + fn visit_rust_sources(directory: &Path, sources: &mut Vec) -> Result<()> { + for entry in fs::read_dir(directory) + .with_context(|| format!("failed to read {}", directory.display()))? + { + let path = entry?.path(); + if path.is_dir() { + visit_rust_sources(&path, sources)?; + } else if path.extension().is_some_and(|extension| extension == "rs") { + sources.push(path); + } + } + Ok(()) + } + + #[test] + fn official_tools_do_not_import_logical_clock_surfaces() -> Result<()> { + let workspace_root = Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .context("xtask manifest directory has no workspace parent")?; + let mut sources = Vec::new(); + visit_rust_sources(&workspace_root.join("tool"), &mut sources)?; + let forbidden = [ + "ClockMode", + "ClockSource", + "SimulationClock", + "StepContext", + "simulation().clock()", + ".clock()", + ]; + + let mut violations = Vec::new(); + for source in sources { + let body = fs::read_to_string(&source)?; + for token in forbidden { + if body.contains(token) { + violations.push(format!("{} imports or calls {token}", source.display())); + } + } + } + + assert!( + violations.is_empty(), + "tools must remain host/event driven:\n{}", + violations.join("\n") + ); + Ok(()) + } + + #[test] + fn official_periodic_tools_skip_missed_ticks() -> Result<()> { + let workspace_root = Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .context("xtask manifest directory has no workspace parent")?; + for tool in ["joypad", "router", "telemetry"] { + let source = workspace_root.join("tool").join(tool).join("src/main.rs"); + let body = fs::read_to_string(&source)?; + assert!( + body.contains("set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip)"), + "periodic tool {tool} must skip missed ticks instead of catching up" + ); + } + Ok(()) + } + + #[test] + fn hello_rover_contains_the_minimal_tool_boundary_world() -> Result<()> { + let workspace_root = Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .context("xtask manifest directory has no workspace parent")?; + let world = fs::read_to_string( + workspace_root.join("examples/hello-rover/worlds/tool-boundary.wbt"), + )?; + assert!(world.starts_with("#VRML_SIM R2025a utf8")); + assert!(world.contains("WorldInfo {")); + assert!(world.contains("basicTimeStep 16")); + assert!(world.contains("RectangleArena {")); + assert!( + !world.contains("controller "), + "the CLI must inject robot controllers; the boundary world stays neutral" + ); + Ok(()) + } + #[test] fn real_workspace_release_scope_is_valid() -> Result<()> { let workspace_manifest = Path::new(env!("CARGO_MANIFEST_DIR")) From 12ef9f7eeca4b40b66162c92cf68b0112065ea88 Mon Sep 17 00:00:00 2001 From: phoxal-release-bot Date: Thu, 16 Jul 2026 15:01:43 +0200 Subject: [PATCH 2/2] refactor(runtime): make tool launches clockless Signed-off-by: phoxal-release-bot --- docs/CONVENTIONS.md | 15 +- docs/GETTING_STARTED.md | 12 +- docs/SIMULATION.md | 6 - .../worlds/{tool-boundary.wbt => default.wbt} | 2 +- phoxal-macros/src/authoring.rs | 13 ++ phoxal/README.md | 13 +- phoxal/src/lib.rs | 2 +- phoxal/src/participant/api.rs | 4 + phoxal/src/participant/launch.rs | 212 ++++++++++++++---- phoxal/src/participant/runner.rs | 136 +++++------ phoxal/tests/interop.rs | 8 +- phoxal/tests/interop_dynamic.rs | 8 +- phoxal/tests/managed_tasks.rs | 17 +- phoxal/tests/robot_root.rs | 6 +- phoxal/tests/runner.rs | 46 ++-- .../default_bus_module_cannot_open_raw_bus.rs | 2 +- ...ault_bus_module_cannot_open_raw_bus.stderr | 15 +- .../trybuild/fail/tool_cannot_access_clock.rs | 12 +- .../fail/tool_cannot_access_clock.stderr | 25 ++- service/navigation/src/main.rs | 4 +- xtask/src/workspace.rs | 19 -- 21 files changed, 335 insertions(+), 242 deletions(-) rename examples/hello-rover/worlds/{tool-boundary.wbt => default.wbt} (97%) diff --git a/docs/CONVENTIONS.md b/docs/CONVENTIONS.md index bef9d7e8..d847f6fa 100644 --- a/docs/CONVENTIONS.md +++ b/docs/CONVENTIONS.md @@ -78,13 +78,14 @@ the only source of keys, and the wire body never appears in the key The runner owns one `ClockSource` and stamps every `StepContext` and every `produced_at_ns` from it, so participants in the robot clock share one domain ([`phoxal/src/participant/clock.rs`](../phoxal/src/participant/clock.rs)). -- Tools are outside that clock. They run from external events and host-monotonic - timers in Run, Webots, and Deploy, even if a generic launch requests - `ClockMode::Simulation`. Tool envelope metadata uses - `phoxal::raw::host_time()`; the runner never gives tools `StepContext` or - enrolls them in simulation scheduling. Official tool sources are checked - against simulation-clock imports; privileged user-authored raw-bus tools must - uphold the same rule and never decide freshness from simulator time. +- Tools are outside that clock. Their process launch contract has no `--clock` + flag or `PHOXAL_CLOCK` binding, and the normal embedding API accepts no clock + argument. They run from external events and host-monotonic timers in every + mode. Tool envelope metadata uses `phoxal::raw::host_time()`; the runner never + gives tools `StepContext` or enrolls them in logical scheduling. Official tool + sources are checked against simulation-clock imports; privileged user-authored + raw-bus tools must uphold the same rule and never decide freshness from robot + logical time. - A logical-time consumer of asynchronous external input owns retention and freshness. Keep only the latest bounded value, record its consumer-local monotonic arrival instant, and sample that value at the logical step. A diff --git a/docs/GETTING_STARTED.md b/docs/GETTING_STARTED.md index 8b801a17..29b2e043 100644 --- a/docs/GETTING_STARTED.md +++ b/docs/GETTING_STARTED.md @@ -23,7 +23,7 @@ examples/hello-rover/ robot.yaml # the manifest robot.dev.yaml # dev-only overlay: local path pins structure.urdf # the robot's own URDF (chassis + mounts) - worlds/tool-boundary.wbt # minimal framework-owned Webots world + worlds/default.wbt # minimal Webots world components/wheel_drive/ # one robot-local component definition component.yaml structure.urdf @@ -289,17 +289,15 @@ phoxal-cli validate --report --allow-user-service-drift # lower-level structur `check` resolves every official participant (services, tools, and any catalog-sourced component drivers) plus every user service and prints `ok: N participants validated` once the whole graph is coherent. -`hello-rover` includes the minimal `tool-boundary` Webots world used to verify -that official tools remain host/event driven while robot services follow -logical simulation time: +`hello-rover` includes a minimal Webots world: ```sh -phoxal-cli simulation run tool-boundary --env dev +phoxal-cli simulation run default --env dev ``` The CLI stages the authored world and injects the generated rover and Webots -controllers. Keep additional example worlds equally small and owned by the -behavior they validate; do not use a private robot as the framework harness. +controllers. Keep additional example worlds equally small; do not use a private +robot as the framework harness. ## Where to go from here diff --git a/docs/SIMULATION.md b/docs/SIMULATION.md index 28b6fedb..1c4f86c6 100644 --- a/docs/SIMULATION.md +++ b/docs/SIMULATION.md @@ -46,12 +46,6 @@ compare timestamps in one domain. A pure-bus participant seeds its simulation scheduler at logical zero, because the feed publishes 0-based simulation time and logical time only advances forward. -Official tools are the deliberate exception: they remain host/event driven in -simulation exactly as in Run and Deploy. They do not subscribe to -`simulation/clock`, join the Webots step barrier, or derive external-input -freshness from logical time. Services retain and sample their latest tool input -at their own logical step boundary. - ## Acceptance record - single robot (2026-07-12) Command: `phoxal-cli simulation run default --env dev` in the `robot-v1` reference diff --git a/examples/hello-rover/worlds/tool-boundary.wbt b/examples/hello-rover/worlds/default.wbt similarity index 97% rename from examples/hello-rover/worlds/tool-boundary.wbt rename to examples/hello-rover/worlds/default.wbt index 713658c4..fb68643c 100644 --- a/examples/hello-rover/worlds/tool-boundary.wbt +++ b/examples/hello-rover/worlds/default.wbt @@ -17,5 +17,5 @@ TexturedBackgroundLight { } RectangleArena { floorSize 4 4 - wallHeight 0.25 + wallHeight 1 } diff --git a/phoxal-macros/src/authoring.rs b/phoxal-macros/src/authoring.rs index 8afde060..053180d5 100644 --- a/phoxal-macros/src/authoring.rs +++ b/phoxal-macros/src/authoring.rs @@ -702,6 +702,17 @@ impl ParticipantKind { } } + fn launch_policy(self, phoxal: &TokenStream) -> TokenStream { + match self { + ParticipantKind::Tool => { + quote!(#phoxal::participant::launch::ToolParticipantLaunch) + } + ParticipantKind::Service | ParticipantKind::Driver | ParticipantKind::Simulator => { + quote!(#phoxal::participant::launch::ClockedParticipantLaunch) + } + } + } + /// Default `Api` type when `api = …` is not given: tools stay raw-bus /// only (decided 2026-07-09 - no typed tool `Api` until a real need /// appears), every other kind defaults to the local `Api` struct. @@ -810,6 +821,7 @@ pub fn expand_participant( let phoxal = phoxal(); let artifact_kind = kind.artifact_kind(); let participant_class = kind.participant_class(); + let launch_policy = kind.launch_policy(&phoxal); let marker = kind.marker_impl(&phoxal, struct_name); let metadata_const_ident = Ident::new( &format!( @@ -841,6 +853,7 @@ pub fn expand_participant( const KIND: &'static str = #artifact_kind; const PARTICIPANT_CLASS: &'static str = #participant_class; const ID: &'static str = #id; + type LaunchPolicy = #launch_policy; type Config = #config_ty; type Api = #api_ty; } diff --git a/phoxal/README.md b/phoxal/README.md index 9a424b66..dc890f17 100644 --- a/phoxal/README.md +++ b/phoxal/README.md @@ -84,12 +84,13 @@ The runner also owns the rest of the lifecycle: `#[step(hz = ...)]` is the sched host/event loops, host-monotonic timers for cadence and freshness, and `phoxal::raw::host_time()` only when a generic bus envelope timestamp is required. The macro rejects `#[step]`, the setup context exposes no clock, and -the runner never enrolls a tool in simulation scheduling or a logical-step -barrier. Official tool sources are additionally checked against importing or -subscribing to simulation-clock surfaces; user-authored tools must preserve the -same boundary when using the privileged raw bus. A service that consumes -asynchronous tool input keeps a bounded latest value with consumer-local -monotonic arrival time and samples it at its own logical step. +the tool process launch has no clock option. The normal embedding API likewise +accepts no clock argument; deterministic clock injection is restricted to typed +graph participants. Official tool sources are additionally checked against +importing or subscribing to simulation-clock surfaces; user-authored tools must +preserve the same boundary when using the privileged raw bus. A service that +consumes asynchronous tool input keeps a bounded latest value with +consumer-local monotonic arrival time and samples it at its own logical step. ## Modules diff --git a/phoxal/src/lib.rs b/phoxal/src/lib.rs index 14c86ac2..8e97991b 100644 --- a/phoxal/src/lib.rs +++ b/phoxal/src/lib.rs @@ -193,7 +193,7 @@ pub mod bus { /// graph checker still includes their contracts, but never lets their raw /// access satisfy checked topology. pub mod raw { - pub use crate::participant::runner::run_with_bus; + pub use crate::participant::runner::{run_with_bus, run_with_bus_clock}; pub use phoxal_bus::*; /// Current host wall time for privileged tools and bridges that need an diff --git a/phoxal/src/participant/api.rs b/phoxal/src/participant/api.rs index 7476302c..da1f48c4 100644 --- a/phoxal/src/participant/api.rs +++ b/phoxal/src/participant/api.rs @@ -336,6 +336,10 @@ pub trait Participant: Sized + Send + 'static { const PARTICIPANT_CLASS: &'static str; /// The participant id (`id = "…"`, default kebab of the type name). const ID: &'static str; + /// The process launch contract. Tools use a clockless policy; checked graph + /// participants use the configurable robot-clock policy. + #[doc(hidden)] + type LaunchPolicy: crate::participant::launch::ParticipantLaunchPolicy; /// The participant's typed config (`robot.yaml` input). type Config: ParticipantConfig; /// The participant's bus-facing contract surface (`()` for a raw-bus diff --git a/phoxal/src/participant/launch.rs b/phoxal/src/participant/launch.rs index 2d26ca5b..45941151 100644 --- a/phoxal/src/participant/launch.rs +++ b/phoxal/src/participant/launch.rs @@ -1,9 +1,11 @@ //! `ParticipantLaunch` - the clap/env process launch contract. //! -//! Every participant binary accepts the same documented `--flag` set with -//! matching `PHOXAL_*` env fallbacks. Supervisors and systemd units use env, -//! while humans can use flags for bench runs. Flags win over env through clap's -//! native precedence, and `--help` is the user-facing contract documentation. +//! Participant binaries share one common `--flag` set with matching `PHOXAL_*` +//! env fallbacks. Clocked graph participants additionally accept `--clock` / +//! `PHOXAL_CLOCK`; tools do not expose either input. Supervisors and systemd +//! units use env, while humans can use flags for bench runs. Flags win over env +//! through clap's native precedence, and `--help` is the user-facing contract +//! documentation. use std::path::PathBuf; @@ -30,7 +32,7 @@ pub mod env { pub const CONNECT: &str = "PHOXAL_CONNECT"; /// Inline JSON participant config block. pub const CONFIG: &str = "PHOXAL_CONFIG"; - /// Clock mode: real or simulation. + /// Clock mode for services, drivers, and simulators: real or simulation. pub const CLOCK: &str = "PHOXAL_CLOCK"; /// All env names in contract order. @@ -58,7 +60,7 @@ pub struct ParticipantLaunch { /// The resolved bus profile. #[serde(default)] pub bus: BusProfile, - /// The clock mode. + /// The robot clock mode. Tool launch policies never read this field. #[serde(default)] pub clock: ClockMode, /// The participant's typed config block (`services..config`), if any. @@ -101,16 +103,6 @@ impl ParticipantLaunch { } } - pub(crate) fn from_cli( - default_participant_id: &'static str, - default_robot_id: &'static str, - ) -> crate::Result { - let matches = - LaunchCli::command_for(default_participant_id, default_robot_id).get_matches(); - let cli = LaunchCli::from_arg_matches(&matches)?; - cli.into_launch(default_participant_id, default_robot_id) - } - /// Set the robot model root for this launch. pub fn with_robot_root(mut self, root: impl Into) -> Self { self.robot_root = Some(root.into()); @@ -124,14 +116,9 @@ impl ParticipantLaunch { } } -/// The clap-derived launch contract for participant binaries. -#[derive(Debug, clap::Parser)] -#[command( - name = "phoxal-participant", - about = "Run a Phoxal participant.", - long_about = None -)] -struct LaunchCli { +/// The clap-derived launch fields shared by every participant binary. +#[derive(Debug, clap::Args)] +struct CommonLaunchCli { /// Bus-unique participant id. Defaults to the compiled participant artifact id. #[arg( long, @@ -190,8 +177,20 @@ struct LaunchCli { value_name = "JSON" )] config: Option, +} + +/// Launch contract for services, drivers, and simulators. +#[derive(Debug, clap::Parser)] +#[command( + name = "phoxal-participant", + about = "Run a Phoxal participant.", + long_about = None +)] +struct ClockedLaunchCli { + #[command(flatten)] + common: CommonLaunchCli, - /// Clock mode for the runner. + /// Clock mode for robot-state execution. #[arg( long, env = env::CLOCK, @@ -202,18 +201,20 @@ struct LaunchCli { clock: ClockMode, } -impl LaunchCli { - fn command_for( - default_participant_id: &'static str, - default_robot_id: &'static str, - ) -> clap::Command { - Self::command() - .mut_arg("participant_id", |arg| { - arg.default_value(default_participant_id) - }) - .mut_arg("robot_id", |arg| arg.default_value(default_robot_id)) - } +/// Launch contract for host/event-driven tools. It intentionally has no clock +/// flag or environment binding. +#[derive(Debug, clap::Parser)] +#[command( + name = "phoxal-tool", + about = "Run a Phoxal tool.", + long_about = None +)] +struct ToolLaunchCli { + #[command(flatten)] + common: CommonLaunchCli, +} +impl CommonLaunchCli { fn into_launch( self, default_participant_id: &'static str, @@ -241,11 +242,77 @@ impl LaunchCli { .context("PHOXAL_CONFIG must be valid JSON for the participant config")?, ); } - launch.clock = self.clock; Ok(launch) } } +fn command_for( + default_participant_id: &'static str, + default_robot_id: &'static str, +) -> clap::Command { + C::command() + .mut_arg("participant_id", |arg| { + arg.default_value(default_participant_id) + }) + .mut_arg("robot_id", |arg| arg.default_value(default_robot_id)) +} + +/// Type-level launch contract emitted by the participant macros. +#[doc(hidden)] +pub trait ParticipantLaunchPolicy: Send + Sync + 'static { + fn from_cli( + default_participant_id: &'static str, + default_robot_id: &'static str, + ) -> crate::Result; + + fn clock_mode(launch: &ParticipantLaunch) -> ClockMode; +} + +/// Launch policy for services, drivers, and simulators. +#[doc(hidden)] +pub struct ClockedParticipantLaunch; + +impl ParticipantLaunchPolicy for ClockedParticipantLaunch { + fn from_cli( + default_participant_id: &'static str, + default_robot_id: &'static str, + ) -> crate::Result { + let matches = + command_for::(default_participant_id, default_robot_id).get_matches(); + let cli = ClockedLaunchCli::from_arg_matches(&matches)?; + let mut launch = cli + .common + .into_launch(default_participant_id, default_robot_id)?; + launch.clock = cli.clock; + Ok(launch) + } + + fn clock_mode(launch: &ParticipantLaunch) -> ClockMode { + launch.clock + } +} + +/// Clockless launch policy for tools. +#[doc(hidden)] +pub struct ToolParticipantLaunch; + +impl ParticipantLaunchPolicy for ToolParticipantLaunch { + fn from_cli( + default_participant_id: &'static str, + default_robot_id: &'static str, + ) -> crate::Result { + let matches = + command_for::(default_participant_id, default_robot_id).get_matches(); + let cli = ToolLaunchCli::from_arg_matches(&matches)?; + cli.common + .into_launch(default_participant_id, default_robot_id) + } + + fn clock_mode(_launch: &ParticipantLaunch) -> ClockMode { + ClockMode::Real + } +} + fn nonempty_or(value: Option, default: impl FnOnce() -> String) -> String { value .filter(|value| !value.is_empty()) @@ -299,19 +366,29 @@ mod tests { } } - fn parse_from(args: &[&str]) -> crate::Result { - let matches = LaunchCli::command_for("default-id", "robot") + fn parse_clocked_from(args: &[&str]) -> crate::Result { + let matches = command_for::("default-id", "robot") + .try_get_matches_from(args) + .map_err(anyhow::Error::from)?; + let cli = ClockedLaunchCli::from_arg_matches(&matches).map_err(anyhow::Error::from)?; + let mut launch = cli.common.into_launch("default-id", "robot")?; + launch.clock = cli.clock; + Ok(launch) + } + + fn parse_tool_from(args: &[&str]) -> crate::Result { + let matches = command_for::("default-id", "robot") .try_get_matches_from(args) .map_err(anyhow::Error::from)?; - let cli = LaunchCli::from_arg_matches(&matches).map_err(anyhow::Error::from)?; - cli.into_launch("default-id", "robot") + let cli = ToolLaunchCli::from_arg_matches(&matches).map_err(anyhow::Error::from)?; + cli.common.into_launch("default-id", "robot") } #[test] #[serial] fn cli_with_nothing_set_matches_local_defaults() { clear_env(); - let launch = parse_from(&["participant-bin"]).unwrap(); + let launch = parse_clocked_from(&["participant-bin"]).unwrap(); assert_eq!(launch.participant_id, "default-id"); assert_eq!(launch.robot_id, "robot"); assert_eq!(launch.namespace, "dev"); @@ -336,7 +413,7 @@ mod tests { std::env::set_var(env::CONFIG, r#"{"rate_hz":10}"#); std::env::set_var(env::CLOCK, "simulation"); } - let launch = parse_from(&["participant-bin"]).unwrap(); + let launch = parse_clocked_from(&["participant-bin"]).unwrap(); assert_eq!(launch.participant_id, "tof-3"); assert_eq!(launch.robot_id, "robot-a"); assert_eq!(launch.namespace, "lab"); @@ -373,7 +450,7 @@ mod tests { std::env::set_var(env::CLOCK, "simulation"); } - let launch = parse_from(&[ + let launch = parse_clocked_from(&[ "participant-bin", "--participant-id", "flag-participant", @@ -414,12 +491,12 @@ mod tests { clear_env(); // SAFETY: serialized test; see clear_env. unsafe { std::env::set_var(env::CONFIG, "not json") }; - assert!(parse_from(&["participant-bin"]).is_err()); + assert!(parse_clocked_from(&["participant-bin"]).is_err()); unsafe { std::env::remove_var(env::CONFIG); std::env::set_var(env::CLOCK, "wallclock"); } - let err = LaunchCli::command_for("default-id", "robot") + let err = command_for::("default-id", "robot") .try_get_matches_from(["participant-bin"]) .unwrap_err(); assert_eq!(err.kind(), ErrorKind::InvalidValue); @@ -434,7 +511,7 @@ mod tests { unsafe { std::env::set_var(env::CONFIG, r#"{"secret":"do-not-print"}"#) }; let mut help = Vec::new(); - LaunchCli::command_for("default-id", "robot") + command_for::("default-id", "robot") .write_long_help(&mut help) .unwrap(); let help = String::from_utf8(help).unwrap(); @@ -455,4 +532,45 @@ mod tests { assert!(!help.contains("do-not-print")); clear_env(); } + + #[test] + #[serial] + fn tool_cli_has_no_clock_input() { + clear_env(); + // A generic supervisor setting is invisible to the tool launch parser. + // SAFETY: serialized test; see clear_env. + unsafe { std::env::set_var(env::CLOCK, "simulation") }; + let launch = parse_tool_from(&["tool-bin"]).unwrap(); + assert_eq!(launch.clock, ClockMode::Real); + + let mut help = Vec::new(); + command_for::("default-id", "robot") + .write_long_help(&mut help) + .unwrap(); + let help = String::from_utf8(help).unwrap(); + assert!(!help.contains("--clock")); + assert!(!help.contains(env::CLOCK)); + + for arguments in [ + vec!["tool-bin", "--clock", "simulation"], + vec!["tool-bin", "--simulation"], + ] { + let error = command_for::("default-id", "robot") + .try_get_matches_from(arguments) + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::UnknownArgument); + } + + let mut programmatic = ParticipantLaunch::local("tool", "robot"); + programmatic.clock = ClockMode::Simulation; + assert_eq!( + ToolParticipantLaunch::clock_mode(&programmatic), + ClockMode::Real + ); + assert_eq!( + ClockedParticipantLaunch::clock_mode(&programmatic), + ClockMode::Simulation + ); + clear_env(); + } } diff --git a/phoxal/src/participant/runner.rs b/phoxal/src/participant/runner.rs index 8e227845..029afdcf 100644 --- a/phoxal/src/participant/runner.rs +++ b/phoxal/src/participant/runner.rs @@ -78,7 +78,7 @@ use crate::participant::bus_log::{self, BusLogState}; use crate::participant::clock::{ClockSource, RealClock}; use crate::participant::context::{SetupContext, ShutdownContext, StepContext}; use crate::participant::heartbeat::{self, HeartbeatPublisher}; -use crate::participant::launch::{ClockMode, ParticipantLaunch}; +use crate::participant::launch::{ClockMode, ParticipantLaunch, ParticipantLaunchPolicy}; use crate::participant::managed::{ManagedTaskExit, ManagedTasks}; use crate::participant::process_metrics::{self, ProcessMetricsPublisher}; use crate::participant::scheduler::{ @@ -104,27 +104,16 @@ pub fn run() -> crate::Result<()> { /// Async host runner for custom Tokio mains /// (`phoxal::tokio::run::().await`). pub async fn run_async() -> crate::Result<()> { - let launch = ParticipantLaunch::from_cli(R::ID, "robot")?; + let launch = R::LaunchPolicy::from_cli(R::ID, "robot")?; init_tracing(); - let clock = launch_clock(&launch)?; - run_with::(launch, clock, shutdown_signal()).await -} - -/// The caller-provided clock for a launch. Real mode uses it directly for -/// timestamps and scheduler anchoring. Simulation mode replaces it with the -/// clock derived from its [`SimulationScheduler`] so step release and every -/// lifecycle timestamp share the authoritative `simulation/clock` domain. -pub(crate) fn launch_clock(launch: &ParticipantLaunch) -> crate::Result { - match launch.clock { - ClockMode::Real | ClockMode::Simulation => Ok(RealClock::new()), - } + run_with::(launch, shutdown_signal()).await } /// Select the step scheduler for `clock_mode` (D34/#09): the seam that /// answers "when should the next `#[step]` tick fire", separate from the -/// [`ClockSource`] used for timestamps (see [`launch_clock`]). +/// [`ClockSource`] used for timestamps. /// /// [`ClockMode::Real`] preserves the runner's pre-#09 wall-clock cadence /// exactly, returning no driving handle (`None`). [`ClockMode::Simulation`] @@ -226,16 +215,12 @@ pub(crate) fn spawn_simulation_clock_feed( })) } -/// Run a participant against an explicit launch, clock, and shutdown trigger. The -/// seam the test harness + integration tests drive (D41). -pub async fn run_with( - launch: ParticipantLaunch, - clock: C, - shutdown: S, -) -> crate::Result<()> +/// Run a participant against an explicit launch and shutdown trigger. The +/// runner owns the host clock; tools therefore have no clock parameter to +/// receive or override. +pub async fn run_with(launch: ParticipantLaunch, shutdown: S) -> crate::Result<()> where R: ParticipantLifecycle, - C: ClockSource, S: Future, { init_tracing(); @@ -249,7 +234,7 @@ where }) .await?; - let result = run_with_bus::(&bus, launch, clock, shutdown).await; + let result = run_with_bus::(&bus, launch, shutdown).await; if let Err(e) = bus.close().await { tracing::warn!(target: "phoxal.runtime", error = %e, "bus close failed"); @@ -257,7 +242,7 @@ where result } -/// Run a participant on a **caller-owned** bus, against an explicit launch, clock, and +/// Run a participant on a **caller-owned** bus, against an explicit launch and /// shutdown trigger. Unlike [`run_with`], this does not open or close the bus - the /// caller controls its lifecycle. /// @@ -268,7 +253,39 @@ where /// sharing one [`Bus`] publish under that bus's participant id, so distinct /// per-participant source attribution still requires a bus per participant. The /// `launch` here drives config, robot-model, and component-instance resolution. -pub async fn run_with_bus( +pub async fn run_with_bus( + bus: &Bus, + launch: ParticipantLaunch, + shutdown: S, +) -> crate::Result<()> +where + R: ParticipantLifecycle, + S: Future, +{ + run_with_bus_inner::(bus, launch, RealClock::new(), shutdown).await +} + +/// Deterministic clock-injection seam for checked graph participants. A tool's +/// fixed [`ToolParticipantLaunch`](crate::participant::launch::ToolParticipantLaunch) +/// policy excludes it even if user code manually adds the public +/// [`TypedGraphSurface`](crate::participant::TypedGraphSurface) marker. +#[doc(hidden)] +pub async fn run_with_bus_clock( + bus: &Bus, + launch: ParticipantLaunch, + clock: C, + shutdown: S, +) -> crate::Result<()> +where + R: ParticipantLifecycle + + crate::participant::TypedGraphSurface, + C: ClockSource, + S: Future, +{ + run_with_bus_inner::(bus, launch, clock, shutdown).await +} + +async fn run_with_bus_inner( bus: &Bus, launch: ParticipantLaunch, clock: C, @@ -300,19 +317,11 @@ where S: Future, { let schedule = R::__step_schedule(); - let clock_mode = effective_clock_mode(R::KIND, launch.clock); - let tool_clock = (R::KIND == "tool").then(RealClock::new); - let scheduler_now = tool_clock - .as_ref() - .map_or_else(|| clock.now(), ClockSource::now); - let (scheduler, clock_handle) = step_scheduler_for(clock_mode, schedule, scheduler_now); - let effective_clock = Arc::new(if let Some(tool_clock) = tool_clock { - RunnerClock::Host(tool_clock) - } else { - match &scheduler { - AnyStepScheduler::Simulation(sim) => RunnerClock::Simulation(sim.simulation_clock()), - AnyStepScheduler::Real(_) => RunnerClock::Delegated(clock), - } + let clock_mode = R::LaunchPolicy::clock_mode(&launch); + let (scheduler, clock_handle) = step_scheduler_for(clock_mode, schedule, clock.now()); + let effective_clock = Arc::new(match &scheduler { + AnyStepScheduler::Simulation(sim) => RunnerClock::Simulation(sim.simulation_clock()), + AnyStepScheduler::Real(_) => RunnerClock::Delegated(clock), }); // Subscribe before setup so every lifecycle heartbeat uses the effective // clock domain and the simulation clock can advance while setup runs. @@ -343,31 +352,16 @@ where result } -/// Tools are host/event driven in every launch mode. A generic orchestrator -/// may still pass `PHOXAL_CLOCK=simulation`, but the runner structurally -/// prevents a tool from creating a simulation scheduler, subscribing to the -/// simulation clock, or stamping lifecycle state in that domain. -fn effective_clock_mode(participant_kind: &str, requested: ClockMode) -> ClockMode { - if participant_kind == "tool" { - ClockMode::Real - } else { - requested - } -} - /// The runner's effective timestamp clock, chosen once the scheduler is built. /// -/// Tools always use an internally owned host [`RealClock`], even if an embedding -/// caller injects a different source. For other participants, real mode delegates -/// to the caller-provided [`ClockSource`] (the host [`RealClock`], or a -/// [`TestClock`](crate::participant::clock::TestClock) under the test harness). -/// In simulation mode it is the +/// In real mode it delegates to the runner-owned host [`RealClock`] (or a +/// [`TestClock`](crate::participant::clock::TestClock) through the checked-graph +/// test seam). In simulation mode it is the /// [`SimulationClock`](crate::participant::clock::SimulationClock) that shares /// the `SimulationScheduler`'s live `simulation/clock` feed, so stamped time and /// step-release time stay in the one simulation domain (see the `SimulationClock` /// docs for why wall-stamping a simulated participant is wrong). enum RunnerClock { - Host(RealClock), Delegated(C), Simulation(crate::participant::clock::SimulationClock), } @@ -375,7 +369,6 @@ enum RunnerClock { impl ClockSource for RunnerClock { fn now(&self) -> LogicalTime { match self { - RunnerClock::Host(clock) => clock.now(), RunnerClock::Delegated(clock) => clock.now(), RunnerClock::Simulation(clock) => clock.now(), } @@ -900,17 +893,6 @@ pub(crate) fn bus_log_state() -> Arc { mod tests { use super::*; - #[test] - fn simulation_clock_launch_is_accepted() { - // #09: PHOXAL_CLOCK=simulation is no longer rejected - the runner - // selects a `SimulationScheduler` for it (see `step_scheduler_for`) - // instead of bailing before the bus even opens. - let mut launch = ParticipantLaunch::local("participant", "robot"); - launch.clock = ClockMode::Simulation; - - launch_clock(&launch).expect("simulation clock launch should be accepted"); - } - #[test] fn step_scheduler_for_selects_real_or_simulation_by_clock_mode() { let now = LogicalTime::new(0, 0); @@ -932,22 +914,6 @@ mod tests { ); } - #[test] - fn tools_always_use_real_clock_mode() { - assert_eq!( - effective_clock_mode("tool", ClockMode::Simulation), - ClockMode::Real - ); - assert_eq!( - effective_clock_mode("tool", ClockMode::Real), - ClockMode::Real - ); - assert_eq!( - effective_clock_mode("service", ClockMode::Simulation), - ClockMode::Simulation - ); - } - #[test] fn logical_step_deadline_skips_collapsed_ticks() { let next = advance_logical_deadline(LogicalTime::new(0, 10), Duration::from_nanos(10), 3); diff --git a/phoxal/tests/interop.rs b/phoxal/tests/interop.rs index 6fddb36b..67b34e11 100644 --- a/phoxal/tests/interop.rs +++ b/phoxal/tests/interop.rs @@ -10,7 +10,7 @@ use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::time::Duration; -use phoxal::participant::{ParticipantLaunch, RealClock}; +use phoxal::participant::ParticipantLaunch; use phoxal::prelude::*; use phoxal::raw::{Bus, BusConfig, run_with_bus}; use phoxal_api::v1 as api; @@ -108,16 +108,14 @@ async fn two_runtimes_exchange_a_contract_on_one_bus() { // Run both runtimes concurrently on the shared bus; each stops after a short // wall-clock window (enough for the 50 Hz producer to emit many samples). - let producer = run_with_bus::( + let producer = run_with_bus::( &bus, ParticipantLaunch::local("producer-1", "robot"), - RealClock::new(), async { tokio::time::sleep(Duration::from_millis(500)).await }, ); - let consumer = run_with_bus::( + let consumer = run_with_bus::( &bus, ParticipantLaunch::local("consumer-1", "robot"), - RealClock::new(), async { tokio::time::sleep(Duration::from_millis(500)).await }, ); diff --git a/phoxal/tests/interop_dynamic.rs b/phoxal/tests/interop_dynamic.rs index 9074d204..f4d9ac9b 100644 --- a/phoxal/tests/interop_dynamic.rs +++ b/phoxal/tests/interop_dynamic.rs @@ -12,7 +12,7 @@ use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::time::Duration; -use phoxal::participant::{ParticipantLaunch, RealClock}; +use phoxal::participant::ParticipantLaunch; use phoxal::prelude::*; use phoxal::raw::{Bus, BusConfig, run_with_bus}; use phoxal_api::v1 as api; @@ -118,16 +118,14 @@ async fn two_runtimes_exchange_a_dynamic_topic_on_one_bus() { .await .expect("open shared bus"); - let producer = run_with_bus::( + let producer = run_with_bus::( &bus, ParticipantLaunch::local("encoder-producer-1", "robot"), - RealClock::new(), async { tokio::time::sleep(Duration::from_millis(500)).await }, ); - let consumer = run_with_bus::( + let consumer = run_with_bus::( &bus, ParticipantLaunch::local("encoder-consumer-1", "robot"), - RealClock::new(), async { tokio::time::sleep(Duration::from_millis(500)).await }, ); diff --git a/phoxal/tests/managed_tasks.rs b/phoxal/tests/managed_tasks.rs index cf40b08d..f11e8b44 100644 --- a/phoxal/tests/managed_tasks.rs +++ b/phoxal/tests/managed_tasks.rs @@ -6,7 +6,7 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::Duration; -use phoxal::participant::{ManagedTaskPolicy, ParticipantLaunch, RealClock, run_with}; +use phoxal::participant::{ManagedTaskPolicy, ParticipantLaunch, run_with}; use phoxal::prelude::*; static NAMESPACE_SEQ: AtomicU64 = AtomicU64::new(0); @@ -49,7 +49,7 @@ async fn managed_task_panic_faults_the_runtime() { let result = tokio::time::timeout( Duration::from_secs(5), - run_with::(launch, RealClock::new(), shutdown), + run_with::(launch, shutdown), ) .await .expect("runner must not hang after a managed task panics"); @@ -92,7 +92,7 @@ async fn managed_task_early_return_faults_by_default() { let result = tokio::time::timeout( Duration::from_secs(5), - run_with::(launch, RealClock::new(), shutdown), + run_with::(launch, shutdown), ) .await .expect("runner must not hang after a managed task returns early"); @@ -138,8 +138,7 @@ async fn managed_task_early_return_under_allow_exit_does_not_fault() { tokio::time::sleep(Duration::from_millis(200)).await; }; - let result = - run_with::(launch, RealClock::new(), shutdown).await; + let result = run_with::(launch, shutdown).await; assert!(EARLY_RETURN_ALLOWED_TASK_RAN.load(Ordering::Relaxed)); result.expect("an AllowExit managed task returning early must not fault the runtime"); @@ -183,7 +182,7 @@ async fn managed_task_panic_under_allow_exit_does_not_fault() { tokio::time::sleep(Duration::from_millis(200)).await; }; - let result = run_with::(launch, RealClock::new(), shutdown).await; + let result = run_with::(launch, shutdown).await; assert!(ALLOWED_PANIC_TASK_RAN.load(Ordering::Relaxed)); result.expect("an AllowExit managed task panicking must not fault the runtime"); @@ -233,7 +232,7 @@ async fn clean_shutdown_cancels_and_joins_managed_tasks() { let result = tokio::time::timeout( Duration::from_secs(5), - run_with::(launch, RealClock::new(), shutdown), + run_with::(launch, shutdown), ) .await .expect("runner must not hang joining a cooperative managed task"); @@ -279,7 +278,7 @@ async fn setup_failure_cancels_spawned_managed_tasks() { let result = tokio::time::timeout( Duration::from_secs(5), - run_with::(launch, RealClock::new(), shutdown), + run_with::(launch, shutdown), ) .await .expect("runner must not hang cleaning up a managed task after setup fails"); @@ -342,7 +341,7 @@ async fn shutdown_reports_unjoined_managed_tasks_after_grace_elapses() { let started = std::time::Instant::now(); let result = tokio::time::timeout( Duration::from_secs(10), - run_with::(launch, RealClock::new(), shutdown), + run_with::(launch, shutdown), ) .await .expect("runner must not hang past its own timeout budget waiting on a stuck task"); diff --git a/phoxal/tests/robot_root.rs b/phoxal/tests/robot_root.rs index 42077c4a..f9454b85 100644 --- a/phoxal/tests/robot_root.rs +++ b/phoxal/tests/robot_root.rs @@ -5,7 +5,7 @@ use std::path::PathBuf; use std::sync::OnceLock; use std::time::Duration; -use phoxal::participant::{ParticipantLaunch, RealClock, run_with}; +use phoxal::participant::{ParticipantLaunch, run_with}; use phoxal::prelude::*; use serial_test::serial; @@ -40,7 +40,7 @@ async fn setup_reads_robot_model_from_the_root() { tokio::time::sleep(Duration::from_millis(50)).await; }; - run_with::(launch, RealClock::new(), shutdown) + run_with::(launch, shutdown) .await .expect("runner should complete with a robot root"); @@ -58,7 +58,7 @@ async fn robot_is_absent_without_a_root() { let shutdown = async { tokio::time::sleep(Duration::from_millis(50)).await; }; - let result = run_with::(launch, RealClock::new(), shutdown).await; + let result = run_with::(launch, shutdown).await; assert!( result.is_err(), "setup should fail when no robot model is bound" diff --git a/phoxal/tests/runner.rs b/phoxal/tests/runner.rs index e3b8c1c5..58022d9f 100644 --- a/phoxal/tests/runner.rs +++ b/phoxal/tests/runner.rs @@ -29,9 +29,9 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::Duration; use phoxal::bus::{DEFAULT_QUERY_TIMEOUT, Latest, LogicalTime, OwnerCap, Publisher, Querier}; -use phoxal::participant::{ClockMode, ParticipantLaunch, RealClock, TestClock}; +use phoxal::participant::{ClockMode, ParticipantLaunch, TestClock}; use phoxal::prelude::*; -use phoxal::raw::{Bus, BusConfig, run_with_bus}; +use phoxal::raw::{Bus, BusConfig, run_with_bus, run_with_bus_clock}; use phoxal_api::ContractBody; use phoxal_api::v1 as api; use phoxal_api::v2; @@ -192,7 +192,7 @@ async fn new_model_participant_runs_through_a_real_bus() { .expect("build submap querier"); let launch = ParticipantLaunch::local("wall-follower-v2-1", "robot"); - let runner = run_with_bus::(&bus, launch, RealClock::new(), async { + let runner = run_with_bus::(&bus, launch, async { tokio::time::sleep(Duration::from_millis(600)).await }); @@ -434,7 +434,7 @@ async fn subscriber_and_latest_survive_the_owned_arc_split() { .expect("build submap querier"); let launch = ParticipantLaunch::local("drain-proof-v2-1", "robot"); - let runner = run_with_bus::(&bus, launch, RealClock::new(), async { + let runner = run_with_bus::(&bus, launch, async { tokio::time::sleep(Duration::from_millis(800)).await }); @@ -753,18 +753,14 @@ impl ConfiguredInspector { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn configless_tool_accepts_absent_config_but_configured_tool_rejects_it() { let configless = ParticipantLaunch::local("robot-inspector", "robot"); - phoxal::participant::run_with::(configless, RealClock::new(), async {}) + phoxal::participant::run_with::(configless, async {}) .await .expect("a tool with omitted config type should accept absent PHOXAL_CONFIG"); let configured = ParticipantLaunch::local("configured-inspector", "robot"); - let error = phoxal::participant::run_with::( - configured, - RealClock::new(), - async {}, - ) - .await - .expect_err("a tool with an explicit non-optional config should require PHOXAL_CONFIG"); + let error = phoxal::participant::run_with::(configured, async {}) + .await + .expect_err("a tool with an explicit non-optional config should require PHOXAL_CONFIG"); assert!( error.to_string().contains("invalid type: null"), "unexpected absent-config error: {error:#}" @@ -772,7 +768,7 @@ async fn configless_tool_accepts_absent_config_but_configured_tool_rejects_it() } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn tool_ignores_simulation_clock_mode_and_keeps_host_work_running() { +async fn clockless_tool_keeps_host_work_and_raw_subscriptions_running() { HOST_TOOL_TICKS.store(0, Ordering::Relaxed); HOST_TOOL_MESSAGES.store(0, Ordering::Relaxed); let participant_id = "host-driven-tool-1"; @@ -791,9 +787,7 @@ async fn tool_ignores_simulation_clock_mode_and_keeps_host_work_running() { let mut launch = ParticipantLaunch::local(participant_id, "robot"); launch.namespace = namespace; - launch.clock = ClockMode::Simulation; - let injected_clock = TestClock::new(); - run_with_bus::(&bus, launch, injected_clock, async move { + run_with_bus::(&bus, launch, async move { tokio::time::sleep(Duration::from_millis(20)).await; manual .publish_at( @@ -804,20 +798,20 @@ async fn tool_ignores_simulation_clock_mode_and_keeps_host_work_running() { }, ) .await - .expect("raw tool input should publish while logical time is paused"); + .expect("raw tool input should publish"); tokio::time::sleep(Duration::from_millis(80)).await; }) .await - .expect("tool should not wait for a simulation clock feed"); + .expect("tool should run without a logical clock input"); assert!( HOST_TOOL_TICKS.load(Ordering::Relaxed) >= 2, - "host ticker must continue without simulation clock samples" + "host ticker must run without a logical clock" ); assert_eq!( HOST_TOOL_MESSAGES.load(Ordering::Relaxed), 1, - "raw tool subscriptions must continue without simulation clock samples" + "raw tool subscriptions must run without a logical clock" ); let mut produced_at = Vec::new(); while let Some(received) = heartbeats.try_recv() { @@ -843,7 +837,7 @@ async fn runner_runs_steps_then_shuts_down_cleanly() { tokio::time::sleep(Duration::from_millis(200)).await; }; - phoxal::participant::run_with::(launch, RealClock::new(), shutdown) + phoxal::participant::run_with::(launch, shutdown) .await .expect("runner should complete cleanly"); @@ -874,7 +868,7 @@ async fn runner_publishes_presence_heartbeats_from_idle_loop() { let mut launch = ParticipantLaunch::local(participant_id, "robot"); launch.namespace = namespace; - let runner = run_with_bus::(&bus, launch, RealClock::new(), async { + let runner = run_with_bus::(&bus, launch, async { tokio::time::sleep(Duration::from_millis(2200)).await; }); let collector = async { @@ -946,7 +940,7 @@ async fn simulation_setup_failure_heartbeats_use_simulation_time() { let injected_clock = TestClock::new(); injected_clock.advance(Duration::from_secs(123)); - let error = run_with_bus::(&bus, launch, injected_clock, async {}) + let error = run_with_bus_clock::(&bus, launch, injected_clock, async {}) .await .expect_err("setup should fail"); assert!(error.to_string().contains("intentional setup failure")); @@ -986,7 +980,7 @@ async fn slow_shutdown_hook_is_bounded_by_grace() { let started = std::time::Instant::now(); tokio::time::timeout( Duration::from_secs(10), - phoxal::participant::run_with::(launch, RealClock::new(), shutdown), + phoxal::participant::run_with::(launch, shutdown), ) .await .expect("runner must not hang on a slow shutdown hook") @@ -1063,7 +1057,7 @@ async fn simulation_mode_step_advances_only_with_the_clock_feed() { // mode must ignore it for StepContext and publication timestamps. let injected_clock = TestClock::new(); injected_clock.advance(Duration::from_secs(123)); - let runner = run_with_bus::(&bus, launch, injected_clock, async { + let runner = run_with_bus_clock::(&bus, launch, injected_clock, async { // No steps should have released yet: the feed has not published a // single sample, so the scheduler's logical time has never left // `start`. @@ -1190,7 +1184,7 @@ async fn driver_reads_its_bound_component_instance() { tokio::time::sleep(Duration::from_millis(50)).await; }; - phoxal::participant::run_with::(launch, RealClock::new(), shutdown) + phoxal::participant::run_with::(launch, shutdown) .await .expect("driver should read its bound component instance and run cleanly"); } diff --git a/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.rs b/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.rs index d0a7ed5f..099e785f 100644 --- a/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.rs +++ b/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.rs @@ -15,7 +15,7 @@ impl DefaultRawBusOpen { async fn setup(_ctx: &mut SetupContext) -> Result<(Self, Self::Api)> { let _config = phoxal::bus::BusConfig::in_process("dev", "robot"); let _open = phoxal::bus::Bus::open; - let _run_with_bus = phoxal::participant::run_with_bus::; + let _run_with_bus = phoxal::participant::run_with_bus::; Ok((Self, ())) } } diff --git a/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.stderr b/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.stderr index c9f69f0e..6347bb88 100644 --- a/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.stderr +++ b/phoxal/tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.stderr @@ -13,22 +13,19 @@ error[E0433]: cannot find `Bus` in `bus` error[E0425]: cannot find value `run_with_bus` in module `phoxal::participant` --> tests/trybuild/fail/default_bus_module_cannot_open_raw_bus.rs:18:50 | -18 | let _run_with_bus = phoxal::participant::run_with_bus::; +18 | let _run_with_bus = phoxal::participant::run_with_bus::; | ^^^^^^^^^^^^ | ::: src/participant/runner.rs | - | / pub async fn run_with( - | | launch: ParticipantLaunch, - | | clock: C, - | | shutdown: S, -... | - | | C: ClockSource, + | / pub async fn run_with(launch: ParticipantLaunch, shutdown: S) -> crate::Result<()> + | | where + | | R: ParticipantLifecycle, | | S: Future, | |___________________________- similarly named function `run_with` defined here | help: a function with a similar name exists | -18 - let _run_with_bus = phoxal::participant::run_with_bus::; -18 + let _run_with_bus = phoxal::participant::run_with::; +18 - let _run_with_bus = phoxal::participant::run_with_bus::; +18 + let _run_with_bus = phoxal::participant::run_with::; | diff --git a/phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs index 933d3bab..2cc71ca4 100644 --- a/phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs +++ b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.rs @@ -3,6 +3,10 @@ use phoxal::prelude::*; #[phoxal::tool(id = "clocked-tool")] struct ClockedTool; +// Even manually adding the public graph marker cannot change the launch policy +// that #[phoxal::tool] fixed in the Participant impl. +impl phoxal::participant::TypedGraphSurface for ClockedTool {} + #[phoxal::behavior] impl ClockedTool { #[setup] @@ -12,4 +16,10 @@ impl ClockedTool { } } -fn main() {} +fn main() { + let _inject = phoxal::raw::run_with_bus_clock::< + ClockedTool, + phoxal::participant::TestClock, + std::future::Ready<()>, + >; +} diff --git a/phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr index 125afeab..c70e6b4f 100644 --- a/phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr +++ b/phoxal/tests/trybuild/fail/tool_cannot_access_clock.stderr @@ -1,5 +1,26 @@ error[E0599]: no method named `clock` found for mutable reference `&mut phoxal::participant::SetupContext` in the current scope - --> tests/trybuild/fail/tool_cannot_access_clock.rs:10:26 + --> tests/trybuild/fail/tool_cannot_access_clock.rs:14:26 | -10 | let _clock = ctx.clock(); +14 | let _clock = ctx.clock(); | ^^^^^ method not found in `&mut phoxal::participant::SetupContext` + +error[E0271]: type mismatch resolving `::LaunchPolicy == ClockedParticipantLaunch` + --> tests/trybuild/fail/tool_cannot_access_clock.rs:21:9 + | +21 | ClockedTool, + | ^^^^^^^^^^^ type mismatch resolving `::LaunchPolicy == ClockedParticipantLaunch` + | +note: expected this to be `phoxal::participant::launch::ClockedParticipantLaunch` + --> tests/trybuild/fail/tool_cannot_access_clock.rs:3:1 + | + 3 | #[phoxal::tool(id = "clocked-tool")] + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +note: required by a bound in `phoxal::raw::run_with_bus_clock` + --> src/participant/runner.rs + | + | pub async fn run_with_bus_clock( + | ------------------ required by a bound in this function +... + | R: ParticipantLifecycle + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `run_with_bus_clock` + = note: this error originates in the attribute macro `phoxal::tool` (in Nightly builds, run with -Z macro-backtrace for more info) diff --git a/service/navigation/src/main.rs b/service/navigation/src/main.rs index 869d87f7..8f6dfe74 100644 --- a/service/navigation/src/main.rs +++ b/service/navigation/src/main.rs @@ -570,7 +570,7 @@ mod tests { use phoxal::participant::{ ClockSource, ContractRole, Participant, ParticipantApi, ParticipantLaunch, TestClock, }; - use phoxal::raw::{Bus, BusConfig, OwnerCap, Publisher, Subscriber, run_with_bus}; + use phoxal::raw::{Bus, BusConfig, OwnerCap, Publisher, Subscriber, run_with_bus_clock}; use phoxal_api::ContractBody; use super::*; @@ -706,7 +706,7 @@ mod tests { let clock = TestClock::new(); let runner_clock = clock.clone(); - let runner = run_with_bus::( + let runner = run_with_bus_clock::( &bus, ParticipantLaunch::local("navigation-1", "robot"), runner_clock, diff --git a/xtask/src/workspace.rs b/xtask/src/workspace.rs index 4c96c602..0188d684 100644 --- a/xtask/src/workspace.rs +++ b/xtask/src/workspace.rs @@ -645,25 +645,6 @@ mod tests { Ok(()) } - #[test] - fn hello_rover_contains_the_minimal_tool_boundary_world() -> Result<()> { - let workspace_root = Path::new(env!("CARGO_MANIFEST_DIR")) - .parent() - .context("xtask manifest directory has no workspace parent")?; - let world = fs::read_to_string( - workspace_root.join("examples/hello-rover/worlds/tool-boundary.wbt"), - )?; - assert!(world.starts_with("#VRML_SIM R2025a utf8")); - assert!(world.contains("WorldInfo {")); - assert!(world.contains("basicTimeStep 16")); - assert!(world.contains("RectangleArena {")); - assert!( - !world.contains("controller "), - "the CLI must inject robot controllers; the boundary world stays neutral" - ); - Ok(()) - } - #[test] fn real_workspace_release_scope_is_valid() -> Result<()> { let workspace_manifest = Path::new(env!("CARGO_MANIFEST_DIR"))