From 87813a72bcc85f7b0e160b6c46e45cc73a255e82 Mon Sep 17 00:00:00 2001 From: Yulong Han Date: Sun, 25 Oct 2026 01:11:11 +0800 Subject: [PATCH] feat(rbnx): declare each capability's transport in the package manifest A capabilities entry may say how the package serves the contract: transport: mcp | grpc | ros2, or a non-empty list for a contract served over several transports. rbnx rejects an invalid or duplicate value, a repeated contract and a transport the contract's mode cannot use, and warns when the field is missing. During boot it compares the manifest with what each provider registers in Atlas and warns on every mismatch, without failing the start. The packages in this repository declare what their code registers. --- Cargo.lock | 1 + .../piper_health/package_manifest.yaml | 3 + .../tiago_camera/package_manifest.yaml | 6 + .../tiago_chassis/package_manifest.yaml | 3 + .../tiago_health/package_manifest.yaml | 3 + .../tiago_lidar/package_manifest.yaml | 2 + .../services/simple_nav/package_manifest.yaml | 3 + services/memory/package_manifest.yaml | 3 + services/memsearch/package_manifest.yaml | 4 + services/speech/package_manifest.yaml | 9 + .../scene_verifier/package_manifest.yaml | 3 +- .../vlm_verifier/package_manifest.yaml | 1 + services/voiceprint/package_manifest.yaml | 5 + .../scene/package_manifest.jetson-docker.yaml | 18 ++ .../scene/package_manifest.jetson-native.yaml | 18 ++ system/scene/package_manifest.yaml | 14 + tools/rbnx/Cargo.toml | 3 +- tools/rbnx/README.md | 12 + tools/rbnx/src/cmd/deploy.rs | 104 ++++++- tools/rbnx/src/launch.rs | 290 +++++++++++++++++- tools/rbnx/src/manifest.rs | 238 +++++++++++++- 21 files changed, 729 insertions(+), 14 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 451a59eb5..82b6a84a6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3154,6 +3154,7 @@ dependencies = [ "sha2", "tokio", "tokio-stream", + "toml", "tonic", "tonic-build", "tonic-prost", diff --git a/examples/piper_vitals/primitives/piper_health/package_manifest.yaml b/examples/piper_vitals/primitives/piper_health/package_manifest.yaml index 111309dc3..9b480bfb1 100644 --- a/examples/piper_vitals/primitives/piper_health/package_manifest.yaml +++ b/examples/piper_vitals/primitives/piper_health/package_manifest.yaml @@ -17,7 +17,10 @@ stop: | capabilities: - name: robonix/primitive/health/state + transport: grpc - name: robonix/primitive/health/stream + transport: grpc - name: robonix/primitive/health/driver + transport: grpc depends: [] diff --git a/examples/webots/primitives/tiago_camera/package_manifest.yaml b/examples/webots/primitives/tiago_camera/package_manifest.yaml index 3b54c659a..a2e6ea600 100644 --- a/examples/webots/primitives/tiago_camera/package_manifest.yaml +++ b/examples/webots/primitives/tiago_camera/package_manifest.yaml @@ -20,18 +20,24 @@ stop: | capabilities: - name: robonix/primitive/camera/snapshot + transport: mcp - name: robonix/primitive/camera/depth_snapshot + transport: mcp # ROS2 streams (data plane). Topic names are runtime-resolved # through atlas — consumers (scene 3D fusion, mapping RGBD ingest) # discover them via QueryCapabilities, they never read the topic # strings declared in the driver source. - name: robonix/primitive/camera/rgb + transport: ros2 - name: robonix/primitive/camera/depth + transport: ros2 # Compatibility `base_link → camera_optical` transform. Scene uses the # complete TF tree first and composes pose + this latched transform only # when that authoritative lookup is unavailable. - name: robonix/primitive/camera/extrinsics + transport: ros2 # Static pinhole intrinsics (sensor_msgs/CameraInfo, latched). Lets # scene back-project depth with the real camera's K instead of a # hardcoded default — see capabilities/primitive/camera/intrinsics.v1.toml. - name: robonix/primitive/camera/intrinsics + transport: ros2 diff --git a/examples/webots/primitives/tiago_chassis/package_manifest.yaml b/examples/webots/primitives/tiago_chassis/package_manifest.yaml index 9f2f3f20a..76df75ac6 100644 --- a/examples/webots/primitives/tiago_chassis/package_manifest.yaml +++ b/examples/webots/primitives/tiago_chassis/package_manifest.yaml @@ -39,5 +39,8 @@ stop: | capabilities: - name: robonix/primitive/chassis/move + transport: grpc - name: robonix/primitive/chassis/odom + transport: ros2 - name: robonix/primitive/chassis/twist_in + transport: ros2 diff --git a/examples/webots/primitives/tiago_health/package_manifest.yaml b/examples/webots/primitives/tiago_health/package_manifest.yaml index db82258ad..671e42e33 100644 --- a/examples/webots/primitives/tiago_health/package_manifest.yaml +++ b/examples/webots/primitives/tiago_health/package_manifest.yaml @@ -20,5 +20,8 @@ stop: | capabilities: - name: robonix/primitive/health/state + transport: grpc - name: robonix/primitive/health/stream + transport: grpc - name: robonix/lifecycle/driver + transport: grpc diff --git a/examples/webots/primitives/tiago_lidar/package_manifest.yaml b/examples/webots/primitives/tiago_lidar/package_manifest.yaml index 4bd9aad26..42632f522 100644 --- a/examples/webots/primitives/tiago_lidar/package_manifest.yaml +++ b/examples/webots/primitives/tiago_lidar/package_manifest.yaml @@ -23,4 +23,6 @@ stop: | capabilities: - name: robonix/primitive/lidar/lidar + transport: ros2 - name: robonix/primitive/lidar/snapshot + transport: mcp diff --git a/examples/webots/services/simple_nav/package_manifest.yaml b/examples/webots/services/simple_nav/package_manifest.yaml index 3ca733fe7..623783f81 100644 --- a/examples/webots/services/simple_nav/package_manifest.yaml +++ b/examples/webots/services/simple_nav/package_manifest.yaml @@ -28,5 +28,8 @@ stop: | capabilities: - name: robonix/service/navigation/navigate + transport: mcp - name: robonix/service/navigation/navigate/status + transport: mcp - name: robonix/service/navigation/navigate/cancel + transport: mcp diff --git a/services/memory/package_manifest.yaml b/services/memory/package_manifest.yaml index 0a2022661..1ba4f1139 100644 --- a/services/memory/package_manifest.yaml +++ b/services/memory/package_manifest.yaml @@ -30,5 +30,8 @@ start: | capabilities: - name: robonix/service/memory/remember + transport: mcp - name: robonix/service/memory/hybrid_search + transport: mcp - name: robonix/service/memory/promote + transport: mcp diff --git a/services/memsearch/package_manifest.yaml b/services/memsearch/package_manifest.yaml index e645bf489..8a77963d8 100644 --- a/services/memsearch/package_manifest.yaml +++ b/services/memsearch/package_manifest.yaml @@ -30,6 +30,10 @@ stop: | capabilities: - name: robonix/service/memory/driver + transport: grpc - name: robonix/service/memory/save + transport: mcp - name: robonix/service/memory/search + transport: mcp - name: robonix/service/memory/compact + transport: mcp diff --git a/services/speech/package_manifest.yaml b/services/speech/package_manifest.yaml index f10defdf2..a902a78f1 100644 --- a/services/speech/package_manifest.yaml +++ b/services/speech/package_manifest.yaml @@ -62,11 +62,20 @@ stop: | capabilities: - name: robonix/service/speech/driver + transport: grpc - name: robonix/service/speech/asr + transport: grpc - name: robonix/service/speech/asr_stream + transport: grpc - name: robonix/service/speech/wake_word + transport: grpc - name: robonix/service/speech/tts + transport: grpc - name: robonix/service/speech/tts_stream + transport: grpc - name: robonix/service/speech/dialog + transport: grpc - name: robonix/service/speech/list_speakers + transport: mcp - name: robonix/service/speech/speak + transport: mcp diff --git a/services/verifiers/scene_verifier/package_manifest.yaml b/services/verifiers/scene_verifier/package_manifest.yaml index e594f9d0f..c8cfa2bb2 100644 --- a/services/verifiers/scene_verifier/package_manifest.yaml +++ b/services/verifiers/scene_verifier/package_manifest.yaml @@ -12,4 +12,5 @@ build: bash scripts/build.sh start: bash scripts/start.sh capabilities: - - name: robonix/service/verifier/verify \ No newline at end of file + - name: robonix/service/verifier/verify + transport: mcp diff --git a/services/verifiers/vlm_verifier/package_manifest.yaml b/services/verifiers/vlm_verifier/package_manifest.yaml index 67f761374..67bd23643 100644 --- a/services/verifiers/vlm_verifier/package_manifest.yaml +++ b/services/verifiers/vlm_verifier/package_manifest.yaml @@ -11,3 +11,4 @@ build: bash scripts/build.sh start: bash scripts/start.sh capabilities: - name: robonix/service/verifier/verify + transport: mcp diff --git a/services/voiceprint/package_manifest.yaml b/services/voiceprint/package_manifest.yaml index 9fa5d23b9..2fe2411b3 100644 --- a/services/voiceprint/package_manifest.yaml +++ b/services/voiceprint/package_manifest.yaml @@ -56,7 +56,12 @@ stop: | capabilities: - name: robonix/service/voiceprint/driver + transport: grpc - name: robonix/service/voiceprint/identify + transport: grpc - name: robonix/service/voiceprint/enroll + transport: grpc - name: robonix/service/voiceprint/list + transport: [grpc, mcp] - name: robonix/service/voiceprint/delete + transport: grpc diff --git a/system/scene/package_manifest.jetson-docker.yaml b/system/scene/package_manifest.jetson-docker.yaml index 64d77a6ee..af15059ab 100644 --- a/system/scene/package_manifest.jetson-docker.yaml +++ b/system/scene/package_manifest.jetson-docker.yaml @@ -22,13 +22,31 @@ start: ROBONIX_SCENE_FORCE=docker bash scripts/start.sh capabilities: - name: robonix/system/scene/list_objects + transport: mcp - name: robonix/system/scene/get_robot_context + transport: mcp - name: robonix/system/scene/list_regions + transport: mcp - name: robonix/system/scene/goal_near + transport: mcp - name: robonix/system/scene/goal_region + transport: mcp # Deprecated alias of goal_region, kept for older callers. - name: robonix/system/scene/goal_room + transport: mcp - name: robonix/system/scene/find + transport: mcp - name: robonix/system/scene/get_scene_graph + transport: mcp - name: robonix/system/scene/get_object_context + transport: mcp - name: robonix/system/scene/list_relations + transport: mcp + - name: robonix/system/scene/update_object_label + transport: mcp + - name: robonix/system/scene/update_object_geometry + transport: mcp + - name: robonix/system/scene/delete_object + transport: mcp + - name: robonix/system/scene/flush_objects + transport: mcp diff --git a/system/scene/package_manifest.jetson-native.yaml b/system/scene/package_manifest.jetson-native.yaml index c20151a8a..8e179638b 100644 --- a/system/scene/package_manifest.jetson-native.yaml +++ b/system/scene/package_manifest.jetson-native.yaml @@ -26,13 +26,31 @@ start: ROBONIX_SCENE_FORCE=native bash scripts/start.sh capabilities: - name: robonix/system/scene/list_objects + transport: mcp - name: robonix/system/scene/get_robot_context + transport: mcp - name: robonix/system/scene/list_regions + transport: mcp - name: robonix/system/scene/goal_near + transport: mcp - name: robonix/system/scene/goal_region + transport: mcp # Deprecated alias of goal_region, kept for older callers. - name: robonix/system/scene/goal_room + transport: mcp - name: robonix/system/scene/find + transport: mcp - name: robonix/system/scene/get_scene_graph + transport: mcp - name: robonix/system/scene/get_object_context + transport: mcp - name: robonix/system/scene/list_relations + transport: mcp + - name: robonix/system/scene/update_object_label + transport: mcp + - name: robonix/system/scene/update_object_geometry + transport: mcp + - name: robonix/system/scene/delete_object + transport: mcp + - name: robonix/system/scene/flush_objects + transport: mcp diff --git a/system/scene/package_manifest.yaml b/system/scene/package_manifest.yaml index 20f8ec80f..15697cbb2 100644 --- a/system/scene/package_manifest.yaml +++ b/system/scene/package_manifest.yaml @@ -32,17 +32,31 @@ stop: docker rm -f "${ROBONIX_SCENE_CONTAINER:-robonix_scene}" capabilities: - name: robonix/system/scene/list_objects + transport: mcp - name: robonix/system/scene/get_robot_context + transport: mcp - name: robonix/system/scene/list_regions + transport: mcp - name: robonix/system/scene/goal_near + transport: mcp - name: robonix/system/scene/goal_region + transport: mcp # Deprecated alias of goal_region, kept for older callers. - name: robonix/system/scene/goal_room + transport: mcp - name: robonix/system/scene/find + transport: mcp - name: robonix/system/scene/get_scene_graph + transport: mcp - name: robonix/system/scene/get_object_context + transport: mcp - name: robonix/system/scene/list_relations + transport: mcp - name: robonix/system/scene/update_object_label + transport: mcp - name: robonix/system/scene/update_object_geometry + transport: mcp - name: robonix/system/scene/delete_object + transport: mcp - name: robonix/system/scene/flush_objects + transport: mcp diff --git a/tools/rbnx/Cargo.toml b/tools/rbnx/Cargo.toml index fcc2c372b..03486369a 100644 --- a/tools/rbnx/Cargo.toml +++ b/tools/rbnx/Cargo.toml @@ -14,6 +14,7 @@ clap.workspace = true serde.workspace = true serde_yaml.workspace = true serde_json.workspace = true +toml.workspace = true sha2.workspace = true tokio.workspace = true anyhow.workspace = true @@ -23,7 +24,7 @@ prost.workspace = true tonic-prost.workspace = true ratatui = { workspace = true, features = ["unstable-rendered-line-info"] } crossterm.workspace = true -tokio-stream = "0.1" +tokio-stream = { version = "0.1", features = ["net"] } dirs = "6.0" chrono = "0.4" hostname = "0.4" diff --git a/tools/rbnx/README.md b/tools/rbnx/README.md index 54b78553a..677a1bb3b 100644 --- a/tools/rbnx/README.md +++ b/tools/rbnx/README.md @@ -55,6 +55,12 @@ cargo install --force --path tools/rbnx --bin rbnx Run `rbnx --help` for full flags. +Each package manifest capability declares `transport: mcp`, `grpc`, or `ros2`, +or a non-empty list such as `transport: [grpc, ros2]`. `rbnx validate` rejects +invalid or duplicate transports, repeated contracts, and transports incompatible +with an available contract TOML's mode. An omitted transport warns and remains +accepted while existing packages migrate; `manifestVersion` stays `1`. + `rbnx install [--github REPO \| --path DIR]`, `rbnx list`, and `rbnx info NAME` manage system-installed packages. They are legacy commands, hidden from `rbnx --help`; deployment manifests with `rbnx build` and `rbnx boot` replace @@ -67,6 +73,12 @@ consumers. ### What a boot leaves behind +During boot, `rbnx` compares package manifest transport declarations with Atlas +registrations, including Soma-managed primitives and skills. It warns once per +mismatch; missing registrations are reported only while the package is ACTIVE. +A manifest without transport declarations gets one warning and skips comparison. +These diagnostics do not fail startup. + `rbnx boot` writes `/rbnx-boot/deployment.lock` before it spawns anything. Each package records what the manifest asked for and what that resolved to, kept apart the way `flake.lock` keeps them: diff --git a/tools/rbnx/src/cmd/deploy.rs b/tools/rbnx/src/cmd/deploy.rs index 5de323f8d..d3829bedc 100644 --- a/tools/rbnx/src/cmd/deploy.rs +++ b/tools/rbnx/src/cmd/deploy.rs @@ -33,9 +33,9 @@ use anyhow::{Context, Result}; use robonix_atlas::client::AtlasClient; use robonix_atlas::pb as atlas_pb; use robonix_cli::launch::{ - PackageRuntimeRecord, ProviderRegistrationSnapshot, RegistrationOutcome, - contract_id_to_service_name, resolve_runtime_driver_contract, snapshot_provider_ids, - terminate_process_group, + CapabilityTransportCheck, PackageRuntimeRecord, ProviderRegistrationSnapshot, + RegistrationOutcome, contract_id_to_service_name, resolve_runtime_driver_contract, + snapshot_provider_ids, terminate_process_group, }; use robonix_cli::output; use serde::Deserialize; @@ -885,6 +885,8 @@ struct Spawned { /// True only for an explicit legacy selection, permitting a current shared /// runtime Driver while the manifest is migrated. allow_shared_driver_upgrade: bool, + /// Manifest transport comparison for package-managed processes. + transport_check: Option, config_json: Option, package_dir: Option, stop: Option, @@ -965,6 +967,7 @@ async fn spawn_system_binary( driver_contract: None, expected_driver_contract: None, allow_shared_driver_upgrade: false, + transport_check: None, config_json: None, package_dir: None, stop: None, @@ -1113,6 +1116,7 @@ async fn spawn_soma_binary( driver_contract: None, expected_driver_contract: None, allow_shared_driver_upgrade: false, + transport_check: None, config_json: None, package_dir: None, stop: None, @@ -1308,6 +1312,10 @@ async fn spawn_package( driver_contract: None, expected_driver_contract, allow_shared_driver_upgrade, + transport_check: Some(CapabilityTransportCheck::new( + &name, + &package_manifest.manifest, + )), config_json: None, package_dir: Some(pkg_path), stop, @@ -1676,6 +1684,7 @@ pub async fn execute( .last_mut() .expect("soma Spawned pushed above") .child; + let mut transport_checks = load_transport_checks(&deploy.primitive, &spawn_env); wait_for_soma_stage1( &mut stage1_atlas, &deploy @@ -1685,6 +1694,7 @@ pub async fn execute( .collect::>(), soma_child, &log_dir, + &mut transport_checks, ) .await?; @@ -1852,6 +1862,7 @@ pub async fn execute( output::boot_wait(&entry.name, "waiting for registration"); } } + let mut transport_checks = load_transport_checks(&deploy.skill, &spawn_env); if let Err(e) = write_stage2_trigger(&mut soma_stage_writer) { failures.push(( "system".to_string(), @@ -1865,6 +1876,7 @@ pub async fn execute( .iter() .map(|entry| entry.name.clone()) .collect::>(), + &mut transport_checks, ) .await { @@ -2413,6 +2425,9 @@ async fn init_spawned( } }; let provider_id = registration.provider_id.clone(); + if let Some(check) = sp.transport_check.as_mut() { + warn_capability_transports(check, ®istration.provider); + } // The exact-id waiter makes this an internal invariant. Keep the check as // defense in depth so a future launcher refactor cannot deliver config to // a provider other than the manifest instance. @@ -2499,6 +2514,7 @@ async fn init_spawned( }; if component == "skill" { + observe_package_transports(atlas, sp).await; // Skills stop at INACTIVE post-INIT; the executor sends // CMD_ACTIVATE on first MCP call (lazy-activate). output::boot_ok( @@ -2537,6 +2553,7 @@ async fn init_spawned( // got their own spinner lines and OK ticks above. provider_id is the // leftmost label so we don't repeat it here. let _ = init_state; // intermediate, only kept for the assertion below + observe_package_transports(atlas, sp).await; output::boot_ok(display_label, &activate_state.to_uppercase()); Ok(()) @@ -2579,11 +2596,79 @@ where } } +fn warn_capability_transports( + check: &mut CapabilityTransportCheck, + provider: &atlas_pb::CapabilityProvider, +) { + for warning in check.observe(provider) { + output::warning(&warning); + } +} + +fn load_transport_checks( + entries: &[PackageEntry], + env: &PackageSpawnEnv<'_>, +) -> HashMap { + entries + .iter() + .filter_map(|entry| { + let manifest = resolve_entry_path(entry, env.cache_root, env.manifest_dir).and_then( + |path| robonix_cli::manifest::detect_and_load(&path, entry.manifest.as_deref()), + ); + match manifest { + Ok(detected) => Some(( + entry.name.clone(), + CapabilityTransportCheck::new(&entry.name, &detected.manifest), + )), + Err(error) => { + output::warning(&format!( + "package '{}': could not read manifest for runtime transport comparison: {error:#}", + entry.name + )); + None + } + } + }) + .collect() +} + +async fn observe_package_transports(atlas: &mut AtlasClient, spawned: &mut Spawned) { + let (Some(check), Some(provider_id)) = ( + spawned.transport_check.as_mut(), + spawned.provider_id.as_deref(), + ) else { + return; + }; + match tokio::time::timeout( + Duration::from_secs(1), + atlas.query_capabilities(provider_id, "", atlas_pb::Transport::Unspecified), + ) + .await + { + Ok(Ok(providers)) => { + if let Some(provider) = providers.iter().find(|provider| provider.id == provider_id) { + warn_capability_transports(check, provider); + } else { + output::warning(&format!( + "package '{provider_id}': no Atlas registration available for runtime transport comparison" + )); + } + } + Ok(Err(error)) => output::warning(&format!( + "package '{provider_id}': could not observe registered capability transports: {error:#}" + )), + Err(_) => output::warning(&format!( + "package '{provider_id}': runtime transport observation timed out after 1s" + )), + } +} + async fn wait_for_soma_stage1( atlas: &mut AtlasClient, primitive_names: &[String], soma_child: &mut Child, log_dir: &Path, + transport_checks: &mut HashMap, ) -> Result<()> { const SPINNER_TICK: Duration = Duration::from_millis(100); const POLLS_PER_TICK: usize = 5; // poll atlas every 500 ms @@ -2687,6 +2772,9 @@ async fn wait_for_soma_stage1( else { continue; }; + if let Some(check) = transport_checks.get_mut(name) { + warn_capability_transports(check, &provider); + } let previous = observed_states.insert(name.clone(), provider.state); if previous != Some(provider.state) { let state = lifecycle_state_label(provider.state); @@ -2756,7 +2844,11 @@ fn lifecycle_state_label(state: i32) -> &'static str { } } -async fn wait_for_soma_skills(atlas: &mut AtlasClient, skill_names: &[String]) -> Result<()> { +async fn wait_for_soma_skills( + atlas: &mut AtlasClient, + skill_names: &[String], + transport_checks: &mut HashMap, +) -> Result<()> { const TIMEOUT: Duration = Duration::from_secs(180); if skill_names.is_empty() { return Ok(()); @@ -2773,6 +2865,9 @@ async fn wait_for_soma_skills(atlas: &mut AtlasClient, skill_names: &[String]) - let Some(provider) = providers.into_iter().find(|provider| provider.id == *name) else { continue; }; + if let Some(check) = transport_checks.get_mut(name) { + warn_capability_transports(check, &provider); + } if observed_states.insert(name.clone(), provider.state) != Some(provider.state) { let state = lifecycle_state_label(provider.state); if provider.state == atlas_pb::LifecycleState::StateInactive as i32 @@ -3187,6 +3282,7 @@ async fn wait_for_registration( driver_contracts.sort(); driver_contracts.dedup(); return Ok(RegistrationOutcome { + provider: current.clone(), provider_id, provider_kind: current.kind, provider_namespace: current.namespace, diff --git a/tools/rbnx/src/launch.rs b/tools/rbnx/src/launch.rs index afb6e03c1..5d6c5e417 100644 --- a/tools/rbnx/src/launch.rs +++ b/tools/rbnx/src/launch.rs @@ -47,7 +47,7 @@ use robonix_atlas::client::AtlasClient; use robonix_atlas::pb as atlas_pb; use robonix_scribe::{info, warn}; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap}; use std::path::PathBuf; use std::process::Stdio; use std::time::{Duration, Instant}; @@ -600,6 +600,102 @@ pub fn contract_id_to_service_name(id: &str) -> String { /// on heartbeat/state/capability updates. pub type ProviderRegistrationSnapshot = HashMap; +/// Compares one package's declared transports with Atlas observations during bring-up. +pub struct CapabilityTransportCheck { + /// Deployment instance used to identify the package in diagnostics. + package: String, + /// Expected contract and transport pairs from the selected package manifest. + declared: BTreeSet<(String, String)>, + /// Contracts whose omitted transport leaves their runtime transport unspecified. + unspecified: BTreeSet, + /// Whether the framework supplies the shared lifecycle Driver implicitly. + implicit_driver: bool, + /// Diagnostics already returned, bounded by the package's observed capabilities. + warned: BTreeSet, +} + +impl CapabilityTransportCheck { + pub fn new(package: &str, manifest: &crate::manifest::Manifest) -> Self { + Self { + package: package.to_owned(), + declared: manifest + .capabilities + .iter() + .flat_map(|capability| { + capability + .transport + .iter() + .map(|transport| (capability.name.clone(), transport.clone())) + }) + .collect(), + unspecified: manifest + .capabilities + .iter() + .filter(|capability| capability.transport.is_empty()) + .map(|capability| capability.name.clone()) + .collect(), + implicit_driver: manifest + .capabilities + .iter() + .all(|capability| !capability.name.ends_with("/driver")), + warned: BTreeSet::new(), + } + } + + pub fn observe(&mut self, provider: &atlas_pb::CapabilityProvider) -> Vec { + let mut warnings = Vec::new(); + if self.declared.is_empty() { + warnings.push(format!( + "package '{}': manifest declares no capability transports; runtime transport comparison skipped", + self.package + )); + } else { + let registered = provider + .capabilities + .iter() + .map(|capability| { + let transport = match atlas_pb::Transport::try_from(capability.transport) { + Ok(atlas_pb::Transport::Grpc) => "grpc".to_owned(), + Ok(atlas_pb::Transport::Ros2) => "ros2".to_owned(), + Ok(atlas_pb::Transport::Mcp) => "mcp".to_owned(), + _ => format!("unknown({})", capability.transport), + }; + (capability.contract_id.clone(), transport) + }) + .collect::>(); + for (contract, transport) in registered.difference(&self.declared) { + if self.implicit_driver + && contract == crate::manifest::SHARED_LIFECYCLE_DRIVER_CONTRACT + && transport == "grpc" + { + continue; + } + warnings.push(if self.unspecified.contains(contract) { + format!( + "package '{}': registered capability contract '{contract}' over '{transport}', but the manifest gives it no `transport`", + self.package + ) + } else { + format!( + "package '{}': registered capability contract '{contract}' transport '{transport}' is not declared in the manifest", + self.package + ) + }); + } + if provider.state == atlas_pb::LifecycleState::StateActive as i32 { + for (contract, transport) in self.declared.difference(®istered) { + warnings.push(format!( + "package '{}': declared capability contract '{contract}' transport '{transport}' is not registered while ACTIVE", + self.package + )); + } + } + } + warnings.retain(|warning| self.warned.insert(warning.clone())); + warnings + } +} + pub async fn snapshot_provider_ids( atlas: &mut AtlasClient, ) -> Result { @@ -648,6 +744,8 @@ pub fn is_expected_provider_registration( /// sending lifecycle commands. #[derive(Debug, Clone)] pub struct RegistrationOutcome { + /// Full Atlas snapshot after declarations settle, before lifecycle commands. + pub provider: atlas_pb::CapabilityProvider, pub provider_id: String, pub provider_kind: i32, pub provider_namespace: String, @@ -832,6 +930,7 @@ pub async fn wait_for_registration_core( } let driver_contracts = runtime_driver_contracts(¤t); return Ok(RegistrationOutcome { + provider: current.clone(), provider_id, provider_kind: current.kind, provider_namespace: current.namespace, @@ -939,17 +1038,194 @@ pub async fn call_driver_cmd( #[cfg(test)] mod tests { use super::{ - PackageRuntimeRecord, atlas_pb, is_expected_provider_registration, - is_new_provider_registration, linux_proc_status_is_not_zombie, - macos_procargs_has_environment_entry, macos_ps_state_is_not_zombie, proc_start_time_ticks, - process_group_has_members, resolve_runtime_driver_contract, - shutdown_package_runtime_checked, terminate_process_group_guarded, - validate_runtime_driver_contracts, + CapabilityTransportCheck, PackageRuntimeRecord, atlas_pb, + is_expected_provider_registration, is_new_provider_registration, + linux_proc_status_is_not_zombie, macos_procargs_has_environment_entry, + macos_ps_state_is_not_zombie, proc_start_time_ticks, process_group_has_members, + resolve_runtime_driver_contract, shutdown_package_runtime_checked, + terminate_process_group_guarded, validate_runtime_driver_contracts, }; const SHARED: &str = "robonix/lifecycle/driver"; const LEGACY: &str = "robonix/primitive/camera/driver"; + fn transport_check(capabilities: &[(&str, &[&str])]) -> CapabilityTransportCheck { + let manifest = crate::manifest::Manifest { + capabilities: capabilities + .iter() + .map(|(name, transport)| crate::manifest::CapabilityRef { + name: (*name).to_owned(), + transport: transport.iter().map(|value| (*value).to_owned()).collect(), + path: None, + }) + .collect(), + ..Default::default() + }; + CapabilityTransportCheck::new("camera", &manifest) + } + + fn transport_provider( + state: atlas_pb::LifecycleState, + capabilities: &[(&str, atlas_pb::Transport)], + ) -> atlas_pb::CapabilityProvider { + atlas_pb::CapabilityProvider { + id: "camera".to_owned(), + state: state as i32, + capabilities: capabilities + .iter() + .map(|(contract, transport)| atlas_pb::Capability { + contract_id: (*contract).to_owned(), + transport: *transport as i32, + ..Default::default() + }) + .collect(), + ..Default::default() + } + } + + #[test] + fn transport_comparison_warns_for_pairs_once_and_defers_missing_until_active() { + use atlas_pb::{LifecycleState as State, Transport}; + let mut check = transport_check(&[("camera/rgb", &["grpc", "ros2"])]); + let mut provider = transport_provider( + State::StateRegistered, + &[ + ("camera/rgb", Transport::Grpc), + ("camera/rgb", Transport::Mcp), + ], + ); + let warnings = check.observe(&provider); + assert_eq!(warnings.len(), 1); + assert!(warnings[0].contains("package 'camera'")); + assert!(warnings[0].contains("contract 'camera/rgb' transport 'mcp'")); + assert!(warnings[0].contains("not declared")); + assert!(check.observe(&provider).is_empty()); + provider.state = State::StateInactive as i32; + assert!(check.observe(&provider).is_empty()); + provider.state = State::StateActive as i32; + let warnings = check.observe(&provider); + assert_eq!(warnings.len(), 1); + assert!(warnings[0].contains("contract 'camera/rgb' transport 'ros2'")); + assert!(warnings[0].contains("not registered while ACTIVE")); + assert!(check.observe(&provider).is_empty()); + } + + #[test] + fn exact_registered_transport_set_and_implicit_driver_need_no_warning() { + use atlas_pb::{LifecycleState as State, Transport}; + let mut check = transport_check(&[("camera/rgb", &["grpc", "ros2"])]); + let provider = transport_provider( + State::StateActive, + &[ + ("camera/rgb", Transport::Grpc), + ("camera/rgb", Transport::Ros2), + (SHARED, Transport::Grpc), + ], + ); + assert!(check.observe(&provider).is_empty()); + let mut explicit = transport_check(&[(SHARED, &["ros2"])]); + assert_eq!(explicit.observe(&provider).len(), 4); + } + + #[test] + fn a_contract_without_transport_is_still_reported_when_it_registers() { + use atlas_pb::{LifecycleState as State, Transport}; + let provider = transport_provider(State::StateActive, &[("camera/rgb", Transport::Ros2)]); + let mut legacy = transport_check(&[("camera/rgb", &[])]); + let warnings = legacy.observe(&provider); + assert_eq!(warnings.len(), 1); + assert!(warnings[0].contains("declares no capability transports")); + assert!(legacy.observe(&provider).is_empty()); + + let mut mixed = transport_check(&[("camera/rgb", &[]), ("camera/status", &["grpc"])]); + let warnings = mixed.observe(&provider); + assert_eq!(warnings.len(), 2); + assert!(warnings.iter().any(|w| { + w.contains("'camera/rgb' over 'ros2', but the manifest gives it no `transport`") + })); + assert!( + warnings + .iter() + .any(|w| w.contains("contract 'camera/status' transport 'grpc'")) + ); + } + + #[tokio::test] + async fn transport_comparison_uses_real_atlas_registration_and_lifecycle_snapshots() { + use atlas_pb::{LifecycleState as State, Transport}; + use robonix_atlas::client::{AtlasClient, grpc_params}; + use robonix_atlas::service::{AtlasRegistry, AtlasService}; + use std::sync::Arc; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn( + tonic::transport::Server::builder() + .add_service(atlas_pb::atlas_server::AtlasServer::new(AtlasService::new( + Arc::new(AtlasRegistry::default()), + ))) + .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)), + ); + let mut atlas = AtlasClient::connect(format!("http://{address}")) + .await + .unwrap(); + atlas + .register_service("camera", "camera", "") + .await + .unwrap(); + atlas + .declare_capability( + "camera", + "camera/rgb", + Transport::Grpc, + "127.0.0.1:51000", + grpc_params("camera.proto", "Camera", "/Camera/Rgb"), + ) + .await + .unwrap(); + let mut check = transport_check(&[("camera/rgb", &["ros2"])]); + let registered = atlas + .query_capabilities("camera", "", Transport::Unspecified) + .await + .unwrap(); + let warnings = check.observe(®istered[0]); + assert_eq!(warnings.len(), 1); + assert!(warnings[0].contains("transport 'grpc' is not declared")); + + atlas + .declare_capability( + "camera", + SHARED, + Transport::Grpc, + "127.0.0.1:51000", + grpc_params( + "lifecycle.proto", + "robonix.contracts.RobonixLifecycleDriver", + "/robonix.contracts.RobonixLifecycleDriver/Driver", + ), + ) + .await + .unwrap(); + atlas + .set_lifecycle_state("camera", State::StateInactive, "") + .await + .unwrap(); + atlas + .set_lifecycle_state("camera", State::StateActive, "") + .await + .unwrap(); + let active = atlas + .query_capabilities("camera", "", Transport::Unspecified) + .await + .unwrap(); + let warnings = check.observe(&active[0]); + assert_eq!(warnings.len(), 1); + assert!(warnings[0].contains("transport 'ros2' is not registered while ACTIVE")); + assert!(check.observe(&active[0]).is_empty()); + server.abort(); + let _ = server.await; + } + #[cfg(any(target_os = "linux", target_os = "macos"))] #[test] fn live_process_start_identity_is_available_and_stable() { diff --git a/tools/rbnx/src/manifest.rs b/tools/rbnx/src/manifest.rs index 8839e49ca..e8578e4bc 100644 --- a/tools/rbnx/src/manifest.rs +++ b/tools/rbnx/src/manifest.rs @@ -352,10 +352,14 @@ pub struct Package { pub license: String, } +/// A contract and the transports served by this package. #[derive(Debug, Clone, Deserialize)] pub struct CapabilityRef { /// Contract id (matches one of the TOMLs under `capabilities/`). pub name: String, + /// Declared transports; empty only for manifests that omit the field. + #[serde(default, deserialize_with = "deserialize_transports")] + pub transport: Vec, /// Optional path to a package-local TOML that overrides / defines /// this capability (for experimental providers not yet in the official /// capabilities directory). Relative to the package root. @@ -363,6 +367,39 @@ pub struct CapabilityRef { pub path: Option, } +fn deserialize_transports<'de, D>(deserializer: D) -> std::result::Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + #[derive(Deserialize)] + #[serde(untagged)] + enum Transports { + One(String), + Many(Vec), + } + let transports = match Transports::deserialize(deserializer)? { + Transports::One(value) => vec![value], + Transports::Many(values) => values, + }; + if transports.is_empty() { + return Err(serde::de::Error::custom("transport list must not be empty")); + } + let mut seen = HashSet::new(); + for transport in &transports { + if !matches!(transport.as_str(), "mcp" | "grpc" | "ros2") { + return Err(serde::de::Error::custom(format!( + "invalid transport '{transport}'; expected mcp, grpc or ros2" + ))); + } + if !seen.insert(transport) { + return Err(serde::de::Error::custom(format!( + "duplicate transport '{transport}'" + ))); + } + } + Ok(transports) +} + /// One entry under a package's `depends:` list. Models a *source / lib* /// dependency (think Linux kernel module SOFT_DEPS) — i.e. another /// package whose codegen output / Python package this package needs at @@ -483,7 +520,9 @@ pub fn load_from_path(manifest_path: &Path) -> Result { .with_context(|| format!("Failed to read manifest: {}", manifest_path.display()))?; let raw: RawManifest = serde_yaml::from_str(&content) .with_context(|| format!("Failed to parse manifest: {}", manifest_path.display()))?; - Ok(normalize(raw, manifest_path)) + let manifest = normalize(raw, manifest_path); + manifest.validate_capabilities(manifest_path.parent().unwrap_or(Path::new(".")))?; + Ok(manifest) } fn normalize(raw: RawManifest, manifest_path: &Path) -> Manifest { @@ -557,6 +596,99 @@ fn normalize(raw: RawManifest, manifest_path: &Path) -> Manifest { } impl Manifest { + fn validate_capabilities(&self, package_root: &Path) -> Result<()> { + let mut names = HashSet::new(); + for capability in &self.capabilities { + if !names.insert(&capability.name) { + anyhow::bail!( + "package '{}': repeated capability contract '{}'", + self.package.name, + capability.name + ); + } + if capability.transport.is_empty() { + crate::output::warning(&format!( + "package '{}': capability '{}' is missing `transport`; declare mcp, grpc or ros2 (or a non-empty list)", + self.package.name, capability.name + )); + } + } + + if self.capabilities.iter().all(|c| c.transport.is_empty()) { + return Ok(()); + } + let mut roots = Vec::new(); + let source = std::env::var_os("ROBONIX_SOURCE_PATH") + .map(PathBuf::from) + .or_else(|| { + package_root + .ancestors() + .find(|root| { + root.join("Cargo.toml").is_file() && root.join("capabilities").is_dir() + }) + .map(Path::to_path_buf) + }) + .or_else(|| { + let path = crate::Config::config_file_path().ok()?; + let content = std::fs::read_to_string(path).ok()?; + serde_yaml::from_str::(&content) + .ok()? + .robonix_source_path + }); + if let Some(source) = source { + roots.push(source.join("capabilities")); + } + roots.push(package_root.join("capabilities")); + let roots: Vec<&Path> = roots + .iter() + .filter(|root| root.is_dir()) + .map(PathBuf::as_path) + .collect(); + let contracts = + robonix_atlas::contract_registry::ContractRegistry::load_from_capability_roots(&roots)?; + for capability in &self.capabilities { + let local = capability.path.as_ref().map(|path| package_root.join(path)); + let explicit_mode = if let Some(path) = local.filter(|path| path.is_file()) { + let value: toml::Value = toml::from_str(&std::fs::read_to_string(&path)?) + .with_context(|| format!("parse contract {}", path.display()))?; + value + .get("mode") + .and_then(|v| v.get("type")) + .and_then(toml::Value::as_str) + .map(str::to_string) + } else { + None + }; + let Some(mode) = explicit_mode.as_deref().or_else(|| { + contracts + .get(&capability.name) + .map(|contract| contract.mode.as_str()) + }) else { + continue; + }; + for transport in &capability.transport { + let allowed = match mode { + "rpc" => true, + "rpc_server_stream" | "rpc_client_stream" | "rpc_bidirectional_stream" => { + transport == "grpc" + } + "topic_out" | "topic_in" => transport == "grpc" || transport == "ros2", + _ => false, + }; + if !allowed { + anyhow::bail!( + "package '{}': capability '{}' transport '{}' is incompatible with contract mode '{}'", + self.package.name, + capability.name, + transport, + mode + ); + } + } + } + Ok(()) + } + /// Return the Driver written in `capabilities:`, if any. pub fn explicit_lifecycle_driver_contract(&self) -> Result> { let drivers = self @@ -685,6 +817,109 @@ mod tests { use super::*; + fn load_transport_manifest(capabilities: &str, mode: Option<&str>) -> Result { + let root = std::env::temp_dir().join(format!("rbnx-transport-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(root.join("capabilities")).unwrap(); + if let Some(mode) = mode { + std::fs::write( + root.join("capabilities/example.v1.toml"), + format!("[contract]\nid = 'test/contract'\n[mode]\ntype = '{mode}'\n"), + ) + .unwrap(); + } + let path = root.join(MANIFEST_FILE); + std::fs::write( + &path, + format!( + "manifestVersion: 1\npackage:\n name: test.package\n version: 1.0.0\n description: Test package\n license: MIT\nstart: 'true'\ncapabilities:\n{capabilities}\n" + ), + ) + .unwrap(); + let result = load_from_path(&path); + std::fs::remove_dir_all(root).unwrap(); + result + } + + #[test] + fn capability_transports_accept_scalar_list_and_missing() { + for (field, expected) in [ + ("transport: mcp", vec!["mcp"]), + ("transport: [grpc, ros2]", vec!["grpc", "ros2"]), + ("", vec![]), + ] { + let manifest = load_transport_manifest( + &format!(" - name: test/contract\n {field}"), + Some("rpc"), + ) + .unwrap(); + assert_eq!(manifest.manifest_version, 1); + assert_eq!(manifest.capabilities[0].transport, expected); + manifest.validate_and_summarize().unwrap(); + } + } + + #[test] + fn invalid_or_duplicate_transports_are_rejected() { + for field in [ + "transport: http", + "transport: [grpc, http]", + "transport: []", + "transport: [grpc, grpc]", + "transport: 42", + "transport: null", + ] { + assert!( + load_transport_manifest(&format!(" - name: test/contract\n {field}"), None) + .is_err(), + "{field}" + ); + } + } + + #[test] + fn repeated_contracts_are_rejected_even_with_different_transports() { + let error = load_transport_manifest( + " - {name: test/contract, transport: grpc}\n - {name: test/contract, transport: ros2}", + None, + ) + .unwrap_err(); + assert!( + error + .to_string() + .contains("repeated capability contract 'test/contract'") + ); + } + + #[test] + fn transports_must_fit_contract_mode_when_known() { + for (mode, allowed) in [ + ("rpc", vec!["mcp", "grpc", "ros2"]), + ("rpc_server_stream", vec!["grpc"]), + ("rpc_client_stream", vec!["grpc"]), + ("rpc_bidirectional_stream", vec!["grpc"]), + ("topic_out", vec!["grpc", "ros2"]), + ("topic_in", vec!["grpc", "ros2"]), + ] { + for transport in ["mcp", "grpc", "ros2"] { + let result = load_transport_manifest( + &format!(" - {{name: test/contract, transport: {transport}}}"), + Some(mode), + ); + if allowed.contains(&transport) { + result.unwrap(); + } else { + let message = result.unwrap_err().to_string(); + assert!( + message.contains("incompatible with contract mode"), + "{message}" + ); + assert!(message.contains(mode) && message.contains(transport)); + } + } + } + load_transport_manifest(" - {name: test/contract, transport: mcp}", None).unwrap(); + } + #[test] fn package_vendor_is_backward_compatible() { let raw: RawManifest = serde_yaml::from_str( @@ -870,6 +1105,7 @@ primitive: .into_iter() .map(|name| CapabilityRef { name: name.to_string(), + transport: Vec::new(), path: None, }) .collect(),