Function bodies 699 total

async fn check_node_subscriptions(session: &Session) -> Vec<DiagnosticResult> {
    let mut results = Vec::new();

    // Try to subscribe to daemon nodes topic
    match session.declare_subscriber("bubbaloop/daemon/nodes").await {
        Ok(_subscriber) => {
            results.push(DiagnosticResult::pass(
                "Node subscription",
                "can subscribe to bubbaloop/daemon/nodes",
            ));
        }
        Err(e) => {
            results.push(DiagnosticResult::fail(
                "Node subscription",
                &format!("failed: {}", e),
                "Check Zenoh router configuration",
            ));
        }
    }

    results
}

async fn check_security() -> Vec<DiagnosticResult> {
    let mut results = Vec::new();

    let home = match dirs::home_dir() {
        Some(h) => h,
        None => return results,
    };

    let bubbaloop_dir = home.join(".bubbaloop");

    // Check 1: TLS enabled/disabled
    results.push(check_tls_status(&bubbaloop_dir));

    // Check 2: ACL configuration present
    results.push(check_acl_status(&bubbaloop_dir));

    // Check 3: Scouting disabled in zenoh config
    results.push(check_scouting_disabled(&bubbaloop_dir));

    // Check 4: Localhost-only binding
    results.push(check_localhost_binding(&bubbaloop_dir));

    // Check 5: Python node sandbox check
    results.extend(check_python_sandbox());

    // Check 6: Node name validation
    results.extend(check_node_names(&bubbaloop_dir));

    results
}

fn check_tls_status(bubbaloop_dir: &std::path::Path) -> DiagnosticResult {
    let zenoh_config = bubbaloop_dir.join("zenoh/zenohd.json5");
    if let Ok(content) = std::fs::read_to_string(&zenoh_config) {
        if content.contains("tls") && content.contains("server_certificate") {
            DiagnosticResult::pass("TLS config", "TLS is configured")
        } else {
            DiagnosticResult::fail(
                "TLS config",
                "TLS not configured (plaintext transport)",
                "Run: bubbaloop init-tls",
            )
        }
    } else {
        DiagnosticResult::fail(
            "TLS config",
            "Zenoh config not found",
            "Run: bubbaloop doctor --fix",
        )
    }
}

fn check_acl_status(bubbaloop_dir: &std::path::Path) -> DiagnosticResult {
    let zenoh_config = bubbaloop_dir.join("zenoh/zenohd.json5");
    if let Ok(content) = std::fs::read_to_string(&zenoh_config) {
        if content.contains("access_control") {
            DiagnosticResult::pass("ACL config", "access control rules present")
        } else {
            DiagnosticResult::fail(
                "ACL config",
                "no access control rules configured",
                "See: configs/zenoh/acl-example.json5",
            )
        }
    } else {
        DiagnosticResult::fail(
            "ACL config",
            "Zenoh config not found",
            "Run: bubbaloop doctor --fix",
        )
    }
}

fn check_scouting_disabled(bubbaloop_dir: &std::path::Path) -> DiagnosticResult {
    let zenoh_config = bubbaloop_dir.join("zenoh/zenohd.json5");
    if let Ok(content) = std::fs::read_to_string(&zenoh_config) {
        // Check for multicast disabled
        let multicast_disabled = content.contains("multicast")
            && content.contains("enabled")
            && content.contains("false");
        if multicast_disabled {
            DiagnosticResult::pass("Scouting", "multicast scouting disabled")
        } else {
            DiagnosticResult::fail(
                "Scouting",
                "multicast scouting may be enabled (security risk on shared networks)",
                "Add scouting.multicast.enabled: false to zenohd.json5",
            )
        }
    } else {
        DiagnosticResult::fail(
            "Scouting",
            "Zenoh config not found",
            "Run: bubbaloop doctor --fix",
        )
    }
}

