Function bodies 699 total

    fn test_manifest_validation_null_bytes_in_command() {
        let manifest = NodeManifest {
            name: "test-node".to_string(),
            version: "1.0.0".to_string(),
            node_type: "rust".to_string(),
            description: "Test".to_string(),
            command: Some("./target\0/release/test".to_string()),
            ..Default::default()
        };
        assert!(manifest.validate().is_err());
    }

    fn test_effective_name_with_override() {
        let entry = NodeEntry {
            path: "/opt/nodes/rtsp-camera".to_string(),
            added_at: "1700000000000".to_string(),
            name_override: Some("rtsp-camera-terrace".to_string()),
            config_override: Some("/etc/bubbaloop/terrace.yaml".to_string()),
        };
        let manifest = NodeManifest {
            name: "rtsp-camera".to_string(),
            version: "0.1.0".to_string(),
            node_type: "rust".to_string(),
            description: "RTSP camera".to_string(),
            ..Default::default()
        };
        assert_eq!(effective_name(&entry, &manifest), "rtsp-camera-terrace");
    }

    fn test_effective_name_without_override() {
        let entry = NodeEntry {
            path: "/opt/nodes/rtsp-camera".to_string(),
            added_at: "1700000000000".to_string(),
            name_override: None,
            config_override: None,
        };
        let manifest = NodeManifest {
            name: "rtsp-camera".to_string(),
            version: "0.1.0".to_string(),
            node_type: "rust".to_string(),
            description: "RTSP camera".to_string(),
            ..Default::default()
        };
        assert_eq!(effective_name(&entry, &manifest), "rtsp-camera");
    }

    fn test_multi_instance_registry_entries() {
        let manifest = NodeManifest {
            name: "rtsp-camera".to_string(),
            version: "0.2.0".to_string(),
            node_type: "rust".to_string(),
            description: "RTSP camera node".to_string(),
            ..Default::default()
        };

        let entries = vec![
            NodeEntry {
                path: "/opt/nodes/rtsp-camera".to_string(),
                added_at: "1700000000000".to_string(),
                name_override: Some("rtsp-camera-terrace".to_string()),
                config_override: Some("/etc/bubbaloop/terrace.yaml".to_string()),
            },
            NodeEntry {
                path: "/opt/nodes/rtsp-camera".to_string(),
                added_at: "1700000001000".to_string(),
                name_override: Some("rtsp-camera-garage".to_string()),
                config_override: Some("/etc/bubbaloop/garage.yaml".to_string()),
            },
            NodeEntry {
                

    fn from(s: &str) -> Self {
        match s {
            "active" => ActiveState::Active,
            "reloading" => ActiveState::Reloading,
            "inactive" => ActiveState::Inactive,
            "failed" => ActiveState::Failed,
            "activating" => ActiveState::Activating,
            "deactivating" => ActiveState::Deactivating,
            other => ActiveState::Unknown(other.to_string()),
        }
    }

    fn from(s: &str) -> Self {
        match s {
            "enabled" => UnitFileState::Enabled,
            "disabled" => UnitFileState::Disabled,
            "static" => UnitFileState::Static,
            "masked" => UnitFileState::Masked,
            "generated" => UnitFileState::Generated,
            "transient" => UnitFileState::Transient,
            other => UnitFileState::Unknown(other.to_string()),
        }
    }

    async fn with_timeout<F, Fut, T>(secs: u64, op: &str, f: F) -> Result<T>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = Result<T>>,
    {
        match tokio::time::timeout(std::time::Duration::from_secs(secs), f()).await {
            Ok(result) => result,
            Err(_) => {
                log::warn!("D-Bus {} timed out", op);
                Err(SystemdError::Timeout)
            }
        }
    }

    pub async fn get_active_state(&self, unit_name: &str) -> Result<ActiveState> {
        Self::with_timeout(5, &format!("get_active_state({unit_name})"), || {
            self.get_active_state_inner(unit_name)
        })
        .await
    }

    async fn get_active_state_inner(&self, unit_name: &str) -> Result<ActiveState> {
        let manager = self.manager().await?;
        match manager.load_unit(unit_name).await {
            Ok(path) => {
                let unit = SystemdUnitProxy::builder(&self.connection)
                    .path(path)?
                    .build()
                    .await?;

                let load_state = unit.load_state().await?;
                if load_state == "not-found" {
                    return Ok(ActiveState::Unknown("not-found".to_string()));
                }

                let state = unit.active_state().await?;
                Ok(ActiveState::from(state.as_str()))
            }
            Err(_) => Ok(ActiveState::Unknown("not-found".to_string())),
        }
    }

    pub async fn is_enabled(&self, unit_name: &str) -> Result<bool> {
        match Self::with_timeout(5, &format!("is_enabled({unit_name})"), || async {
            let manager = self.manager().await?;
            match manager.get_unit_file_state(unit_name).await {
                Ok(state) => Ok(state == "enabled"),
                Err(_) => Ok(false),
            }
        })
        .await
        {
            Ok(v) => Ok(v),
            Err(_) => Ok(false), // timeout → treat as not enabled
        }
    }

    pub async fn get_unit_file_state(&self, unit_name: &str) -> Result<UnitFileState> {
        let manager = self.manager().await?;
        match manager.get_unit_file_state(unit_name).await {
            Ok(state) => Ok(UnitFileState::from(state.as_str())),
            Err(_) => Ok(UnitFileState::Unknown("not-found".to_string())),
        }
    }

    pub async fn start_unit(&self, unit_name: &str) -> Result<()> {
        Self::with_timeout(30, &format!("start_unit({unit_name})"), || async {
            let manager = self.manager().await?;
            manager
                .start_unit(unit_name, "replace")
                .await
                .map_err(|e| {
                    SystemdError::OperationFailed(format!("Failed to start {unit_name}: {e}"))
                })?;
            Ok(())
        })
        .await
    }

    pub async fn stop_unit(&self, unit_name: &str) -> Result<()> {
        Self::with_timeout(10, &format!("stop_unit({unit_name})"), || async {
            let manager = self.manager().await?;
            manager.stop_unit(unit_name, "replace").await.map_err(|e| {
                SystemdError::OperationFailed(format!("Failed to stop {unit_name}: {e}"))
            })?;
            Ok(())
        })
        .await
    }

    pub async fn restart_unit(&self, unit_name: &str) -> Result<()> {
        Self::with_timeout(30, &format!("restart_unit({unit_name})"), || async {
            let manager = self.manager().await?;
            manager
                .restart_unit(unit_name, "replace")
                .await
                .map_err(|e| {
                    SystemdError::OperationFailed(format!("Failed to restart {unit_name}: {e}"))
                })?;
            Ok(())
        })
        .await
    }

    pub async fn enable_unit(&self, unit_name: &str) -> Result<()> {
        Self::with_timeout(30, &format!("enable_unit({unit_name})"), || async {
            let manager = self.manager().await?;
            manager
                .enable_unit_files(&[unit_name], false, false)
                .await
                .map_err(|e| {
                    SystemdError::OperationFailed(format!("Failed to enable {unit_name}: {e}"))
                })?;
            Ok(())
        })
        .await
    }

    pub async fn disable_unit(&self, unit_name: &str) -> Result<()> {
        Self::with_timeout(30, &format!("disable_unit({unit_name})"), || async {
            let manager = self.manager().await?;
            manager
                .disable_unit_files(&[unit_name], false)
                .await
                .map_err(|e| {
                    SystemdError::OperationFailed(format!("Failed to disable {unit_name}: {e}"))
                })?;
            Ok(())
        })
        .await
    }

    pub async fn daemon_reload(&self) -> Result<()> {
        Self::with_timeout(30, "daemon_reload", || async {
            let manager = self.manager().await?;
            manager.reload().await?;
            Ok(())
        })
        .await
    }

    pub async fn subscribe_to_signals(&self) -> Result<mpsc::Receiver<SystemdSignalEvent>> {
        let manager = self.manager().await?;

        // Subscribe to signals first
        manager.subscribe().await?;
        log::info!("Subscribed to systemd signals");

        let (tx, rx) = mpsc::channel(100);

        // Clone connection for the signal listener task
        let connection = self.connection.clone();

        // Spawn signal listener task
        tokio::spawn(async move {
            let manager_proxy = match SystemdManagerProxy::new(&connection).await {
                Ok(proxy) => proxy,
                Err(e) => {
                    log::error!("Failed to create manager proxy for signals: {}", e);
                    return;
                }
            };

            // Spawn separate tasks for each signal type
            let tx_job = tx.clone();
            let mut job_removed = match manager_proxy.receive_job_removed().await {
                Ok(stream) => strea

fn extract_node_name(unit: &str) -> Option<String> {
    if unit.starts_with("bubbaloop-") && unit.ends_with(".service") {
        let name = unit.strip_prefix("bubbaloop-")?.strip_suffix(".service")?;
        if name.is_empty() {
            None
        } else {
            Some(name.to_string())
        }
    } else {
        None
    }
}

fn validate_node_name(name: &str) -> Result<()> {
    // Node names should be alphanumeric with hyphens/underscores only
    if !name
        .chars()
        .all(|c| c.is_alphanumeric() || c == '-' || c == '_')
    {
        return Err(SystemdError::InvalidNodeName(format!(
            "'{}' contains invalid characters (only alphanumeric, hyphens, and underscores allowed)",
            name
        )));
    }
    if name.is_empty() || name.len() > 64 {
        return Err(SystemdError::InvalidNodeName(format!(
            "'{}' has invalid length (must be 1-64 characters)",
            name
        )));
    }
    Ok(())
}

fn sanitize_description(s: &str) -> String {
    // Remove newlines and special characters that could break unit file parsing
    s.chars()
        .filter(|c| !matches!(c, '\n' | '\r' | '[' | ']'))
        .take(200) // Reasonable length limit for descriptions
        .collect()
}

fn sanitize_path(path: &str) -> Result<String> {
    // Basic validation - paths should not contain newlines or null bytes
    if path.contains('\n') || path.contains('\r') || path.contains('\0') {
        return Err(SystemdError::InvalidInput(format!(
            "Path '{}' contains invalid characters",
            path
        )));
    }
    // Ensure it's a valid UTF-8 path
    Ok(path.to_string())
}

fn sanitize_command(cmd: &str) -> Result<String> {
    // Commands should not contain newlines or null bytes
    if cmd.contains('\n') || cmd.contains('\r') || cmd.contains('\0') {
        return Err(SystemdError::InvalidInput(
            "Command contains invalid characters".to_string(),
        ));
    }
    // Prevent injection of systemd directives by checking for suspicious patterns
    if cmd.contains("[Unit]") || cmd.contains("[Service]") || cmd.contains("[Install]") {
        return Err(SystemdError::InvalidInput(
            "Command contains systemd unit section markers".to_string(),
        ));
    }
    Ok(cmd.to_string())
}

pub fn generate_service_unit(
    node_path: &str,
    name: &str,
    node_type: &str,
    command: Option<&str>,
    depends_on: &[String],
) -> Result<String> {
    // Validate and sanitize inputs
    validate_node_name(name)?;
    let safe_node_path = sanitize_path(node_path)?;
    let safe_name = sanitize_description(name);
    let home = dirs::home_dir().unwrap_or_else(|| PathBuf::from("/home/user"));
    let cargo_path = home.join(".cargo/bin/cargo");
    let pixi_bin_dir = home.join(".pixi/bin");
    let pixi_path = pixi_bin_dir.join("pixi");
    let path_env = format!(
        "PATH={}:{}:/usr/local/bin:/usr/bin:/bin",
        home.join(".cargo/bin").display(),
        pixi_bin_dir.display()
    );

    let (exec_start, environment) = if let Some(cmd) = command {
        let safe_cmd = sanitize_command(cmd)?;
        if safe_cmd.starts_with("cargo ") {
            (
                safe_cmd.replacen("cargo ", &format!("{} ", cargo_path.display()), 1),
                "RUST_LOG

pub async fn install_service(
    node_path: &str,
    name: &str,
    node_type: &str,
    command: Option<&str>,
    depends_on: &[String],
) -> Result<()> {
    let service_dir = get_systemd_user_dir();
    std::fs::create_dir_all(&service_dir)?;

    let service_path = get_service_path(name);
    let content = generate_service_unit(node_path, name, node_type, command, depends_on)?;
    std::fs::write(&service_path, &content)?;

    // Reload systemd to pick up the new unit
    let client = SystemdClient::new().await?;
    client.daemon_reload().await?;

    Ok(())
}

pub async fn uninstall_service(name: &str) -> Result<()> {
    let client = SystemdClient::new().await?;
    let service_name = get_service_name(name);

    // Stop and disable first (ignore errors)
    let _ = client.stop_unit(&service_name).await;
    let _ = client.disable_unit(&service_name).await;

    // Remove the unit file
    let service_path = get_service_path(name);
    if service_path.exists() {
        std::fs::remove_file(&service_path)?;
    }

    // Reload daemon
    client.daemon_reload().await?;

    Ok(())
}

    fn test_active_state_from_known_values() {
        assert_eq!(ActiveState::from("active"), ActiveState::Active);
        assert_eq!(ActiveState::from("reloading"), ActiveState::Reloading);
        assert_eq!(ActiveState::from("inactive"), ActiveState::Inactive);
        assert_eq!(ActiveState::from("failed"), ActiveState::Failed);
        assert_eq!(ActiveState::from("activating"), ActiveState::Activating);
        assert_eq!(ActiveState::from("deactivating"), ActiveState::Deactivating);
    }

    fn test_active_state_from_unknown() {
        assert_eq!(
            ActiveState::from("unknown-state"),
            ActiveState::Unknown("unknown-state".to_string())
        );
        assert_eq!(
            ActiveState::from("weird"),
            ActiveState::Unknown("weird".to_string())
        );
    }

    fn test_unit_file_state_from_known_values() {
        assert_eq!(UnitFileState::from("enabled"), UnitFileState::Enabled);
        assert_eq!(UnitFileState::from("disabled"), UnitFileState::Disabled);
        assert_eq!(UnitFileState::from("static"), UnitFileState::Static);
        assert_eq!(UnitFileState::from("masked"), UnitFileState::Masked);
        assert_eq!(UnitFileState::from("generated"), UnitFileState::Generated);
        assert_eq!(UnitFileState::from("transient"), UnitFileState::Transient);
    }

    fn test_unit_file_state_from_unknown() {
        assert_eq!(
            UnitFileState::from("not-found"),
            UnitFileState::Unknown("not-found".to_string())
        );
        assert_eq!(
            UnitFileState::from("custom"),
            UnitFileState::Unknown("custom".to_string())
        );
    }

    fn test_extract_node_name_valid() {
        assert_eq!(
            extract_node_name("bubbaloop-rtsp-camera.service"),
            Some("rtsp-camera".to_string())
        );
        assert_eq!(
            extract_node_name("bubbaloop-weather.service"),
            Some("weather".to_string())
        );
        assert_eq!(
            extract_node_name("bubbaloop-my-node.service"),
            Some("my-node".to_string())
        );
    }

    fn test_get_service_name() {
        assert_eq!(
            get_service_name("rtsp-camera"),
            "bubbaloop-rtsp-camera.service"
        );
        assert_eq!(get_service_name("weather"), "bubbaloop-weather.service");
        assert_eq!(get_service_name("my_node"), "bubbaloop-my_node.service");
    }

    fn test_validate_node_name_valid() {
        assert!(validate_node_name("rtsp-camera").is_ok());
        assert!(validate_node_name("weather").is_ok());
        assert!(validate_node_name("my_node").is_ok());
        assert!(validate_node_name("node123").is_ok());
        assert!(validate_node_name("a-b-c_d").is_ok());
        assert!(validate_node_name("a").is_ok()); // minimum length
        assert!(validate_node_name(&"a".repeat(64)).is_ok()); // maximum length
    }

    fn test_validate_node_name_special_chars() {
        assert!(validate_node_name("node.with.dots").is_err());
        assert!(validate_node_name("node/with/slash").is_err());
        assert!(validate_node_name("node with spaces").is_err());
        assert!(validate_node_name("node@email").is_err());
        assert!(validate_node_name("node$var").is_err());
        assert!(validate_node_name("node!").is_err());
    }

    fn test_validate_node_name_empty() {
        let result = validate_node_name("");
        assert!(result.is_err());
        if let Err(SystemdError::InvalidNodeName(msg)) = result {
            assert!(msg.contains("invalid length"));
        }
    }

    fn test_validate_node_name_too_long() {
        let long_name = "a".repeat(65);
        let result = validate_node_name(&long_name);
        assert!(result.is_err());
        if let Err(SystemdError::InvalidNodeName(msg)) = result {
            assert!(msg.contains("invalid length"));
        }
    }

    fn test_sanitize_description_removes_brackets() {
        assert_eq!(
            sanitize_description("description [with] brackets"),
            "description with brackets"
        );
        assert_eq!(sanitize_description("[Unit]Description"), "UnitDescription");
    }

    fn test_sanitize_description_truncates_long() {
        let long_desc = "a".repeat(300);
        let sanitized = sanitize_description(&long_desc);
        assert_eq!(sanitized.len(), 200);
        assert_eq!(sanitized, "a".repeat(200));
    }

    fn test_generate_service_unit_rust_default() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/test-node",
            "test-node",
            "rust",
            None,
            &[],
        );
        assert!(result.is_ok());
        let content = result.unwrap();

        assert!(content.contains("Description=Bubbaloop Node: test-node"));
        assert!(content.contains("After=network.target"));
        assert!(!content.contains("Requires="));
        assert!(content.contains("WorkingDirectory=/home/user/.bubbaloop/nodes/test-node"));
        assert!(content
            .contains("ExecStart=/home/user/.bubbaloop/nodes/test-node/target/release/test-node"));
        assert!(content.contains("Environment=RUST_LOG=info"));
        assert!(content.contains("WantedBy=default.target"));
    }

    fn test_generate_service_unit_python_default() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/py-node",
            "py-node",
            "python",
            None,
            &[],
        );
        assert!(result.is_ok());
        let content = result.unwrap();

        assert!(content.contains("Description=Bubbaloop Node: py-node"));
        assert!(content.contains("WorkingDirectory=/home/user/.bubbaloop/nodes/py-node"));
        assert!(content
            .contains("ExecStart=/home/user/.bubbaloop/nodes/py-node/venv/bin/python main.py"));
        assert!(content.contains("Environment=PYTHONUNBUFFERED=1"));
    }

    fn test_generate_service_unit_cargo_command() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/dev-node",
            "dev-node",
            "rust",
            Some("cargo run --release"),
            &[],
        );
        assert!(result.is_ok());
        let content = result.unwrap();

        // Should replace "cargo " with absolute path
        assert!(content.contains("/.cargo/bin/cargo run --release"));
        assert!(content.contains("Environment=RUST_LOG=info"));
    }

    fn test_generate_service_unit_pixi_command() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/pixi-node",
            "pixi-node",
            "python",
            Some("pixi run start"),
            &[],
        );
        assert!(result.is_ok());
        let content = result.unwrap();

        // Should replace "pixi " with absolute path
        assert!(content.contains("/.pixi/bin/pixi run start"));
        assert!(content.contains("Environment=PYTHONUNBUFFERED=1"));
    }

    fn test_generate_service_unit_with_dependencies() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/dependent-node",
            "dependent-node",
            "rust",
            None,
            &["dep1".to_string(), "dep2".to_string()],
        );
        assert!(result.is_ok());
        let content = result.unwrap();

        assert!(
            content.contains("After=network.target bubbaloop-dep1.service bubbaloop-dep2.service")
        );
        assert!(content.contains("Requires=bubbaloop-dep1.service bubbaloop-dep2.service"));
    }

    fn test_generate_service_unit_invalid_name() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/bad-node",
            "bad node with spaces",
            "rust",
            None,
            &[],
        );
        assert!(result.is_err());
        if let Err(SystemdError::InvalidNodeName(_)) = result {
            // Expected error type
        } else {
            panic!("Expected InvalidNodeName error");
        }
    }

    fn test_generate_service_unit_invalid_command() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/test-node",
            "test-node",
            "rust",
            Some("cargo run\n[Unit]\nDescription=evil"),
            &[],
        );
        assert!(result.is_err());
        if let Err(SystemdError::InvalidInput(_)) = result {
            // Expected error type
        } else {
            panic!("Expected InvalidInput error");
        }
    }

    fn test_generate_service_unit_invalid_dependency_name() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/test-node",
            "test-node",
            "rust",
            None,
            &["valid-dep".to_string(), "bad dep".to_string()],
        );
        assert!(result.is_err());
        if let Err(SystemdError::InvalidNodeName(_)) = result {
            // Expected error type
        } else {
            panic!("Expected InvalidNodeName error");
        }
    }

    fn test_generate_service_unit_security_hardening() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/secure-node",
            "secure-node",
            "rust",
            None,
            &[],
        );
        assert!(result.is_ok());
        let content = result.unwrap();

        // Check for security hardening directives
        assert!(content.contains("NoNewPrivileges=true"));
        assert!(content.contains("ProtectSystem=strict"));
        assert!(content.contains("PrivateTmp=true"));
        assert!(content.contains("ProtectKernelTunables=true"));
        assert!(content.contains("ProtectControlGroups=true"));

        // Robotics-compatible settings
        assert!(content.contains("RestrictRealtime=false"));
        assert!(content.contains("MemoryDenyWriteExecute=false"));
    }

    fn test_generate_service_unit_restart_policy() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/restart-node",
            "restart-node",
            "rust",
            None,
            &[],
        );
        assert!(result.is_ok());
        let content = result.unwrap();

        assert!(content.contains("Restart=on-failure"));
        assert!(content.contains("RestartSec=5"));
    }

    fn test_sanitize_description_preserves_valid_chars() {
        let desc = "My node with numbers 123 and symbols: - _ / @ !";
        let sanitized = sanitize_description(desc);
        // Should keep everything except brackets and newlines
        assert!(sanitized.contains("My node with numbers 123"));
        assert!(sanitized.contains("- _ / @ !"));
    }

    fn test_generate_service_unit_absolute_path_command() {
        let result = generate_service_unit(
            "/home/user/.bubbaloop/nodes/abs-node",
            "abs-node",
            "python",
            Some("/usr/bin/python3 script.py"),
            &[],
        );
        assert!(result.is_ok());
        let content = result.unwrap();

        // Absolute paths should be preserved
        assert!(content.contains("ExecStart=/usr/bin/python3 script.py"));
    }