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(),