fn check_localhost_binding(bubbaloop_dir: &std::path::Path) -> DiagnosticResult {
    let zenoh_config = bubbaloop_dir.join("zenoh/zenohd.json5");
    if let Ok(content) = std::fs::read_to_string(&zenoh_config) {
        // Check for 0.0.0.0 binding (allows remote connections)
        if content.contains("0.0.0.0") {
            DiagnosticResult::fail(
                "Localhost binding",
                "Zenoh binds to 0.0.0.0 (all interfaces)",
                "Use 127.0.0.1 for local-only or enable TLS for remote access",
            )
        } else {
            DiagnosticResult::pass("Localhost binding", "not binding to all interfaces")
        }
    } else {
        DiagnosticResult::fail(
            "Localhost binding",
            "Zenoh config not found",
            "Run: bubbaloop doctor --fix",
        )
    }
}

fn check_python_sandbox() -> Vec<DiagnosticResult> {
    let mut results = Vec::new();

    let systemd_dir = dirs::home_dir()
        .map(|h| h.join(".config/systemd/user"))
        .unwrap_or_default();

    if !systemd_dir.exists() {
        return results;
    }

    let entries = match std::fs::read_dir(&systemd_dir) {
        Ok(e) => e,
        Err(_) => return results,
    };

    for entry in entries.flatten() {
        let path = entry.path();
        let name = entry.file_name().to_string_lossy().to_string();
        if !name.starts_with("bubbaloop-") || !name.ends_with(".service") {
            continue;
        }

        if let Ok(content) = std::fs::read_to_string(&path) {
            // Only check Python nodes (they have PYTHONUNBUFFERED)
            if !content.contains("PYTHONUNBUFFERED") {
                continue;
            }

            let has_sandbox = content.contains("ProtectHome=read-only")
                && content.contains("MemoryMax=")
                

fn is_valid_node_name(name: &str) -> bool {
    !name.is_empty()
        && name.len() <= 64
        && name
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-')
}

fn check_node_names(bubbaloop_dir: &std::path::Path) -> Vec<DiagnosticResult> {
    let mut results = Vec::new();
    let nodes_json = bubbaloop_dir.join("nodes.json");

    if let Ok(content) = std::fs::read_to_string(&nodes_json) {
        // Simple check: parse as JSON and validate each node name
        if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&content) {
            if let Some(nodes) = parsed.get("nodes").and_then(|n| n.as_array()) {
                for node in nodes {
                    if let Some(name) = node.get("name").and_then(|n| n.as_str()) {
                        if is_valid_node_name(name) {
                            results.push(DiagnosticResult::pass(
                                &format!("Node name ({})", name),
                                "valid",
                            ));
                        } else {
                            results.push(DiagnosticResult::fail(
                                &format!("Node name ({})", 

async fn query_with_timeout<T: for<'de> Deserialize<'de>>(
    session: &Session,
    key_expr: &str,
) -> Result<T> {
    let replies = session
        .get(key_expr)
        .target(QueryTarget::BestMatching)
        .timeout(Duration::from_secs(TIMEOUT_SECS))
        .await
        .map_err(|e| anyhow!("Zenoh query failed: {}", e))?;

    let timeout_duration = Duration::from_secs(TIMEOUT_SECS);

    match tokio::time::timeout(timeout_duration, replies.recv_async()).await {
        Ok(Ok(reply)) => {
            if let Ok(sample) = reply.result() {
                let bytes = sample.payload().to_bytes();
                let text = String::from_utf8_lossy(&bytes);
                let result: T = serde_json::from_str(&text)?;
                return Ok(result);
            }
        }
        Ok(Err(e)) => return Err(anyhow!("Failed to receive reply: {}", e)),
        Err(_) => { /* timeout - fall through to error below */ }
    }

    Err(anyhow!(
        "No reply received for {} (t

    fn test_diagnostic_result_pass() {
        let result = DiagnosticResult::pass("test", "all good");
        assert!(result.passed);
        assert_eq!(result.check, "test");
        assert_eq!(result.message, "all good");
        assert!(result.fix.is_none());
    }

    fn test_diagnostic_result_fail() {
        let result = DiagnosticResult::fail("test", "something wrong", "do this");
        assert!(!result.passed);
        assert_eq!(result.check, "test");
        assert_eq!(result.message, "something wrong");
        assert_eq!(result.fix, Some("do this".to_string()));
    }

    fn test_node_list_response_deserialization() {
        let json = r#"{
            "nodes": [],
            "timestamp_ms": 1234567890
        }"#;
        let response: NodeListResponse = serde_json::from_str(json).unwrap();
        assert_eq!(response.nodes.len(), 0);
        assert_eq!(response.timestamp_ms, 1234567890);
    }

    fn test_node_state_deserialization() {
        let json = r#"{
            "name": "test-node",
            "path": "/path",
            "status": "running",
            "installed": true,
            "autostart_enabled": false,
            "version": "1.0.0",
            "description": "Test",
            "node_type": "rust",
            "is_built": true,
            "build_output": []
        }"#;
        let node: NodeState = serde_json::from_str(json).unwrap();
        assert_eq!(node.name, "test-node");
        assert_eq!(node.status, "running");
    }

    fn test_is_valid_node_name_accepts_valid() {
        assert!(is_valid_node_name("camera"));
        assert!(is_valid_node_name("rtsp-camera"));
        assert!(is_valid_node_name("my_node_123"));
        assert!(is_valid_node_name("A"));
        assert!(is_valid_node_name("a".repeat(64).as_str()));
    }

    fn test_is_valid_node_name_rejects_invalid() {
        assert!(!is_valid_node_name(""));
        assert!(!is_valid_node_name("bad node"));
        assert!(!is_valid_node_name("bad/node"));
        assert!(!is_valid_node_name("bad.node"));
        assert!(!is_valid_node_name("bad\0node"));
        assert!(!is_valid_node_name(&"a".repeat(65)));
    }

    fn test_check_tls_status_with_tls_config() {
        let dir = tempfile::tempdir().unwrap();
        let zenoh_dir = dir.path().join("zenoh");
        std::fs::create_dir_all(&zenoh_dir).unwrap();
        std::fs::write(
            zenoh_dir.join("zenohd.json5"),
            r#"{ transport: { link: { tls: { server_certificate: "/path/cert.pem" } } } }"#,
        )
        .unwrap();

        let result = check_tls_status(dir.path());
        assert!(result.passed, "Expected TLS check to pass");
    }

    fn test_check_tls_status_without_tls() {
        let dir = tempfile::tempdir().unwrap();
        let zenoh_dir = dir.path().join("zenoh");
        std::fs::create_dir_all(&zenoh_dir).unwrap();
        std::fs::write(zenoh_dir.join("zenohd.json5"), r#"{ mode: "router" }"#).unwrap();

        let result = check_tls_status(dir.path());
        assert!(!result.passed, "Expected TLS check to fail");
    }

    fn test_check_scouting_disabled_passes() {
        let dir = tempfile::tempdir().unwrap();
        let zenoh_dir = dir.path().join("zenoh");
        std::fs::create_dir_all(&zenoh_dir).unwrap();
        std::fs::write(
            zenoh_dir.join("zenohd.json5"),
            r#"{ scouting: { multicast: { enabled: false } } }"#,
        )
        .unwrap();

        let result = check_scouting_disabled(dir.path());
        assert!(result.passed, "Expected scouting check to pass");
    }

    fn test_check_scouting_disabled_fails_when_enabled() {
        let dir = tempfile::tempdir().unwrap();
        let zenoh_dir = dir.path().join("zenoh");
        std::fs::create_dir_all(&zenoh_dir).unwrap();
        std::fs::write(zenoh_dir.join("zenohd.json5"), r#"{ mode: "router" }"#).unwrap();

        let result = check_scouting_disabled(dir.path());
        assert!(!result.passed, "Expected scouting check to fail");
    }

    fn test_check_localhost_binding_passes() {
        let dir = tempfile::tempdir().unwrap();
        let zenoh_dir = dir.path().join("zenoh");
        std::fs::create_dir_all(&zenoh_dir).unwrap();
        std::fs::write(
            zenoh_dir.join("zenohd.json5"),
            r#"{ listen: { endpoints: ["tcp/127.0.0.1:7447"] } }"#,
        )
        .unwrap();

        let result = check_localhost_binding(dir.path());
        assert!(result.passed, "Expected localhost check to pass");
    }

    fn test_check_localhost_binding_fails_wildcard() {
        let dir = tempfile::tempdir().unwrap();
        let zenoh_dir = dir.path().join("zenoh");
        std::fs::create_dir_all(&zenoh_dir).unwrap();
        std::fs::write(
            zenoh_dir.join("zenohd.json5"),
            r#"{ listen: { endpoints: ["tcp/0.0.0.0:7447"] } }"#,
        )
        .unwrap();

        let result = check_localhost_binding(dir.path());
        assert!(
            !result.passed,
            "Expected localhost check to fail for 0.0.0.0"
        );
    }

    fn test_check_acl_status_passes() {
        let dir = tempfile::tempdir().unwrap();
        let zenoh_dir = dir.path().join("zenoh");
        std::fs::create_dir_all(&zenoh_dir).unwrap();
        std::fs::write(
            zenoh_dir.join("zenohd.json5"),
            r#"{ access_control: { enabled: true } }"#,
        )
        .unwrap();

        let result = check_acl_status(dir.path());
        assert!(result.passed, "Expected ACL check to pass");
    }

    fn test_check_acl_status_fails() {
        let dir = tempfile::tempdir().unwrap();
        let zenoh_dir = dir.path().join("zenoh");
        std::fs::create_dir_all(&zenoh_dir).unwrap();
        std::fs::write(zenoh_dir.join("zenohd.json5"), r#"{ mode: "router" }"#).unwrap();

        let result = check_acl_status(dir.path());
        assert!(!result.passed, "Expected ACL check to fail");
    }

    fn test_check_node_names_valid() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(
            dir.path().join("nodes.json"),
            r#"{"nodes": [{"name": "camera"}, {"name": "rtsp-cam"}]}"#,
        )
        .unwrap();

        let results = check_node_names(dir.path());
        assert_eq!(results.len(), 2);
        assert!(results.iter().all(|r| r.passed));
    }

    fn test_check_node_names_invalid() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(
            dir.path().join("nodes.json"),
            r#"{"nodes": [{"name": "good-node"}, {"name": "bad node!"}]}"#,
        )
        .unwrap();

        let results = check_node_names(dir.path());
        assert_eq!(results.len(), 2);
        assert!(results[0].passed);
        assert!(!results[1].passed);
    }

fn parse_launch_file(content: &str) -> Result<LaunchFile> {
    let launch: LaunchFile = serde_yaml::from_str(content)?;
    if launch.name.is_empty() {
        return Err(LaunchError::Instance(
            "launch file must have a non-empty 'name' field".to_string(),
        ));
    }
    Ok(launch)
}

fn write_config(
    configs_dir: &std::path::Path,
    name: &str,
    config: &serde_yaml::Value,
) -> Result<PathBuf> {
    std::fs::create_dir_all(configs_dir)?;
    let dest = configs_dir.join(format!("{}.yaml", name));
    let yaml = serde_yaml::to_string(config)?;
    std::fs::write(&dest, &yaml)?;
    Ok(dest)
}

fn default_configs_dir() -> PathBuf {
    dirs::home_dir()
        .unwrap_or_else(|| PathBuf::from("."))
        .join(".bubbaloop")
        .join("configs")
}

    pub async fn run(self) -> Result<()> {
        // 1. Read and parse the launch file
        let file_path = std::path::Path::new(&self.file);
        if !file_path.exists() {
            return Err(LaunchError::FileNotFound(self.file.clone()));
        }

        let content = std::fs::read_to_string(file_path)?;
        let launch = parse_launch_file(&content)?;

        if self.dry_run {
            println!("[DRY RUN] Instance: {}", launch.name);
            println!("  Base node: {}", self.node);
            println!("  Launch file: {}", self.file);
            if launch.config.is_some() {
                println!(
                    "  Config: would write to ~/.bubbaloop/configs/{}.yaml",
                    launch.name
                );
            }
            if self.build {
                println!("  Would: build");
            }
            if self.install {
                println!("  Would: install as systemd service");
            }
            if self.start {
    

    async fn resolve_node_path(&self, node_name: &str) -> Result<String> {
        let session = node::get_zenoh_session().await?;
        let replies = session
            .get("bubbaloop/daemon/api/nodes")
            .target(zenoh::query::QueryTarget::BestMatching)
            .timeout(std::time::Duration::from_secs(10))
            .await
            .map_err(|e| LaunchError::Node(node::NodeError::Zenoh(e.to_string())))?;

        #[derive(Deserialize)]
        struct NodeInfo {
            name: String,
            path: String,
        }
        #[derive(Deserialize)]
        struct NodeListResponse {
            nodes: Vec<NodeInfo>,
        }

        for reply in replies {
            if let Ok(sample) = reply.into_result() {
                let payload = sample.payload().to_bytes();
                if let Ok(response) = serde_json::from_slice::<NodeListResponse>(&payload) {
                    for info in response.nodes {
                        if info.name == node_name {
 

    async fn register_instance(
        &self,
        node_path: &str,
        name_override: Option<&str>,
        config_path: Option<&str>,
    ) -> Result<()> {
        let session = node::get_zenoh_session().await?;

        let payload = serde_json::to_string(&serde_json::json!({
            "command": "add",
            "node_path": node_path,
            "name": name_override,
            "config": config_path,
        }))?;

        let replies = session
            .get("bubbaloop/daemon/api/nodes/add")
            .payload(payload)
            .target(zenoh::query::QueryTarget::BestMatching)
            .timeout(std::time::Duration::from_secs(30))
            .await
            .map_err(|e| LaunchError::Node(node::NodeError::Zenoh(e.to_string())))?;

        let mut success = false;
        for reply in replies {
            if let Ok(sample) = reply.into_result() {
                let data: serde_json::Value = serde_json::from_slice(&sample.payload().to_bytes())?;
       

    fn test_parse_full_launch_file() {
        let yaml = r#"
name: rtsp-camera-entrance
config:
  name: entrance
  publish_topic: camera/entrance/compressed
  url: "rtsp://user:[email protected]:554/stream2"
  latency: 200
"#;
        let launch = parse_launch_file(yaml).unwrap();
        assert_eq!(launch.name, "rtsp-camera-entrance");
        assert!(launch.config.is_some());
    }

    fn test_parse_minimal_launch_file() {
        let yaml = "name: my-instance\n";
        let launch = parse_launch_file(yaml).unwrap();
        assert_eq!(launch.name, "my-instance");
        assert!(launch.config.is_none());
    }

    fn test_write_config_to_tempdir() {
        let dir = tempfile::tempdir().unwrap();
        let config: serde_yaml::Value =
            serde_yaml::from_str("name: terrace\npublish_topic: camera/terrace/compressed\n")
                .unwrap();

        let dest = write_config(dir.path(), "rtsp-camera-terrace", &config).unwrap();
        assert!(dest.exists());
        assert!(dest.to_string_lossy().ends_with("rtsp-camera-terrace.yaml"));

        let content = std::fs::read_to_string(&dest).unwrap();
        assert!(content.contains("terrace"));
        assert!(content.contains("publish_topic"));
    }

    fn test_write_config_nested_values() {
        let dir = tempfile::tempdir().unwrap();
        let config: serde_yaml::Value =
            serde_yaml::from_str("location:\n  latitude: 41.39\n  longitude: 2.17\n").unwrap();

        let dest = write_config(dir.path(), "openmeteo", &config).unwrap();
        let content = std::fs::read_to_string(&dest).unwrap();
        assert!(content.contains("latitude"));
        assert!(content.contains("41.39"));
    }

fn sources_path() -> PathBuf {
    dirs::home_dir()
        .unwrap_or_else(|| PathBuf::from("."))
        .join(".bubbaloop")
        .join("sources.json")
}

fn load_sources() -> SourcesRegistry {
    let path = sources_path();
    if path.exists() {
        fs::read_to_string(&path)
            .ok()
            .and_then(|s| serde_json::from_str(&s).ok())
            .unwrap_or_default()
    } else {
        SourcesRegistry::default()
    }
}

fn save_sources(registry: &SourcesRegistry) -> Result<()> {
    let path = sources_path();
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
    }
    let json = serde_json::to_string_pretty(registry)?;
    fs::write(path, json)?;
    Ok(())
}

fn find_source_mut<'a>(
    registry: &'a mut SourcesRegistry,
    name_or_path: &str,
) -> Option<&'a mut SourceEntry> {
    registry
        .sources
        .iter_mut()
        .find(|s| s.name == name_or_path || s.path == name_or_path)
}

    pub async fn run(self) -> Result<()> {
        match self.action {
            MarketplaceAction::List(args) => list_sources(args),
            MarketplaceAction::Add(args) => add_source(args),
            MarketplaceAction::Remove(args) => remove_source(args),
            MarketplaceAction::Enable(args) => enable_source(args),
            MarketplaceAction::Disable(args) => disable_source(args),
        }
    }

fn list_sources(args: ListArgs) -> Result<()> {
    let registry = load_sources();

    if args.format == "json" {
        println!("{}", serde_json::to_string_pretty(&registry.sources)?);
        return Ok(());
    }

    if registry.sources.is_empty() {
        println!("No marketplace sources configured.");
        println!("Add one with: bubbaloop marketplace add <name> <path>");
        return Ok(());
    }

    println!("{:<3} {:<20} {:<10} PATH", "ON", "NAME", "TYPE");
    println!("{}", "-".repeat(70));
    for source in &registry.sources {
        let enabled = if source.enabled { "yes" } else { "no" };
        println!(
            "{:<3} {:<20} {:<10} {}",
            enabled, source.name, source.source_type, source.path
        );
    }

    Ok(())
}

fn add_source(args: AddArgs) -> Result<()> {
    let mut registry = load_sources();

    if registry.sources.iter().any(|s| s.path == args.path) {
        return Err(MarketplaceError::Other(
            "Source with this path already exists".into(),
        ));
    }

    registry.sources.push(SourceEntry {
        name: args.name.clone(),
        path: args.path.clone(),
        source_type: args.source_type,
        enabled: true,
    });

    save_sources(&registry)?;
    println!("Added marketplace source: {}", args.name);
    Ok(())
}

fn remove_source(args: RemoveArgs) -> Result<()> {
    let mut registry = load_sources();

    // Prevent removing builtin sources
    if let Some(source) = registry
        .sources
        .iter()
        .find(|s| s.name == args.name_or_path || s.path == args.name_or_path)
    {
        if source.source_type == OFFICIAL_SOURCE_TYPE {
            return Err(MarketplaceError::Other(
                "Cannot remove builtin source".into(),
            ));
        }
    }

    let before = registry.sources.len();
    registry
        .sources
        .retain(|s| s.name != args.name_or_path && s.path != args.name_or_path);

    if registry.sources.len() == before {
        return Err(MarketplaceError::Other(format!(
            "Source '{}' not found",
            args.name_or_path
        )));
    }

    save_sources(&registry)?;
    println!("Removed marketplace source: {}", args.name_or_path);
    Ok(())
}

fn update_source_enabled(name_or_path: &str, enabled: bool, action: &str) -> Result<()> {
    let mut registry = load_sources();

    let source = find_source_mut(&mut registry, name_or_path)
        .ok_or_else(|| MarketplaceError::Other(format!("Source '{}' not found", name_or_path)))?;

    source.enabled = enabled;
    save_sources(&registry)?;
    println!("{} marketplace source: {}", action, name_or_path);
    Ok(())
}

    fn test_sources_registry_round_trip() {
        let registry = SourcesRegistry {
            sources: vec![
                SourceEntry {
                    name: "Official Nodes".into(),
                    path: "kornia/bubbaloop-nodes-official".into(),
                    source_type: "builtin".into(),
                    enabled: true,
                },
                SourceEntry {
                    name: "My Local".into(),
                    path: "/home/user/nodes".into(),
                    source_type: "local".into(),
                    enabled: false,
                },
            ],
        };

        let json = serde_json::to_string_pretty(&registry).unwrap();
        let parsed: SourcesRegistry = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.sources.len(), 2);
        assert_eq!(parsed.sources[0].name, "Official Nodes");
        assert_eq!(parsed.sources[0].source_type, "builtin");
        assert!(parsed.sources[0].enabled);
        assert_e

    fn test_serde_rename_type_field() {
        let json = r#"{"name":"test","path":"/path","type":"local","enabled":true}"#;
        let entry: SourceEntry = serde_json::from_str(json).unwrap();
        assert_eq!(entry.source_type, "local");

        // Verify it serializes back with "type" key
        let serialized = serde_json::to_string(&entry).unwrap();
        assert!(serialized.contains("\"type\""));
        assert!(!serialized.contains("\"source_type\""));
    }

    fn test_cannot_remove_builtin() {
        let mut registry = SourcesRegistry {
            sources: vec![SourceEntry {
                name: "Official Nodes".into(),
                path: "kornia/bubbaloop-nodes-official".into(),
                source_type: "builtin".into(),
                enabled: true,
            }],
        };

        // Verify the source is found as builtin
        let source = registry
            .sources
            .iter()
            .find(|s| s.name == "Official Nodes")
            .unwrap();
        assert_eq!(source.source_type, OFFICIAL_SOURCE_TYPE);

        // Simulate removal check
        let is_builtin = registry
            .sources
            .iter()
            .find(|s| s.name == "Official Nodes")
            .map(|s| s.source_type == OFFICIAL_SOURCE_TYPE)
            .unwrap_or(false);
        assert!(is_builtin);

        // Non-builtin should be removable
        registry.sources.push(SourceEntry {
            name: "Removable".into(),

    fn test_find_source_by_name_or_path() {
        let mut registry = SourcesRegistry {
            sources: vec![SourceEntry {
                name: "My Nodes".into(),
                path: "/home/user/nodes".into(),
                source_type: "local".into(),
                enabled: true,
            }],
        };

        assert!(find_source_mut(&mut registry, "My Nodes").is_some());
        assert!(find_source_mut(&mut registry, "/home/user/nodes").is_some());
        assert!(find_source_mut(&mut registry, "nonexistent").is_none());
    }

fn copy_canonical_header_proto(node_path: &Path) {
    let protos_dir = node_path.join("protos");
    if !protos_dir.exists() {
        return;
    }

    let dest = protos_dir.join("header.proto");
    if let Err(e) = std::fs::write(&dest, crate::HEADER_PROTO) {
        log::warn!(
            "Could not copy canonical header.proto to {}: {}",
            dest.display(),
            e
        );
    } else {
        log::info!("Copied canonical header.proto to {}", dest.display());
    }
}